113 lines
4.5 KiB
Python
113 lines
4.5 KiB
Python
"""
|
|
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() |