feat: 全局cron job健康监控, failed时主动推XMPP

2026-07-07 11:29:56 +08:00
parent 5ffc487659
commit 43bd84a7c0
1 changed files with 116 additions and 0 deletions
@@ -0,0 +1,116 @@
+#!/usr/bin/env python3
+"""cron_health_monitor.py — 全局cron job健康监控
+
+每10分钟跑一次，检查所有cron job的last_status。
+发现failed → 立即推XMPP告警到老爸。
+"""
+import json, os, sys, sqlite3, time
+from pathlib import Path
+from datetime import datetime
+from urllib.request import Request, urlopen
+
+XMPP_BRIDGE = "http://127.0.0.1:5805/"
+XMPP_USER = "hmo@yoin.fun"
+STATE_FILE = Path.home() / ".hermes" / ".cron_health_state.json"
+
+def xmpp_push(text):
+    try:
+        payload = json.dumps({"to": XMPP_USER, "body": text, "type": "chat"}).encode()
+        req = Request(XMPP_BRIDGE, data=payload, headers={"Content-Type": "application/json"})
+        urlopen(req, timeout=5)
+        return True
+    except Exception as e:
+        print(f"[XMPP推送失败] {e}", file=sys.stderr)
+        return False
+
+def scan_all_jobs():
+    """扫描所有cron jobs.json，返回全量job列表"""
+    jobs = []
+    seen = set()
+    for jf in [
+        "/home/hmo/.hermes/cron/jobs.json",
+        "/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json",
+    ]:
+        try:
+            data = json.load(open(jf))
+            for job in data.get("jobs", []):
+                jid = job.get("id", "")
+                if jid in seen:
+                    continue
+                seen.add(jid)
+                last_status = job.get("last_status", "")
+                if not last_status:
+                    continue
+                jobs.append({
+                    "id": jid,
+                    "name": job.get("name", jid[:12]),
+                    "status": last_status,
+                    "last_run": job.get("last_run_at", "?"),
+                    "enabled": job.get("enabled", True),
+                    "no_agent": job.get("no_agent", False),
+                })
+        except Exception:
+            pass
+    return jobs
+
+def load_state():
+    try:
+        with open(STATE_FILE) as f:
+            return json.load(f)
+    except Exception:
+        return {}
+
+def save_state(state):
+    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
+    with open(STATE_FILE, "w") as f:
+        json.dump(state, f, indent=2)
+
+def main():
+    now = datetime.now().strftime("%H:%M")
+    jobs = scan_all_jobs()
+    state = load_state()
+    
+    failed_jobs = [j for j in jobs if j["status"] == "failed" and j["enabled"]]
+    
+    if not failed_jobs:
+        # 全部正常
+        print("[SILENT]")
+        # 记录全部分辨率，用于跟踪从failed→ok的恢复
+        save_state({j["id"]: {"status": j["status"], "notified": False} for j in jobs})
+        return
+    
+    # 有failed job → 过滤出未通知过的（避免重复推送）
+    new_failures = []
+    ongoing_failures = []
+    for j in failed_jobs:
+        prev = state.get(j["id"], {})
+        prev_status = prev.get("status", "")
+        if prev_status != "failed":
+            # 新发现失败（之前正常或未记录）
+            new_failures.append(j)
+        else:
+            ongoing_failures.append(j)
+    
+    if not new_failures:
+        # 没有新增失败 → 静默（上次已通知过）
+        print("[SILENT] 已有失败未恢复")
+        save_state({j["id"]: {"status": j["status"], "notified": True} for j in jobs})
+        return
+    
+    # 新发现的失败 → 推XMPP
+    msg_lines = [f"🔴 {len(new_failures)}个cron job失败 ({now}):"]
+    for j in new_failures:
+        tag = "no_agent" if j["no_agent"] else "LLM"
+        msg_lines.append(f"  ❌ {j['name']}({j['id'][:8]}) [{tag}] last_run={j['last_run']}")
+    if ongoing_failures:
+        msg_lines.append(f"  (另有{len(ongoing_failures)}个上次已报过)")
+    
+    text = "\n".join(msg_lines)
+    print(text)
+    xmpp_push(text)
+    
+    # 记录状态
+    save_state({j["id"]: {"status": j["status"], "notified": True} for j in jobs})
+
+if __name__ == "__main__":
+    main()