vent_agent_2.0/api/schedule_routes.py

"""定时任务路由（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)}