import requests import json import pandas as pd import numpy as np from datetime import datetime, timedelta from requests.auth import HTTPBasicAuth from sqlalchemy import create_engine from airflow import DAG from airflow.operators.python import PythonOperator default_args = { 'owner': 'airflow', 'depends_on_past': False, 'start_date': datetime(2023, 1, 1), 'retries': 1, } def get_db_engine(): """Создает подключение к PostgreSQL""" DF_CONFIG = { 'dbname': "postgres", 'user': "postgres", 'password': "4a00d4b90cd830da0796", 'host': "postgresql", 'port': "5432" } return create_engine( f"postgresql+psycopg2://{DF_CONFIG['user']}:{DF_CONFIG['password']}@" f"{DF_CONFIG['host']}:{DF_CONFIG['port']}/{DF_CONFIG['dbname']}", pool_size=10, max_overflow=20 ) def read_data_1C(**kwargs): """ Создает DataFrame с ежедневным календарем и останавливается на дате T-2 Параметры: start_year (int): начальный год (по умолчанию 2019) end_year (int): конечный год (по умолчанию 2025) Возвращает: pandas.DataFrame: DataFrame с колонками Год, Месяц, День """ start_year = 2019 # Вычисляем дату T-2 (текущая дата минус 2 дня) t_minus_2 = datetime.now() - timedelta(days=2) t_minus_2 = t_minus_2.replace(hour=0, minute=0, second=0, microsecond=0) print(f"Текущая дата: {datetime.now().strftime('%Y-%m-%d')}") print(f"Дата T-2: {t_minus_2.strftime('%Y-%m-%d')}") # Создаем диапазон дат от начала до T-2 start_date = f'{start_year}-01-01' # date_range = pd.date_range(start=start_date, end=t_minus_2, freq='D') # Создаем пустой список для хранения данных calendar_data = [] params = {"СписокСчетов": ["66","66.01","66.02", "66.03","66.04","66.21","66.22","66.23","66.24", "67","67.01","67.02", "67.03","67.04","67.21","67.22","67.23","67.24"]} engine = get_db_engine() table_name = 'oborotno_salbdovaya_vedomostb_with_docs' temp_table_name = f'temp_{table_name}' # Проходим циклом по каждому дню # print("\nПрохождение по дням календаря:") # for i, date in enumerate(date_range): # print(f'{date}') query = f"""ВЫБРАТЬ ОстаткиОбороты.Счет, ОстаткиОбороты.Регистратор, ОстаткиОбороты.Субконто1, ОстаткиОбороты.Субконто2, ОстаткиОбороты.Организация, ОстаткиОбороты.Субконто2.Номер КАК НомерДоговора, ОстаткиОбороты.Субконто2.Дата КАК ДатаДоговора, ОстаткиОбороты.Субконто2.СрокДействия КАК СрокДействияДоговора, ОстаткиОбороты.Субконто1.Инн КАК ИннКонтрагента, ОстаткиОбороты.Организация.Инн КАК ИннКлиента, ОстаткиОбороты.СуммаОборот, ОстаткиОбороты.СуммаОборотДт, ОстаткиОбороты.СуммаОборотКт, ОстаткиОбороты.СуммаКонечныйОстаток, ОстаткиОбороты.СуммаКонечныйОстатокДт, ОстаткиОбороты.СуммаКонечныйОстатокКт, ОстаткиОбороты.СуммаКонечныйРазвернутыйОстатокДт, ОстаткиОбороты.СуммаКонечныйРазвернутыйОстатокКт, UUID(ОстаткиОбороты.Регистратор.ССылка) КАК ИдентификаторРегистратора, UUID(ОстаткиОбороты.Субконто2.Ссылка) КАК ИдентификаторДоговора, UUID(ОстаткиОбороты.Субконто1.Ссылка) КАК ИдентификаторКонтрагента, UUID(ОстаткиОбороты.Организация.Ссылка) КАК ИдентификаторКлиента, ОстаткиОбороты.Период КАК ДатаОтчета ИЗ РегистрБухгалтерии.Хозрасчетный.ОстаткиИОбороты( , ДАТАВРЕМЯ({t_minus_2.year},{t_minus_2.month},{t_minus_2.day}), Регистратор, , Счет.Код в (&СписокСчетов), , ) КАК ОстаткиОбороты """ # РегистрБухгалтерии.Хозрасчетный.ОстаткиИОбороты( ДАТАВРЕМЯ({date.year},{date.month},{date.day}), ДАТАВРЕМЯ({date.year},{date.month},{date.day}), Регистратор, , Счет.Код в (&СписокСчетов), , ) КАК ОстаткиОбороты auth = HTTPBasicAuth('obmen', 'bOR2W7w4') response = requests.post( # http://адрес сервера/название БД/hs/services/query?ProcessPostedRequest url=r'http://192.168.1.75/chupd/hs/services/query?ProcessPostedRequest', json={"query":query, "params": params}, auth=auth, verify=False ) data_from_1c = response.json() df = pd.DataFrame(data_from_1c['data']) field_mapping = { # Основные поля 'Счет': 'schet', 'Регистратор': 'registrator', 'Субконто1': 'subkonto1', 'Субконто2': 'subkonto2', 'Организация': 'organizaciya', 'НомерДоговора': 'nomer', 'ДатаДоговора': 'date_begin', 'СрокДействияДоговора': 'date_end', 'ИннКонтрагента': 'inn_subkonto1', 'ИннКлиента': 'inn_organizaciya', # Суммовые остатки и обороты 'СуммаОборот': 'summa_oborot', 'СуммаОборотДт': 'summa_oborot_dt', 'СуммаОборотКт': 'summa_oborot_kt', 'СуммаКонечныйОстаток': 'summa_konechnyy_ostatok', 'СуммаКонечныйОстатокДт': 'summa_konechnyy_ostatok_dt', 'СуммаКонечныйОстатокКт': 'summa_konechnyy_ostatok_kt', 'СуммаКонечныйРазвернутыйОстатокДт': 'summa_konechnyy_razvernutyy_ostatok_dt', 'СуммаКонечныйРазвернутыйОстатокКт': 'summa_konechnyy_razvernutyy_ostatok_kt', #ID 'ИдентификаторРегистратора': 'uid_registrator', 'ИдентификаторДоговора': 'uid_subkonto2', 'ИдентификаторКонтрагента': 'uid_subkonto1', 'ИдентификаторКлиента': 'uid_organizaciya', #DATE 'ДатаОтчета': 'get_date' } df = df.rename(columns=field_mapping) # # Удаляем дубликаты по ключевым полям перед вставкой # group_columns = [ # 'schet', 'subkonto1', 'subkonto2', 'organizaciya', 'nomer', # 'date_begin', 'date_end', 'inn_subkonto1', 'inn_organizaciya', # 'uid_subkonto2', 'uid_subkonto1', 'uid_organizaciya', 'get_date' # ] # columns_for_sum = [ # 'summa_oborot', 'summa_oborot_dt', 'summa_oborot_kt', # 'summa_konechnyy_ostatok', 'summa_konechnyy_ostatok_dt', 'summa_konechnyy_ostatok_kt', # 'summa_konechnyy_razvernutyy_ostatok_dt', 'summa_konechnyy_razvernutyy_ostatok_kt' # ] # sum_columns = { # 'summa_oborot': 'sum', 'summa_oborot_dt': 'sum', 'summa_oborot_kt': 'sum', # 'summa_konechnyy_ostatok': 'sum', 'summa_konechnyy_ostatok_dt': 'sum', 'summa_konechnyy_ostatok_kt': 'sum', # 'summa_konechnyy_razvernutyy_ostatok_dt': 'sum', 'summa_konechnyy_razvernutyy_ostatok_kt': 'sum' # } # for col in columns_for_sum: # if col in df.columns: # # Очищаем и преобразуем числовые значения # df[col] = (df[col] # .str.replace('\xa0', '', regex=False) # Убираем неразрывные пробелы # .str.replace(' ', '', regex=False) # Убираем обычные пробелы # .str.replace(',', '.', regex=False) # Запятые -> точки для десятичных # .apply(lambda x: x if x.replace('.', '').replace('-', '').isdigit() else None) # .apply(pd.to_numeric, errors='coerce')) # # Диагностика # original = df[col].copy() # cleaned = df[col] # failed = cleaned.isna() & original.notna() # if failed.any(): # display(f" {col}: не преобразовано {failed.sum()} значений") # display(f" Примеры: {original[failed].head(3).tolist()}") # df = df.groupby(group_columns, as_index=False).agg(sum_columns) # Преобразуем даты из формата DD.MM.YYYY HH:MM:SS в datetime date_columns = ['date_begin', 'date_end', 'get_date'] for col in date_columns: if col in df.columns: # Пробуем разные форматы дат df[col] = pd.to_datetime(df[col], format='%d.%m.%Y %H:%M:%S', errors='coerce') # Если не сработало, пробуем без времени if df[col].isna().any(): df[col] = pd.to_datetime(df[col], format='%d.%m.%Y', errors='coerce') # Убедимся, что все даты преобразованы for col in date_columns: if col in df.columns: failed_count = df[col].isna().sum() if failed_count > 0: print(f" Внимание: в колонке '{col}' осталось {failed_count} некорректных дат") with engine.begin() as conn: if not df.empty: conn.execute(f"DROP TABLE IF EXISTS {temp_table_name}") conn.execute(f"CREATE TEMP TABLE {temp_table_name} (LIKE public.{table_name})") df.to_sql( f'temp_{table_name}', con=conn, if_exists='append', index=False, method='multi' ) conn.execute(f"TRUNCATE TABLE public.{table_name}") conn.execute(f""" INSERT INTO public.{table_name} SELECT schet , registrator , subkonto1 , subkonto2 , organizaciya , case when nomer = '' then null else nomer end nomer , case when date_begin = '0001-01-01' then null else date_begin end date_begin , case when date_begin = '0001-01-01' then null else date_end end date_end , inn_subkonto1 as inn_subkonto1 , inn_organizaciya as inn_organizaciya , summa_oborot , summa_oborot_dt , summa_oborot_kt , summa_konechnyy_ostatok , summa_konechnyy_ostatok_dt , summa_konechnyy_ostatok_kt , summa_konechnyy_razvernutyy_ostatok_dt , summa_konechnyy_razvernutyy_ostatok_kt , uid_registrator , uid_subkonto2 , uid_subkonto1 , uid_organizaciya , get_date::date::timestamp FROM temp_{table_name} """ ) def upsert_list_fin_portfel(**kwargs): engine = get_db_engine() query = """ select distinct osv.uid_subkonto2 uid_dogovor , osv.schet as schet , osv.subkonto2 as name , null as summa_dogovora , null as percent_value from public.oborotno_salbdovaya_vedomostb osv left join public.fin_porfel fp on fp.schet = osv.schet and fp.uid_dogovor = osv.uid_subkonto2 where (osv.schet like '%%01%%' or osv.schet like '%%03%%') and fp.uid_dogovor is null """ df = pd.read_sql(query, engine) with engine.begin() as conn: if not df.empty: conn.execute("CREATE TEMP TABLE temp_fin_porfel (uid_dogovor text null, schet text null, name text null, summa_dogovora text null, percent_value text null)") df.to_sql('temp_fin_porfel', con=conn, if_exists='append', index=False, method='multi') conn.execute(""" INSERT INTO public.fin_porfel (uid_dogovor, schet, name, summa_dogovora, percent_value) SELECT DISTINCT uid_dogovor , schet , name , summa_dogovora , percent_value FROM temp_fin_porfel ON CONFLICT (uid_dogovor, schet) DO UPDATE SET name = EXCLUDED.name, summa_dogovora = EXCLUDED.summa_dogovora, percent_value = EXCLUDED.percent_value """ ) conn.execute(""" UPDATE public.fin_porfel fp SET id = subquery.new_id FROM ( SELECT id, ROW_NUMBER() OVER (ORDER BY uid_dogovor, schet, summa_dogovora) as new_id FROM public.fin_porfel ) AS subquery WHERE fp.id = subquery.id; """ ) return 'Список обновлен.' else: return 'Обновлять нечего.' def pogasheniya(**kwargs): engine = get_db_engine() query = text(""" SELECT DISTINCT osv.schet , osv.uid_subkonto2 as uid_dogovor , osv.subkonto2 as name , osv.nomer , osv.date_begin , osv.date_end FROM oborotno_salbdovaya_vedomostb osv LEFT JOIN pogasheniya p on p.uid_dogovor = osv.uid_subkonto2 and p.schet = osv.schet WHERE p.uid_dogovor is null and (osv.schet::text like '%03%' OR osv.schet::text like '%01%') UNION ALL SELECT '76.07.1' as schet , '00000000-0000-0000-0000-000000000000' as uid_dogovor , 'Лизинг' as name , osv.agreement_num as nomer , osv.agreement_date as date_begin , osv.redemption_date as date_end FROM lizingi_garantii osv LEFT JOIN pogasheniya p on p.nomer = osv.agreement_num WHERE p.uid_dogovor is null """) df = pd.read_sql(query, engine) with engine.begin() as conn: if not df.empty: conn.execute("CREATE TEMP TABLE temp_pogasheniya (schet text null, uid_dogovor text null, name text null, nomer text null, date_begin text null, date_end text null)") df.to_sql('temp_pogasheniya', con=conn, if_exists='append', index=False, method='multi') conn.execute(""" INSERT INTO public.pogasheniya (schet, uid_dogovor, name, nomer, date_begin, date_end) SELECT * FROM temp_pogasheniya ON CONFLICT (schet, uid_dogovor, nomer) DO UPDATE SET name = EXCLUDED.name, nomer = EXCLUDED.nomer, date_begin = EXCLUDED.date_begin, date_end = EXCLUDED.date_end """) # conn.execute(""" # UPDATE public.pogasheniya fp # SET id = subquery.new_id # FROM ( # SELECT # id, # ROW_NUMBER() OVER (ORDER BY uid_dogovor, schet, nomer) as new_id # FROM public.pogasheniya # ) AS subquery # WHERE fp.id = subquery.id; # """ # ) return 'Список обновлен.' else: return 'Обновлять нечего.' with DAG( dag_id='data_download_from_1C_source_with_docs', default_args=default_args, description='Выгрузка данных из 1С', schedule_interval=None, #"0,30 01-10 * * *", catchup=False, tags=['sigma'], ) as dag: read_data_1C_task = PythonOperator( task_id="read_data_1C", python_callable=read_data_1C, provide_context=True ) upsert_list_fin_portfel_task = PythonOperator( task_id='upsert_list_fin_portfel', python_callable=upsert_list_fin_portfel, provide_context=True ) pogasheniya_task = PythonOperator( task_id='pogasheniya', python_callable=pogasheniya, provide_context=True ) read_data_1C_task >> [upsert_list_fin_portfel_task, pogasheniya_task]