"""Compatibility facade over the durable pipeline control plane.""" from __future__ import annotations from typing import Any from api.services.pipeline import get_run, pipeline_health from supply_infra.config import get_infra_settings from supply_infra.pipeline.dag import PIPELINE_STEPS from supply_infra.pipeline.dates import CHINA_TIMEZONE, next_schedule_china from supply_infra.pipeline.run_service import submit_pipeline_run from supply_infra.scheduler.constants import SUPPLY_PIPELINE_JOB_ID, SUPPLY_PIPELINE_JOB_NAME def list_triggerable_jobs() -> list[dict[str, Any]]: return [ { "id": SUPPLY_PIPELINE_JOB_ID, "name": SUPPLY_PIPELINE_JOB_NAME, "description": "提交严格门禁的 12 步持久化供给流水线", "accepts_biz_dt": True, "deprecated": False, "steps": [step.key for step in PIPELINE_STEPS], } ] def run_scheduler_job( job_id: str, *, biz_dt: str | None = None, ) -> dict[str, Any]: if job_id != SUPPLY_PIPELINE_JOB_ID: raise KeyError(job_id) return submit_pipeline_run( biz_dt=biz_dt, trigger_type="api", trigger_source="legacy_scheduler_api", ).to_dict() def get_scheduler_job_run(run_id: str) -> dict[str, Any] | None: return get_run(run_id) def run_supply_pipeline(*, biz_dt: str | None = None) -> dict[str, Any]: return run_scheduler_job(SUPPLY_PIPELINE_JOB_ID, biz_dt=biz_dt) def scheduler_status() -> dict[str, Any]: settings = get_infra_settings() health = pipeline_health() return { "enabled": settings.scheduler_enabled, "running": health["scheduler_running"], "embedded": False, "deprecated": True, "timezone": settings.scheduler_timezone, "jobs": [ { "id": SUPPLY_PIPELINE_JOB_ID, "name": SUPPLY_PIPELINE_JOB_NAME, "next_run_time": next_schedule_china(settings=settings) .replace(tzinfo=CHINA_TIMEZONE) .isoformat(), } ], **health, }