"""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)