宜博 / pixelphone

Commit 20815fb380e0380377c6ccc52e75f9d465974931

Authored by yibo
1 parent b339a5dd

feat(wechat-auto-reply): 编排器串起 1→5 全链路 + watcher 接线 

wechat_reply_orchestrator.py：消费 watcher incoming.jsonl(cursor增量+单实例锁+
首跑基线)→ 候选事件(私聊/@我/引用我)→ env2 查上下文(rcontact身份+历史,一次拷库
全批复用)→ env3 hermes -z 生成 → env4 policy.decide(白名单读 circle.db)→ env5
落地(auto:send_text_via_hook+auto_sent / review:pending 进 /approvals)。生成失败
即静默不发,绝不误发。watcher._trigger_orchestrator 读到新消息后 fire-and-forget
拉起 drain(独立进程不阻塞采集)。真机验证:接线自动触发、真实猎头上下文生成火星人
口吻草稿并正确进审核队列。auto_send 默认关=全审核(保守起步)。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Inline Side-by-side
Showing 3 changed files with 303 additions and 0 deletions
md/wechat-auto-reply-requirements.md
  View file @20815fb
... ... @@ -93,6 +93,13 @@
93 93 - 落地新增：①「待回复队列」存储（sqlite/json）；② wechat viewer 的"发送"按钮接到发送链路。
94 94 - 安全红线（贯穿）：涉钱 / 承诺 / 约时间 / 低置信 → 即使白名单也建议进审核。
95 95  
  96 +### 编排器 · 把 1→5 串起来 —— ✅ **已实现并真机验证（2026-07-10）**
  97 +`wechat_plain/scripts/auto-reply/wechat_reply_orchestrator.py`（设备 `~/.hermes/wechat_reply/`）：
  98 +- watcher 读到新消息 → 写 `incoming.jsonl` → **fire-and-forget 触发** `orchestrator --drain`（watcher `_trigger_orchestrator`，独立进程不阻塞采集）。
  99 +- drain（非阻塞单实例锁 + cursor 增量 + 首跑基线）：对候选事件（私聊/群@我/群引用我）→ env2 查上下文（rcontact 身份 + 最近历史，一次拷库全批复用）→ env3 `hermes -z` 生成 → env4 `policy.decide()`（白名单读 circle.db `reply_whitelist`）→ env5 落地：auto → `send_text_via_hook` + `reply_queue(auto_sent)` / review → `reply_queue(pending)`。
  100 +- 安全：生成超时/失败 → 无草稿 → policy skip → 不发（失败即静默，绝不误发）。
  101 +- 真机验证：接线（watcher inotify → 自动拉起编排器 drain）通过；真实上下文（猎头 婷婷）生成「哈哈婷婷，我不是什么总，就是个写代码的，不负责招人这块哈」→ 正确进 /approvals 待审。
  102 +
96 103 ### 环节 5 · 发送回复带引用（原生气泡）
97 104 需求 = 点原文→引用→发送后显示引用气泡。两条实现路：
98 105 - **(a) A11y UI 自动化（一期推荐）**：长按目标气泡 → 点"引用" → 输入 → 发送。驱动微信自己 UI → **产出货真价实的原生引用气泡，零逆向**。代价：需微信前台、脚本略脆、比 API 慢。已有 A11y 发送服务底子。
... ...
wechat_plain/scripts/auto-reply/wechat_reply_orchestrator.py 0 → 100644
  View file @20815fb
  1 +#!/usr/bin/env python3
  2 +"""环节 1→5 编排器 —— 把整条自动回复链串起来
  3 +
  4 +消费 watcher 的 incoming.jsonl（cursor 增量）→ 对候选事件（私聊/群@我/群引用我）：
  5 + env2 查上下文（发送者身份 + 最近历史）→ env3 hermes -z 生成草稿 →
  6 + env4 policy.decide()（白名单读 wechat-viewer 的 circle.db）→
  7 + env5 落地：auto 直接 send_text_via_hook + 入队(auto_sent) / review 入队(pending，在 /approvals 等人工)
  8 +
  9 +作为 root 在 Termux 跑（同 watcher 上下文）。由 watcher 每次读到新消息后触发 --drain。
  10 +
  11 +安全：policy 默认 auto_send_enabled=False → 一切进 review（不自动发）。生成超时/失败 →
  12 +无草稿 → policy skip → 不发（失败即静默，绝不误发）。
  13 +
  14 +ponytail: 每次 drain 拷一次整库(~96MB)供该批所有事件查上下文；事件稀疏可接受。
  15 +升级：与 watcher 共享一次拷贝；incoming.jsonl 增长需按 offset 游标+轮转。
  16 +"""
  17 +import json
  18 +import os
  19 +import sqlite3
  20 +import subprocess
  21 +import sys
  22 +import time
  23 +
  24 +_HERE = os.path.dirname(os.path.abspath(__file__))
  25 +sys.path.insert(0, _HERE)
  26 +import wechat_reply_watcher as w # 复用解密/查询/账号推导
  27 +import wechat_reply_policy as policy
  28 +
  29 +WORK_DIR = w.WORK_DIR
  30 +QUEUE_FILE = f"{WORK_DIR}/incoming.jsonl"
  31 +CURSOR_FILE = f"{WORK_DIR}/orch_state.json"
  32 +ORCH_LOG = f"{WORK_DIR}/orchestrator.log"
  33 +CIRCLE_DB = "/data/data/com.termux/files/home/wechat-viewer/circle.db"
  34 +EVENT_BUS = "/data/data/com.termux/files/home/.hermes/event_bus"
  35 +HERMES = "/data/data/com.termux/files/usr/bin/hermes"
  36 +
  37 +CANDIDATE_KINDS = {"private", "group_at_me", "group_quote_me"}
  38 +HISTORY_N = 12
  39 +GEN_TIMEOUT = 160
  40 +
  41 +
  42 +def log(msg):
  43 + line = f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}"
  44 + print(line, flush=True)
  45 + try:
  46 + with open(ORCH_LOG, "a") as f:
  47 + f.write(line + "\n")
  48 + except OSError:
  49 + pass
  50 +
  51 +
  52 +# ── env2：查上下文 ─────────────────────────────────────
  53 +def _display_name(stage, key, wxid):
  54 + if not wxid:
  55 + return ""
  56 + rows = w.sqlcipher_query(
  57 + stage, key,
  58 + f"SELECT conRemark, nickname, alias FROM rcontact WHERE username='{wxid}';")
  59 + if rows and rows[0]:
  60 + r = rows[0] + ["", "", ""]
  61 + return r[0] or r[1] or r[2] or wxid
  62 + return wxid
  63 +
  64 +
  65 +def _placeholder(mtype, text):
  66 + if mtype == 1:
  67 + return text
  68 + return {3: "[图片]", 34: "[语音]", 43: "[视频]", 47: "[表情]",
  69 + 42: "[名片]", 49: "[链接/文件]", 48: "[位置]"}.get(mtype, text[:40] if text else "[消息]")
  70 +
  71 +
  72 +def build_context(stage, key, event, self_wxid):
  73 + talker = event.get("talker", "")
  74 + is_group = event.get("isGroup") or talker.endswith("@chatroom")
  75 + sender_wxid = event.get("sender") or talker
  76 + sender_disp = _display_name(stage, key, sender_wxid)
  77 + lines = [f"对方：{sender_disp}"]
  78 + if is_group:
  79 + gname = _display_name(stage, key, talker)
  80 + lines.insert(0, f"[群] {gname}")
  81 + # 最近历史
  82 + rows = w.sqlcipher_query(
  83 + stage, key,
  84 + "SELECT isSend, type, createTime, hex(content) FROM message "
  85 + f"WHERE talker='{talker}' ORDER BY msgId DESC LIMIT {HISTORY_N};")
  86 + hist = []
  87 + for r in reversed(rows):
  88 + if len(r) < 4:
  89 + continue
  90 + is_send, mtype, _, chex = r
  91 + try:
  92 + raw = bytes.fromhex(chex).decode("utf-8", "ignore")
  93 + except ValueError:
  94 + raw = ""
  95 + if is_send == "1":
  96 + who, body = "我", raw
  97 + else:
  98 + s, body = w.parse_group_content(raw) if is_group else (sender_wxid, raw)
  99 + who = _display_name(stage, key, s) if is_group and s else sender_disp
  100 + txt = _placeholder(int(mtype) if mtype.isdigit() else 0, body).replace("\n", " ").strip()
  101 + if txt:
  102 + hist.append(f" {who}: {txt}")
  103 + ctx = "\n".join(lines)
  104 + if hist:
  105 + ctx += "\n最近对话：\n" + "\n".join(hist[-HISTORY_N:])
  106 + ctx += f"\n本条：{sender_disp}: {(event.get('body') or '').strip()[:200]}"
  107 + return ctx, sender_disp
  108 +
  109 +
  110 +# ── env3：hermes 生成 ─────────────────────────────────
  111 +def generate(context_text):
  112 + prompt = (
  113 + "【任务】下面是别人发到微信来的一条消息，请以你本人（火星人）的口吻，写一条你会真的发出去的"
  114 + "微信回复。只输出这条回复本身，不要解释、不要引号、不要用任何工具、不要真的发送。\n\n"
  115 + "【上下文】\n" + context_text)
  116 + env = dict(os.environ)
  117 + env.update({
  118 + "HOME": "/data/data/com.termux/files/home",
  119 + "PREFIX": "/data/data/com.termux/files/usr",
  120 + "PATH": "/data/data/com.termux/files/usr/bin:/system/bin",
  121 + "LD_LIBRARY_PATH": "/data/data/com.termux/files/usr/lib",
  122 + })
  123 + try:
  124 + p = subprocess.run([HERMES, "-z", prompt], capture_output=True, text=True,
  125 + timeout=GEN_TIMEOUT, env=env)
  126 + out = (p.stdout or "").strip()
  127 + # 取最后一段非空文本（躲开可能的日志行）
  128 + return out.split("\n")[-1].strip() if out else ""
  129 + except subprocess.TimeoutExpired:
  130 + log("生成超时")
  131 + return ""
  132 + except Exception as e: # noqa: BLE001
  133 + log(f"生成异常: {e}")
  134 + return ""
  135 +
  136 +
  137 +# ── env4→5：落地 ─────────────────────────────────────
  138 +def _load_whitelist():
  139 + contacts, groups = [], []
  140 + try:
  141 + con = sqlite3.connect(CIRCLE_DB)
  142 + for kind, wxid in con.execute("SELECT kind, wxid FROM reply_whitelist"):
  143 + (groups if kind == "group" else contacts).append(wxid)
  144 + con.close()
  145 + except Exception:
  146 + pass
  147 + return contacts, groups
  148 +
  149 +
  150 +def _queue_insert(event, draft, sender_disp, action, reason, status):
  151 + con = sqlite3.connect(CIRCLE_DB)
  152 + con.execute(
  153 + """INSERT INTO reply_queue
  154 + (msg_id, talker, sender, sender_name, kind, incoming, draft, action, reason, status, sent_at)
  155 + VALUES (?,?,?,?,?,?,?,?,?,?, CASE WHEN ?='auto_sent' THEN datetime('now','localtime') ELSE NULL END)""",
  156 + (event.get("msgId"), event.get("talker"), event.get("sender"), sender_disp,
  157 + event.get("kind"), event.get("body"), draft, action, reason, status, status))
  158 + con.commit()
  159 + con.close()
  160 +
  161 +
  162 +def _send_text(to_wxid, text):
  163 + sys.path.insert(0, EVENT_BUS)
  164 + from wechat_hook_send import send_text_via_hook
  165 + return send_text_via_hook(to_wxid, text)
  166 +
  167 +
  168 +def process_event(event, stage, key, self_wxid, cfg):
  169 + ctx, sender_disp = build_context(stage, key, event, self_wxid)
  170 + draft = generate(ctx)
  171 + d = policy.decide(event, draft, self_wxid=self_wxid, cfg=cfg)
  172 + action, reason = d["action"], d["reason"]
  173 + if action == "skip":
  174 + log(f"skip #{event.get('msgId')} {sender_disp}: {reason}")
  175 + return
  176 + if action == "auto":
  177 + res = _send_text(event.get("talker"), draft)
  178 + ok = bool(res.get("ok") or res.get("status") == "sent")
  179 + _queue_insert(event, draft, sender_disp, "auto",
  180 + reason, "auto_sent" if ok else "failed")
  181 + log(f"auto {'OK' if ok else 'FAIL'} #{event.get('msgId')} {sender_disp} → {draft!r}")
  182 + else: # review
  183 + _queue_insert(event, draft, sender_disp, "review", reason, "pending")
  184 + log(f"review #{event.get('msgId')} {sender_disp}: {reason} → 草稿 {draft!r}")
  185 +
  186 +
  187 +# ── drain：消费 incoming.jsonl ─────────────────────────
  188 +def _read_events():
  189 + evs = []
  190 + try:
  191 + with open(QUEUE_FILE) as f:
  192 + for line in f:
  193 + line = line.strip()
  194 + if line:
  195 + try:
  196 + evs.append(json.loads(line))
  197 + except ValueError:
  198 + pass
  199 + except OSError:
  200 + pass
  201 + return evs
  202 +
  203 +
  204 +def drain():
  205 + """非阻塞单实例：已有 drain 在跑就跳过（后续消息的 inotify 会再触发）。"""
  206 + lock = f"{WORK_DIR}/orch.lock"
  207 + try:
  208 + fd = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
  209 + os.close(fd)
  210 + except FileExistsError:
  211 + try: # 5min 以上视为残留锁，抢占
  212 + if time.time() - os.path.getmtime(lock) > 300:
  213 + os.remove(lock)
  214 + os.close(os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY))
  215 + else:
  216 + return
  217 + except OSError:
  218 + return
  219 + try:
  220 + _drain_inner()
  221 + finally:
  222 + try:
  223 + os.remove(lock)
  224 + except OSError:
  225 + pass
  226 +
  227 +
  228 +def _drain_inner():
  229 + evs = _read_events()
  230 + if not evs:
  231 + return
  232 + try:
  233 + with open(CURSOR_FILE) as f:
  234 + cursor = json.load(f).get("last_msgId", 0)
  235 + except (OSError, ValueError):
  236 + cursor = None
  237 + max_id = max((int(e.get("msgId", 0)) for e in evs), default=0)
  238 + if cursor is None: # 首跑：基线，不回放
  239 + with open(CURSOR_FILE, "w") as f:
  240 + json.dump({"last_msgId": max_id}, f)
  241 + w.chown_to_hermes(CURSOR_FILE)
  242 + log(f"编排器首跑基线 last_msgId={max_id}")
  243 + return
  244 + todo = [e for e in evs if int(e.get("msgId", 0)) > cursor
  245 + and e.get("kind") in CANDIDATE_KINDS]
  246 + if todo:
  247 + _, db, key = w.account_paths()
  248 + stage = w.copy_db(db)
  249 + try:
  250 + self_wxid, _ = w.detect_self(stage, key)
  251 + cfg = policy.load_config()
  252 + cfg["whitelist_contacts"], cfg["whitelist_groups"] = _load_whitelist()
  253 + log(f"drain：{len(todo)} 条候选（auto_send={cfg['auto_send_enabled']}）")
  254 + for e in todo:
  255 + try:
  256 + process_event(e, stage, key, self_wxid, cfg)
  257 + except Exception as ex: # noqa: BLE001
  258 + log(f"处理 #{e.get('msgId')} 失败: {ex}")
  259 + finally:
  260 + for p in __import__("glob").glob(f"{WORK_DIR}/_stage/en.db*"):
  261 + try:
  262 + os.remove(p)
  263 + except OSError:
  264 + pass
  265 + with open(CURSOR_FILE, "w") as f:
  266 + json.dump({"last_msgId": max_id}, f)
  267 + w.chown_to_hermes(CURSOR_FILE)
  268 +
  269 +
  270 +def main():
  271 + args = sys.argv[1:]
  272 + if "--drain" in args:
  273 + drain()
  274 + else:
  275 + print("用法: wechat_reply_orchestrator.py --drain")
  276 +
  277 +
  278 +if __name__ == "__main__":
  279 + main()
... ...
wechat_plain/scripts/auto-reply/wechat_reply_watcher.py
  View file @20815fb
... ... @@ -310,12 +310,29 @@ def daemon():
310 310 _watch_loop_mtime(db)
311 311  
312 312  
  313 +def _trigger_orchestrator():
  314 + """读到新消息后 fire-and-forget 触发编排器 drain（生成/策略/发送在独立进程，
  315 + 不阻塞 watcher；编排器自身有单实例锁）。缺文件/异常都不影响 watcher。"""
  316 + orch = os.path.join(os.path.dirname(os.path.abspath(__file__)),
  317 + "wechat_reply_orchestrator.py")
  318 + if not os.path.exists(orch):
  319 + return
  320 + try:
  321 + import sys
  322 + subprocess.Popen([sys.executable, orch, "--drain"],
  323 + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
  324 + start_new_session=True)
  325 + except Exception as e: # noqa: BLE001
  326 + log(f"触发编排器失败: {e}")
  327 +
  328 +
313 329 def _process(reason):
314 330 try:
315 331 evs = read_new()
316 332 if evs:
317 333 kinds = ",".join(e["kind"] for e in evs)
318 334 log(f"[{reason}] +{len(evs)} 条 → {kinds} (至 msgId {evs[-1]['msgId']})")
  335 + _trigger_orchestrator()
319 336 except Exception as e: # noqa: BLE001 — daemon 不能因单次读失败退出
320 337 log(f"[{reason}] 读失败: {e}")
321 338  
... ...