diff --git a/md/wechat-auto-reply-requirements.md b/md/wechat-auto-reply-requirements.md index 45c70e4..50ee38f 100644 --- a/md/wechat-auto-reply-requirements.md +++ b/md/wechat-auto-reply-requirements.md @@ -93,6 +93,13 @@ - 落地新增：①「待回复队列」存储（sqlite/json）；② wechat viewer 的"发送"按钮接到发送链路。 - 安全红线（贯穿）：涉钱 / 承诺 / 约时间 / 低置信 → 即使白名单也建议进审核。 +### 编排器 · 把 1→5 串起来 —— ✅ **已实现并真机验证（2026-07-10）** +`wechat_plain/scripts/auto-reply/wechat_reply_orchestrator.py`（设备 `~/.hermes/wechat_reply/`）： +- watcher 读到新消息 → 写 `incoming.jsonl` → **fire-and-forget 触发** `orchestrator --drain`（watcher `_trigger_orchestrator`，独立进程不阻塞采集）。 +- 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)`。 +- 安全：生成超时/失败 → 无草稿 → policy skip → 不发（失败即静默，绝不误发）。 +- 真机验证：接线（watcher inotify → 自动拉起编排器 drain）通过；真实上下文（猎头 婷婷）生成「哈哈婷婷，我不是什么总，就是个写代码的，不负责招人这块哈」→ 正确进 /approvals 待审。 + ### 环节 5 · 发送回复带引用（原生气泡） 需求 = 点原文→引用→发送后显示引用气泡。两条实现路： - **(a) A11y UI 自动化（一期推荐）**：长按目标气泡 → 点"引用" → 输入 → 发送。驱动微信自己 UI → **产出货真价实的原生引用气泡，零逆向**。代价：需微信前台、脚本略脆、比 API 慢。已有 A11y 发送服务底子。 diff --git a/wechat_plain/scripts/auto-reply/wechat_reply_orchestrator.py b/wechat_plain/scripts/auto-reply/wechat_reply_orchestrator.py new file mode 100644 index 0000000..a8e6660 --- /dev/null +++ b/wechat_plain/scripts/auto-reply/wechat_reply_orchestrator.py @@ -0,0 +1,279 @@ +#!/usr/bin/env python3 +"""环节 1→5 编排器 —— 把整条自动回复链串起来 + +消费 watcher 的 incoming.jsonl（cursor 增量）→ 对候选事件（私聊/群@我/群引用我）： + env2 查上下文（发送者身份 + 最近历史）→ env3 hermes -z 生成草稿 → + env4 policy.decide()（白名单读 wechat-viewer 的 circle.db）→ + env5 落地：auto 直接 send_text_via_hook + 入队(auto_sent) / review 入队(pending，在 /approvals 等人工) + +作为 root 在 Termux 跑（同 watcher 上下文）。由 watcher 每次读到新消息后触发 --drain。 + +安全：policy 默认 auto_send_enabled=False → 一切进 review（不自动发）。生成超时/失败 → +无草稿 → policy skip → 不发（失败即静默，绝不误发）。 + +ponytail: 每次 drain 拷一次整库(~96MB)供该批所有事件查上下文；事件稀疏可接受。 +升级：与 watcher 共享一次拷贝；incoming.jsonl 增长需按 offset 游标+轮转。 +""" +import json +import os +import sqlite3 +import subprocess +import sys +import time + +_HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, _HERE) +import wechat_reply_watcher as w # 复用解密/查询/账号推导 +import wechat_reply_policy as policy + +WORK_DIR = w.WORK_DIR +QUEUE_FILE = f"{WORK_DIR}/incoming.jsonl" +CURSOR_FILE = f"{WORK_DIR}/orch_state.json" +ORCH_LOG = f"{WORK_DIR}/orchestrator.log" +CIRCLE_DB = "/data/data/com.termux/files/home/wechat-viewer/circle.db" +EVENT_BUS = "/data/data/com.termux/files/home/.hermes/event_bus" +HERMES = "/data/data/com.termux/files/usr/bin/hermes" + +CANDIDATE_KINDS = {"private", "group_at_me", "group_quote_me"} +HISTORY_N = 12 +GEN_TIMEOUT = 160 + + +def log(msg): + line = f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}" + print(line, flush=True) + try: + with open(ORCH_LOG, "a") as f: + f.write(line + "\n") + except OSError: + pass + + +# ── env2：查上下文 ───────────────────────────────────── +def _display_name(stage, key, wxid): + if not wxid: + return "" + rows = w.sqlcipher_query( + stage, key, + f"SELECT conRemark, nickname, alias FROM rcontact WHERE username='{wxid}';") + if rows and rows[0]: + r = rows[0] + ["", "", ""] + return r[0] or r[1] or r[2] or wxid + return wxid + + +def _placeholder(mtype, text): + if mtype == 1: + return text + return {3: "[图片]", 34: "[语音]", 43: "[视频]", 47: "[表情]", + 42: "[名片]", 49: "[链接/文件]", 48: "[位置]"}.get(mtype, text[:40] if text else "[消息]") + + +def build_context(stage, key, event, self_wxid): + talker = event.get("talker", "") + is_group = event.get("isGroup") or talker.endswith("@chatroom") + sender_wxid = event.get("sender") or talker + sender_disp = _display_name(stage, key, sender_wxid) + lines = [f"对方：{sender_disp}"] + if is_group: + gname = _display_name(stage, key, talker) + lines.insert(0, f"[群] {gname}") + # 最近历史 + rows = w.sqlcipher_query( + stage, key, + "SELECT isSend, type, createTime, hex(content) FROM message " + f"WHERE talker='{talker}' ORDER BY msgId DESC LIMIT {HISTORY_N};") + hist = [] + for r in reversed(rows): + if len(r) < 4: + continue + is_send, mtype, _, chex = r + try: + raw = bytes.fromhex(chex).decode("utf-8", "ignore") + except ValueError: + raw = "" + if is_send == "1": + who, body = "我", raw + else: + s, body = w.parse_group_content(raw) if is_group else (sender_wxid, raw) + who = _display_name(stage, key, s) if is_group and s else sender_disp + txt = _placeholder(int(mtype) if mtype.isdigit() else 0, body).replace("\n", " ").strip() + if txt: + hist.append(f" {who}: {txt}") + ctx = "\n".join(lines) + if hist: + ctx += "\n最近对话：\n" + "\n".join(hist[-HISTORY_N:]) + ctx += f"\n本条：{sender_disp}: {(event.get('body') or '').strip()[:200]}" + return ctx, sender_disp + + +# ── env3：hermes 生成 ───────────────────────────────── +def generate(context_text): + prompt = ( + "【任务】下面是别人发到微信来的一条消息，请以你本人（火星人）的口吻，写一条你会真的发出去的" + "微信回复。只输出这条回复本身，不要解释、不要引号、不要用任何工具、不要真的发送。\n\n" + "【上下文】\n" + context_text) + env = dict(os.environ) + env.update({ + "HOME": "/data/data/com.termux/files/home", + "PREFIX": "/data/data/com.termux/files/usr", + "PATH": "/data/data/com.termux/files/usr/bin:/system/bin", + "LD_LIBRARY_PATH": "/data/data/com.termux/files/usr/lib", + }) + try: + p = subprocess.run([HERMES, "-z", prompt], capture_output=True, text=True, + timeout=GEN_TIMEOUT, env=env) + out = (p.stdout or "").strip() + # 取最后一段非空文本（躲开可能的日志行） + return out.split("\n")[-1].strip() if out else "" + except subprocess.TimeoutExpired: + log("生成超时") + return "" + except Exception as e: # noqa: BLE001 + log(f"生成异常: {e}") + return "" + + +# ── env4→5：落地 ───────────────────────────────────── +def _load_whitelist(): + contacts, groups = [], [] + try: + con = sqlite3.connect(CIRCLE_DB) + for kind, wxid in con.execute("SELECT kind, wxid FROM reply_whitelist"): + (groups if kind == "group" else contacts).append(wxid) + con.close() + except Exception: + pass + return contacts, groups + + +def _queue_insert(event, draft, sender_disp, action, reason, status): + con = sqlite3.connect(CIRCLE_DB) + con.execute( + """INSERT INTO reply_queue + (msg_id, talker, sender, sender_name, kind, incoming, draft, action, reason, status, sent_at) + VALUES (?,?,?,?,?,?,?,?,?,?, CASE WHEN ?='auto_sent' THEN datetime('now','localtime') ELSE NULL END)""", + (event.get("msgId"), event.get("talker"), event.get("sender"), sender_disp, + event.get("kind"), event.get("body"), draft, action, reason, status, status)) + con.commit() + con.close() + + +def _send_text(to_wxid, text): + sys.path.insert(0, EVENT_BUS) + from wechat_hook_send import send_text_via_hook + return send_text_via_hook(to_wxid, text) + + +def process_event(event, stage, key, self_wxid, cfg): + ctx, sender_disp = build_context(stage, key, event, self_wxid) + draft = generate(ctx) + d = policy.decide(event, draft, self_wxid=self_wxid, cfg=cfg) + action, reason = d["action"], d["reason"] + if action == "skip": + log(f"skip #{event.get('msgId')} {sender_disp}: {reason}") + return + if action == "auto": + res = _send_text(event.get("talker"), draft) + ok = bool(res.get("ok") or res.get("status") == "sent") + _queue_insert(event, draft, sender_disp, "auto", + reason, "auto_sent" if ok else "failed") + log(f"auto {'OK' if ok else 'FAIL'} #{event.get('msgId')} {sender_disp} → {draft!r}") + else: # review + _queue_insert(event, draft, sender_disp, "review", reason, "pending") + log(f"review #{event.get('msgId')} {sender_disp}: {reason} → 草稿 {draft!r}") + + +# ── drain：消费 incoming.jsonl ───────────────────────── +def _read_events(): + evs = [] + try: + with open(QUEUE_FILE) as f: + for line in f: + line = line.strip() + if line: + try: + evs.append(json.loads(line)) + except ValueError: + pass + except OSError: + pass + return evs + + +def drain(): + """非阻塞单实例：已有 drain 在跑就跳过（后续消息的 inotify 会再触发）。""" + lock = f"{WORK_DIR}/orch.lock" + try: + fd = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY) + os.close(fd) + except FileExistsError: + try: # 5min 以上视为残留锁，抢占 + if time.time() - os.path.getmtime(lock) > 300: + os.remove(lock) + os.close(os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY)) + else: + return + except OSError: + return + try: + _drain_inner() + finally: + try: + os.remove(lock) + except OSError: + pass + + +def _drain_inner(): + evs = _read_events() + if not evs: + return + try: + with open(CURSOR_FILE) as f: + cursor = json.load(f).get("last_msgId", 0) + except (OSError, ValueError): + cursor = None + max_id = max((int(e.get("msgId", 0)) for e in evs), default=0) + if cursor is None: # 首跑：基线，不回放 + with open(CURSOR_FILE, "w") as f: + json.dump({"last_msgId": max_id}, f) + w.chown_to_hermes(CURSOR_FILE) + log(f"编排器首跑基线 last_msgId={max_id}") + return + todo = [e for e in evs if int(e.get("msgId", 0)) > cursor + and e.get("kind") in CANDIDATE_KINDS] + if todo: + _, db, key = w.account_paths() + stage = w.copy_db(db) + try: + self_wxid, _ = w.detect_self(stage, key) + cfg = policy.load_config() + cfg["whitelist_contacts"], cfg["whitelist_groups"] = _load_whitelist() + log(f"drain：{len(todo)} 条候选（auto_send={cfg['auto_send_enabled']}）") + for e in todo: + try: + process_event(e, stage, key, self_wxid, cfg) + except Exception as ex: # noqa: BLE001 + log(f"处理 #{e.get('msgId')} 失败: {ex}") + finally: + for p in __import__("glob").glob(f"{WORK_DIR}/_stage/en.db*"): + try: + os.remove(p) + except OSError: + pass + with open(CURSOR_FILE, "w") as f: + json.dump({"last_msgId": max_id}, f) + w.chown_to_hermes(CURSOR_FILE) + + +def main(): + args = sys.argv[1:] + if "--drain" in args: + drain() + else: + print("用法: wechat_reply_orchestrator.py --drain") + + +if __name__ == "__main__": + main() diff --git a/wechat_plain/scripts/auto-reply/wechat_reply_watcher.py b/wechat_plain/scripts/auto-reply/wechat_reply_watcher.py index f2ba86e..74a3663 100644 --- a/wechat_plain/scripts/auto-reply/wechat_reply_watcher.py +++ b/wechat_plain/scripts/auto-reply/wechat_reply_watcher.py @@ -310,12 +310,29 @@ def daemon(): _watch_loop_mtime(db) +def _trigger_orchestrator(): + """读到新消息后 fire-and-forget 触发编排器 drain（生成/策略/发送在独立进程， + 不阻塞 watcher；编排器自身有单实例锁）。缺文件/异常都不影响 watcher。""" + orch = os.path.join(os.path.dirname(os.path.abspath(__file__)), + "wechat_reply_orchestrator.py") + if not os.path.exists(orch): + return + try: + import sys + subprocess.Popen([sys.executable, orch, "--drain"], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + start_new_session=True) + except Exception as e: # noqa: BLE001 + log(f"触发编排器失败: {e}") + + def _process(reason): try: evs = read_new() if evs: kinds = ",".join(e["kind"] for e in evs) log(f"[{reason}] +{len(evs)} 条 → {kinds} (至 msgId {evs[-1]['msgId']})") + _trigger_orchestrator() except Exception as e: # noqa: BLE001 — daemon 不能因单次读失败退出 log(f"[{reason}] 读失败: {e}")