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

dags/OSV_with_docs.py

@@ -238,6 +238,59 @@ def read_data_1C(**kwargs):
             """
             )
 
+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 'Обновлять нечего.'
+
 with DAG(
     dag_id='data_download_from_1C_source_with_docs',
     default_args=default_args,
@@ -253,4 +306,10 @@ with DAG(
         provide_context=True
     )
 
-read_data_1C_task
+    upsert_list_fin_portfel_task = PythonOperator(
+        task_id='upsert_list_fin_portfel',
+        python_callable=upsert_list_fin_portfel,
+        provide_context=True
+    )
+
+read_data_1C_task >> upsert_list_fin_portfel_task