#!/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()