105 lines
3.8 KiB
Python
105 lines
3.8 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
|
||
|
||
# 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())
|