diff --git a/dags/OSV_with_docs.py b/dags/OSV_with_docs.py index d32d6a9..ccd2d26 100644 --- a/dags/OSV_with_docs.py +++ b/dags/OSV_with_docs.py @@ -347,6 +347,64 @@ def pogasheniya(**kwargs): else: return 'Обновлять нечего.' +def poruchitelstva(**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 poruchitelstva p on p.uid_dogovor = osv.uid_subkonto2 and p.schet = osv.schet + LEFT JOIN sigma_gk sg_sub ON sg_sub.inn = osv.inn_subkonto1::text + LEFT JOIN sigma_gk sg_org ON sg_org.inn = osv.inn_organizaciya::text + WHERE p.uid_dogovor is null and (osv.schet::text like '%03%' OR osv.schet::text like '%01%') and (case when sg_sub.inn is not NULL AND sg_org.inn is not null then 1 else 0 end = 0) + UNION ALL + SELECT + case when osv.category_name = 'Лизинг' then '76.07.1' else '76.09' end as schet + , osv.id as uid_dogovor + , osv.category_name 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 poruchitelstva 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_poruchitelstva (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_poruchitelstva', con=conn, if_exists='append', index=False, method='multi') + conn.execute(""" + INSERT INTO public.poruchitelstva (schet, uid_dogovor, name, nomer, date_begin, date_end) + SELECT * FROM temp_poruchitelstva + """) + # ON CONFLICT (schet, uid_dogovor) + # DO UPDATE SET + # name = EXCLUDED.name, + # nomer = EXCLUDED.nomer, + # date_begin = EXCLUDED.date_begin, + # date_end = EXCLUDED.date_end + conn.execute(""" + UPDATE public.poruchitelstva fp + SET id = subquery.new_id + FROM ( + SELECT + id, + ROW_NUMBER() OVER (ORDER BY uid_dogovor, schet, nomer) as new_id + FROM public.poruchitelstva + ) 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, @@ -374,4 +432,10 @@ with DAG( provide_context=True ) -read_data_1C_task >> [upsert_list_fin_portfel_task, pogasheniya_task] \ No newline at end of file + poruchitelstva_task = PythonOperator( + task_id='poruchitelstva', + python_callable=poruchitelstva, + provide_context=True + ) + +read_data_1C_task >> [upsert_list_fin_portfel_task, pogasheniya_task, poruchitelstva_task] \ No newline at end of file