更新

wechat_rpa/tmp/clean_dead_tasks.py

@@ -0,0 +1,58 @@
+"""按现场探针结论清掉 4 个永远服务不了的待回复任务，给监听一个干净起点。
+
+两个是订阅号/系统号行（打卡、行业资讯），指纹分毫不差但 _is_real_conversation
+永远否掉；两个在整份会话列表里已经不存在。代码现在会自己淘汰它们，但要熬过
+退避，这里一次性清完。
+"""
+import json
+import os
+import shutil
+import time
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+PENDING = os.path.join(ROOT, "pending_replies.json")
+UNREPLIABLE = os.path.join(ROOT, "unrepliable_sessions.json")
+
+NOT_A_CONVERSATION = [
+    "071f171f101f1f00",  # 行业资讯（订阅号行）
+    "87878f8f8f878682",  # 打卡（系统号行）
+]
+GONE_FROM_LIST = [
+    "f0e0f0f0f0f0f0f8",
+    "787818181f0f0f00",
+]
+
+
+def main() -> None:
+    stamp = time.strftime("%Y%m%d_%H%M%S")
+    shutil.copy2(PENDING, f"{PENDING}.{stamp}.bak")
+
+    with open(PENDING, encoding="utf-8") as handle:
+        data = json.load(handle)
+    pending = data.get("pending", data)
+    doomed = tuple(NOT_A_CONVERSATION + GONE_FROM_LIST)
+    removed = [k for k in pending if k.startswith(doomed)]
+    for key in removed:
+        pending.pop(key)
+    with open(PENDING, "w", encoding="utf-8") as handle:
+        json.dump(data, handle, ensure_ascii=False, indent=2)
+    print(f"已清理 {len(removed)} 个任务，剩余 {len(pending)} 个")
+
+    blacklist = {}
+    if os.path.exists(UNREPLIABLE):
+        try:
+            with open(UNREPLIABLE, encoding="utf-8") as handle:
+                blacklist = json.load(handle) or {}
+        except Exception:
+            blacklist = {}
+    now = time.time()
+    for key in removed:
+        if key.startswith(tuple(NOT_A_CONVERSATION)):
+            blacklist[key] = now
+    with open(UNREPLIABLE, "w", encoding="utf-8") as handle:
+        json.dump(blacklist, handle, ensure_ascii=False, indent=2)
+    print(f"订阅号/系统号名单 {len(blacklist)} 个，6 小时后自动复查")
+
+
+if __name__ == "__main__":
+    main()

wechat_rpa/tmp/probe_identity_drift.py

@@ -0,0 +1,76 @@
+"""身份会不会随时间漂？这是改用昵称之后最要命的失败方式。
+
+列表里的时间戳一直在走（「1分钟前」→「12:08」）。如果它混进了昵称的裁剪框，
+同一个人过一会儿就换一个 md5，任务永远对不上。隔一段时间连读几轮，看 ID 变不变；
+顺带对照聊天标题读出来的名字，是否和列表行算出同一个身份。
+"""
+import os
+import sys
+import time
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import wechat_bot as bot_module  # noqa: E402
+
+ROUNDS = 4
+GAP_SECONDS = 20
+
+
+def main() -> None:
+    bot = bot_module.WeChatBot()
+    bot.safe_window_mode = True
+    bot.auto_activate_window = True
+    if not bot.connect(activate=False, wait_if_missing=True):
+        print("[-] 未挂载到企业微信窗口")
+        return
+    if not bot._ensure_visible():
+        print("[-] 企业微信主界面未还原")
+        return
+
+    history = {}
+    for attempt in range(ROUNDS):
+        listing = bot.capture_session_list()
+        item_h = max(1, int(bot.session_item_h))
+        rows = max(1, listing.shape[0] // item_h)
+        for index in range(min(rows, 8)):
+            center = index * item_h + item_h // 2
+            name = bot._row_display_name(listing, center, row_center=True)
+            fp = bot._session_fingerprint(listing, center, row_center=True)
+            history.setdefault(index, []).append((name, bot.session_id_of(fp)))
+        print(f"  第 {attempt + 1}/{ROUNDS} 轮已采集（{time.strftime('%H:%M:%S')}）")
+        if attempt < ROUNDS - 1:
+            time.sleep(GAP_SECONDS)
+
+    print()
+    print("  行  昵称                 结果")
+    print("  " + "-" * 70)
+    unstable = 0
+    for index, samples in sorted(history.items()):
+        ids = {sid for _name, sid in samples}
+        names = {name for name, _sid in samples}
+        first = samples[0][0] or "(读不到)"
+        if len(ids) == 1:
+            verdict = f"稳定  {samples[0][1][:16]}…"
+        else:
+            unstable += 1
+            verdict = f"★漂移★ 读到过 {names}"
+        print(f"  {index:<3} {first:<20} {verdict}")
+
+    print()
+    title_name = bot._open_chat_display_name()
+    title_fp = bot._fp_from_name(title_name)
+    print(f"  聊天标题读到: {title_name!r} → {bot.session_id_of(title_fp)}")
+    listing = bot.capture_session_list()
+    selected_y = bot.detect_selected_row(listing)
+    if selected_y >= 0:
+        row_fp = bot._session_fingerprint(listing, selected_y, row_center=True)
+        row_name = bot._row_display_name(listing, selected_y, row_center=True)
+        print(f"  选中行读到:   {row_name!r} → {bot.session_id_of(row_fp)}")
+        print(f"  标题与列表行是同一个身份？{bot._session_fp_matches(title_fp, row_fp)}")
+
+    print()
+    print("★ 漂移行数:" if unstable else "全部稳定，没有一行的身份随时间变化。", unstable or "")
+
+
+if __name__ == "__main__":
+    main()

wechat_rpa/tmp/probe_resume_click.py

@@ -0,0 +1,98 @@
+"""为什么每次 [待回复恢复] 之后都跟着 [页面校验] 点击前会话列表已变化？
+
+定位器 _pending_rows_on_page 找到了行，click_session 用同一个 y 复核却不过。
+两边的判据不一样，这里把 click_session 的三个子条件拆开单独打印，看是哪一个
+把点击否掉的。纯读，不点击、不发送。
+"""
+import json
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import wechat_bot as bot_module  # noqa: E402
+from wechat_bot import (  # noqa: E402
+    _AVATAR_FP_BYTES,
+    _SESSION_FP_BYTES,
+    _LEGACY_SESSION_FP_BYTES,
+)
+
+
+def hamming(a: bytes, b: bytes) -> int:
+    if len(a) != len(b) or not a:
+        return -1
+    return (int.from_bytes(a, "big") ^ int.from_bytes(b, "big")).bit_count()
+
+
+def main() -> None:
+    bot = bot_module.WeChatBot()
+    bot.safe_window_mode = True
+    bot.auto_activate_window = False
+    if not bot.connect(activate=False, wait_if_missing=False):
+        print("[-] 未挂载到企业微信窗口")
+        return
+    pending = dict(getattr(bot, "_pending_reply_sessions", {}))
+    print(f"待回复任务 {len(pending)} 个\n")
+
+    for key, state in pending.items():
+        target_fp = bytes.fromhex(key)
+        print("=" * 72)
+        print(f"任务 {key[:16]}…  send_state={state.get('send_state')!r}")
+        found = bot._find_pending_session(target_fp)
+        if found is None:
+            incomplete = getattr(bot, "_pending_scan_incomplete", False)
+            print(
+                f"  _find_pending_session: 没找到"
+                f"（扫描{'被时间预算截断，结论不可信' if incomplete else '已走完全部分页，确实不存在'}）"
+            )
+            continue
+        _page, row_center = found
+        print(f"  _find_pending_session: 命中 row_center={row_center}")
+
+        latest = bot.capture_session_list()
+        if latest is None:
+            print("  重新截取列表失败")
+            continue
+        legacy = len(target_fp) == _LEGACY_SESSION_FP_BYTES
+        if legacy:
+            actual = bot._legacy_session_fingerprint(latest, row_center, row_center=True)
+            fp_ok = actual == target_fp
+        else:
+            actual = bot._session_fingerprint(latest, row_center, row_center=True)
+            fp_ok = bot._session_fp_matches(actual, target_fp)
+
+        print(f"  条件1 指纹匹配 = {fp_ok}")
+        if actual and len(actual) == _SESSION_FP_BYTES and not legacy:
+            name_ok, name_d = bot._name_fp_matches(
+                actual[_AVATAR_FP_BYTES:], target_fp[_AVATAR_FP_BYTES:]
+            )
+            print(
+                f"        头像距离={hamming(actual[:_AVATAR_FP_BYTES], target_fp[:_AVATAR_FP_BYTES])} "
+                f"(容差 {bot._FP_HAMMING_TOL_NAMED}/{bot._FP_HAMMING_TOL})  "
+                f"名称匹配={name_ok} 距离={name_d}  "
+                f"名称有效={bot._name_fp_is_substantive(actual[_AVATAR_FP_BYTES:])}"
+            )
+            pair = tuple(sorted((actual.hex(), target_fp.hex())))
+            print(f"        别名已登记={pair in getattr(bot, '_live_session_fp_aliases', set())}")
+
+        pending_state = bot._pending_reply_state(target_fp)
+        allow_flat = bool(
+            (pending_state or {}).get("requires_visual_proof", False)
+            or bot._flat_session_is_known(target_fp)
+        )
+        allow_unrounded = bool(
+            len(target_fp) == _SESSION_FP_BYTES
+            and (pending_state or {}).get("requires_visual_proof", False)
+            and not bot._flat_session_rejected(target_fp)
+        )
+        shape_ok = bot._is_real_conversation(
+            latest, int(row_center), quiet=True, row_center=True, allow_flat=allow_flat
+        )
+        print(f"  条件2 会话形态 = {shape_ok}  (allow_flat={allow_flat})")
+        print(f"  条件3 兜底放行 = {allow_unrounded}")
+        verdict = fp_ok and (shape_ok or allow_unrounded)
+        print(f"  → click_session 会{'放行' if verdict else '取消这次点击'}")
+
+
+if __name__ == "__main__":
+    main()

wechat_rpa/tmp/probe_security_gate.py

@@ -0,0 +1,76 @@
+"""复现"把广告文本当成登录二维码"的误判，并验证替代判据。
+
+纯读：只截图、只计算，不点击、不发送。
+"""
+import os
+import sys
+
+import cv2
+import numpy as np
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import wechat_bot as bot_module  # noqa: E402
+
+OUT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "secgate")
+
+
+def report(img: np.ndarray, label: str) -> None:
+    height, width = img.shape[:2]
+    rgb = img[:, :, :3].astype(np.float32)
+    gray = rgb.mean(axis=2)
+    nav_width = max(16, int(width * 0.06))
+    nav = gray[int(height * 0.08):int(height * 0.92), :nav_width]
+    center = gray[
+        int(height * 0.32):int(height * 0.70),
+        int(width * 0.32):int(width * 0.68),
+    ]
+    nav_dark_ratio = float((nav < 120).mean())
+    dark = center < 75
+    light_ratio = float((center > 220).mean())
+    transition_ratio = float(
+        (dark[:, 1:] != dark[:, :-1]).mean() + (dark[1:, :] != dark[:-1, :]).mean()
+    )
+    qr_like = (
+        float(dark.mean()) >= 0.025 and light_ratio >= 0.45 and transition_ratio >= 0.035
+    )
+    print(f"--- {label}  {width}x{height}")
+    print(f"  nav_dark_ratio  = {nav_dark_ratio:.4f}   (需 < 0.12)")
+    print(f"  center dark     = {float(dark.mean()):.4f}   (需 >= 0.025)")
+    print(f"  light_ratio     = {light_ratio:.4f}   (需 >= 0.45)")
+    print(f"  transition      = {transition_ratio:.4f}   (需 >= 0.035)")
+    print(f"  qr_like         = {qr_like}")
+    print(f"  旧判据结论      = {nav_dark_ratio < 0.12 and qr_like}")
+
+    gray8 = cv2.cvtColor(img[:, :, :3], cv2.COLOR_BGR2GRAY)
+    detector = cv2.QRCodeDetector()
+    ok, points = detector.detect(gray8)
+    print(f"  cv2 真的找到二维码 = {bool(ok)}")
+    if ok and points is not None:
+        for quad in np.asarray(points).reshape(-1, 4, 2):
+            w = np.linalg.norm(quad[0] - quad[1])
+            h = np.linalg.norm(quad[1] - quad[2])
+            print(f"      四角={quad.astype(int).tolist()}  边长≈{w:.0f}x{h:.0f}")
+
+
+def main() -> None:
+    os.makedirs(OUT, exist_ok=True)
+    bot = bot_module.WeChatBot()
+    bot.safe_window_mode = True
+    bot.auto_activate_window = False
+    if not bot.connect(activate=False, wait_if_missing=False):
+        print("[-] 未挂载到企业微信窗口")
+        return
+    img = bot._capture_full_window()
+    if img is None:
+        print("[-] 截屏失败")
+        return
+    cv2.imwrite(os.path.join(OUT, "full.png"), img[:, :, :3])
+    report(img, "当前企业微信主界面（不该被判为验证页）")
+    print()
+    print(f"looks_like_security_verification() = "
+          f"{bot_module.looks_like_security_verification(img)}")
+
+
+if __name__ == "__main__":
+    main()

wechat_rpa/tmp/probe_session_name.py

@@ -0,0 +1,71 @@
+"""用 session_name 模块跑一遍真实窗口：标题和列表各行的昵称、md5、耗时。"""
+import os
+import sys
+import time
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import numpy as np  # noqa: E402
+import session_name  # noqa: E402
+import wechat_bot as bot_module  # noqa: E402
+
+
+def main() -> None:
+    bot = bot_module.WeChatBot()
+    bot.safe_window_mode = True
+    bot.auto_activate_window = False
+    if not bot.connect(activate=False, wait_if_missing=False):
+        print("[-] 未挂载到企业微信窗口")
+        return
+
+    reader = session_name.NameReader()
+    if not reader.available:
+        print("[-] OCR 引擎不可用")
+        return
+
+    scale = max(0.75, float(bot.scale or 1.0))
+    full = bot._capture_full_window()
+    x1 = int(bot._list_x + bot._list_w)
+
+    print("=" * 78)
+    print("当前打开的会话（聊天标题）")
+    print("=" * 78)
+    panel = np.ascontiguousarray(
+        full[int(20 * scale):int(78 * scale),
+             x1 + int(8 * scale):x1 + int(600 * scale), :3]
+    )
+    start = time.perf_counter()
+    name = reader.read(panel)
+    cost = (time.perf_counter() - start) * 1000
+    print(f"  昵称   {name!r}")
+    print(f"  会话ID {session_name.session_id_for(name)}")
+    print(f"  耗时   {cost:.0f} ms")
+
+    print()
+    print("=" * 78)
+    print("会话列表")
+    print("=" * 78)
+    listing = bot.capture_session_list()
+    item_h = max(1, int(bot.session_item_h))
+    rows = max(1, listing.shape[0] // item_h)
+    total = 0.0
+    for index in range(min(rows, 10)):
+        y0 = index * item_h
+        panel = np.ascontiguousarray(
+            listing[y0 + int(item_h * 0.14):y0 + int(item_h * 0.50),
+                    int(72 * scale):int(310 * scale), :3]
+        )
+        start = time.perf_counter()
+        name = reader.read(panel)
+        cost = (time.perf_counter() - start) * 1000
+        total += cost
+        sid = session_name.session_id_for(name)
+        print(f"  第 {index} 行  {cost:5.0f} ms  {name!r:<24} {sid[:16] if sid else '(读不到)'}")
+    print(f"  整列共 {total:.0f} ms")
+
+    print()
+    print(f"已认识的昵称: {reader.known_names()}")
+
+
+if __name__ == "__main__":
+    main()

wechat_rpa/tmp/reset_identity_data.py

@@ -0,0 +1,47 @@
+"""会话身份从「像素哈希」换成「昵称 md5」，旧键再也匹配不上，全部作废重建。
+
+留着不清的后果就是日志里那个循环：待回复恢复找到一堆旧任务 → 按旧指纹去开 →
+「实际打开对象与目标不一致」→ 任务留在队列里，下一轮再来一遍，真正的新消息
+被挤在后面。原文件都留 .bak。
+"""
+import json
+import os
+import shutil
+import time
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+STAMP = time.strftime("%Y%m%d_%H%M%S")
+
+TARGETS = {
+    "pending_replies.json": {"pending": {}},
+    "unrepliable_sessions.json": {},
+    "conversations.json": {},
+    "false_pos_cache.json": {},
+}
+
+
+def main() -> None:
+    for name, empty in TARGETS.items():
+        path = os.path.join(ROOT, name)
+        if not os.path.exists(path):
+            print(f"  {name:28} 不存在，跳过")
+            continue
+        size = os.path.getsize(path)
+        try:
+            with open(path, encoding="utf-8") as handle:
+                data = json.load(handle)
+            count = len(data.get("pending", data)) if isinstance(data, dict) else 0
+        except Exception:
+            count = -1
+        if size > 2:
+            shutil.copy2(path, f"{path}.{STAMP}.bak")
+        with open(path, "w", encoding="utf-8") as handle:
+            json.dump(empty, handle, ensure_ascii=False, indent=2)
+        print(f"  {name:28} 清空（原有 {count} 条，{size} 字节，已备份）")
+
+    print()
+    print("旧身份数据已作废。下一轮监听会按昵称重新建立会话身份和档案。")
+
+
+if __name__ == "__main__":
+    main()