Add DAG and migration_reporter module

2026-10-29 16:14:13 +03:00
parent d89623b013
commit 16ee4d41e4
4 changed files with 6674 additions and 0 deletions
@@ -0,0 +1,113 @@
+"""
+DAG: пайплайн отчетов по миграциям.
+
+Этапы (каждая задача видна отдельно в Airflow UI):
+    1. pull_repo      — git pull локальной копии репозитория миграций
+    2. generate_excel — анализ .kt файлов + Excel-отчет + кэш изменений (JSON)
+    3. generate_html  — HTML-отчет из кэша (батчи по дате загрузки + фильтры)
+
+Расписание: Пн / Ср / Пт в 08:00 (cron "0 8 * * 1,3,5").
+
+Требования к окружению:
+    - пакет openpyxl в образе (_PIP_ADDITIONAL_REQUIREMENTS=openpyxl или кастомный Dockerfile);
+    - git в контейнере + доступ до origin (token в remote-url или ssh-key, см. AIRFLOW_SETUP.md);
+    - volume-монтирования: репозиторий миграций и папка отчетов (общая).
+
+Конфигурация пути (приоритет: env -> Airflow Variable -> default):
+    MIGRATIONS_REPO_DIR / Variable "migrations_repo_dir"   — корень репозитория
+    MIGRATIONS_REPORTS_DIR / Variable "migrations_reports_dir" — общая папка отчетов
+"""
+
+import os
+import re
+import subprocess
+import sys
+from datetime import datetime, timedelta
+
+from airflow.decorators import dag, task
+
+
+def _setting(var_key: str, env_key: str, default: str) -> str:
+    value = os.environ.get(env_key)
+    if value:
+        return value
+    try:
+        from airflow.models import Variable
+        return Variable.get(var_key)
+    except Exception:
+        return default
+
+
+REPO_DIR = _setting("migrations_repo_dir", "MIGRATIONS_REPO_DIR", "/opt/airflow/migrations_tool")
+REPORTS_DIR = _setting("migrations_reports_dir", "MIGRATIONS_REPORTS_DIR", "/opt/airflow/reports")
+MIGRATIONS_DIR = os.path.join(REPO_DIR, "pgsql")
+SCRIPT_PATH = os.path.join(REPO_DIR, "migration_reporter.py")
+
+
+def _run(cmd: list, timeout: int = 3600) -> str:
+    """Запускает команду, падает с полным выводом при ошибке"""
+    result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
+    if result.returncode != 0:
+        raise RuntimeError(
+            f"Ошибка выполнения ({result.returncode}): {' '.join(map(str, cmd))}\n"
+            f".stdout: {result.stdout}\n.stderr: {result.stderr}"
+        )
+    return result.stdout
+
+
+@dag(
+    dag_id="migration_reporter_pipeline",
+    description="Отчет по миграциям: git pull → Excel → HTML",
+    schedule="0 8 * * 1,3,5",  # Пн, Ср, Пт в 08:00
+    start_date=datetime(2026, 1, 1),
+    catchup=False,
+    max_active_runs=1,
+    default_args={
+        "owner": "migrations",
+        "retries": 2,
+        "retry_delay": timedelta(minutes=5),
+    },
+    tags=["migrations", "reporting"],
+)
+def migration_reporter_pipeline():
+
+    @task
+    def pull_repo() -> str:
+        """Обновляет локальную копию репозитория миграций"""
+        stdout = _run(["git", "-C", REPO_DIR, "pull", "--ff-only"], timeout=600)
+        return stdout.strip() or "already up to date"
+
+    @task
+    def generate_excel() -> dict:
+        """Анализ миграций + Excel-отчет; кэш изменений идет в XCom"""
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        output = os.path.join(REPORTS_DIR, f"migration_report_{timestamp}.xlsx")
+        stdout = _run([
+            sys.executable, SCRIPT_PATH,
+            "excel", "--dir", MIGRATIONS_DIR, "--output", output,
+        ])
+        match = re.search(r"^CACHE_FILE: (.+)$", stdout, re.MULTILINE)
+        cache_path = match.group(1).strip() if match else output.replace(".xlsx", ".changes.json")
+        return {"report": output, "cache": cache_path}
+
+    @task
+    def generate_html(excel_info: dict) -> str:
+        """HTML-отчет из кэша изменений (в файлы latest.html / latest.xlsx)."""
+        output = os.path.join(
+            REPORTS_DIR, os.path.basename(excel_info["report"]).replace(".xlsx", ".html")
+        )
+        stdout = _run([
+            sys.executable, SCRIPT_PATH,
+            "html", "--cache", excel_info["cache"], "--output", output,
+        ])
+        match = re.search(r"^HTML_FILE: (.+)$", stdout, re.MULTILINE)
+        return match.group(1).strip() if match else output
+
+    pull = pull_repo()
+    excel = generate_excel()
+    html = generate_html(excel)
+
+    pull >> excel >> html
+
+
+migration_reporter_pipeline()