SupplyAgent/api/services/scheduler.py

"""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,
            "accepts_partition_date": False,
            "deprecated": False,
            "steps": [step.key for step in PIPELINE_STEPS],
        }
    ]


def run_scheduler_job(
    job_id: str,
    *,
    biz_dt: str | None = None,
    partition_date: str | None = None,
    wait: bool = False,
) -> dict[str, Any]:
    del partition_date
    if job_id != SUPPLY_PIPELINE_JOB_ID:
        raise KeyError(job_id)
    if wait:
        raise ValueError("wait=true is disabled for the durable API; poll by run_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,
    }