140 lines
5.1 KiB
Python
140 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
"""alert_helper.py — MoFin 统一告警网关（信噪比控制）
|
|
|
|
原则（老爸 2026-07-21 定）：
|
|
- 真正有意义的信息（重点推荐操作/买入信号/需人工处理）绝不能被淹没
|
|
- 纯通知型信息（系统报备/日常状态）必须控制频率和篇幅
|
|
|
|
两级通道：
|
|
- ACTION（行动级）：买入信号、重点推荐、需人工核查的故障
|
|
→ 直通，不限速，🚨 醒目前缀，独立成条
|
|
- INFO（通知级）：部署报备、卫生审计、修复报备、螺旋嫌疑
|
|
→ 同类 30 分钟内最多 1 条；篇幅 ≤8 行；24h 内容去重（同一问题不重复报）
|
|
|
|
所有系统消息统一 📟【MoFin系统·类别】前缀，与知微本人消息一眼区分。
|
|
|
|
用法:
|
|
from alert_helper import notify, ACTION, INFO
|
|
notify("信号", "📈 300308 买入信号...", level=ACTION) # 直通
|
|
notify("部署守卫", "自动部署完成...", level=INFO) # 限速
|
|
"""
|
|
import json, os, time, hashlib
|
|
from datetime import datetime
|
|
|
|
ACTION = "action"
|
|
INFO = "info"
|
|
|
|
STATE_FILE = "/home/hmo/MoFin/gateway/logs/alert_state.json"
|
|
LOG = "/home/hmo/MoFin/gateway/logs/alert_helper.log"
|
|
|
|
INFO_MIN_INTERVAL = 1800 # 同类 info 30 分钟最多 1 条
|
|
INFO_MAX_LINES = 8 # info 篇幅上限
|
|
ACTION_MAX_LINES = 30 # action 篇幅上限（宽松但不失控）
|
|
ACTION_DEDUP_SEC = 300 # action 相同内容 5 分钟内不重复发（防同秒双发/竞态）
|
|
DEDUP_SEC = 24 * 3600 # 相同内容 24h 不重复
|
|
|
|
|
|
def _log(msg):
|
|
line = f"[{datetime.now().isoformat(timespec='seconds')}] {msg}"
|
|
print(line, flush=True)
|
|
try:
|
|
os.makedirs(os.path.dirname(LOG), exist_ok=True)
|
|
with open(LOG, "a", encoding="utf-8") as f:
|
|
f.write(line + "\n")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _load_state():
|
|
try:
|
|
with open(STATE_FILE, encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def _save_state(st):
|
|
try:
|
|
# 只保留最近 100 个类别条目
|
|
if len(st) > 100:
|
|
st = dict(sorted(st.items(), key=lambda kv: kv[1].get("last_ts", 0))[-100:])
|
|
with open(STATE_FILE, "w", encoding="utf-8") as f:
|
|
json.dump(st, f)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _send(body):
|
|
import urllib.request
|
|
req = urllib.request.Request(
|
|
"http://127.0.0.1:5805/",
|
|
data=json.dumps({"body": body, "to": "hmo@yoin.fun", "type": "chat"}).encode(),
|
|
headers={"Content-Type": "application/json"})
|
|
urllib.request.urlopen(req, timeout=5)
|
|
|
|
|
|
def notify(category, body, level=INFO):
|
|
"""统一告警入口。返回 True=已发送, False=被限速/去重静默。"""
|
|
now = time.time()
|
|
st = _load_state()
|
|
entry = st.get(category, {"last_ts": 0, "last_hash": "", "suppressed": 0})
|
|
|
|
body_hash = hashlib.md5(body.encode()).hexdigest()
|
|
|
|
if level == INFO:
|
|
# 24h 内容去重：同一问题不重复报
|
|
if body_hash == entry.get("last_hash") and now - entry.get("last_ts", 0) < DEDUP_SEC:
|
|
_log(f"[{category}] 内容重复(24h内已报)，静默")
|
|
return False
|
|
# 频率限制：同类 30min 最多 1 条
|
|
if now - entry.get("last_ts", 0) < INFO_MIN_INTERVAL:
|
|
entry["suppressed"] = entry.get("suppressed", 0) + 1
|
|
st[category] = entry
|
|
_save_state(st)
|
|
_log(f"[{category}] 30min 频率限制，静默(累计压制{entry['suppressed']}条)")
|
|
return False
|
|
# 篇幅截断
|
|
lines = body.splitlines()
|
|
body = "\n".join(lines)
|
|
# 告知压制历史
|
|
if entry.get("suppressed", 0) > 0:
|
|
body += f"\n(注: 上次以来另有 {entry['suppressed']} 条同类通知已按频率策略静默)"
|
|
entry["suppressed"] = 0
|
|
prefix = f"📟【MoFin系统·{category}】（非知微本人）"
|
|
else:
|
|
# ACTION: 不限速，但相同内容 5 分钟内去重（防竞态双发）
|
|
if body_hash == entry.get("last_hash") and now - entry.get("last_ts", 0) < ACTION_DEDUP_SEC:
|
|
_log(f"[{category}] ACTION 内容重复(5min内)，静默")
|
|
return False
|
|
lines = body.splitlines()
|
|
body = "\n".join(lines)
|
|
prefix = f"🚨【MoFin·{category}】"
|
|
|
|
# 写入 broadcast_messages（统一消息源，告警类）
|
|
try:
|
|
from messenger import _classify, DB
|
|
import sqlite3 as _sq3
|
|
from datetime import datetime as _dt
|
|
_cat = "system_error" if level == ACTION else _classify(category, body)
|
|
_conn = _sq3.connect(DB, timeout=30)
|
|
_conn.execute(
|
|
"INSERT INTO broadcast_messages (ts, category, title, content, source) VALUES (?,?,?,?,?)",
|
|
(_dt.now().isoformat(), _cat, f"MoFin·{category}", body, category))
|
|
_conn.commit()
|
|
_conn.close()
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
_send(f"{prefix}\n{body}")
|
|
except Exception as e:
|
|
_log(f"[{category}] XMPP发送失败: {e}")
|
|
return False
|
|
|
|
entry["last_ts"] = now
|
|
entry["last_hash"] = body_hash
|
|
st[category] = entry
|
|
_save_state(st)
|
|
_log(f"[{category}] 已发送({level})")
|
|
return True
|