#!/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, time, 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") class _MessengerStdout: # 2026-08-24 事件块聚合：逐行碎片→一块一条消息（老莫：broadcast每条消息详情=摘要,无信息量） # 块边界：静默>2秒 或 满25行；atexit强制flush。xmpp实时输出不缓冲(hermes靠stdout推送)。 MAX_BLOCK_LINES = 25 MAX_IDLE_SEC = 2.0 def __init__(self, orig, script): self.orig = orig self.script = script self.channel = get_channel(script) self._buf = [] self._last = 0.0 import atexit atexit.register(self._flush_buf) def write(self, s): if self.channel in ("xmpp", "both"): # xmpp 通道: 实时输出给 hermes(推xmpp)——不缓冲 self.orig.write(s); self.orig.flush() now = time.time() if self._buf and now - self._last > self.MAX_IDLE_SEC: self._flush_buf() # 静默超2秒: 上一个事件块结束 self._buf.append(s) self._last = now if len(self._buf) >= self.MAX_BLOCK_LINES: self._flush_buf() return len(s) def _flush_buf(self): if self._buf: text = "".join(self._buf) self._buf = [] _flush_broadcast_block(text.split("\n"), self.script) def flush(self): if self.channel in ("xmpp", "both"): self.orig.flush() def _flush_broadcast_block(lines, script): """把一批 stdout 行聚合成一条 broadcast 消息（2026-08-24 事件块化）。 过滤噪音前缀 + 相邻重复行去重（根治逐行碎片+双写重复）+ 单连接一次写入。""" skip_prefix = ("[DB]", "[SYNC", "[guard", "[regime] ", "⏱", "📊 ", "📥 ", "💾 ", "🔄 ", "[SILENT]") out = [] for line in lines: line = line.strip() if not line or len(line) < 3: continue if line.startswith(skip_prefix): continue if line.startswith("【") and "】" in line and "MoFin·" in line: continue # 跳过已格式化的 messenger 输出, 防递归 if out and out[-1] == line: continue # 相邻重复去重 out.append(line) if not out: return content = "\n".join(out) try: category = _classify(_job_title(script), content) conn = sqlite3.connect(DB, timeout=30) conn.execute( "INSERT INTO broadcast_messages (ts, category, title, content, source) VALUES (?,?,?,?,?)", (datetime.now().isoformat(), category, _job_title(script), content, script)) conn.commit() conn.close() except Exception: pass def _write_broadcast_from_stdout(text, script): """把 stdout 内容直接写入 broadcast_messages(不经 send, 避免 xmpp 递归)""" if not text or not text.strip(): return _flush_broadcast_block(text.strip().split("\n"), script) def _job_title(script): return script.replace(".py", "") def install_stdio_hook(script_name=None): """在脚本入口调用：重定向 stdout 到 messenger 通道控制""" if script_name is None: script_name = os.path.basename(sys.argv[0]) if sys.argv else "mofin" sys.stdout = _MessengerStdout(sys.stdout, script_name) return get_channel(script_name) 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}")