diff --git a/wechat_plain/scripts/auto-reply/wechat-viewer-watchdog.sh b/wechat_plain/scripts/auto-reply/wechat-viewer-watchdog.sh new file mode 100755 index 0000000..b81d6ad --- /dev/null +++ b/wechat_plain/scripts/auto-reply/wechat-viewer-watchdog.sh @@ -0,0 +1,91 @@ +#!/data/data/com.termux/files/usr/bin/bash +# WeChat Viewer 看门狗 — 检查 Flask + Hermes Gateway + 微信 Watcher 是否存活 +# 由 Hermes cron 每 5 分钟调用一次 + +export PATH=/data/data/com.termux/files/usr/bin:$PATH + +FLASK_PORT=5001 +APP_DIR=/data/data/com.termux/files/home/wechat-viewer +LOG=$APP_DIR/watchdog.log +HOME_DIR=/data/data/com.termux/files/home + +# ── 1. Flask ────────────────────────────────────── + +ALIVE=$(su -c "ss -tlnp" 2>/dev/null | grep -c ":$FLASK_PORT") + +if [ "$ALIVE" -eq 0 ]; then + echo "[$(date)] Flask 不在线，重启中..." | tee -a "$LOG" + cd "$APP_DIR" + pkill -f 'python3 app.py' 2>/dev/null + sleep 1 + fuser -k "$FLASK_PORT/tcp" 2>/dev/null + nohup python3 app.py > $APP_DIR/flask.log 2>&1 & + sleep 3 + ALIVE2=$(su -c "ss -tlnp" 2>/dev/null | grep -c ":$FLASK_PORT") + if [ "$ALIVE2" -gt 0 ]; then + echo "[$(date)] Flask 重启成功" | tee -a "$LOG" + else + echo "[$(date)] Flask 重启失败！" | tee -a "$LOG" + fi +fi + +# ── 2. Hermes Gateway ────────────────────────────── + +GATEWAY_ALIVE=$(pgrep -f "hermes.*gateway" 2>/dev/null | wc -l) + +if [ "$GATEWAY_ALIVE" -eq 0 ]; then + echo "[$(date)] Gateway 不在线，重启中..." | tee -a "$LOG" + nohup hermes gateway > /tmp/hermes-gateway.log 2>&1 & + sleep 3 + GATEWAY_ALIVE2=$(pgrep -f "hermes.*gateway" 2>/dev/null | wc -l) + if [ "$GATEWAY_ALIVE2" -gt 0 ]; then + echo "[$(date)] Gateway 重启成功" | tee -a "$LOG" + else + echo "[$(date)] Gateway 重启失败！" | tee -a "$LOG" + fi +fi + +# ── 3. Stuck Job Reaper ────────────────────────── +# 清理因 Flask 重启而孤儿化的 hermes_jobs +python3 "$APP_DIR/reap_stuck_jobs.py" >> "$LOG" 2>&1 + +# ── 4. Orphan Report Reaper ───────────────────── +# 清理因 Flask 重启丢回调而未入库的日报文件 +python3 "$APP_DIR/reap_orphan_reports.py" >> "$LOG" 2>&1 + +# ── 5. 微信自动回复 Watcher ──────────────────────── +# 监听微信 DB 变化，写入 incoming.jsonl 队列 +# 单例保护：每次都先 pkill 清场再起一个，不能只信一次 ps|grep 的探测结果—— +# 探测在系统过载时会漏检/超时返回空，若不清场会在卡死的旧实例上再叠一个， +# 雪球式堆积出几百个进程拖垮整机（2026-07-25 事故：堆到 877 个吃满内存+swap）。 + +REPLY_DIR=$HOME_DIR/.hermes/wechat_reply +WATCHER_COUNT=$(su -c 'ps -A' 2>/dev/null | grep -c '[w]echat_reply_watcher') + +if [ "$WATCHER_COUNT" -ne 1 ]; then + echo "[$(date)] 微信 Watcher 存活数=$WATCHER_COUNT(期望 1)，重置中..." | tee -a "$LOG" + su -c "pkill -f wechat_reply_watcher.py" 2>/dev/null + sleep 1 + su -c "/data/data/com.termux/files/usr/bin/python3 $REPLY_DIR/wechat_reply_watcher.py >> $REPLY_DIR/watcher.log 2>&1 &" + sleep 2 + WATCHER_COUNT2=$(su -c 'ps -A' 2>/dev/null | grep -c '[w]echat_reply_watcher') + if [ "$WATCHER_COUNT2" -eq 1 ]; then + echo "[$(date)] 微信 Watcher 重置成功" | tee -a "$LOG" + else + echo "[$(date)] 微信 Watcher 重置后存活数异常: $WATCHER_COUNT2" | tee -a "$LOG" + fi +fi + +# ── 6. 微信自动回复 Orchestrator Drain ────────────── +# 消费 incoming.jsonl 队列，处理候选事件 + +ORCH_LOCK=$REPLY_DIR/orch.lock +# 清理超过 5 分钟的残留锁 +if [ -f "$ORCH_LOCK" ]; then + LOCK_AGE=$(($(date +%s) - $(stat -c %Y "$ORCH_LOCK" 2>/dev/null || echo 0))) + if [ "$LOCK_AGE" -gt 300 ]; then + rm -f "$ORCH_LOCK" + fi +fi + +su -c "/data/data/com.termux/files/usr/bin/python3 $REPLY_DIR/wechat_reply_orchestrator.py --drain >> $REPLY_DIR/orchestrator.log 2>&1" diff --git a/wechat_plain/scripts/auto-reply/wechat_reply_orchestrator.py b/wechat_plain/scripts/auto-reply/wechat_reply_orchestrator.py index db07e38..ebd5f4a 100644 --- a/wechat_plain/scripts/auto-reply/wechat_reply_orchestrator.py +++ b/wechat_plain/scripts/auto-reply/wechat_reply_orchestrator.py @@ -2,17 +2,11 @@ """环节 1→5 编排器 —— 把整条自动回复链串起来 消费 watcher 的 incoming.jsonl（cursor 增量）→ 对候选事件（私聊/群@我/群引用我）： - env2 查上下文（发送者身份 + 最近历史）→ env3 hermes -z 生成草稿 → + env2 查上下文（发送者身份 + 最近历史）→ env3 hermes 生成草稿 → env4 policy.decide()（白名单读 wechat-viewer 的 circle.db）→ - env5 落地：auto 直接 send_text_via_hook + 入队(auto_sent) / review 入队(pending，在 /approvals 等人工) + env5 落地：auto 直接 send_text_via_hook + 入队(auto_sent) / review 入队(pending) -作为 root 在 Termux 跑（同 watcher 上下文）。由 watcher 每次读到新消息后触发 --drain。 - -安全：policy 默认 auto_send_enabled=False → 一切进 review（不自动发）。生成超时/失败 → -无草稿 → policy skip → 不发（失败即静默，绝不误发）。 - -ponytail: 每次 drain 拷一次整库(~96MB)供该批所有事件查上下文；事件稀疏可接受。 -升级：与 watcher 共享一次拷贝；incoming.jsonl 增长需按 offset 游标+轮转。 +策略权威参考：wechat-auto-reply skill (v3.0)。代码是参考实现，skill 定义逻辑。 """ import json import os @@ -36,7 +30,30 @@ HERMES = "/data/data/com.termux/files/usr/bin/hermes" CANDIDATE_KINDS = {"private", "group_at_me", "group_quote_me"} HISTORY_N = 12 -GEN_TIMEOUT = 160 +GEN_TIMEOUT = 600 # admin 任务超时（10分钟），GFW 下 GitHub 访问慢 +PROCESSED_FILE = f"{WORK_DIR}/processed_ids.json" +# 去重：已处理的 msgId 集合。drain 时加载，处理后持久化 +_processed_ids = None + + +def _load_processed(): + global _processed_ids + if _processed_ids is not None: + return _processed_ids + try: + with open(PROCESSED_FILE) as f: + _processed_ids = set(json.load(f)) + except (OSError, ValueError): + _processed_ids = set() + return _processed_ids + + +def _save_processed(): + try: + with open(PROCESSED_FILE, "w") as f: + json.dump(list(_processed_ids or set()), f) + except OSError: + pass def log(msg): @@ -108,11 +125,43 @@ def build_context(stage, key, event, self_wxid): # ── env3：hermes 生成 ───────────────────────────────── -def generate(context_text): - prompt = ( - "【任务】下面是别人发到微信来的一条消息，请以你本人（火星人）的口吻，写一条你会真的发出去的" - "微信回复。只输出这条回复本身，不要解释、不要引号、不要用任何工具、不要真的发送。\n\n" - "【上下文】\n" + context_text) +def generate(context_text, event=None, role="member"): + """调用 Hermes 生成回复。 + + role='admin' → hermes chat（完整 CLI，可调工具/skill/API，执行复杂任务） + role='member' → hermes -z（零工具纯文本补全，简单闲聊回复） + """ + talker = (event or {}).get("talker", "") + body = (event or {}).get("body", "") or "" + is_group = talker.endswith("@chatroom") + + if role == "admin": + is_task = _is_task_request(body) + result_file = f"{WORK_DIR}/_reply_result_{int(time.time())}.txt" + if is_task: + prompt = ( + f"你是火星人的微信助手。用 reply-task-executor skill 执行任务。\n\n" + f"任务：{body}\n\n" + f"上下文：\n{context_text}\n\n" + f"重要：先执行任务（调工具、写文件、调API），不要先写报告。\n" + f"任务完成后，把结果报告写入 {result_file}。" + ) + else: + prompt = ( + f"你是火星人的微信助手。有人发了一条消息，分析内容，以火星人口吻回复。\n\n" + f"【消息内容】{body}\n\n" + f"【上下文】\n{context_text}\n\n" + f"把你要发送的回复文本写入文件 {result_file}。只写回复本身，不要解释。" + ) + cmd = [HERMES, "chat", "-q", prompt, "-s", "reply-task-executor", "--yolo", "-t", "terminal,file,web"] + else: + # 简单模式：零工具，纯粹文本补全 + prompt = ( + "【任务】下面是别人发到微信来的一条消息，请以你本人（火星人）的口吻，写一条你会真的发出去的" + "微信回复。只输出这条回复本身，不要解释、不要引号、不要用任何工具、不要真的发送。\n\n" + "【上下文】\n" + context_text) + cmd = [HERMES, "-z", prompt] + env = dict(os.environ) env.update({ "HOME": "/data/data/com.termux/files/home", @@ -121,40 +170,114 @@ def generate(context_text): "LD_LIBRARY_PATH": "/data/data/com.termux/files/usr/lib", }) try: - p = subprocess.run([HERMES, "-z", prompt], capture_output=True, text=True, + p = subprocess.run(cmd, capture_output=True, text=True, timeout=GEN_TIMEOUT, env=env) - out = (p.stdout or "").strip() - # 取最后一段非空文本（躲开可能的日志行） - return out.split("\n")[-1].strip() if out else "" + # admin 模式从文件读结果，不用 stdout + if role == "admin" and result_file: + try: + with open(result_file) as f: + draft = f.read().strip() + try: os.remove(result_file) + except: pass + except OSError: + draft = "" + else: + # member 模式：从 stdout 取最后一行 + out = (p.stdout or "").strip() + draft = out.split("\n")[-1].strip() if out else "" + stderr = (p.stderr or "")[:200] + return draft, p.returncode, stderr, result_file if role == "admin" else "" except subprocess.TimeoutExpired: log("生成超时") - return "" - except Exception as e: # noqa: BLE001 + return "", -1, "timeout", result_file if role == "admin" else "" + except Exception as e: log(f"生成异常: {e}") - return "" - - -# ── env4→5：落地 ───────────────────────────────────── + return "", -1, str(e)[:200], result_file if role == "admin" else "" + + +# ── V 层：独立验证 ────────────────────────────────── +def _verify_draft(draft, role="member"): + """独立验证回复质量。不能由 agent 自己宣布完成。 + + admin 生成的回复需要更严格检查（因为可能调了错误工具/产生幻觉）。 + member 的回复比较简单，主要防胡话。 + 返回 (ok: bool, reason: str) + """ + if not draft or not draft.strip(): + return False, "empty_draft" + text = draft.strip() + # 噪音行——hermes 输出里的元数据/分隔线 + if all(c in "─═━═─ \t" for c in text): + return False, "noise_line" + for noise_start in ["Messages:", "Session:", "Duration:", "Tool:", + "Exit code:", "📱", "Model:", "Provider:", + "Tokens:", "Cost:", "$", "hermes ", + "Resume this", "Session saved"]: + if text.startswith(noise_start): + return False, f"noise:{noise_start}" + # 错误标记——hermes 调用失败时可能输出的内容 + for marker in ["[Hermes", "[调用失败", "[超时", "[错误", "[Error", + "抱歉，我无法", "I cannot", "I'm unable"]: + if marker in text: + return False, f"error_marker:{marker}" + # 过短——不太可能是有效回复（admin 尤甚，它应该产出实质性内容） + if role == "admin" and len(text) < 5: + return False, "admin_too_short" + if len(text) < 2: + return False, "too_short" + # admin 回复不应该只是 emoji/单字（说明 hermes 没认真执行） + if role == "admin" and len(text) < 10 and not any('\u4e00' <= c <= '\u9fff' for c in text): + return False, "admin_no_chinese" + return True, "ok" + + +# ── L 层：Task Contract ────────────────────────────── +def _write_contract(event, role, exit_code, output, err): + """admin 的 hermes chat 失败时写 task contract——包含复现所需全部信息。""" + import __main__ + contract_dir = os.path.join(WORK_DIR, "_q") + os.makedirs(contract_dir, exist_ok=True) + contract = { + "ts": time.time(), + "msgId": event.get("msgId"), + "talker": event.get("talker"), + "sender": event.get("sender"), + "body": (event.get("body") or "")[:300], + "role": role, + "exit_code": exit_code, + "output": (output or "")[:500], + "stderr": (err or "")[:200], + } + fname = os.path.join(contract_dir, f"ct_{int(time.time())}_{event.get('msgId')}.json") + try: + with open(fname, "w") as f: + json.dump(contract, f, ensure_ascii=False, indent=2) + w.chown_to_hermes(fname) + log(f"contract written: {fname}") + except Exception as e: # noqa: BLE001 + log(f"contract write failed: {e}") def _load_whitelist(): - contacts, groups = [], [] + """返回 {wxid: role} 映射。role: 'admin' | 'member'。""" + roles = {} 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) + for wxid, role in con.execute("SELECT wxid, COALESCE(role,'member') FROM reply_whitelist"): + roles[wxid] = role or "member" con.close() except Exception: pass - return contacts, groups + return roles -def _queue_insert(event, draft, sender_disp, action, reason, status): +def _queue_insert(event, draft, sender_disp, action, reason, status, result_file="", override_talker=""): + talker = override_talker or event.get("talker") 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)) + (msg_id, talker, sender, sender_name, kind, incoming, draft, result_file, action, reason, status, sent_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?, CASE WHEN ?='auto_sent' THEN datetime('now','localtime') ELSE NULL END)""", + (event.get("msgId"), talker, event.get("sender"), sender_disp, + event.get("kind"), event.get("body"), draft, result_file, action, reason, status, status)) con.commit() con.close() @@ -209,6 +332,73 @@ def _sent_status(db, key, talker, base_msgid, text): return None +def _run_hermes_z(prompt): + """快速 hermes -z 调用（零工具，纯文本补全），用于生成失败报告""" + cmd = [HERMES, "-z", prompt] + 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(cmd, capture_output=True, text=True, + timeout=60, env=env) + return (p.stdout or "").strip(), p.returncode, (p.stderr or "")[:200] + except subprocess.TimeoutExpired: + return "", -1, "timeout" + except Exception as e: + return "", -1, str(e)[:200] + + +# ── 两段式回复：任务识别 + ack 生成 ────────────── + +TASK_KEYWORDS = ["研究", "学习", "查一下", "分析", "帮我", "看看", + "安装", "配置", "搭建", "写一个", "改一下", + "怎么", "什么是", "解释", "总结", "调查", + "research", "install", "setup", "analyze"] + +def _is_task_request(body): + """判断是否是需要后台执行的复杂任务（vs 简单闲聊）""" + text = body.lower() + score = sum(1 for kw in TASK_KEYWORDS if kw in text) + return score >= 1 + +def _gen_ack(ctx, body, sender_disp): + """生成确认收到消息，根据任务内容个性化""" + if "研究" in body or "学习" in body or "看看" in body: + return "收到，我来研究，搞定了给你报告" + if "安装" in body or "配置" in body or "搭建" in body: + return "好的，我来处理，完了给你结果" + if "查" in body or "分析" in body: + return "收到，我查一下，稍等给你报告" + if "写" in body or "改" in body or "帮" in body: + return "好的，我来弄，弄好给你报告" + return "收到，我来处理，有结果跟你说" + + +def _is_complete_report(draft): + """检查 admin 任务报告是否完整（不只是简短回复）""" + if not draft: + return False + text = draft.strip() + # 太短 → 不完整 + if len(text) < 80: + return False + # 只有一句话没有结构 → 不完整 + lines = [l for l in text.split("\n") if l.strip()] + if len(lines) < 2: + return False + # 至少含一项具体信息（数字/URL/skill名/文件路径/工具名） + has_specific = any( + kw in text + for kw in ["http", "github.com", "skill", "memory", "创建", + "安装", "pip", "npm", "实现了", "测试", "验证", + "Star", "star", "⭐"] + ) + return has_specific + def _send_text_verified(to_wxid, text, key): """可靠发送：切前台唤醒微信 → 广播发送 → 轮询 status 直到 =2(确认送达)。 超时仍卡在发送中/失败 → 返回 ok=False,由调用方标 failed 进审核(不假装成功)。 @@ -260,12 +450,89 @@ def _send_text_verified(to_wxid, text, key): def process_event(event, stage, key, self_wxid, cfg): + # ═══ Step 0: 去重 ═══ + msg_id = event.get("msgId") + processed = _load_processed() + if msg_id in processed: + return # 已处理过，跳过 + ctx, sender_disp = build_context(stage, key, event, self_wxid) - draft = generate(ctx) - d = policy.decide(event, draft, self_wxid=self_wxid, cfg=cfg) + # 查 sender 的 role：admin → 完整 Hermes CLI，member → 简单 hermes -z + sender_wxid = event.get("sender") or event.get("talker") or "" + role = cfg.get("whitelist_roles", {}).get(sender_wxid, "member") + + body = (event.get("body") or "").strip() + talker = event.get("talker", "") + + # ═══ 两段式回复：admin 复杂任务先回 ack ═══ + if role == "admin" and _is_task_request(body): + ack = _gen_ack(ctx, body, sender_disp) + ack_text = f"@{sender_disp} {ack}" if event.get("isGroup") and sender_disp else ack + _send_text_verified(talker, ack_text, key) + _queue_insert(event, ack_text, sender_disp, "auto", "ack", "auto_sent") + log(f"ack #{event.get('msgId')} {sender_disp}: {ack!r}") + + # ── 生成回复（V 层：独立验证 + 回退）── + draft, exit_code, stderr, result_file = generate(ctx, event=event, role=role) + ok, vreason = _verify_draft(draft, role) + + # ═══ 报告完整性检查（admin 任务必须产出完整报告） ═══ + if ok and role == "admin" and _is_task_request(body): + if not _is_complete_report(draft): + ok = False + vreason = f"incomplete_report(len={len(draft)})" + + if not ok: + log(f"V-verify FAIL #{event.get('msgId')} role={role}: {vreason} draft={draft!r}") + _write_contract(event, role, exit_code, draft, stderr) + + if role == "admin": + is_task = _is_task_request(body) + if is_task: + # admin 任务失败 → 调 reply-failure-report skill 生成报告，写入队列 + fail_prompt = ( + f"任务内容：{body}\\n" + f"失败原因：{vreason}\\n" + f"对方：{sender_disp}\\n\\n" + f"按 reply-failure-report skill 生成一条要发给对方的失败报告。只输出消息本身。" + ) + fail_draft, _, _ = _run_hermes_z(fail_prompt) + fail_report = (fail_draft or "").strip() + if not fail_report or len(fail_report) < 5: + fail_report = f"这个任务没跑成（{vreason}），我加到 to-do list 了。" + fail_text = f"@{sender_disp} {fail_report}" if event.get("isGroup") and sender_disp else fail_report + # 写入队列，cron 会发送 + _queue_insert(event, fail_text, sender_disp, "auto", + f"task_failed:{vreason}", "failed") + log(f"task failed queued #{event.get('msgId')}: {vreason}") + if msg_id: + loaded = _load_processed() + loaded.add(msg_id) + _save_processed() + return + # admin 闲聊失败 → 回退到 member 模式 + log(f"admin fallback → member mode for #{event.get('msgId')}") + draft, exit_code, stderr, _ = generate(ctx, event=event, role="member") + ok, vreason = _verify_draft(draft, "member") + if not ok: + log(f"V-verify FAIL (member fallback) #{event.get('msgId')}: {vreason}") + _queue_insert(event, draft or "(验证失败)", sender_disp, "auto", + f"verify_failed:{vreason}", "failed") + return + else: + _queue_insert(event, draft or "(验证失败)", sender_disp, "auto", + f"verify_failed:{vreason}", "failed") + return + + d = policy.decide(event, draft, self_wxid=self_wxid, cfg=cfg, sender_role=role) action, reason = d["action"], d["reason"] if action == "skip": log(f"skip #{event.get('msgId')} {sender_disp}: {reason}") + # ═══ 去重持久化（skip 也记录） ═══ + if msg_id: + loaded = _load_processed() + loaded.add(msg_id) + _save_processed() return # 群里回复自动 @ 触发者：昵称前缀 + U+2005（微信原生 @ 分隔符）。安全判定(decide)已在 # 原始 draft 上跑过,前缀只影响发送/展示的文本。 @@ -275,12 +542,28 @@ def process_event(event, stage, key, self_wxid, cfg): if event.get("isGroup") and sender_disp: send_text = f"@{sender_disp} {draft}" if action == "auto": - res = _send_text_verified(event.get("talker"), send_text, key) - ok = res["ok"] - _queue_insert(event, send_text, sender_disp, "auto", - reason, "auto_sent" if ok else "failed") - log(f"auto {'OK ' + res['reason'] if ok else 'FAILED ' + res['reason']} " - f"#{event.get('msgId')} {sender_disp} → {send_text!r}") + is_task = role == "admin" and _is_task_request(body) + if is_task: + # admin 任务结果 → 私聊发 sender，直接发送 + task_talker = sender_wxid if event.get("isGroup") else talker + res = _send_text_verified(task_talker, send_text, key) + ok_send = res["ok"] + _queue_insert(event, send_text, sender_disp, "auto", + reason, "auto_sent" if ok_send else "failed", + result_file=result_file, override_talker=task_talker) + log(f"task result sent #{event.get('msgId')} → {task_talker}: {'OK' if ok_send else 'FAILED'}") + else: + # 闲聊 / member → 直接发送 + res = _send_text_verified(event.get("talker"), send_text, key) + ok_send = res["ok"] + _queue_insert(event, send_text, sender_disp, "auto", + reason, "auto_sent" if ok_send else "failed") + log(f"auto {'OK ' + res['reason'] if ok_send else 'FAILED ' + res['reason']} " + f"#{event.get('msgId')} {sender_disp} → {send_text!r}") + if msg_id: + loaded = _load_processed() + loaded.add(msg_id) + _save_processed() else: # review _queue_insert(event, send_text, sender_disp, "review", reason, "pending") log(f"review #{event.get('msgId')} {sender_disp}: {reason} → 草稿 {send_text!r}") @@ -351,7 +634,10 @@ def _drain_inner(): try: self_wxid, _ = w.detect_self(stage, key) cfg = policy.load_config() - cfg["whitelist_contacts"], cfg["whitelist_groups"] = _load_whitelist() + cfg["whitelist_roles"] = _load_whitelist() + # backward compat: policy.decide still checks these lists for whitelist membership + cfg["whitelist_contacts"] = [w for w, r in cfg["whitelist_roles"].items() if not w.endswith("@chatroom")] + cfg["whitelist_groups"] = [w for w, r in cfg["whitelist_roles"].items() if w.endswith("@chatroom")] log(f"drain：{len(todo)} 条候选（auto_send={cfg['auto_send_enabled']}）") for e in todo: try: diff --git a/wechat_plain/scripts/auto-reply/wechat_reply_policy.py b/wechat_plain/scripts/auto-reply/wechat_reply_policy.py index b98aef0..5cb82c8 100644 --- a/wechat_plain/scripts/auto-reply/wechat_reply_policy.py +++ b/wechat_plain/scripts/auto-reply/wechat_reply_policy.py @@ -64,8 +64,12 @@ def _hit_keyword(text, keywords): return None -def decide(event, draft, confidence=None, should_reply=True, self_wxid="", cfg=None): - """核心判定。返回 {action, reason, hit?}。纯函数、无副作用（写队列在 route()）。""" +def decide(event, draft, confidence=None, should_reply=True, self_wxid="", cfg=None, sender_role="member"): + """核心判定。返回 {action, reason, hit?}。纯函数、无副作用（写队列在 route()）。 + + sender_role='admin' → 允许执行命令/任务类消息（跳过部分敏感词、自动发送） + sender_role='member' → 默认保守策略 + """ cfg = cfg or load_config() kind = event.get("kind", "") sender = event.get("sender") or event.get("talker") or "" @@ -89,15 +93,19 @@ def decide(event, draft, confidence=None, should_reply=True, self_wxid="", cfg=N return {"action": "skip", "reason": f"nontext:非文本(type={mtype})且无草稿"} return {"action": "skip", "reason": "empty:草稿为空"} - # ── 硬安全override：命中敏感词一律 review（白名单也拉回）── + # ── 硬安全override：命中敏感词一律 review（admin 豁免——可执行命令/任务）── hit = _hit_keyword(body + "\n" + (draft or ""), cfg["block_keywords"]) - if hit: + if hit and sender_role != "admin": return {"action": "review", "reason": f"sensitive:命中敏感词「{hit}」", "hit": hit} # ── 低置信 → review ── if confidence is not None and confidence < cfg["min_confidence"]: return {"action": "review", "reason": f"lowconf:置信{confidence}<{cfg['min_confidence']}"} + # ── admin + 总开关 → 直接 auto（信任，无需白名单匹配）── + if sender_role == "admin" and cfg["auto_send_enabled"]: + return {"action": "auto", "reason": "admin:管理员信任→直接发"} + # ── 白名单 + 总开关 → auto ── is_group = event.get("isGroup") or str(talker).endswith("@chatroom") whitelisted = ( diff --git a/wechat_plain/scripts/auto-reply/wechat_reply_watcher.py b/wechat_plain/scripts/auto-reply/wechat_reply_watcher.py index 5ab85cb..f430cb5 100644 --- a/wechat_plain/scripts/auto-reply/wechat_reply_watcher.py +++ b/wechat_plain/scripts/auto-reply/wechat_reply_watcher.py @@ -73,6 +73,7 @@ def md5hex(s: str) -> str: # ── 账号 / key 推导 ──────────────────────────────────────────────────────── def read_uin() -> str: + # 先尝试直接 glob（root 环境下有效） for xml in glob.glob(f"{MM_BASE}/shared_prefs/*.xml"): try: with open(xml, "r", errors="ignore") as f: @@ -81,6 +82,25 @@ def read_uin() -> str: return m.group(1) except OSError: continue + + # fallback: su -c 读取（非root Termux 环境） + try: + result = subprocess.run( + ["su", "-c", f"grep -rl last_login_uin {MM_BASE}/shared_prefs/*.xml"], + capture_output=True, text=True, timeout=10 + ) + if result.returncode == 0 and result.stdout.strip(): + xml_file = result.stdout.strip().split('\n')[0] + result2 = subprocess.run( + ["su", "-c", f"cat {xml_file}"], + capture_output=True, text=True, timeout=10 + ) + m = re.search(r'last_login_uin">(-?\d+)', result2.stdout) + if m: + return m.group(1) + except Exception: + pass + raise RuntimeError("没读到微信 last_login_uin(未登录?没 root?)") @@ -91,8 +111,28 @@ def account_paths(): if not os.path.exists(db): # 兜底:定位不到就取任一个 cands = glob.glob(f"{MM_BASE}/MicroMsg/*/EnMicroMsg.db") if not cands: - raise RuntimeError("找不到 EnMicroMsg.db") - db = cands[0] + # fallback: su -c 找 db（非root Termux 环境） + try: + result = subprocess.run( + ["su", "-c", f"ls {db}"], + capture_output=True, text=True, timeout=10 + ) + if result.returncode == 0 and result.stdout.strip(): + return uin, result.stdout.strip(), md5hex(IMEI + uin)[:7] + # 兜底:用 su 找任一个 EnMicroMsg.db + result = subprocess.run( + ["su", "-c", f"ls {MM_BASE}/MicroMsg/*/EnMicroMsg.db"], + capture_output=True, text=True, timeout=10 + ) + if result.returncode == 0 and result.stdout.strip(): + cands = result.stdout.strip().split('\n') + db = cands[0] + else: + raise RuntimeError("找不到 EnMicroMsg.db") + except RuntimeError: + raise + except Exception as e: + raise RuntimeError(f"找不到 EnMicroMsg.db: {e}") key = md5hex(IMEI + uin)[:7] return uin, db, key @@ -132,13 +172,27 @@ def sqlcipher_query(db_copy: str, key: str, sql: str): def copy_db(db: str, subdir: str = "_stage") -> str: """拷当前库(含 journal)到独立暂存目录,返回副本路径。root 直接文件拷,不经 /sdcard。 subdir 隔离并发方(watcher=_stage / orchestrator=_stage_orch),避免共用文件互相删/覆盖 - 导致 'no such table' 竞态。""" + 导致 'no such table' 竞态。非root Termux 环境用 su -c cp fallback + chown。""" os.makedirs(f"{WORK_DIR}/{subdir}", exist_ok=True) dst = f"{WORK_DIR}/{subdir}/en.db" - subprocess.run(["cp", db, dst], check=True, timeout=60) + # 先尝试普通 cp（root 环境下有效） + p = subprocess.run(["cp", db, dst], capture_output=True, timeout=60) + if p.returncode != 0: + # fallback: su -c cp（非root Termux 环境） + subprocess.run(["su", "-c", f"cp {db} {dst}"], check=True, timeout=60) + # su cp 的文件属主是 root，需要 chown 给 hermes 用户 + chown_to_hermes(dst) for ext in ("-journal", "-wal", "-shm"): - if os.path.exists(db + ext): - subprocess.run(["cp", db + ext, dst + ext], timeout=60) + src_ext = db + ext + dst_ext = dst + ext + if os.path.exists(src_ext): + subprocess.run(["cp", src_ext, dst_ext], timeout=60) + else: + # 非root 看不到 src_ext 时，用 su 探测并 cp + pe = subprocess.run(["su", "-c", f"test -f {src_ext} && cp {src_ext} {dst_ext}"], + capture_output=True, timeout=60) + if pe.returncode == 0 and os.path.exists(dst_ext): + chown_to_hermes(dst_ext) return dst @@ -193,7 +247,14 @@ def chown_to_hermes(path): try: os.chown(path, o[0], o[1]) except OSError: - pass + # 非root 环境下 os.chown 无权限，fallback 到 su -c chown + try: + subprocess.run( + ["su", "-c", f"chown {o[0]}:{o[1]} {path}"], + capture_output=True, timeout=10 + ) + except Exception: + pass def save_state(state):