"""工单定时执行调度器 — 基于 APScheduler + SQLite 持久化。 优势（对比纯 asyncio）: 1. 服务重启后定时任务不丢失（SQLite job store 持久化） 2. 启动时自动恢复 scheduled 状态的工单 3. 支持任务重试、misfire_grace_time 等生产级特性 """ import asyncio import logging import os from datetime import datetime, timezone from typing import Optional from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore from apscheduler.triggers.date import DateTrigger logger = logging.getLogger(__name__) # 全局调度器实例（延迟初始化） _scheduler: Optional[AsyncIOScheduler] = None DB_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), '..', 'apscheduler.db') def _get_scheduler() -> AsyncIOScheduler: """获取或创建 APScheduler 实例（单例）。""" global _scheduler if _scheduler is not None and _scheduler.running: return _scheduler jobstore_url = 'sqlite:///' + os.path.abspath(DB_DIR) _scheduler = AsyncIOScheduler( jobstores={'default': SQLAlchemyJobStore(url=jobstore_url)}, job_defaults={ 'coalesce': True, # 同一 job 只执行一次 'max_instances': 1, # 不并发 'misfire_grace_time': 300, # 5 分钟内迟到仍执行 }, ) return _scheduler def start_scheduler() -> None: """启动调度器（应用启动时调用）。""" scheduler = _get_scheduler() if not scheduler.running: scheduler.start() logger.info("APScheduler 已启动 (SQLite 持久化)") def shutdown_scheduler() -> None: """关闭调度器。""" global _scheduler if _scheduler and _scheduler.running: _scheduler.shutdown(wait=False) logger.info("APScheduler 已关闭") async def schedule_ticket_execution(ticket_id: int, scheduled_at: datetime) -> None: """注册一个定时执行任务。 Args: ticket_id: 工单 ID scheduled_at: 定时执行时间 (UTC) """ scheduler = _get_scheduler() job_id = f"ticket_{ticket_id}" now = datetime.now(timezone.utc) run_date = scheduled_at if scheduled_at.tzinfo else scheduled_at.replace(tzinfo=timezone.utc) if run_date <= now: # 已到时间，立即执行 logger.info("工单 %d 定时时间已到，立即执行", ticket_id) await _execute_scheduled_ticket(ticket_id) return delay = (run_date - now).total_seconds() logger.info("工单 %d 定时执行: %s (%.1f 秒后)", ticket_id, scheduled_at, delay) scheduler.add_job( _run_async_execution, trigger=DateTrigger(run_date=run_date), args=[ticket_id], id=job_id, replace_existing=True, name=f"定时执行工单 #{ticket_id}", ) async def _run_async_execution(ticket_id: int) -> None: """APScheduler 调用的入口（同步 wrapper → 异步执行）。""" await _execute_scheduled_ticket(ticket_id) async def _execute_scheduled_ticket(ticket_id: int) -> None: """执行定时工单 — 创建独立数据库会话执行。""" from app.core.database import async_session from app.models.ticket import Ticket as TicketModel from app.models.instance import Instance from app.services.ticket_executor import TicketExecutor from app.core.security import decrypt_password from sqlalchemy import select import json logger.info("开始执行定时工单 %d", ticket_id) async with async_session() as db: try: result = await db.execute(select(TicketModel).where(TicketModel.id == ticket_id)) ticket = result.scalar_one_or_none() if not ticket: logger.error("定时工单 %d 不存在", ticket_id) return if ticket.status != "scheduled": logger.warning("工单 %d 状态已变更 (%s)，跳过执行", ticket_id, ticket.status) return ticket.status = "executing" await db.commit() await db.refresh(ticket) inst_result = await db.execute(select(Instance).where(Instance.id == ticket.instance_id)) instance = inst_result.scalar_one_or_none() if not instance: ticket.status = "failed" ticket.execute_result = json.dumps({"status": "failed", "errors": ["数据源实例不存在"]}, ensure_ascii=False) ticket.execute_time = datetime.now(timezone.utc) await db.commit() return instance.password = decrypt_password(instance.password) executor = TicketExecutor() report = await executor.execute(ticket, instance) ticket.status = "success" if report.status == "success" else "failed" ticket.execute_result = report.to_json() ticket.execute_time = datetime.now(timezone.utc) await db.commit() logger.info("定时工单 %d 执行完成: %s", ticket_id, report.status) except Exception as e: logger.exception("定时工单 %d 执行异常", ticket_id) try: result = await db.execute(select(TicketModel).where(TicketModel.id == ticket_id)) ticket = result.scalar_one_or_none() if ticket: ticket.status = "failed" ticket.execute_result = json.dumps({"status": "failed", "errors": [f"{type(e).__name__}: {e}"]}, ensure_ascii=False) ticket.execute_time = datetime.now(timezone.utc) await db.commit() except Exception: logger.exception("更新定时工单 %d 失败状态也失败", ticket_id) def cancel_ticket_execution(ticket_id: int) -> bool: """取消定时执行任务。 Args: ticket_id: 工单 ID Returns: 是否成功取消 """ scheduler = _get_scheduler() job_id = f"ticket_{ticket_id}" try: scheduler.remove_job(job_id) logger.info("已取消工单 %d 的定时执行", ticket_id) return True except Exception: return False async def recover_scheduled_tickets() -> int: """启动时恢复所有 scheduled 状态的工单。 扫描 DB 中 status=scheduled 的工单，重新注册到 APScheduler。 Returns: 恢复的任务数量 """ from app.core.database import async_session from app.models.ticket import Ticket as TicketModel from sqlalchemy import select count = 0 async with async_session() as db: result = await db.execute( select(TicketModel).where(TicketModel.status == "scheduled") ) tickets = result.scalars().all() now = datetime.now(timezone.utc) for ticket in tickets: if not ticket.scheduled_at: continue scheduled = ticket.scheduled_at if scheduled.tzinfo is None: scheduled = scheduled.replace(tzinfo=timezone.utc) if scheduled <= now: # 已过期，立即执行 logger.info("恢复过期定时工单 %d，立即执行", ticket.id) asyncio.create_task(_execute_scheduled_ticket(ticket.id)) else: await schedule_ticket_execution(ticket.id, scheduled) count += 1 if count > 0: logger.info("恢复了 %d 个定时执行工单", count) return count def get_active_scheduled_count() -> int: """返回当前 APScheduler 中的定时任务数量。""" scheduler = _get_scheduler() if not scheduler.running: return 0 jobs = scheduler.get_jobs() return len([j for j in jobs if j.id.startswith("ticket_")]) def get_active_scheduled_ids() -> list[int]: """返回当前定时工单 ID 列表。""" scheduler = _get_scheduler() if not scheduler.running: return [] jobs = scheduler.get_jobs() ids = [] for j in jobs: if j.id.startswith("ticket_"): try: ids.append(int(j.id.split("_")[1])) except (ValueError, IndexError): pass return ids