Обновить dags/OSV_with_docs.py

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]
+    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]