From 19a10ed8f9c77d9f7346d4063ffd9da0f1a4c1cf Mon Sep 17 00:00:00 2001
From: xxm <xxm@mofin.local>
Date: Mon, 24 Aug 2026 11:41:48 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20broadcast=E6=B6=88=E6=81=AF=E4=BA=8B?=
 =?UTF-8?q?=E4=BB=B6=E5=9D=97=E5=8C=96=E2=80=94=E2=80=94stdout=20hook?=
 =?UTF-8?q?=E4=BB=8E=E9=80=90=E8=A1=8C=E7=A2=8E=E7=89=87=E6=94=B9=E4=B8=BA?=
 =?UTF-8?q?=E7=BC=93=E5=86=B2=E8=81=9A=E5=90=88(=E9=9D=99=E9=BB=982s/?=
 =?UTF-8?q?=E6=BB=A125=E8=A1=8C/atexit=E5=88=87=E5=9D=97),=E7=9B=B8?=
 =?UTF-8?q?=E9=82=BB=E9=87=8D=E5=A4=8D=E8=A1=8C=E5=8E=BB=E9=87=8D,?=
 =?UTF-8?q?=E5=8D=95=E8=BF=9E=E6=8E=A5=E5=86=99=E5=85=A5(=E8=80=81?=
 =?UTF-8?q?=E8=8E=AB:=E6=AF=8F=E6=9D=A1=E6=B6=88=E6=81=AF=E8=AF=A6?=
 =?UTF-8?q?=E6=83=85=3D=E6=91=98=E8=A6=81=E6=97=A0=E4=BF=A1=E6=81=AF?=
 =?UTF-8?q?=E9=87=8F)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 deploy/profile-scripts/messenger.py | 76 ++++++++++++++++++++---------
 1 file changed, 54 insertions(+), 22 deletions(-)

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):