demand_server/app/scheduler/manager.py

from datetime import datetime
from zoneinfo import ZoneInfo

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.date import DateTrigger
from apscheduler.triggers.interval import IntervalTrigger

from app.core.config import settings
from app.scheduler.jobs import (
    demand_pool_daily_strategy_alert_job,
    demand_pool_today_incremental_sync_job,
    heartbeat_job,
)

# 与 scheduler 一致。APScheduler 3.x 在 add_job 里传入 CronTrigger 实例时，若未显式指定
# timezone，会按 UTC 等默认时区解析 hour/minute，导致「配置的 9:40」实际在 UTC 触发，
# 北京时间观测为 17:40（+8 小时）。
_CRON_TZ = "Asia/Shanghai"

scheduler = AsyncIOScheduler(timezone=_CRON_TZ)


def setup_jobs() -> None:
    scheduler.add_job(
        heartbeat_job,
        trigger=IntervalTrigger(seconds=settings.scheduler_heartbeat_seconds),
        id="heartbeat_job",
        replace_existing=True,
    )
    if settings.demand_pool_hourly_sync_enabled:
        scheduler.add_job(
            demand_pool_today_incremental_sync_job,
            trigger=CronTrigger(
                minute=settings.demand_pool_hourly_sync_minute,
                timezone=_CRON_TZ,
            ),
            id="demand_pool_hourly_sync_job",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
    if settings.demand_pool_daily_strategy_alert_enabled:
        scheduler.add_job(
            demand_pool_daily_strategy_alert_job,
            trigger=CronTrigger(
                hour=settings.demand_pool_daily_strategy_alert_hour,
                minute=settings.demand_pool_daily_strategy_alert_minute,
                timezone=_CRON_TZ,
            ),
            id="demand_pool_daily_strategy_alert_job",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )


def start_scheduler() -> None:
    if scheduler.running:
        return
    setup_jobs()
    scheduler.start()
    if settings.demand_pool_hourly_sync_enabled:
        scheduler.add_job(
            demand_pool_today_incremental_sync_job,
            trigger=DateTrigger(
                run_date=datetime.now(ZoneInfo(_CRON_TZ)),
                timezone=_CRON_TZ,
            ),
            id="demand_pool_startup_sync_job",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )


def shutdown_scheduler() -> None:
    if not scheduler.running:
        return
    scheduler.shutdown(wait=False)