ai-agent-admin/backend-fastapi/scheduler/api.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Scheduler API - 定时任务管理接口
提供定时任务的 CRUD 操作和管理功能
"""
from datetime import datetime, timedelta
from typing import List, Optional

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import select, func, or_
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_db
from app.config import settings
from app.base_schema import PaginatedResponse, ResponseModel
from scheduler.model import SchedulerJob, SchedulerLog
from scheduler.schema import (
    SchedulerJobCreate,
    SchedulerJobUpdate,
    SchedulerJobResponse,
    SchedulerJobSimple,
    SchedulerJobBatchDeleteIn,
    SchedulerJobBatchDeleteOut,
    SchedulerJobBatchUpdateStatusIn,
    SchedulerJobBatchUpdateStatusOut,
    SchedulerJobExecuteIn,
    SchedulerJobExecuteOut,
    SchedulerJobStatisticsOut,
    SchedulerJobSearchRequest,
    SchedulerLogResponse,
    SchedulerLogBatchDeleteIn,
    SchedulerLogBatchDeleteOut,
    SchedulerLogCleanIn,
    SchedulerLogCleanOut,
    SchedulerStatusOut,
)
from scheduler.service import scheduler_service

router = APIRouter(prefix="/scheduler", tags=["定时任务管理"])


def _build_job_response(job: SchedulerJob) -> SchedulerJobResponse:
    """构建任务响应"""
    return SchedulerJobResponse(
        id=job.id,
        application_id=job.application_id,
        name=job.name,
        code=job.code,
        description=job.description,
        group=job.group,
        trigger_type=job.trigger_type,
        trigger_type_display=job.get_trigger_type_display(),
        cron_expression=job.cron_expression,
        interval_seconds=job.interval_seconds,
        run_date=job.run_date,
        task_func=job.task_func,
        task_args=job.task_args,
        task_kwargs=job.task_kwargs,
        status=job.status,
        status_display=job.get_status_display(),
        priority=job.priority,
        max_instances=job.max_instances,
        max_retries=job.max_retries,
        timeout=job.timeout,
        coalesce=job.coalesce,
        allow_concurrent=job.allow_concurrent,
        total_run_count=job.total_run_count,
        success_count=job.success_count,
        failure_count=job.failure_count,
        success_rate=job.get_success_rate(),
        last_run_time=job.last_run_time,
        next_run_time=job.next_run_time,
        last_run_status=job.last_run_status,
        last_run_result=job.last_run_result,
        remark=job.remark,
        sort=job.sort,
        sys_create_datetime=job.sys_create_datetime,
        sys_update_datetime=job.sys_update_datetime,
    )


def _build_log_response(log: SchedulerLog) -> SchedulerLogResponse:
    """构建日志响应"""
    return SchedulerLogResponse(
        id=log.id,
        job_id=log.job_id,
        job_name=log.job_name,
        job_code=log.job_code,
        status=log.status,
        status_display=log.get_status_display(),
        start_time=log.start_time,
        end_time=log.end_time,
        duration=log.duration,
        result=log.result,
        exception=log.exception,
        traceback=log.traceback,
        hostname=log.hostname,
        process_id=log.process_id,
        retry_count=log.retry_count,
        sys_create_datetime=log.sys_create_datetime,
    )


# ==================== SchedulerJob APIs ====================

@router.post("/job", response_model=SchedulerJobResponse, summary="创建定时任务")
async def create_scheduler_job(data: SchedulerJobCreate, db: AsyncSession = Depends(get_db)):
    """创建新的定时任务"""
    # 检查任务编码是否已存在
    result = await db.execute(
        select(SchedulerJob).where(
            SchedulerJob.code == data.code,
            SchedulerJob.is_deleted == False  # noqa: E712
        )
    )
    if result.scalar_one_or_none():
        raise HTTPException(status_code=400, detail=f"任务编码已存在: {data.code}")

    # 创建任务
    job = SchedulerJob(**data.model_dump())
    db.add(job)
    await db.commit()
    await db.refresh(job)

    # 如果任务是启用状态，添加到调度器
    if job.is_enabled() and scheduler_service.is_running():
        await scheduler_service.add_job(job)

    return _build_job_response(job)


@router.get("/job/all", response_model=List[SchedulerJobSimple], summary="获取所有定时任务（简化版）")
async def get_all_scheduler_jobs(
    application_id: str = Query(None, alias="applicationId", description="所属应用ID"),
    db: AsyncSession = Depends(get_db),
):
    """获取所有定时任务（不分页，简化版）"""
    conditions = [SchedulerJob.is_deleted == False]  # noqa: E712
    if application_id:
        conditions.append(SchedulerJob.application_id == application_id)
    else:
        conditions.append(SchedulerJob.application_id.is_(None))
    
    result = await db.execute(
        select(SchedulerJob).where(*conditions).order_by(SchedulerJob.priority.desc(), SchedulerJob.name)
    )
    jobs = result.scalars().all()
    return jobs


@router.get("/job", response_model=PaginatedResponse[SchedulerJobResponse], summary="获取定时任务列表")
async def get_scheduler_job_list(
    page: int = Query(default=1, ge=1, description="页码"),
    page_size: int = Query(default=settings.PAGE_SIZE, ge=1, le=settings.PAGE_MAX_SIZE, alias="pageSize", description="每页数量"),
    application_id: str = Query(None, alias="applicationId", description="所属应用ID"),
    name: Optional[str] = Query(default=None, description="任务名称"),
    code: Optional[str] = Query(default=None, description="任务编码"),
    group: Optional[str] = Query(default=None, description="任务分组"),
    trigger_type: Optional[str] = Query(default=None, description="触发器类型"),
    status: Optional[int] = Query(default=None, description="任务状态"),
    db: AsyncSession = Depends(get_db)
):
    """获取定时任务列表（分页）"""
    filters = [SchedulerJob.is_deleted == False]  # noqa: E712
    # 应用过滤
    if application_id:
        filters.append(SchedulerJob.application_id == application_id)
    else:
        filters.append(SchedulerJob.application_id.is_(None))
    if name:
        filters.append(SchedulerJob.name.ilike(f"%{name}%"))
    if code:
        filters.append(SchedulerJob.code.ilike(f"%{code}%"))
    if group:
        filters.append(SchedulerJob.group == group)
    if trigger_type:
        filters.append(SchedulerJob.trigger_type == trigger_type)
    if status is not None:
        filters.append(SchedulerJob.status == status)

    # 查询总数
    count_result = await db.execute(
        select(func.count(SchedulerJob.id)).where(*filters)
    )
    total = count_result.scalar()

    # 查询数据
    offset = (page - 1) * page_size
    result = await db.execute(
        select(SchedulerJob).where(*filters)
        .order_by(SchedulerJob.priority.desc(), SchedulerJob.sys_update_datetime.desc())
        .offset(offset).limit(page_size)
    )
    jobs = result.scalars().all()

    return PaginatedResponse(
        items=[_build_job_response(job) for job in jobs],
        total=total
    )


@router.post("/job/batch/delete", response_model=SchedulerJobBatchDeleteOut, summary="批量删除定时任务")
async def batch_delete_scheduler_jobs(
    data: SchedulerJobBatchDeleteIn,
    db: AsyncSession = Depends(get_db)
):
    """批量删除定时任务"""
    success_count = 0
    failed_ids = []

    for job_id in data.ids:
        try:
            result = await db.execute(
                select(SchedulerJob).where(SchedulerJob.id == job_id)
            )
            job = result.scalar_one_or_none()

            if job:
                # 从调度器移除
                if scheduler_service.is_running():
                    await scheduler_service.remove_job(job.code)

                # 软删除
                job.is_deleted = True
                success_count += 1
            else:
                failed_ids.append(job_id)
        except Exception:
            failed_ids.append(job_id)

    await db.commit()
    return SchedulerJobBatchDeleteOut(count=success_count, failed_ids=failed_ids)


@router.post("/job/batch/update_status", response_model=SchedulerJobBatchUpdateStatusOut, summary="批量更新任务状态")
async def batch_update_scheduler_job_status(
    data: SchedulerJobBatchUpdateStatusIn,
    db: AsyncSession = Depends(get_db)
):
    """批量启用、禁用或暂停任务"""
    result = await db.execute(
        select(SchedulerJob).where(SchedulerJob.id.in_(data.ids))
    )
    jobs = result.scalars().all()

    count = 0
    for job in jobs:
        job.status = data.status
        count += 1

        # 同步更新调度器
        if scheduler_service.is_running():
            if job.is_enabled():
                await scheduler_service.add_job(job)
            elif job.is_paused():
                await scheduler_service.pause_job(job.code)
            else:
                await scheduler_service.remove_job(job.code)

    await db.commit()
    return SchedulerJobBatchUpdateStatusOut(count=count)


@router.post("/job/execute", response_model=SchedulerJobExecuteOut, summary="立即执行任务")
async def execute_scheduler_job(
    data: SchedulerJobExecuteIn,
    db: AsyncSession = Depends(get_db)
):
    """立即执行指定任务（不影响正常调度）"""
    result = await db.execute(
        select(SchedulerJob).where(SchedulerJob.id == data.job_id)
    )
    job = result.scalar_one_or_none()

    if not job:
        raise HTTPException(status_code=404, detail="任务不存在")

    if not scheduler_service.is_running():
        raise HTTPException(status_code=400, detail="调度器未运行")

    # 立即执行任务
    success = await scheduler_service.run_job_now(job.code)

    if success:
        return SchedulerJobExecuteOut(
            success=True,
            message=f"任务 {job.name} 将立即执行"
        )
    else:
        return SchedulerJobExecuteOut(
            success=False,
            message=f"任务 {job.name} 执行失败，可能任务未在调度器中"
        )


@router.post("/job/search", response_model=PaginatedResponse[SchedulerJobResponse], summary="搜索定时任务")
async def search_scheduler_jobs(
    data: SchedulerJobSearchRequest,
    page: int = Query(default=1, ge=1, description="页码"),
    page_size: int = Query(default=settings.PAGE_SIZE, ge=1, le=settings.PAGE_MAX_SIZE, alias="pageSize", description="每页数量"),
    db: AsyncSession = Depends(get_db)
):
    """搜索定时任务"""
    keyword = data.keyword
    filters = [
        SchedulerJob.is_deleted == False,  # noqa: E712
        or_(
            SchedulerJob.name.ilike(f"%{keyword}%"),
            SchedulerJob.code.ilike(f"%{keyword}%"),
            SchedulerJob.description.ilike(f"%{keyword}%"),
        )
    ]

    # 查询总数
    count_result = await db.execute(
        select(func.count(SchedulerJob.id)).where(*filters)
    )
    total = count_result.scalar()

    # 查询数据
    offset = (page - 1) * page_size
    result = await db.execute(
        select(SchedulerJob).where(*filters)
        .order_by(SchedulerJob.priority.desc())
        .offset(offset).limit(page_size)
    )
    jobs = result.scalars().all()

    return PaginatedResponse(
        items=[_build_job_response(job) for job in jobs],
        total=total
    )


@router.get("/job/statistics/data", response_model=SchedulerJobStatisticsOut, summary="获取任务统计信息")
async def get_scheduler_job_statistics(db: AsyncSession = Depends(get_db)):
    """获取任务统计信息"""
    # 任务统计
    total_result = await db.execute(
        select(func.count(SchedulerJob.id)).where(SchedulerJob.is_deleted == False)  # noqa: E712
    )
    total_jobs = total_result.scalar() or 0

    enabled_result = await db.execute(
        select(func.count(SchedulerJob.id)).where(
            SchedulerJob.is_deleted == False,  # noqa: E712
            SchedulerJob.status == 1
        )
    )
    enabled_jobs = enabled_result.scalar() or 0

    disabled_result = await db.execute(
        select(func.count(SchedulerJob.id)).where(
            SchedulerJob.is_deleted == False,  # noqa: E712
            SchedulerJob.status == 0
        )
    )
    disabled_jobs = disabled_result.scalar() or 0

    paused_result = await db.execute(
        select(func.count(SchedulerJob.id)).where(
            SchedulerJob.is_deleted == False,  # noqa: E712
            SchedulerJob.status == 2
        )
    )
    paused_jobs = paused_result.scalar() or 0

    # 执行统计
    total_exec_result = await db.execute(
        select(func.count(SchedulerLog.id))
    )
    total_executions = total_exec_result.scalar() or 0

    success_exec_result = await db.execute(
        select(func.count(SchedulerLog.id)).where(SchedulerLog.status == 'success')
    )
    success_executions = success_exec_result.scalar() or 0

    failed_exec_result = await db.execute(
        select(func.count(SchedulerLog.id)).where(SchedulerLog.status == 'failed')
    )
    failed_executions = failed_exec_result.scalar() or 0

    # 计算成功率
    success_rate = round(success_executions / total_executions * 100, 2) if total_executions > 0 else 0

    return SchedulerJobStatisticsOut(
        total_jobs=total_jobs,
        enabled_jobs=enabled_jobs,
        disabled_jobs=disabled_jobs,
        paused_jobs=paused_jobs,
        total_executions=total_executions,
        success_executions=success_executions,
        failed_executions=failed_executions,
        success_rate=success_rate,
    )


@router.get("/job/{job_id}", response_model=SchedulerJobResponse, summary="获取定时任务详情")
async def get_scheduler_job(job_id: str, db: AsyncSession = Depends(get_db)):
    """获取单个定时任务的详细信息"""
    result = await db.execute(
        select(SchedulerJob).where(SchedulerJob.id == job_id)
    )
    job = result.scalar_one_or_none()

    if not job:
        raise HTTPException(status_code=404, detail="任务不存在")

    return _build_job_response(job)


@router.put("/job/{job_id}", response_model=SchedulerJobResponse, summary="更新定时任务")
async def update_scheduler_job(
    job_id: str,
    data: SchedulerJobUpdate,
    db: AsyncSession = Depends(get_db)
):
    """更新定时任务"""
    result = await db.execute(
        select(SchedulerJob).where(SchedulerJob.id == job_id)
    )
    job = result.scalar_one_or_none()

    if not job:
        raise HTTPException(status_code=404, detail="任务不存在")

    # 检查任务编码是否已存在（排除自身）
    if data.code:
        code_result = await db.execute(
            select(SchedulerJob).where(
                SchedulerJob.code == data.code,
                SchedulerJob.id != job_id,
                SchedulerJob.is_deleted == False  # noqa: E712
            )
        )
        if code_result.scalar_one_or_none():
            raise HTTPException(status_code=400, detail=f"任务编码已存在: {data.code}")

    # 更新字段
    update_data = data.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(job, key, value)

    await db.commit()
    await db.refresh(job)

    # 同步更新调度器
    if scheduler_service.is_running():
        await scheduler_service.modify_job(job)

    return _build_job_response(job)


@router.delete("/job/{job_id}", response_model=ResponseModel, summary="删除定时任务")
async def delete_scheduler_job(
    job_id: str,
    hard: bool = Query(default=False, description="是否物理删除"),
    db: AsyncSession = Depends(get_db)
):
    """删除定时任务"""
    result = await db.execute(
        select(SchedulerJob).where(SchedulerJob.id == job_id)
    )
    job = result.scalar_one_or_none()

    if not job:
        raise HTTPException(status_code=404, detail="任务不存在")

    # 从调度器移除
    if scheduler_service.is_running():
        await scheduler_service.remove_job(job.code)

    if hard:
        await db.delete(job)
    else:
        job.is_deleted = True

    await db.commit()
    return ResponseModel(message="删除成功")


# ==================== SchedulerLog APIs ====================

@router.get("/log", response_model=PaginatedResponse[SchedulerLogResponse], summary="获取任务执行日志列表")
async def get_scheduler_log_list(
    page: int = Query(default=1, ge=1, description="页码"),
    page_size: int = Query(default=settings.PAGE_SIZE, ge=1, le=settings.PAGE_MAX_SIZE, alias="pageSize", description="每页数量"),
    job_id: Optional[str] = Query(default=None, description="任务ID"),
    job_code: Optional[str] = Query(default=None, description="任务编码"),
    job_name: Optional[str] = Query(default=None, description="任务名称"),
    status: Optional[str] = Query(default=None, description="执行状态"),
    start_time_gte: Optional[datetime] = Query(default=None, alias="startTimeGte", description="开始时间>="),
    start_time_lte: Optional[datetime] = Query(default=None, alias="startTimeLte", description="开始时间<="),
    db: AsyncSession = Depends(get_db)
):
    """获取任务执行日志列表（分页）"""
    filters = []
    if job_id:
        filters.append(SchedulerLog.job_id == job_id)
    if job_code:
        filters.append(SchedulerLog.job_code.ilike(f"%{job_code}%"))
    if job_name:
        filters.append(SchedulerLog.job_name.ilike(f"%{job_name}%"))
    if status:
        filters.append(SchedulerLog.status == status)
    if start_time_gte:
        filters.append(SchedulerLog.start_time >= start_time_gte)
    if start_time_lte:
        filters.append(SchedulerLog.start_time <= start_time_lte)

    # 查询总数
    count_result = await db.execute(
        select(func.count(SchedulerLog.id)).where(*filters) if filters else select(func.count(SchedulerLog.id))
    )
    total = count_result.scalar()

    # 查询数据
    offset = (page - 1) * page_size
    query = select(SchedulerLog).order_by(SchedulerLog.start_time.desc()).offset(offset).limit(page_size)
    if filters:
        query = select(SchedulerLog).where(*filters).order_by(SchedulerLog.start_time.desc()).offset(offset).limit(page_size)
    
    result = await db.execute(query)
    logs = result.scalars().all()

    return PaginatedResponse(
        items=[_build_log_response(log) for log in logs],
        total=total
    )


@router.get("/log/by/job/{job_id}", response_model=PaginatedResponse[SchedulerLogResponse], summary="获取指定任务的执行日志")
async def get_scheduler_logs_by_job(
    job_id: str,
    page: int = Query(default=1, ge=1, description="页码"),
    page_size: int = Query(default=settings.PAGE_SIZE, ge=1, le=settings.PAGE_MAX_SIZE, alias="pageSize", description="每页数量"),
    db: AsyncSession = Depends(get_db)
):
    """获取指定任务的所有执行日志"""
    # 查询总数
    count_result = await db.execute(
        select(func.count(SchedulerLog.id)).where(SchedulerLog.job_id == job_id)
    )
    total = count_result.scalar()

    # 查询数据
    offset = (page - 1) * page_size
    result = await db.execute(
        select(SchedulerLog).where(SchedulerLog.job_id == job_id)
        .order_by(SchedulerLog.start_time.desc())
        .offset(offset).limit(page_size)
    )
    logs = result.scalars().all()

    return PaginatedResponse(
        items=[_build_log_response(log) for log in logs],
        total=total
    )


@router.get("/log/{log_id}", response_model=SchedulerLogResponse, summary="获取任务执行日志详情")
async def get_scheduler_log(log_id: str, db: AsyncSession = Depends(get_db)):
    """获取单个任务执行日志的详细信息"""
    result = await db.execute(
        select(SchedulerLog).where(SchedulerLog.id == log_id)
    )
    log = result.scalar_one_or_none()

    if not log:
        raise HTTPException(status_code=404, detail="日志不存在")

    return _build_log_response(log)


@router.delete("/log/{log_id}", response_model=ResponseModel, summary="删除任务执行日志")
async def delete_scheduler_log(log_id: str, db: AsyncSession = Depends(get_db)):
    """删除任务执行日志"""
    result = await db.execute(
        select(SchedulerLog).where(SchedulerLog.id == log_id)
    )
    log = result.scalar_one_or_none()

    if not log:
        raise HTTPException(status_code=404, detail="日志不存在")

    await db.delete(log)
    await db.commit()
    return ResponseModel(message="删除成功")


@router.post("/log/batch/delete", response_model=SchedulerLogBatchDeleteOut, summary="批量删除任务执行日志")
async def batch_delete_scheduler_logs(
    data: SchedulerLogBatchDeleteIn,
    db: AsyncSession = Depends(get_db)
):
    """批量删除任务执行日志"""
    result = await db.execute(
        select(SchedulerLog).where(SchedulerLog.id.in_(data.ids))
    )
    logs = result.scalars().all()

    count = len(logs)
    for log in logs:
        await db.delete(log)

    await db.commit()
    return SchedulerLogBatchDeleteOut(count=count)


@router.post("/log/clean", response_model=SchedulerLogCleanOut, summary="清理旧日志")
async def clean_scheduler_logs(
    data: SchedulerLogCleanIn,
    db: AsyncSession = Depends(get_db)
):
    """清理旧日志"""
    cutoff_date = datetime.now() - timedelta(days=data.days)

    filters = [SchedulerLog.start_time < cutoff_date]
    if data.status:
        filters.append(SchedulerLog.status == data.status)

    result = await db.execute(
        select(SchedulerLog).where(*filters)
    )
    logs = result.scalars().all()

    count = len(logs)
    for log in logs:
        await db.delete(log)

    await db.commit()
    return SchedulerLogCleanOut(count=count)


# ==================== Scheduler Control APIs ====================

@router.post("/start", response_model=ResponseModel, summary="启动调度器")
async def start_scheduler():
    """启动调度器（注意：调度器应在应用启动时自动启动）"""
    if scheduler_service.is_running():
        raise HTTPException(status_code=400, detail="调度器已在运行中")

    # APScheduler 4.x 中调度器应在 lifespan 中启动
    raise HTTPException(status_code=400, detail="请重启应用以启动调度器")


@router.post("/shutdown", response_model=ResponseModel, summary="关闭调度器")
async def shutdown_scheduler():
    """关闭调度器（注意：调度器会在应用关闭时自动关闭）"""
    if not scheduler_service.is_running():
        raise HTTPException(status_code=400, detail="调度器未运行")

    # APScheduler 4.x 中调度器应在 lifespan 中关闭
    raise HTTPException(status_code=400, detail="请关闭应用以停止调度器")


@router.post("/pause", response_model=ResponseModel, summary="暂停调度器")
async def pause_scheduler():
    """暂停调度器（暂不支持）"""
    raise HTTPException(status_code=400, detail="APScheduler 4.x 暂不支持暂停整个调度器")


@router.post("/resume", response_model=ResponseModel, summary="恢复调度器")
async def resume_scheduler():
    """恢复调度器（暂不支持）"""
    raise HTTPException(status_code=400, detail="APScheduler 4.x 暂不支持恢复整个调度器")


@router.get("/status", response_model=SchedulerStatusOut, summary="获取调度器状态")
async def get_scheduler_status():
    """获取调度器状态"""
    is_running = scheduler_service.is_running()
    jobs = await scheduler_service.get_all_jobs() if is_running else []

    return SchedulerStatusOut(
        is_running=is_running,
        job_count=len(jobs),
        jobs=jobs,
    )


@router.get("/log/{log_id}/stream", summary="实时日志流")
async def stream_task_log(log_id: str):
    """
    通过 SSE 实时推送任务执行日志
    
    Args:
        log_id: 日志记录 ID
        
    Returns:
        SSE 事件流，包含任务执行过程中的日志消息
    """
    from fastapi.responses import StreamingResponse
    from scheduler.task_log_service import TaskLogService
    import json
    
    async def event_generator():
        try:
            async for log_entry in TaskLogService.subscribe(log_id):
                yield f"data: {json.dumps(log_entry, ensure_ascii=False)}\n\n"
        except Exception as e:
            yield f"data: {json.dumps({'level': 'error', 'message': str(e)}, ensure_ascii=False)}\n\n"
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        }
    )