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

dags/pogasheniya.py

@@ -30,62 +30,98 @@ def get_db_engine():
         pool_size=10,
         max_overflow=20
     )
-
+    
-def pogasheniya(**kwargs):
+def update_pogasheniya_structure_with_detailed_errors(**kwargs):
+    """
+    Версия с подробными сообщениями об ошибках
+    """
     engine = get_db_engine()
-    query = text("""
+    
-    SELECT DISTINCT
+    logger.info("Запуск обновления структуры pogasheniya с проверками")
-          osv.schet
+    
-        , osv.uid_subkonto2 as uid_dogovor
+    # Шаг 1: Проверяем существование функции get_years()
-        , osv.subkonto2 as name
+    try:
-        , osv.nomer
+        with engine.connect() as conn:
-        , osv.date_begin
+            # Проверяем, существует ли функция
-        , osv.date_end
+            func_exists = conn.execute("""
-    FROM oborotno_salbdovaya_vedomostb osv
+                SELECT EXISTS(
-    LEFT JOIN pogasheniya p on p.uid_dogovor = osv.uid_subkonto2 and p.schet = osv.schet
+                    SELECT 1 FROM pg_proc 
-    WHERE p.uid_dogovor is null and (osv.schet::text like '%03%' OR osv.schet::text like '%01%')
+                    JOIN pg_namespace ON pg_proc.pronamespace = pg_namespace.oid 
-    UNION ALL
+                    WHERE proname = 'get_years' 
-    SELECT 
+                    AND nspname = 'public'
-          '76.07.1' as schet
+                )
-        , '00000000-0000-0000-0000-000000000000' as uid_dogovor
+            """).scalar()
-        , 'Лизинг' as name
+            
-        , osv.agreement_num as nomer
+            if not func_exists:
-        , osv.agreement_date as date_begin
+                return "ERROR: Функция get_years() не существует в схеме public"
-        , osv.redemption_date as date_end
+            
-    FROM lizingi_garantii osv
+            # Проверяем существование таблицы years_for_pogasheniya
-    LEFT JOIN pogasheniya p on p.nomer = osv.agreement_num
+            table_exists = conn.execute("""
-    WHERE p.uid_dogovor is null
+                SELECT EXISTS(
-    """)
+                    SELECT 1 FROM information_schema.tables 
-    df = pd.read_sql(query, engine)
+                    WHERE table_name = 'years_for_pogasheniya' 
-    with engine.begin() as conn:
+                    AND table_schema = 'public'
-        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)")
+            """).scalar()
-            df.to_sql('temp_pogasheniya', con=conn, if_exists='append', index=False, method='multi')
+            
-            conn.execute("""
+            if not table_exists:
-            INSERT INTO public.pogasheniya (schet, uid_dogovor, name, nomer, date_begin, date_end)
+                return "ERROR: Таблица years_for_pogasheniya не существует"
-            SELECT * FROM temp_pogasheniya
+            
-            ON CONFLICT (schet, uid_dogovor, nomer)
+            # Проверяем, есть ли данные в таблице
-            DO UPDATE SET
+            has_data = conn.execute("""
-                name = EXCLUDED.name,
+                SELECT EXISTS(SELECT 1 FROM years_for_pogasheniya LIMIT 1)
-                nomer = EXCLUDED.nomer,
+            """).scalar()
-                date_begin = EXCLUDED.date_begin,
+            
-                date_end = EXCLUDED.date_end
+            if not has_data:
-            """)
+                return "ERROR: Таблица years_for_pogasheniya пустая"
-            # conn.execute("""
+            
-            # UPDATE public.pogasheniya fp
+            # Вызываем функцию get_years()
-            # SET id = subquery.new_id
+            result = conn.execute("SELECT * FROM get_years()").fetchone()
-            # FROM (
+            
-            #     SELECT 
+            if result is None:
-            #         id,
+                return "ERROR: Функция get_years() вернула NULL. Проверьте данные в таблице"
-            #         ROW_NUMBER() OVER (ORDER BY uid_dogovor, schet, nomer) as new_id
+            
-            #     FROM public.pogasheniya
+            start_year, end_year = result
-            # ) AS subquery
+            
-            # WHERE fp.id = subquery.id;
+            # Детальная проверка значений
-            # """
+            error_messages = []
-            # )
+            
-            return 'Список обновлен.'
+            if start_year is None:
-        else:
+                error_messages.append("start_year is NULL")
-            return 'Обновлять нечего.'
+            if end_year is None:
+                error_messages.append("end_year is NULL")
+            
+            if error_messages:
+                return f"ERROR: {', '.join(error_messages)}"
+            
+            # Преобразуем в числа с проверкой
+            try:
+                start_year_int = int(start_year)
+                end_year_int = int(end_year)
+            except ValueError:
+                return f"ERROR: Невозможно преобразовать годы в числа: start_year='{start_year}', end_year='{end_year}'"
+            
+            if start_year_int > end_year_int:
+                return f"ERROR: Неправильный порядок лет: start_year ({start_year}) > end_year ({end_year})"
+            
+            logger.info(f"Валидные настройки лет: {start_year}-{end_year}")
+            
+    except Exception as e:
+        return f"ERROR: Ошибка при проверке данных: {str(e)}"
+    
+    # Шаг 2: Обновляем структуру
+    try:
+        with engine.begin() as conn:
+            update_result = conn.execute(
+                "SELECT update_years_and_rebuild(%s, %s)",
+                (str(start_year), str(end_year))
+            ).scalar()
+            
+            return f"SUCCESS: {update_result}"
+            
+    except Exception as e:
+        return f"ERROR: Ошибка при пересоздании таблицы: {str(e)}"
+
 
 with DAG(
     dag_id='update_pogasheniya',
@@ -97,8 +133,8 @@ with DAG(
 ) as dag:
 
     pogasheniya_task = PythonOperator(
-        task_id='pogasheniya',
+        task_id='update_pogasheniya_structure_with_detailed_errors',
-        python_callable=pogasheniya,
+        python_callable=update_pogasheniya_structure_with_detailed_errors,
         provide_context=True
     )