#!/usr/bin/env python3 # -*- coding: utf-8 -*- """生成「定时任务 + 看板」总览报告 (MD + HTML 双份) -> /root/web_reports/""" import json, sqlite3, subprocess, html, datetime as dt, os OUT_DIR = "/root/web_reports" STAMP = dt.datetime.now().strftime("%Y%m%d") BASE = f"cron_kanban_overview_{STAMP}" TZ = dt.timezone(dt.timedelta(hours=8)) def fmt(ts): if not ts: return "—" try: d = dt.datetime.fromisoformat(str(ts).replace("Z", "+00:00")) return d.astimezone(TZ).strftime("%m-%d %H:%M") except Exception: return str(ts)[:16] def fmt_epoch(ts): if not ts: return "—" return dt.datetime.fromtimestamp(int(ts), TZ).strftime("%m-%d %H:%M") # ---------- 1. Hermes cron ---------- jobs_raw = json.load(open("/root/.hermes/cron/jobs.json")) jobs = jobs_raw["jobs"] if isinstance(jobs_raw, dict) else jobs_raw DOW = {"0": "日", "1": "一", "2": "二", "3": "三", "4": "四", "5": "五", "6": "六"} def human_sched(j): expr = j.get("schedule_display") or (j.get("schedule") or {}).get("expr") or "" if j.get("schedule", {}).get("kind") == "interval": return f"每 {j['schedule'].get('minutes','?')} 分钟" try: mi, ho, dom, mon, dow = expr.split() except Exception: return expr hh = "/".join(f"{int(x):02d}" for x in ho.split(",")) mm = "/".join(f"{int(x):02d}" for x in mi.split(",")) t = f"{hh}:{mm}" if dow == "*": day = "每天" elif dow == "1-5": day = "工作日" elif "," in dow and dom == "*": day = "、".join(f"周{DOW.get(x, x)}" for x in dow.split(",")) else: day = f"周{DOW.get(dow, dow)}" if dom.isdigit(): day = f"每月{int(dom)}日" return f"{day} {t}" for j in jobs: j["_sched"] = human_sched(j) j["_last"] = fmt(j.get("last_run_at")) j["_next"] = fmt(j.get("next_run_at")) j["_deliver"] = {"feishu": "飞书", "local": "仅落盘", "origin": "本会话"}.get(j.get("deliver"), j.get("deliver") or "—") j["_kind"] = "脚本(无LLM)" if j.get("no_agent") else ("Agent" + (f" · {j.get('skill')}" if j.get("skill") else "")) if j.get("no_agent"): j["_model"] = "—" elif j.get("model"): j["_model"] = str(j["model"]) + "（钉死）" else: j["_model"] = f"{j.get('model_snapshot') or '全局默认'} · 跟随主模型" j["_status"] = {"ok": "正常", "error": "异常", "failed": "失败"}.get(j.get("last_status"), j.get("last_status") or "未跑") j["_runs"] = (j.get("repeat") or {}).get("completed") or 0 cron_ok = sum(1 for j in jobs if j.get("last_status") == "ok") cron_bad = len(jobs) - cron_ok cron_feishu = sum(1 for j in jobs if j.get("deliver") == "feishu") # ---------- 2. 系统 crontab ---------- sys_cron = [] try: out = subprocess.run(["crontab", "-l"], capture_output=True, text=True).stdout for line in out.splitlines(): line = line.strip() if not line or line.startswith("#"): continue parts = line.split(None, 5) if len(parts) >= 6: sys_cron.append({"expr": " ".join(parts[:5]), "cmd": parts[5]}) except Exception: pass # ---------- 3. 看板 ---------- con = sqlite3.connect("/root/.hermes/kanban.db") con.row_factory = sqlite3.Row tasks = [dict(r) for r in con.execute("SELECT * FROM tasks ORDER BY created_at DESC")] by_status = {} for t in tasks: by_status[t["status"]] = by_status.get(t["status"], 0) + 1 STATUS_CN = {"done": "已完成", "blocked": "阻塞", "running": "执行中", "ready": "待执行", "todo": "待办", "triage": "待分诊", "scheduled": "已排期", "archived": "归档", "review": "待审核"} for t in tasks: t["_ctime"] = fmt_epoch(t.get("created_at")) t["_dtime"] = fmt_epoch(t.get("completed_at")) t["_dur"] = "—" if t.get("started_at") and t.get("completed_at"): t["_dur"] = f"{int((t['completed_at'] - t['started_at']) / 60)}分" t["_s"] = STATUS_CN.get(t["status"], t["status"]) t["_t"] = t["title"] diags = [] try: for r in con.execute("SELECT * FROM tasks WHERE consecutive_failures > 0 AND status IN ('blocked','ready','todo') "): diags.append(dict(r)) except Exception: pass runs = [] try: runs = [dict(r) for r in con.execute("SELECT * FROM task_runs ORDER BY id DESC LIMIT 6")] except Exception: pass con.close() ran_today = [t for t in tasks if t["_ctime"].startswith(dt.datetime.now(TZ).strftime("%m-%d"))] # ---------- 4. MD ---------- now = dt.datetime.now(TZ).strftime("%Y-%m-%d %H:%M") md = [f"# 定时任务 · 看板总览（{now}）", ""] md.append(f"## 一、Hermes 定时任务（{len(jobs)} 个，正常 {cron_ok}）") md.append("") md.append("| 任务 | 时间 | 类型 | 模型 | 交付 | 最近运行 | 状态 | 下次 |") md.append("|---|---|---|---|---|---|---|---|") for j in sorted(jobs, key=lambda x: x["_sched"]): md.append(f"| {j['name']} | {j['_sched']} | {j['_kind']} | {j['_model']} | {j['_deliver']} | {j['_last']} | {j['_status']} | {j['_next']} |") md += ["", f"## 二、系统 crontab（{len(sys_cron)} 条）", ""] for s in sys_cron: md.append(f"- `{s['expr']}` — {s['cmd']}") md += ["", f"## 三、看板 default 板（共 {len(tasks)} 张卡）", ""] md.append("- 状态分布：" + "，".join(f"{STATUS_CN.get(k,k)} {v}" for k, v in sorted(by_status.items(), key=lambda x: -x[1]))) md.append("") md.append("| 卡片 | 状态 | 创建 | 完成 | 耗时 |") md.append("|---|---|---|---|---|") for t in tasks[:20]: md.append(f"| {t['_t']} | {t['_s']} | {t['_ctime']} | {t['_dtime']} | {t['_dur']} |") if diags: md += ["", "### 阻塞/异常"] for d in diags: md.append(f"- `{d['id']}` {d['title']}：连续失败 {d['consecutive_failures']} 次 — {(d.get('last_failure_error') or '')[:160]}") md += ["", f"> 生成 {now} · MD/HTML 双份 · HTML: http://47.113.231.110/{BASE}.html"] md_text = "\n".join(md) open(f"{OUT_DIR}/{BASE}.md", "w", encoding="utf-8").write(md_text) # ---------- 5. HTML ---------- e = html.escape def row_job(j): dot = "#0a7d4d" if j["_status"] == "正常" else "#c0392b" return f""" {e(j['name'])}{j['id']} {e(j['_sched'])} {e(j['_kind'])} {e(j['_model'])} {e(j['_deliver'])} {e(j['_last'])} {e(j['_status'])} {e(j['_next'])}""" def row_task(t): cls = {"done": "ok", "blocked": "bad", "archived": "faint"}.get(t["status"], "") return f""" {e(t['_t'])}{t['id']} {e(t['_s'])} {e(t['_ctime'])} {e(t['_dtime'])} {e(t['_dur'])}""" kpi = [ ("定时任务", len(jobs), ""), ("运行正常", cron_ok, "green"), ("异常", cron_bad, "red" if cron_bad else ""), ("推送飞书", cron_feishu, ""), ("系统 crontab", len(sys_cron), ""), ("看板卡片", len(tasks), ""), ("已完成", by_status.get("done", 0), "green"), ("阻塞", by_status.get("blocked", 0), "red" if by_status.get("blocked") else ""), ] kpi_html = "".join( f'
{v}
{e(k)}
' for k, v, c in kpi) diag_html = "" if diags: items = "".join( f'
  • {e(d["title"])} {d["id"]}
    连续失败 {d["consecutive_failures"]} 次 · ' f'{e((d.get("last_failure_error") or "")[:220])}
    • ' for d in diags) diag_html = f'
    ⚠ 需要处理
    ' html_doc = f""" 定时任务 · 看板总览 · 钱小兜
    QIANXIAODOU · SYSTEM INVENTORY

    定时任务 · 看板

    数据源：cron/jobs.json · kanban.db · crontab生成 {now} CST
    {kpi_html}

    01Hermes 定时任务{len(jobs)} 个 · 正常 {cron_ok}

    {"".join(row_job(j) for j in sorted(jobs, key=lambda x: x['_sched']))}
    任务时间类型模型交付最近运行状态下次

    交付「飞书」= 结果推送本会话；「仅落盘」= 静默执行，产出留本地。模型列：脚本任务无 LLM；「跟随主模型」= 未钉死、随全局默认（换主模型后跑 hermes cron resnap <job_id> 刷新）。

    02系统 crontab{len(sys_cron)} 条

    03看板 · default 板共 {len(tasks)} 张卡

    {"".join(f'{STATUS_CN.get(k,k)}{v}' for k, v in sorted(by_status.items(), key=lambda x: -x[1]))}
    {"".join(row_task(t) for t in tasks[:22])}
    卡片状态创建完成耗时

    展示最近 22 张（按创建倒序）；今日新增 {len(ran_today)} 张。

    {diag_html}
    """ open(f"{OUT_DIR}/{BASE}.html", "w", encoding="utf-8").write(html_doc) print("OK", BASE) print("cron:", len(jobs), "ok", cron_ok, "bad", cron_bad, "| syscron:", len(sys_cron), "| tasks:", len(tasks), by_status) print("blocked:", [(d["id"], d["title"]) for d in diags])