feat(airflow): пульт стал world_init, добавлен DAG world_next_day (#4)

- Зачем:
  - список DAG'ов должен читаться лесенкой ddl_init → world_init →
    world_next_day, а путь менти — проходиться пустыми формами
    (issue #4, спека редизайна пути менти, решения 2–3).
- Что:
  - generator_control переименован в world_init, дефолт операции —
    import; next-day ушёл из выпадашки в отдельный DAG;
  - новый беспараметрный world_next_day: расписание */30 * * * *,
    создаётся на паузе, catchup=False, max_active_runs=1; общие
    задачи вынесены в airflow/dags/utils/startup_history_tasks.py;
  - доки и контрактные тесты обновлены синхронно; быстрый старт
    README — без make ddl, схему создаёт DAG ddl_init.
- Проверка:
  - make test (210 + 31) и make lint зелёные;
  - живая приёмка на чистом стенде: world_init пустой формой
    импортировал эталонный мир за 217 с (3 дня, 280 437 событий),
    world_next_day после снятия с паузы добавляет ровно один день
    за прогон, дашборд Superset собирается.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

airflow/dags/utils/startup_history_tasks.py

@@ -0,0 +1,59 @@
+"""Общие задачи Airflow для операций над миром стенда."""
+
+from airflow.exceptions import AirflowException
+from airflow.models.dag import DagModel
+from airflow.utils.session import provide_session
+from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook
+
+from clickstream_generator.airflow_control import (
+    assert_clickhouse_matches_manifest,
+    assert_live_generator_not_running_for_next_day,
+    assert_next_day_snapshot,
+    build_next_day_env,
+    load_manifest_from_kafka,
+    load_state_from_kafka,
+    run_next_day,
+    target_dag_trigger_error,
+)
+
+
+def precheck_next_day(**context) -> None:
+    """Проверяет точку продолжения и готовит настройки мира."""
+    assert_live_generator_not_running_for_next_day()
+    manifest = load_manifest_from_kafka()
+    state = load_state_from_kafka()
+    assert_next_day_snapshot(manifest, state)
+    context["ti"].xcom_push(
+        key="generator_env",
+        value=build_next_day_env(manifest),
+    )
+
+
+def run_next_day_task(**context) -> None:
+    """Генерирует следующий модельный день от проверенного state."""
+    env = context["ti"].xcom_pull(
+        task_ids="precheck_next_day",
+        key="generator_env",
+    )
+    run_next_day(env)
+
+
+def check_manifest_task(**context) -> None:
+    """Сверяет витрину ClickHouse с manifest стартовой истории."""
+    manifest = load_manifest_from_kafka()
+    hook = ClickHouseHook(clickhouse_conn_id="clickhouse_default", database="default")
+    assert_clickhouse_matches_manifest(manifest, hook)
+
+
+@provide_session
+def assert_target_dag_not_paused(dag_id: str, session=None) -> None:
+    """
+    Проверяет, что зависимый DAG можно запустить.
+
+    Context7: для Airflow 2.10.5 у старого TriggerDagRunOperator нет надёжного
+    fail_when_dag_is_paused, поэтому паузу проверяем заранее через DagModel.
+    """
+    dag_model = session.query(DagModel).filter(DagModel.dag_id == dag_id).one_or_none()
+    error = target_dag_trigger_error(dag_id, dag_model)
+    if error:
+        raise AirflowException(error)

airflow/dags/world_init_dag.py

@@ -10,36 +10,28 @@ from __future__ import annotations
 from datetime import datetime, timedelta
 
 from airflow import DAG
-from airflow.exceptions import AirflowException
-from airflow.models.dag import DagModel
 from airflow.models.param import Param
 from airflow.operators.empty import EmptyOperator
 from airflow.operators.python import BranchPythonOperator, PythonOperator
 from airflow.operators.trigger_dagrun import TriggerDagRunOperator
-from airflow.utils.session import provide_session
 from airflow.utils.trigger_rule import TriggerRule
 from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook
 
 from clickstream_generator.airflow_control import (
     KAFKA_BOOTSTRAP_SERVERS,
-    assert_clickhouse_matches_manifest,
-    assert_expected_t_end,
     assert_live_generator_not_running,
-    assert_live_generator_not_running_for_next_day,
-    assert_next_day_snapshot,
     assert_stand_clean,
     build_control_env,
-    build_next_day_env,
     default_artifact_path,
-    load_manifest_from_kafka,
-    load_state_from_kafka,
     run_backfill,
     run_import,
-    run_next_day,
-    target_dag_trigger_error,
     validate_import_artifact,
 )
 from clickstream_generator.launch import PROFILES
+from utils.startup_history_tasks import (
+    assert_target_dag_not_paused,
+    check_manifest_task,
+)
 
 
 default_args = {
@@ -107,8 +99,6 @@ def choose_operation(**context) -> str:
         return "check_etl_not_paused_before_backfill"
     if operation == "import":
         return "check_etl_not_paused_before_import"
-    if operation == "next-day":
-        return "check_etl_not_paused_before_next_day"
     if operation == "check":
         return "check_only"
     raise ValueError(f"Неизвестная операция: {operation}")
@@ -151,56 +141,9 @@ def run_import_task(**context) -> None:
     run_import(env, artifact_path)
 
 
-def precheck_next_day(**context) -> None:
-    """Проверяет точку продолжения и готовит неизменные настройки мира."""
-    assert_live_generator_not_running_for_next_day()
-    manifest = load_manifest_from_kafka()
-    state = load_state_from_kafka()
-    assert_next_day_snapshot(manifest, state)
-    assert_expected_t_end(
-        str(_param(context, "expected_t_end") or "").strip(),
-        str(manifest["model_t_end"]),
-    )
-    context["ti"].xcom_push(
-        key="generator_env",
-        value=build_next_day_env(manifest),
-    )
-
-
-def run_next_day_task(**context) -> None:
-    """Генерирует следующий модельный день от проверенного state."""
-    env = context["ti"].xcom_pull(
-        task_ids="precheck_next_day",
-        key="generator_env",
-    )
-    run_next_day(env)
-
-
-def check_manifest_task(**context) -> None:
-    """Сверяет витрину ClickHouse с manifest стартовой истории."""
-    manifest = load_manifest_from_kafka()
-    hook = ClickHouseHook(clickhouse_conn_id="clickhouse_default", database="default")
-    assert_clickhouse_matches_manifest(manifest, hook)
-
-
-@provide_session
-def assert_target_dag_not_paused(dag_id: str, session=None) -> None:
-    """
-    Проверяет, что зависимый DAG можно запустить.
-
-    Context7: для Airflow 2.10.5 у старого TriggerDagRunOperator нет надёжного
-    fail_when_dag_is_paused, поэтому паузу проверяем заранее через DagModel.
-    """
-    dag_model = session.query(DagModel).filter(DagModel.dag_id == dag_id).one_or_none()
-    error = target_dag_trigger_error(dag_id, dag_model)
-    if error:
-        raise AirflowException(error)
-
-
-# Context7, Airflow 2.10.5: Param поддерживает enum, а max_active_runs ограничивает
-# число одновременных DAG run. Поэтому форму и блокировку next-day держим в DAG.
+# Context7, Airflow 2.10.5: Param поддерживает enum.
 with DAG(
-    dag_id="generator_control",
+    dag_id="world_init",
     description="Пульт стартовой истории генератора",
     default_args=default_args,
     schedule=None,
@@ -211,13 +154,13 @@ with DAG(
     tags=["generator", "startup-history"],
     params={
         "operation": Param(
-            "backfill",
+            "import",
             type="string",
-            enum=["backfill", "import", "next-day", "check"],
+            enum=["backfill", "import", "check"],
             title="Операция",
             description=(
-                "Что сделать: создать историю, импортировать артефакт, "
-                "добавить следующий день или проверить витрины."
+                "Что сделать: импортировать артефакт, создать историю "
+                "или проверить витрины."
             ),
         ),
         "profile": Param(
@@ -255,15 +198,6 @@ with DAG(
                 "Import: что читать; пусто — эталонный мир из репозитория."
             ),
         ),
-        "expected_t_end": Param(
-            None,
-            type=["null", "string"],
-            title="Ожидаемая граница next-day",
-            description=(
-                "Необязательный model_t_end до запуска. Защищает от "
-                "повторной доливки того же дня."
-            ),
-        ),
     },
 ) as dag:
     route = BranchPythonOperator(
@@ -289,15 +223,6 @@ with DAG(
         python_callable=run_import_task,
     )
 
-    precheck_next_day_task = PythonOperator(
-        task_id="precheck_next_day",
-        python_callable=precheck_next_day,
-    )
-    next_day_task = PythonOperator(
-        task_id="run_next_day",
-        python_callable=run_next_day_task,
-    )
-
     check_etl_not_paused_before_backfill = PythonOperator(
         task_id="check_etl_not_paused_before_backfill",
         python_callable=assert_target_dag_not_paused,
@@ -308,12 +233,6 @@ with DAG(
         python_callable=assert_target_dag_not_paused,
         op_kwargs={"dag_id": ETL_DAG_ID},
     )
-    check_etl_not_paused_before_next_day = PythonOperator(
-        task_id="check_etl_not_paused_before_next_day",
-        python_callable=assert_target_dag_not_paused,
-        op_kwargs={"dag_id": ETL_DAG_ID},
-    )
-
     trigger_etl = TriggerDagRunOperator(
         task_id="trigger_etl",
         trigger_dag_id=ETL_DAG_ID,
@@ -343,14 +262,11 @@ with DAG(
     route >> [
         check_etl_not_paused_before_backfill,
         check_etl_not_paused_before_import,
-        check_etl_not_paused_before_next_day,
         check_only,
     ]
     check_etl_not_paused_before_backfill >> precheck_backfill_task
     check_etl_not_paused_before_import >> precheck_import_task
-    check_etl_not_paused_before_next_day >> precheck_next_day_task
     precheck_backfill_task >> backfill_task >> trigger_etl
     precheck_import_task >> import_task >> trigger_etl
-    precheck_next_day_task >> next_day_task >> trigger_etl
     trigger_etl >> check_after_etl >> done
     check_only >> done

airflow/dags/world_next_day_dag.py

@@ -0,0 +1,69 @@
+"""DAG добавления одного модельного дня в мир стенда."""
+
+from datetime import datetime, timedelta
+
+from airflow import DAG
+from airflow.operators.python import PythonOperator
+from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+
+from utils.startup_history_tasks import (
+    assert_target_dag_not_paused,
+    check_manifest_task,
+    precheck_next_day,
+    run_next_day_task,
+)
+
+
+ETL_DAG_ID = "etl_pipeline"
+
+default_args = {
+    "owner": "airflow",
+    "depends_on_past": False,
+    "email_on_failure": False,
+    "email_on_retry": False,
+    "retries": 0,
+    "retry_delay": timedelta(minutes=1),
+}
+
+
+# Context7, Airflow 2.10.5: schedule принимает cron-строку. Расписание задано
+# заранее, но новый DAG остаётся на паузе до отдельного решения.
+with DAG(
+    dag_id="world_next_day",
+    description="Добавление одного модельного дня в мир стенда",
+    default_args=default_args,
+    schedule="*/30 * * * *",
+    start_date=datetime(2024, 1, 1),
+    catchup=False,
+    max_active_runs=1,
+    is_paused_upon_creation=True,
+    tags=["generator", "startup-history"],
+) as dag:
+    check_etl_not_paused = PythonOperator(
+        task_id="check_etl_not_paused",
+        python_callable=assert_target_dag_not_paused,
+        op_kwargs={"dag_id": ETL_DAG_ID},
+    )
+    precheck = PythonOperator(
+        task_id="precheck_next_day",
+        python_callable=precheck_next_day,
+    )
+    generate = PythonOperator(
+        task_id="run_next_day",
+        python_callable=run_next_day_task,
+    )
+    trigger_etl = TriggerDagRunOperator(
+        task_id="trigger_etl",
+        trigger_dag_id=ETL_DAG_ID,
+        conf={"full_refresh": True},
+        wait_for_completion=True,
+        allowed_states=["success"],
+        failed_states=["failed"],
+        poke_interval=30,
+    )
+    check = PythonOperator(
+        task_id="check_after_etl",
+        python_callable=check_manifest_task,
+    )
+
+    check_etl_not_paused >> precheck >> generate >> trigger_etl >> check