diff --git a/deploy/profile-scripts/messenger.py b/deploy/profile-scripts/messenger.py index e3a2c0a5..fff7f736 100644 --- a/deploy/profile-scripts/messenger.py +++ b/deploy/profile-scripts/messenger.py @@ -13,7 +13,7 @@ from messenger import send send(title="MoFin持仓异动", content="...") # 自动按当前脚本/job路由 """ -import os, sys, json, sqlite3 +import os, sys, time, json, sqlite3 from datetime import datetime DB = "/home/hmo/MoFin/data/mofin.db" @@ -129,31 +129,50 @@ def send_action_recommendation(title, content, source=None): 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) + 写入 broadcast 表 + # xmpp 通道: 实时输出给 hermes(推xmpp)——不缓冲 self.orig.write(s); self.orig.flush() - _write_broadcast_from_stdout(s, self.script) - else: - # broadcast 通道: 不输出 stdout(不推 xmpp), 只写入 broadcast 表 - _write_broadcast_from_stdout(s, self.script) + 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 _write_broadcast_from_stdout(text, script): - """把 stdout 内容直接写入 broadcast_messages(不经 send, 避免 xmpp 递归)""" - if not text or not text.strip(): - return - t = text.strip() +def _flush_broadcast_block(lines, script): + """把一批 stdout 行聚合成一条 broadcast 消息(2026-08-24 事件块化)。 + 过滤噪音前缀 + 相邻重复行去重(根治逐行碎片+双写重复)+ 单连接一次写入。""" skip_prefix = ("[DB]", "[SYNC", "[guard", "[regime] ", "⏱", "📊 ", "📥 ", "💾 ", "🔄 ", "[SILENT]") - for line in t.split("\n"): + out = [] + for line in lines: line = line.strip() if not line or len(line) < 3: continue @@ -161,16 +180,29 @@ def _write_broadcast_from_stdout(text, script): continue if line.startswith("【") and "】" in line and "MoFin·" in line: continue # 跳过已格式化的 messenger 输出, 防递归 - try: - category = _classify(_job_title(script), line) - 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), line, script)) - conn.commit() - conn.close() - except Exception: - pass + 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):