122 lines
4.4 KiB
Python
122 lines
4.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""alert_logger.py — MoFin 统一异常告警通道（2026-08-12 落地）
|
||
|
||
任何脚本/任务出错时调用 record_alert()：
|
||
- 写入 data/alerts.json（active 异常列表，前端浮窗读取 /api/overview 的 alerts）
|
||
- 追加 data/alerts_history.log（历史异常实时日志，永久保留）
|
||
- 同时通过 xmpp_logger 发送 XMPP 告警（老莫铁律：任何错误/异常 → xmpp告警 + MoFin监控）
|
||
|
||
用法：
|
||
from alert_logger import record_alert
|
||
record_alert(level="error", source="price_monitor", title="重评失败",
|
||
detail="688002 LLM 调用超时", code="688002")
|
||
|
||
level: info / warning / error / critical
|
||
"""
|
||
import json
|
||
import time
|
||
from pathlib import Path
|
||
|
||
DATA_DIR = Path("/home/hmo/MoFin/data")
|
||
ALERTS_FILE = DATA_DIR / "alerts.json"
|
||
HISTORY_FILE = DATA_DIR / "alerts_history.log"
|
||
|
||
# 保留的 active 异常数（浮窗只显示 active，最多保留 20 条）
|
||
MAX_ACTIVE = 20
|
||
|
||
|
||
def _now_str():
|
||
from datetime import datetime
|
||
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
|
||
def record_alert(level="error", source="", title="", detail="", code="", clear_after=0):
|
||
"""记录一条异常。clear_after>0 表示自动清除的秒数（可空）"""
|
||
ts = time.time()
|
||
ts_str = _now_str()
|
||
entry = {
|
||
"ts": ts,
|
||
"ts_str": ts_str,
|
||
"level": level,
|
||
"source": source,
|
||
"title": title,
|
||
"detail": detail,
|
||
"code": code,
|
||
"clear_after": clear_after,
|
||
}
|
||
|
||
# 1. active 列表（alerts.json）
|
||
try:
|
||
alerts = json.loads(ALERTS_FILE.read_text(encoding="utf-8")) if ALERTS_FILE.exists() else []
|
||
if not isinstance(alerts, list):
|
||
alerts = []
|
||
# 同 source+code 的旧告警先移除（避免重复堆积），再插到头部
|
||
alerts = [a for a in alerts if not (a.get("source") == source and a.get("code") == code)]
|
||
alerts.insert(0, entry)
|
||
# 清除超时的
|
||
now = time.time()
|
||
alerts = [a for a in alerts if not a.get("clear_after") or now - a.get("ts", 0) < a["clear_after"]]
|
||
alerts = alerts[:MAX_ACTIVE]
|
||
ALERTS_FILE.write_text(json.dumps(alerts, ensure_ascii=False, indent=1), encoding="utf-8")
|
||
except Exception:
|
||
pass
|
||
|
||
# 2. 历史日志（永久追加）
|
||
try:
|
||
with HISTORY_FILE.open("a", encoding="utf-8") as f:
|
||
f.write(f"[{ts_str}] [{level.upper()}] {source} | {title} | {detail} | {code}\n")
|
||
except Exception:
|
||
pass
|
||
|
||
# 2.5 写入 broadcast_messages 表（统一消息源，异常=system_error类）
|
||
try:
|
||
from messenger import _classify, DB
|
||
import sqlite3 as _sq3
|
||
_cat = _classify(title, detail)
|
||
# 异常/告警统一归为 system_error 类（供右上角异常浮窗筛选）
|
||
if level in ("error", "critical", "warning"):
|
||
_cat = "system_error"
|
||
_conn = _sq3.connect(DB, timeout=30)
|
||
_conn.execute(
|
||
"INSERT INTO broadcast_messages (ts, category, title, content, source) VALUES (?,?,?,?,?)",
|
||
(ts_str, _cat, title, detail, source))
|
||
_conn.commit()
|
||
_conn.close()
|
||
except Exception:
|
||
pass
|
||
|
||
# 3. XMPP 告警（非 info 级别才发，避免轰炸）→ 本地 XMPP 发送服务 :5805
|
||
if level != "info":
|
||
try:
|
||
import urllib.request as _ur
|
||
body = f"⚠️[{level.upper()}] {source}: {title} {detail} {code}".strip()
|
||
req = _ur.Request(
|
||
"http://127.0.0.1:5805/",
|
||
data=body.encode("utf-8"),
|
||
headers={"Content-Type": "text/plain"},
|
||
method="POST",
|
||
)
|
||
_ur.urlopen(req, timeout=5).read()
|
||
except Exception:
|
||
pass
|
||
|
||
return entry
|
||
|
||
|
||
def clear_alert(source="", code=""):
|
||
"""清除指定 source+code 的 active 告警（恢复后调用）"""
|
||
try:
|
||
alerts = json.loads(ALERTS_FILE.read_text(encoding="utf-8")) if ALERTS_FILE.exists() else []
|
||
if not isinstance(alerts, list):
|
||
return
|
||
alerts = [a for a in alerts if not (a.get("source") == source and (not code or a.get("code") == code))]
|
||
ALERTS_FILE.write_text(json.dumps(alerts, ensure_ascii=False, indent=1), encoding="utf-8")
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 自测
|
||
record_alert(level="info", source="alert_logger", title="自测", detail="告警通道正常", code="test")
|
||
print("alerts.json 写入 OK:", ALERTS_FILE.exists())
|