diff --git a/dags/pogasheniya.py b/dags/pogasheniya.py index d9e29cd..ea68369 100644 --- a/dags/pogasheniya.py +++ b/dags/pogasheniya.py @@ -32,107 +32,142 @@ def get_db_engine(): max_overflow=20 ) -def update_pogasheniya_structure_with_detailed_errors(**kwargs): +def update_poruchitelstva_structure_with_detailed_errors(**kwargs): """ - Версия с подробными сообщениями об ошибках + Версия с подробными сообщениями об ошибках для поручительств """ engine = get_db_engine() - # Шаг 1: Проверяем существование функции get_years() + # Шаг 1: Проверяем существование необходимых объектов 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_years' + WHERE proname = 'get_active_companies' AND nspname = 'public' ) """).scalar() if not func_exists: - return "ERROR: Функция get_years() не существует в схеме public" + return "ERROR: Функция get_active_companies() не существует в схеме public" - # Проверяем существование таблицы years_for_pogasheniya + # Проверяем существование таблицы sigma_gk table_exists = conn.execute(""" SELECT EXISTS( SELECT 1 FROM information_schema.tables - WHERE table_name = 'years_for_pogasheniya' + WHERE table_name = 'sigma_gk' AND table_schema = 'public' ) """).scalar() if not table_exists: - return "ERROR: Таблица years_for_pogasheniya не существует" + return "ERROR: Таблица sigma_gk не существует" - # Проверяем, есть ли данные в таблице - has_data = conn.execute(""" - SELECT EXISTS(SELECT 1 FROM years_for_pogasheniya LIMIT 1) + # Проверяем, есть ли активные компании + active_count = conn.execute(""" + SELECT COUNT(*) + FROM sigma_gk + WHERE is_active = '1' + AND poruchitel_name IS NOT NULL + AND poruchitel_name != '' """).scalar() - if not has_data: - return "ERROR: Таблица years_for_pogasheniya пустая" + if active_count == 0: + # Получаем список всех компаний для отладки + all_companies = conn.execute(""" + SELECT name, inn, is_active, poruchitel_name + FROM sigma_gk + LIMIT 10 + """).fetchall() + + companies_info = "\n".join([ + f"{row[0]} (ИНН: {row[1]}, active: {row[2]}, poruchitel: {row[3]})" + for row in all_companies + ]) + + return f"ERROR: Нет активных компаний в sigma_gk\n\nДоступные компании:\n{companies_info}" - # Вызываем функцию get_years() - result = conn.execute("SELECT * FROM get_years()").fetchone() + # Получаем список активных компаний для информации + 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() - if result is None: - return "ERROR: Функция get_years() вернула NULL. Проверьте данные в таблице" - - start_year, end_year = result - - # Детальная проверка значений - 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})" + companies_list = ", ".join([row[0] for row in active_companies]) except Exception as e: return f"ERROR: Ошибка при проверке данных: {str(e)}" - # Шаг 2: Обновляем структуру + # Шаг 2: Обновляем структуру таблицы poruchitelstva try: with engine.begin() as conn: + # Вызываем функцию обновления update_result = conn.execute( - "SELECT update_years_and_rebuild(%s, %s)", - (str(start_year), str(end_year)) + "SELECT update_companies_and_rebuild()" ).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: - return f"ERROR: Ошибка при пересоздании таблицы: {str(e)}" + error_detail = 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_pogasheniya', + dag_id='update_poruchitelstva', default_args=default_args, - description='Обновление списка погашений', + description='Обновление таблицы поручительств', schedule_interval=None, catchup=False, - tags=['sigma'], + tags=['sigma', 'poruchitelstva'], ) as dag: - pogasheniya_task = PythonOperator( - task_id='update_pogasheniya_structure_with_detailed_errors', - python_callable=update_pogasheniya_structure_with_detailed_errors, + poruchitelstva_task = PythonOperator( + task_id='update_poruchitelstva_structure', + python_callable=update_poruchitelstva_structure_with_detailed_errors, provide_context=True ) -pogasheniya_task \ No newline at end of file +poruchitelstva_task \ No newline at end of file