dags_scripts/dags/migration_reporter_dag.py

"""
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:
        import os, subprocess
        print(f"REPO_DIR={REPO_DIR}")
        print(f"Exists: {os.path.exists(REPO_DIR)}")
        print(f"CWD: {os.getcwd()}")
        result = subprocess.run(
            ["git", "-C", REPO_DIR, "pull", "--ff-only"],
            capture_output=True, text=True, timeout=600
        )
        print(f"STDOUT: {result.stdout}")
        print(f"STDERR: {result.stderr}")
        print(f"RC: {result.returncode}")
        if result.returncode != 0:
            raise RuntimeError(f"git pull failed: {result.stderr}")
        return result.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()