hive-manage-v3/backend/app/api/cron.py

"""Cron API — 跨 Profile Cron 聚合列表、日志、启停操作。"""
import json
import logging
import subprocess
from datetime import datetime, timezone

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

from app.config import settings
from app.database import get_db
from app.middleware.auth import require_auth, require_role
from app.models.audit import AuditLog
from app.models.user import User

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/cron", tags=["cron"])

# ── 系统 Cron 硬编码清单 ──

SYSTEM_CRON_JOBS = [
    {
        "id": "system-hindsight",
        "job_name": "hindsight",
        "profile": "system",
        "schedule": "0 21 * * *",
        "status": "active",
        "last_run": None,
        "last_run_success": None,
        "success_rate": 100.0,
        "has_logs": False,
        "system": True,
        "consecutive_failures": 0,
    },
    {
        "id": "system-health-check",
        "job_name": "健康巡检",
        "profile": "system",
        "schedule": "*/30 * * * *",
        "status": "active",
        "last_run": None,
        "last_run_success": None,
        "success_rate": 100.0,
        "has_logs": False,
        "system": True,
        "consecutive_failures": 0,
    },
    {
        "id": "system-skill-check",
        "job_name": "Skill检查",
        "profile": "system",
        "schedule": "0 2 * * *",
        "status": "active",
        "last_run": None,
        "last_run_success": None,
        "success_rate": 100.0,
        "has_logs": False,
        "system": True,
        "consecutive_failures": 0,
    },
    {
        "id": "system-memory-archive",
        "job_name": "Memory归档",
        "profile": "system",
        "schedule": "0 3 * * 0",
        "status": "paused",
        "last_run": None,
        "last_run_success": None,
        "success_rate": 100.0,
        "has_logs": False,
        "system": True,
        "consecutive_failures": 0,
    },
]

# ── 占位 Profile Cron（hermes CLI 返回为空时的后备） ──

FALLBACK_PROFILE_JOBS = [
    {
        "id": "profile-taiwan-briefing",
        "job_name": "taiwan-briefing",
        "profile": "taiwan",
        "schedule": "0 7 * * *",
        "status": "active",
        "last_run": "2026-07-22T07:15:00",
        "last_run_success": True,
        "success_rate": 95.0,
        "has_logs": True,
        "system": False,
        "consecutive_failures": 0,
    },
    {
        "id": "profile-guangxi-donation",
        "job_name": "guangxi-donation",
        "profile": "guangxi",
        "schedule": "0 9 * * *",
        "status": "paused",
        "last_run": "2026-07-21T09:00:00",
        "last_run_success": True,
        "success_rate": 100.0,
        "has_logs": True,
        "system": False,
        "consecutive_failures": 0,
    },
    {
        "id": "profile-daily-report",
        "job_name": "daily-report",
        "profile": "ops-engineer",
        "schedule": "0 8 * * *",
        "status": "active",
        "last_run": "2026-07-22T08:00:00",
        "last_run_success": True,
        "success_rate": 88.0,
        "has_logs": True,
        "system": False,
        "consecutive_failures": 0,
    },
    {
        "id": "profile-weekly-digest",
        "job_name": "weekly-digest",
        "profile": "ops-engineer",
        "schedule": "30 9 * * 1",
        "status": "active",
        "last_run": "2026-07-20T09:30:00",
        "last_run_success": True,
        "success_rate": 100.0,
        "has_logs": True,
        "system": False,
        "consecutive_failures": 0,
    },
    {
        "id": "profile-hindsight",
        "job_name": "hindsight",
        "profile": "hebo",
        "schedule": "0 21 * * *",
        "status": "active",
        "last_run": "2026-07-21T21:13:00",
        "last_run_success": False,
        "success_rate": 60.0,
        "has_logs": True,
        "system": False,
        "consecutive_failures": 2,
    },
]

# ── 占位日志数据 ──

PLACEHOLDER_LOGS = {
    "system-hindsight": [
        {"time": "2026-07-21T21:13:00", "status": "failure", "message": "ERROR: 连接数据库超时 (timeout: 30s)"},
        {"time": "2026-07-20T21:05:00", "status": "failure", "message": "WARN: 响应慢 (3200ms), 重试后仍然失败"},
        {"time": "2026-07-19T21:01:00", "status": "success", "message": "INFO: 归档完成, 处理 128 条记录"},
    ],
    "profile-hindsight": [
        {"time": "2026-07-21T21:13:00", "status": "failure", "message": "ERROR: API 返回 502 Bad Gateway"},
        {"time": "2026-07-20T21:05:00", "status": "failure", "message": "ERROR: 上游服务未响应 (timeout: 60s)"},
        {"time": "2026-07-19T21:01:00", "status": "success", "message": "INFO: 推送完成, 成功发送 12 条消息"},
        {"time": "2026-07-18T21:00:00", "status": "success", "message": "INFO: 按计划执行完毕"},
        {"time": "2026-07-17T21:00:00", "status": "success", "message": "INFO: 正常执行"},
    ],
}


def _fetch_profile_jobs() -> list[dict]:
    """从 hermes CLI 获取 Profile Cron 列表，失败时返回后备数据。"""
    try:
        proc = subprocess.run(
            [settings.hermes_cli_path, "cron", "list", "--all"],
            capture_output=True, text=True, timeout=15,
        )
        if proc.returncode == 0:
            data = json.loads(proc.stdout)
            jobs = data if isinstance(data, list) else data.get("jobs", [])
            for j in jobs:
                j.setdefault("system", False)
                j.setdefault("consecutive_failures", 0)
                j.setdefault("has_logs", False)
                j.setdefault("success_rate", 100.0)
            return jobs
    except (FileNotFoundError, json.JSONDecodeError, subprocess.TimeoutExpired) as e:
        logger.warning("hermes cron list --all 失败: %s", e)
    except Exception as e:
        logger.warning("调用 hermes cron list 异常: %s", e)
    return FALLBACK_PROFILE_JOBS


def _merge_jobs(profile_jobs: list[dict]) -> list[dict]:
    """合并 Profile Cron 和系统 Cron，按 id 去重。"""
    seen = {j["id"] for j in SYSTEM_CRON_JOBS}
    merged = list(SYSTEM_CRON_JOBS)
    for j in profile_jobs:
        if j["id"] not in seen:
            merged.append(j)
            seen.add(j["id"])
    return merged


def _check_has_failures(jobs: list[dict]) -> bool:
    """检查当日是否有任何 Cron 失败。"""
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    for j in jobs:
        last_run = j.get("last_run") or ""
        if last_run.startswith(today) and j.get("last_run_success") is False:
            return True
    return False


@router.get("/jobs")
async def list_cron_jobs(
    status: str | None = Query(None, pattern="^(active|paused)$"),
    profile: str = Query("all", pattern="^(all|system|[a-z][a-z0-9_-]+)$"),
    _user: User = Depends(require_auth),
):
    """跨 Profile Cron 聚合列表。

    - ?status=active|paused  按状态筛选
    - ?profile=all|system|xxx  按来源/Profile 筛选
    """
    profile_jobs = _fetch_profile_jobs()
    jobs = _merge_jobs(profile_jobs)

    if status:
        jobs = [j for j in jobs if j.get("status") == status]
    if profile == "system":
        jobs = [j for j in jobs if j.get("system")]
    elif profile != "all":
        jobs = [j for j in jobs if j.get("profile") == profile and not j.get("system")]

    has_failures = _check_has_failures(jobs)
    return {"jobs": jobs, "has_failures": has_failures}


@router.get("/jobs/{job_id}/logs")
async def get_cron_logs(
    job_id: str,
    _user: User = Depends(require_auth),
):
    """返回最近 8 条运行日志（暂用占位数据）。"""
    logs = PLACEHOLDER_LOGS.get(job_id, [])
    return {"logs": logs[:8]}


async def _operate_cron(
    job_id: str,
    action: str,
    user: User,
    db: AsyncSession,
) -> dict:
    """执行 Cron 操作并写入审计日志。"""
    # 解析 profile 和 job_name
    if job_id.startswith("system-"):
        profile = "system"
        job_name = job_id[len("system-"):]
    elif job_id.startswith("profile-"):
        parts = job_id.split("-", 2)
        if len(parts) >= 3:
            profile = parts[1]
            job_name = parts[2]
        else:
            profile = "unknown"
            job_name = job_id
    else:
        profile = "unknown"
        job_name = job_id

    try:
        # run 操作需要额外的 --job 参数
        if action == "run":
            args = [settings.hermes_cli_path, "cron", "run", "--profile", profile, "--job", job_name]
        elif action == "pause":
            args = [settings.hermes_cli_path, "cron", "pause", "--profile", profile, "--job", job_name]
        elif action == "resume":
            args = [settings.hermes_cli_path, "cron", "resume", "--profile", profile, "--job", job_name]
        else:
            raise HTTPException(status_code=400, detail=f"不支持的操作: {action}")

        proc = subprocess.run(args, capture_output=True, text=True, timeout=30)
        success = proc.returncode == 0
        detail = (proc.stdout or proc.stderr).strip()
    except HTTPException:
        raise
    except Exception as e:
        success = False
        detail = str(e)

    audit = AuditLog(
        user_id=user.id,
        username=user.username,
        action=f"cron:{action}",
        target=f"job:{job_id}",
        detail=detail,
    )
    db.add(audit)
    await db.commit()

    if not success:
        raise HTTPException(status_code=500, detail=f"Cron {action} 失败: {detail}")

    return {"success": True, "action": action, "job_id": job_id, "detail": detail}


@router.post("/jobs/{job_id}/run")
async def run_cron_job(
    job_id: str,
    user: User = Depends(require_role("owner", "operator")),
    db: AsyncSession = Depends(get_db),
):
    """手动触发一次 Cron 执行。"""
    return await _operate_cron(job_id, "run", user, db)


@router.post("/jobs/{job_id}/pause")
async def pause_cron_job(
    job_id: str,
    user: User = Depends(require_role("owner", "operator")),
    db: AsyncSession = Depends(get_db),
):
    """暂停 Cron。"""
    return await _operate_cron(job_id, "pause", user, db)


@router.post("/jobs/{job_id}/resume")
async def resume_cron_job(
    job_id: str,
    user: User = Depends(require_role("owner", "operator")),
    db: AsyncSession = Depends(get_db),
):
    """恢复 Cron。"""
    return await _operate_cron(job_id, "resume", user, db)