From 9ecef54fa1ba4ad9f47f035d6c3fd29b3f32e5a4 Mon Sep 17 00:00:00 2001 From: bn_user Date: Fri, 5 Dec 2025 13:48:00 +0000 Subject: [PATCH] =?UTF-8?q?=D0=9E=D0=B1=D0=BD=D0=BE=D0=B2=D0=B8=D1=82?= =?UTF-8?q?=D1=8C=20dags/pogasheniya.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dags/pogasheniya.py | 141 +++++++++++++++++--------------------------- 1 file changed, 53 insertions(+), 88 deletions(-) diff --git a/dags/pogasheniya.py b/dags/pogasheniya.py index ea68369..d9e29cd 100644 --- a/dags/pogasheniya.py +++ b/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(""" - SELECT COUNT(*) - FROM sigma_gk - WHERE is_active = '1' - AND poruchitel_name IS NOT NULL - AND poruchitel_name != '' + # Проверяем, есть ли данные в таблице + has_data = conn.execute(""" + SELECT EXISTS(SELECT 1 FROM years_for_pogasheniya LIMIT 1) """).scalar() - 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}" + if not has_data: + return "ERROR: Таблица years_for_pogasheniya пустая" - # Получаем список активных компаний для информации - 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() + # Вызываем функцию get_years() + result = conn.execute("SELECT * FROM get_years()").fetchone() - companies_list = ", ".join([row[0] for row in active_companies]) + 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})" 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() - # Добавляем информацию о компаниях в результат - result_message = f"{update_result}\n\nАктивные компании: {companies_list}\nВсего активных компаний: {active_count}" - - return f"SUCCESS: {result_message}" + return f"SUCCESS: {update_result}" except Exception as 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}" + return f"ERROR: Ошибка при пересоздании таблицы: {str(e)}" 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( - task_id='update_poruchitelstva_structure', - python_callable=update_poruchitelstva_structure_with_detailed_errors, + pogasheniya_task = PythonOperator( + task_id='update_pogasheniya_structure_with_detailed_errors', + python_callable=update_pogasheniya_structure_with_detailed_errors, provide_context=True ) -poruchitelstva_task \ No newline at end of file +pogasheniya_task \ No newline at end of file