diff --git a/scripts/cron_health_monitor.py b/scripts/cron_health_monitor.py new file mode 100644 index 00000000..05b8f7ab --- /dev/null +++ b/scripts/cron_health_monitor.py @@ -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()