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

dags/pogasheniya.py

@@ -32,142 +32,107 @@ def get_db_engine():
         max_overflow=20
     )
     
-def update_poruchitelstva_structure_with_detailed_errors(**kwargs):
+def update_pogasheniya_structure_with_detailed_errors(**kwargs):
     """
-    Версия с подробными сообщениями об ошибках для поручительств
+    Версия с подробными сообщениями об ошибках
     """
     engine = get_db_engine()
     
-    # Шаг 1: Проверяем существование необходимых объектов
+    # Шаг 1: Проверяем существование функции get_years()
     try:
         with engine.connect() as conn:
-            # Проверяем существование функции get_active_companies()
+            # Проверяем, существует ли функция
             func_exists = conn.execute("""
                 SELECT EXISTS(
                     SELECT 1 FROM pg_proc 
                     JOIN pg_namespace ON pg_proc.pronamespace = pg_namespace.oid 
-                    WHERE proname = 'get_active_companies' 
+                    WHERE proname = 'get_years' 
                     AND nspname = 'public'
                 )
             """).scalar()
             
             if not func_exists:
-                return "ERROR: Функция get_active_companies() не существует в схеме public"
+                return "ERROR: Функция get_years() не существует в схеме public"
             
-            # Проверяем существование таблицы sigma_gk
+            # Проверяем существование таблицы years_for_pogasheniya
             table_exists = conn.execute("""
                 SELECT EXISTS(
                     SELECT 1 FROM information_schema.tables 
-                    WHERE table_name = 'sigma_gk' 
+                    WHERE table_name = 'years_for_pogasheniya' 
                     AND table_schema = 'public'
                 )
             """).scalar()
             
             if not table_exists:
-                return "ERROR: Таблица sigma_gk не существует"
+                return "ERROR: Таблица years_for_pogasheniya не существует"
             
-            # Проверяем, есть ли активные компании
+            # Проверяем, есть ли данные в таблице
-            active_count = conn.execute("""
+            has_data = conn.execute("""
-                SELECT COUNT(*) 
+                SELECT EXISTS(SELECT 1 FROM years_for_pogasheniya LIMIT 1)
-                FROM sigma_gk 
-                WHERE is_active = '1' 
-                  AND poruchitel_name IS NOT NULL 
-                  AND poruchitel_name != ''
             """).scalar()
             
-            if active_count == 0:
+            if not has_data:
-                # Получаем список всех компаний для отладки
+                return "ERROR: Таблица years_for_pogasheniya пустая"
-                all_companies = conn.execute("""
-                    SELECT name, inn, is_active, poruchitel_name 
-                    FROM sigma_gk 
-                    LIMIT 10
-                """).fetchall()
             
-                companies_info = "\n".join([
+            # Вызываем функцию get_years()
-                    f"{row[0]} (ИНН: {row[1]}, active: {row[2]}, poruchitel: {row[3]})" 
+            result = conn.execute("SELECT * FROM get_years()").fetchone()
-                    for row in all_companies
-                ])
             
-                return f"ERROR: Нет активных компаний в sigma_gk\n\nДоступные компании:\n{companies_info}"
+            if result is None:
+                return "ERROR: Функция get_years() вернула NULL. Проверьте данные в таблице"
             
-            # Получаем список активных компаний для информации
+            start_year, end_year = result
-            active_companies = conn.execute("""
-                SELECT poruchitel_name 
-                FROM sigma_gk 
-                WHERE is_active = '1' 
-                  AND poruchitel_name IS NOT NULL 
-                  AND poruchitel_name != ''
-                ORDER BY poruchitel_name
-            """).fetchall()
             
-            companies_list = ", ".join([row[0] for row in active_companies])
+            # Детальная проверка значений
+            error_messages = []
+            
+            if start_year is None:
+                error_messages.append("start_year is NULL")
+            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})"
             
     except Exception as e:
         return f"ERROR: Ошибка при проверке данных: {str(e)}"
     
-    # Шаг 2: Обновляем структуру таблицы poruchitelstva
+    # Шаг 2: Обновляем структуру
     try:
         with engine.begin() as conn:
-            # Вызываем функцию обновления
             update_result = conn.execute(
-                "SELECT update_companies_and_rebuild()"
+                "SELECT update_years_and_rebuild(%s, %s)",
+                (str(start_year), str(end_year))
             ).scalar()
             
-            # Добавляем информацию о компаниях в результат
+            return f"SUCCESS: {update_result}"
-            result_message = f"{update_result}\n\nАктивные компании: {companies_list}\nВсего активных компаний: {active_count}"
-            
-            return f"SUCCESS: {result_message}"
             
     except Exception as e:
-        error_detail = str(e)
+        return f"ERROR: Ошибка при пересоздании таблицы: {str(e)}"
-        
-        # Дополнительная диагностика при ошибках
-        diagnostic_info = ""
-        try:
-            with engine.connect() as conn2:
-                # Проверяем существование таблицы poruchitelstva
-                por_table_exists = conn2.execute("""
-                    SELECT EXISTS(
-                        SELECT 1 FROM information_schema.tables 
-                        WHERE table_name = 'poruchitelstva' 
-                        AND table_schema = 'public'
-                    )
-                """).scalar()
-                
-                if por_table_exists:
-                    # Проверяем структуру таблицы
-                    columns = conn2.execute("""
-                        SELECT column_name, data_type 
-                        FROM information_schema.columns 
-                        WHERE table_name = 'poruchitelstva' 
-                          AND table_schema = 'public'
-                        ORDER BY ordinal_position
-                    """).fetchall()
-                    
-                    columns_list = "\n".join([f"  - {row[0]} ({row[1]})" for row in columns])
-                    diagnostic_info = f"\n\nТекущая структура poruchitelstva:\n{columns_list}"
-                else:
-                    diagnostic_info = "\n\nТаблица poruchitelstva не существует"
-                    
-        except Exception as diag_error:
-            diagnostic_info = f"\n\nНе удалось получить диагностическую информацию: {str(diag_error)}"
-        
-        return f"ERROR: Ошибка при пересоздании таблицы: {error_detail}{diagnostic_info}"
 
 
 with DAG(
-    dag_id='update_poruchitelstva',
+    dag_id='update_pogasheniya',
     default_args=default_args,
-    description='Обновление таблицы поручительств',
+    description='Обновление списка погашений',
     schedule_interval=None,
     catchup=False,
-    tags=['sigma', 'poruchitelstva'],
+    tags=['sigma'],
 ) as dag:
 
-    poruchitelstva_task = PythonOperator(
+    pogasheniya_task = PythonOperator(
-        task_id='update_poruchitelstva_structure',
+        task_id='update_pogasheniya_structure_with_detailed_errors',
-        python_callable=update_poruchitelstva_structure_with_detailed_errors,
+        python_callable=update_pogasheniya_structure_with_detailed_errors,
         provide_context=True
     )
 
-poruchitelstva_task
+pogasheniya_task