demand_server/app/scheduler/jobs.py

from datetime import datetime
from zoneinfo import ZoneInfo

from app.services.demand_pool_strategy_daily_alert import run_daily_strategy_alert
from app.services.strategy_generate_service import run_strategy_generation
from app.sync.demand_pool_sync import run_full_sync, run_today_incremental_sync


def heartbeat_job() -> None:
    now = datetime.now(ZoneInfo("Asia/Shanghai")).isoformat()
    print(f"[scheduler] heartbeat at {now}")


def demand_pool_full_sync_job() -> None:
    print("[scheduler] start full sync for demand pool")
    result = run_full_sync()
    print(f"[scheduler] full sync done: {result}")


def demand_pool_today_incremental_sync_job() -> None:
    print("[scheduler] start incremental sync for demand pool")
    result = run_today_incremental_sync()
    print(f"[scheduler] incremental sync done: {result}")


def demand_pool_daily_strategy_alert_job(partition_dt: str | None = None) -> None:
    print("[scheduler] start daily ODPS strategy alert for demand pool")
    try:
        result = run_daily_strategy_alert(partition_dt)
        print(f"[scheduler] daily strategy alert done: {result}")
    except Exception as exc:
        print(f"[scheduler] daily strategy alert failed: {exc}")
        raise


def strategy_staging_hourly_generate_job(batch_date: str | None = None) -> None:
    print("[scheduler] start hourly strategy generation for strategy_staging")
    try:
        result = run_strategy_generation(batch_date)
        print(
            "[scheduler] strategy generation done: "
            f"batch_date={result['batch_date']}, "
            f"active={result['active_count']}, "
            f"success={result['success_count']}, "
            f"skipped={result.get('skipped_count', 0)}, "
            f"errors={result['error_count']}"
        )
        if result["warnings"]:
            print(f"[scheduler] strategy generation warnings: {result['warnings']}")
        if result["errors"]:
            print(f"[scheduler] strategy generation errors: {result['errors']}")
    except Exception as exc:
        print(f"[scheduler] strategy generation failed: {exc}")
        raise