feat(messenger): 统一消息处理者——生产者(脚本/job)驱动通道,不按内容判通道

2026-08-21 20:11:57 +08:00
parent 548fd4bab1
commit ed83fddcf4
1 changed files with 135 additions and 0 deletions
@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+"""messenger.py — 统一消息处理者（生产者与消费者解耦）
+
+核心原则：消费者(本模块)根据【生产者是谁】(哪个 cron job) 决定消息去向，
+绝不根据消息内容判断。
+
+通道配置（jobs.json 的 delivery 字段）：
+    broadcast : 只写 broadcast_messages 表（不推 xmpp）
+    xmpp      : 写 broadcast 归档 + 推 xmpp
+    both      : 同上（保留兼容）
+
+用法：
+    from messenger import send
+    send(title="MoFin持仓异动", content="...")   # 自动按当前脚本/job路由
+"""
+import os, sys, json, sqlite3
+from datetime import datetime
+
+DB = "/home/hmo/MoFin/data/mofin.db"
+JOBS_JSON = "/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json"
+DEFAULT_CHANNEL = "broadcast"
+
+# 脚本名 → job delivery 配置缓存 {script_basename: channel}
+_script_channel_cache = None
+
+
+def _load_script_channels():
+    """读取 jobs.json：建立 {脚本basename: delivery通道} 映射"""
+    global _script_channel_cache
+    if _script_channel_cache is not None:
+        return _script_channel_cache
+    mapping = {}
+    try:
+        if os.path.exists(JOBS_JSON):
+            with open(JOBS_JSON, encoding="utf-8") as f:
+                jobs = json.load(f).get("jobs", [])
+            for j in jobs:
+                if not isinstance(j, dict):
+                    continue
+                script = str(j.get("script", "")).split("/")[-1]
+                if not script or not script.endswith(".py"):
+                    continue
+                channel = j.get("delivery")
+                if channel:
+                    if script not in mapping or channel == "xmpp":
+                        mapping[script] = channel
+    except Exception as e:
+        print(f"  [messenger] 读取 job 通道配置失败: {e}", file=sys.stderr)
+    _script_channel_cache = mapping
+    return mapping
+
+
+def get_channel(job_script=None):
+    """获取指定生产者脚本的通道。None=自动识别当前脚本"""
+    mapping = _load_script_channels()
+    if job_script:
+        return mapping.get(job_script, DEFAULT_CHANNEL)
+    script = os.path.basename(sys.argv[0]) if sys.argv else "mofin"
+    return mapping.get(script, DEFAULT_CHANNEL)
+
+
+def _classify(title, content):
+    """内容分类(仅为 broadcast 表展示用，不用于通道决策)"""
+    text = (title + " " + content).lower()
+    words = {
+        "trading": ["买入", "卖出", "止损", "止盈", "加仓", "减仓", "持仓", "区间", "操作", "推荐"],
+        "system_error": ["llm端点", "api错误", "连接失败", "超时", "异常", "失败", "error", "exception", "告警", "故障"],
+        "health": ["健康", "体检", "完整性", "守卫", "cron"],
+        "market": ["大盘", "市场", "板块", "行业", "指数", "行情"],
+        "news": ["新闻", "消息", "资讯", "公告", "政策"],
+        "strategy": ["策略", "重评", "评估", "温区", "regime", "12维"],
+    }
+    for cat, kws in words.items():
+        if any(kw in text for kw in kws):
+            return cat
+    return "general"
+
+
+def send(title="", content="", source=None, channel=None, job_name=None):
+    """统一消息发送。生产者调用即可，不关心去向。
+
+    通道决策优先级：
+      1. 显式 channel 参数
+      2. 当前脚本名对应的 job 配置（自动识别生产者）
+      3. 默认 broadcast
+    """
+    now = datetime.now().isoformat()
+    producer = source or job_name or (os.path.basename(sys.argv[0]) if sys.argv else "mofin")
+    category = _classify(title, content)
+
+    # 决定通道（生产者驱动：按脚本名查 job 配置）
+    if channel is None:
+        mapping = _load_script_channels()
+        script_key = None
+        if source and source.endswith(".py"):
+            script_key = source
+        elif producer and producer.endswith(".py"):
+            script_key = producer
+        else:
+            script_key = os.path.basename(sys.argv[0]) if sys.argv else "mofin"
+        channel = mapping.get(script_key, DEFAULT_CHANNEL)
+
+    # 写 broadcast 表（所有通道归档，保证日志完整）
+    try:
+        conn = sqlite3.connect(DB, timeout=30)
+        conn.execute(
+            "INSERT INTO broadcast_messages (ts, category, title, content, source) VALUES (?,?,?,?,?)",
+            (now, category, title, content, producer))
+        conn.commit()
+        conn.close()
+    except Exception as e:
+        print(f"  [messenger] broadcast 写入失败: {e}", file=sys.stderr)
+
+    # xmpp 通道：输出 stdout（hermes 捕获推 xmpp）
+    if channel in ("xmpp", "both"):
+        if title:
+            print(f"【{title}】{content}", flush=True)
+        else:
+            print(content, flush=True)
+        return "xmpp"
+
+    return "broadcast"
+
+
+def send_action_recommendation(title, content, source=None):
+    """操作推荐：强制 xmpp 通道（必须推送给用户）"""
+    return send(title=title, content=content, source=source, channel="xmpp")
+
+
+if __name__ == "__main__":
+    print("=== messenger 自测 ===")
+    mapping = _load_script_channels()
+    for s, c in sorted(mapping.items()):
+        if "price" in s or "market" in s or "per_stock" in s or "anomaly" in s:
+            print(f"  {s}: {c}")