"""定时任务路由(CRUD + 手动触发 + 执行记录)。 写操作的纪律:**先落 DB,再同步调度器 job**。反过来的话,DB 写失败会留下 一个「内存里有 job、DB 里没有」的幽灵任务,重启就消失——排查起来很费劲。 """ from __future__ import annotations import asyncio from fastapi import APIRouter, HTTPException from api.models import ScheduleCreateRequest, ScheduleUpdateRequest from db import schedule_store from obs import get_logger from scheduler import scheduler as sched logger = get_logger(__name__) router = APIRouter(tags=["定时任务"]) def _validate_cron(expr: str) -> str: """校验 cron 表达式,非法直接 400(不要让脏数据进 DB)。""" value = (expr or "").strip() try: sched._parse_cron(value) except Exception as exc: raise HTTPException( status_code=400, detail={ "code": "invalid_cron", "message": f"cron 表达式非法(需 5 段:分 时 日 月 周):{value!r} — {exc}", }, ) from exc return value def _require_schedule(schedule_id: str): schedule = schedule_store.get_schedule(schedule_id) if schedule is None: raise HTTPException( status_code=404, detail={"code": "schedule_not_found", "message": "定时任务不存在"} ) return schedule @router.get("/schedules") async def list_schedules(user_name: str | None = None): """列出定时任务,附带最近一次执行状态。""" items = await asyncio.to_thread(schedule_store.list_schedules, user_name, False, 200) for item in items: latest = await asyncio.to_thread(schedule_store.latest_run, item["id"]) item["latest_run"] = latest return {"items": items, "count": len(items), "scheduler_running": sched.is_running()} @router.post("/schedules") async def create_schedule(payload: ScheduleCreateRequest): """新建定时任务。""" cron = _validate_cron(payload.cron) if not (payload.prompt or "").strip(): raise HTTPException( status_code=400, detail={"code": "empty_prompt", "message": "prompt 不能为空"} ) user = (payload.user_name or "default").strip() or "default" # 1) 先落 DB(事实源) schedule = await asyncio.to_thread( schedule_store.create_schedule, user, payload.title, cron, payload.prompt, payload.mode, payload.enabled, ) # 2) 再投影到调度器 if schedule.get("enabled"): try: await asyncio.to_thread(sched.add_job, schedule) except Exception as exc: # DB 已写入但 job 装载失败:任务仍在,只是这轮没生效。 # 不回滚 DB——用户重试或重启服务后会自动补上。 logger.error("任务已入库但装载失败 id=%s:%s", schedule["id"], exc) schedule["load_error"] = str(exc) return schedule @router.get("/schedules/{schedule_id}") async def get_schedule(schedule_id: str): """取任务详情与执行记录。""" schedule = await asyncio.to_thread(_require_schedule, schedule_id) runs = await asyncio.to_thread(schedule_store.list_runs, schedule_id, 20) return {**schedule, "runs": runs} @router.put("/schedules/{schedule_id}") async def update_schedule(schedule_id: str, payload: ScheduleUpdateRequest): """更新任务。传了 cron 就先校验;传了 enabled 就同步 job。""" await asyncio.to_thread(_require_schedule, schedule_id) fields: dict = {} if payload.cron is not None: fields["cron"] = _validate_cron(payload.cron) if payload.title is not None: fields["title"] = payload.title if payload.prompt is not None: fields["prompt"] = payload.prompt if payload.mode is not None: fields["mode"] = payload.mode if payload.enabled is not None: fields["enabled"] = payload.enabled if not fields: raise HTTPException( status_code=400, detail={"code": "no_fields", "message": "没有需要更新的字段"} ) schedule = await asyncio.to_thread(schedule_store.update_schedule, schedule_id, **fields) if schedule is None: raise HTTPException( status_code=404, detail={"code": "schedule_not_found", "message": "定时任务不存在"} ) await asyncio.to_thread(sched.sync_job, schedule) return schedule @router.delete("/schedules/{schedule_id}") async def delete_schedule(schedule_id: str): """删除任务及其执行记录。""" await asyncio.to_thread(_require_schedule, schedule_id) await asyncio.to_thread(sched.remove_job, schedule_id) deleted = await asyncio.to_thread(schedule_store.delete_schedule, schedule_id) return {"deleted": deleted, "schedule_id": schedule_id} @router.post("/schedules/{schedule_id}/pause") async def pause_schedule(schedule_id: str): """暂停任务(保留定义,摘掉 job)。""" await asyncio.to_thread(_require_schedule, schedule_id) schedule = await asyncio.to_thread(schedule_store.set_enabled, schedule_id, False) await asyncio.to_thread(sched.remove_job, schedule_id) return {"schedule_id": schedule_id, "enabled": False} @router.post("/schedules/{schedule_id}/resume") async def resume_schedule(schedule_id: str): """恢复任务。""" await asyncio.to_thread(_require_schedule, schedule_id) schedule = await asyncio.to_thread(schedule_store.set_enabled, schedule_id, True) if schedule is not None: # cron 可能在这期间被改坏,装载失败要如实报出来 try: await asyncio.to_thread(sched.add_job, schedule) except Exception as exc: raise HTTPException( status_code=400, detail={"code": "invalid_cron", "message": f"任务已启用但装载失败:{exc}"}, ) from exc return {"schedule_id": schedule_id, "enabled": True} @router.post("/schedules/{schedule_id}/run") async def run_now(schedule_id: str): """立即执行一次(不影响后续 cron 计划)。 同步等待执行完成——手动触发是用户主动点的,他想立刻看到结果。 不放进后台是因为「点了没反应」比「等一会儿」更糟。 """ from scheduler.runner import run_schedule await asyncio.to_thread(_require_schedule, schedule_id) result = await run_schedule(schedule_id, trigger="manual") if not result.get("ok"): raise HTTPException( status_code=500, detail={"code": "run_failed", "message": result.get("error") or "执行失败"}, ) return result @router.get("/schedules/{schedule_id}/runs") async def list_runs(schedule_id: str, limit: int = 20): """任务的执行记录。""" await asyncio.to_thread(_require_schedule, schedule_id) items = await asyncio.to_thread(schedule_store.list_runs, schedule_id, min(limit, 100)) return {"schedule_id": schedule_id, "items": items, "count": len(items)}