Neo-ZQYY/apps/backend/app/routers/trigger_jobs.py

# -*- coding: utf-8 -*-
"""定时任务管理 API

提供 3 个端点：
- GET   /api/trigger-jobs              — 列出所有定时任务
- POST  /api/trigger-jobs/{id}/run     — 手动执行指定任务
- PATCH /api/trigger-jobs/{id}/config  — 编辑触发器配置

所有端点需要 JWT 认证（系统管理后台使用）。
"""

from __future__ import annotations

import json
import logging

from fastapi import APIRouter, Depends, HTTPException, status

from app.auth.dependencies import CurrentUser, get_current_user
from app.database import get_connection
from app.schemas.trigger_jobs import TriggerJobItem, RunJobResult, UpdateTriggerConfigRequest
from app.services.trigger_scheduler import list_trigger_jobs, run_job_by_id, _calculate_next_run
from app.utils.cron_validator import validate_cron_expression

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/trigger-jobs", tags=["定时任务"])


@router.get("", response_model=list[TriggerJobItem])
async def get_trigger_jobs(
    user: CurrentUser = Depends(get_current_user),
) -> list[TriggerJobItem]:
    """返回所有定时任务列表。"""
    try:
        jobs = list_trigger_jobs()
        return [TriggerJobItem(**j) for j in jobs]
    except Exception as exc:
        logger.exception("获取定时任务列表失败")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"获取定时任务列表失败: {str(exc)[:200]}",
        )


@router.post("/{job_id}/run", response_model=RunJobResult)
async def run_trigger_job(
    job_id: int,
    user: CurrentUser = Depends(get_current_user),
) -> RunJobResult:
    """手动执行指定定时任务。"""
    try:
        result = run_job_by_id(job_id)
        return RunJobResult(**result)
    except Exception as exc:
        logger.exception("手动执行任务 %s 失败", job_id)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"执行失败: {str(exc)[:200]}",
        )


@router.patch("/{job_id}/config", response_model=TriggerJobItem)
async def update_trigger_config(
    job_id: int,
    body: UpdateTriggerConfigRequest,
    user: CurrentUser = Depends(get_current_user),
) -> TriggerJobItem:
    """编辑触发器的 cron_expression 或 interval_seconds。

    仅 merge 请求中非 None 的字段到 trigger_config JSONB，
    不覆盖其他已有字段。更新后重新计算 next_run_at。
    """
    # --- 校验 cron_expression 格式 ---
    if body.cron_expression is not None and not validate_cron_expression(body.cron_expression):
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail="cron 表达式格式无效，需要 5 字段格式",
        )

    # --- 校验 interval_seconds >= 1 ---
    if body.interval_seconds is not None and body.interval_seconds < 1:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail="interval_seconds 必须 >= 1",
        )

    conn = get_connection()
    try:
        with conn.cursor() as cur:
            # 查询 trigger_job 是否存在，同时获取当前 trigger_condition 和 trigger_config
            cur.execute(
                "SELECT trigger_condition, trigger_config FROM biz.trigger_jobs WHERE id = %s",
                (job_id,),
            )
            row = cur.fetchone()
            if row is None:
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail=f"任务 {job_id} 不存在",
                )

            trigger_condition, current_config = row
            current_config = current_config or {}

            # 构建 config_updates：仅包含非 None 字段
            config_updates: dict = {}
            if body.cron_expression is not None:
                config_updates["cron_expression"] = body.cron_expression
            if body.interval_seconds is not None:
                config_updates["interval_seconds"] = body.interval_seconds

            # 合并后的 trigger_config 用于计算 next_run_at
            merged_config = {**current_config, **config_updates}
            next_run_at = _calculate_next_run(trigger_condition, merged_config)

            # 使用 || 操作符 merge JSONB，避免覆盖其他字段
            cur.execute(
                """
                UPDATE biz.trigger_jobs
                SET trigger_config = trigger_config || %s::jsonb,
                    next_run_at = %s
                WHERE id = %s
                RETURNING id, job_type, job_name, trigger_condition, trigger_config,
                          last_run_at, next_run_at, status, description, last_error, created_at
                """,
                (json.dumps(config_updates), next_run_at, job_id),
            )
            updated = cur.fetchone()
            conn.commit()

        col_names = [
            "id", "job_type", "job_name", "trigger_condition", "trigger_config",
            "last_run_at", "next_run_at", "status", "description", "last_error", "created_at",
        ]
        result = dict(zip(col_names, updated))
        # 日期时间字段转字符串（psycopg2 返回 datetime 对象）
        for dt_field in ("last_run_at", "next_run_at", "created_at"):
            if result[dt_field] is not None:
                result[dt_field] = str(result[dt_field])

        return TriggerJobItem(**result)

    except HTTPException:
        raise
    except Exception as exc:
        conn.rollback()
        logger.exception("更新任务 %s 触发器配置失败", job_id)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"更新触发器配置失败: {str(exc)[:200]}",
        )
    finally:
        conn.close()