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