From 670d2226650ebda8a8addc13b8a25280b28aebcb Mon Sep 17 00:00:00 2001 From: enzo Date: Wed, 22 Jul 2026 17:19:11 +0800 Subject: [PATCH] feat: Cron management API and frontend page (Phase 5) Backend: CRON API with list/logs/run/pause/resume endpoints, Hermes CLI integration, system cron jobs, audit logging for operations. Frontend: CronPage with filter bar (status/profile), table with 30s polling, log modal with level-colored messages, action buttons (run/pause/resume), run confirmation dialog, 4-state coverage. Sidebar shows cron failure badge. Co-Authored-By: Claude --- backend/app/api/cron.py | 330 +++++++++++++++ backend/app/main.py | 3 +- frontend/src/api/cron.ts | 52 +++ frontend/src/components/layout/Sidebar.tsx | 20 +- frontend/src/pages/CronPage.tsx | 466 ++++++++++++++++++++- 5 files changed, 866 insertions(+), 5 deletions(-) create mode 100644 backend/app/api/cron.py create mode 100644 frontend/src/api/cron.ts diff --git a/backend/app/api/cron.py b/backend/app/api/cron.py new file mode 100644 index 0000000..2ec8d85 --- /dev/null +++ b/backend/app/api/cron.py @@ -0,0 +1,330 @@ +"""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) diff --git a/backend/app/main.py b/backend/app/main.py index 5463c39..51c6eb9 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -4,7 +4,7 @@ from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.database import engine, Base -from app.api import auth, users, audit, dashboard, gateway, health +from app.api import auth, users, audit, dashboard, gateway, health, cron from app.middleware.audit import AuditMiddleware @@ -37,6 +37,7 @@ app.include_router(audit.router) app.include_router(dashboard.router, prefix="/api/v1") app.include_router(gateway.router, prefix="/api/v1") app.include_router(health.router, prefix="/api/v1") +app.include_router(cron.router, prefix="/api/v1") @app.get("/api/v1/health") diff --git a/frontend/src/api/cron.ts b/frontend/src/api/cron.ts new file mode 100644 index 0000000..e84ad20 --- /dev/null +++ b/frontend/src/api/cron.ts @@ -0,0 +1,52 @@ +import client from './client' + +export interface CronJob { + id: string + job_name: string + profile: string + schedule: string + status: string + last_run: string | null + last_run_success: boolean | null + success_rate: number + has_logs: boolean + system: boolean + consecutive_failures: number +} + +export interface CronListResponse { + jobs: CronJob[] + has_failures: boolean +} + +export interface CronLogItem { + time: string + status: string + message: string +} + +export interface CronLogsResponse { + logs: CronLogItem[] +} + +export interface CronActionResponse { + success: boolean + action: string + job_id: string + detail: string +} + +export const getCronJobs = (params?: { status?: string; profile?: string }) => + client.get('/cron/jobs', { params }) + +export const getCronLogs = (jobId: string) => + client.get(`/cron/jobs/${jobId}/logs`) + +export const runCronJob = (jobId: string) => + client.post(`/cron/jobs/${jobId}/run`) + +export const pauseCronJob = (jobId: string) => + client.post(`/cron/jobs/${jobId}/pause`) + +export const resumeCronJob = (jobId: string) => + client.post(`/cron/jobs/${jobId}/resume`) diff --git a/frontend/src/components/layout/Sidebar.tsx b/frontend/src/components/layout/Sidebar.tsx index a3db0dc..6291980 100644 --- a/frontend/src/components/layout/Sidebar.tsx +++ b/frontend/src/components/layout/Sidebar.tsx @@ -1,5 +1,7 @@ +import { useEffect, useState } from 'react' import { useAuth } from '../../hooks/useAuth' import Badge from '../ui/Badge' +import { getDashboardSummary } from '../../api/dashboard' interface NavItem { label: string @@ -77,6 +79,19 @@ function hasAccess(userRole: string | undefined, requiredRole?: string): boolean export default function Sidebar({ activePath, onNavigate }: SidebarProps) { const { user } = useAuth() + const [cronFailures, setCronFailures] = useState(0) + + useEffect(() => { + const fetch = async () => { + try { + const res = await getDashboardSummary() + setCronFailures(res.data.cron_failures_today) + } catch { /* ignore */ } + } + fetch() + const id = setInterval(fetch, 60_000) + return () => clearInterval(id) + }, []) return (