"""定时任务调度器:支持 cron 表达式触发采集任务。""" from __future__ import annotations import asyncio import logging from datetime import datetime, timedelta from typing import Callable, Coroutine from croniter import croniter logger = logging.getLogger(__name__) class ScheduledTask: """一个定时任务。""" #: 运行循环的轮询粒度(秒)。同时决定运行期 cron/next_run 变更的生效延迟上限。 SLEEP_TICK: float = 1.0 def __init__( self, task_id: str, cron: str, fn: Callable[[], Coroutine], enabled: bool = True, ) -> None: self.task_id = task_id self.cron = cron self.fn = fn self.enabled = enabled self.last_run: datetime | None = None self.next_run: datetime | None = None self._task: asyncio.Task | None = None # 构造时立即校验 cron:非法表达式直接报错,不留给 _run_loop 静默吞掉。 # (全量审查 C1: 传入任务类型字符串时 croniter 抛异常被吞 → 任务永不触发) self._calc_next(raise_on_error=True) def _calc_next(self, *, raise_on_error: bool = False) -> None: """计算下次运行时间。 raise_on_error=True 时非法 cron 抛 ValueError(构造期用); 否则仅告警并置 next_run=None(运行期容错)。 """ try: # 必须调用 datetime.now():传方法对象本身会让 croniter 在 # (start_time or now) 的算术里抛 TypeError,导致 next_run 永远为 None。 self.next_run = croniter(self.cron, datetime.now()).get_next(datetime) except Exception as e: self.next_run = None msg = ( f"任务 {self.task_id!r} 的 cron 表达式非法: {self.cron!r} ({e})。" "注意: 此处应为 cron 表达式(如 '0 8 * * *'),不是任务类型名。" ) if raise_on_error: raise ValueError(msg) from e logger.error(msg) def update(self, cron: str | None = None, enabled: bool | None = None) -> None: if cron is not None: self.cron = cron if enabled is not None: self.enabled = enabled self._calc_next() async def _run_loop(self) -> None: """任务运行循环。 睡眠策略: 使用固定的短 tick(SLEEP_TICK 秒)轮询 next_run,而不是 一次性 sleep 到 next_run。原因是运行期可通过 API 更新 cron / 手动 调整 next_run;若按 wait_seconds 长时间沉睡,变更最长要等 wait_seconds 才生效(实测可达 60s),表现为「改了不生效」。 """ while True: try: if not self.enabled or not self.next_run: await asyncio.sleep(self.SLEEP_TICK) self._calc_next() continue now = datetime.now() wait_seconds = (self.next_run - now).total_seconds() if wait_seconds > 0: # 短 tick 轮询,保证 next_run/cron 变更能及时被感知 await asyncio.sleep(min(wait_seconds, self.SLEEP_TICK)) continue # 执行任务:先推进 next_run 再执行,避免任务耗时导致重复触发 self.last_run = datetime.now() self._calc_next() try: logger.info("定时任务触发: %s (cron=%s)", self.task_id, self.cron) await self.fn() logger.info("定时任务完成: %s", self.task_id) except asyncio.CancelledError: raise except Exception: # 单个任务失败不能终止循环,否则一次异常即永久停摆 logger.exception("定时任务失败: %s", self.task_id) except asyncio.CancelledError: raise except Exception: # 循环体自身的意外异常(如 _calc_next)也不能终止调度 logger.exception("调度循环异常: %s", self.task_id) await asyncio.sleep(self.SLEEP_TICK) class DataQualityScheduler: """数据质量检查调度器(独立于采集任务)。""" def __init__(self) -> None: self._task: asyncio.Task | None = None self._running = False async def start(self) -> None: self._running = True self._task = asyncio.create_task(self._run_loop()) logger.info("数据质量检查调度器已启动") async def stop(self) -> None: self._running = False if self._task: self._task.cancel() async def _run_loop(self) -> None: """每小时执行一次数据质量检查。""" while self._running: try: await self._run_checks() except Exception: logger.exception("数据质量检查失败") await asyncio.sleep(3600) # 每小时 async def _run_checks(self) -> None: """执行数据质量检查并写入 DataQualityCheck 表。""" from src.db.base import AsyncSessionLocal from src.db.models import DataQualityCheck, Match, MatchStats, Standing, League from sqlalchemy import func, select async with AsyncSessionLocal() as db: # 检查1: 已完赛但无统计的比赛数 finished_no_stats = ( await db.execute( select(func.count()) .select_from(Match) .outerjoin(MatchStats, Match.id == MatchStats.match_id) .where(Match.match_status == "finished") .where(MatchStats.id.is_(None)) ) ).scalar() or 0 db.add(DataQualityCheck( check_name="finished_without_stats", entity_type="match", actual_value=float(finished_no_stats), passed=finished_no_stats == 0, severity="warning" if finished_no_stats > 0 else "info", detail={"message": f"{finished_no_stats} 场已完赛比赛缺少统计数据"}, )) # 检查2: 积分榜缺失的联赛数 leagues_without_standings = ( await db.execute( select(func.count()) .select_from(League) .outerjoin(Standing, League.id == Standing.league_id) .where(Standing.id.is_(None)) ) ).scalar() or 0 db.add(DataQualityCheck( check_name="league_without_standings", entity_type="league", actual_value=float(leagues_without_standings), passed=leagues_without_standings == 0, severity="warning" if leagues_without_standings > 0 else "info", detail={"message": f"{leagues_without_standings} 个联赛缺少积分榜"}, )) await db.commit() logger.info("数据质量检查完成: stats=%d, standings=%d", finished_no_stats, leagues_without_standings) # 全局单例 quality_scheduler = DataQualityScheduler() class Scheduler: """全局定时任务调度器。""" def __init__(self) -> None: self._tasks: dict[str, ScheduledTask] = {} self._running = False def register( self, task_id: str, cron: str, fn: Callable[[], Coroutine], enabled: bool = True, ) -> ScheduledTask: """注册(或更新)一个定时任务。 若调度器已 start,新任务会立即启动其运行循环 —— 否则运行期通过 API 新建的任务永远不会被执行(全量审查 C1 缺陷 2)。 """ existing = self._tasks.get(task_id) if existing is not None: existing.update(cron=cron, enabled=enabled) # 更新 cron 后重新校验:改坏了要立刻报错,而不是静默失活 existing._calc_next(raise_on_error=True) existing.fn = fn task = existing else: task = ScheduledTask(task_id, cron, fn, enabled) self._tasks[task_id] = task if self._running: self._ensure_loop(task) return task def _ensure_loop(self, task: ScheduledTask) -> None: """为任务启动运行循环(幂等:已在运行则跳过)。""" if task._task is None or task._task.done(): task._task = asyncio.create_task(task._run_loop()) def get(self, task_id: str) -> ScheduledTask | None: return self._tasks.get(task_id) def list_all(self) -> list[ScheduledTask]: return list(self._tasks.values()) def remove(self, task_id: str) -> None: task = self._tasks.pop(task_id, None) if task is not None and task._task is not None: task._task.cancel() async def start(self) -> None: self._running = True for task in self._tasks.values(): self._ensure_loop(task) logger.info("定时调度器已启动, 共 %d 个任务", len(self._tasks)) async def stop(self) -> None: self._running = False for task in self._tasks.values(): if task._task: task._task.cancel() self._tasks.clear() # 全局单例 scheduler = Scheduler()