Files
kefu/wechat_rpa/tmp/diagnose_autoreply.py
2026-07-31 11:48:16 +08:00

161 lines
5.8 KiB
Python

"""只读诊断:逐个检查自动回复链路上的每个闸门,不点击、不发送、不滚动列表。"""
import os
import sys
import time
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import win32gui
import wechat_bot as bot_module
def show(label, value):
print(f"{label:<34}: {value}")
def main():
bot = bot_module.WeChatBot()
bot.safe_window_mode = True
# 不抢前台,纯读。
bot.auto_activate_window = False
print("=" * 70)
print("1) 配置")
print("=" * 70)
from ai_config import (
AI_ENABLED,
AI_USE_VISION,
AI_PROVIDER_TYPE,
AI_API_BASE,
AI_MODEL,
)
show("AI_ENABLED", AI_ENABLED)
show("AI_USE_VISION", AI_USE_VISION)
show("AI_PROVIDER_TYPE", AI_PROVIDER_TYPE)
show("AI_API_BASE", AI_API_BASE)
show("AI_MODEL", AI_MODEL)
show("mouse_idle_enabled", bot.mouse_idle_enabled)
show("mouse_idle_seconds", bot.mouse_idle_seconds)
show("message_batch_window_seconds", bot.message_batch_window_seconds)
print()
print("=" * 70)
print("2) 窗口")
print("=" * 70)
ok = bot.connect(activate=False, wait_if_missing=False)
show("connect()", ok)
if not ok:
print("[结论] 找不到企业微信主窗口,后面全部无法执行。")
return
show("hwnd", hex(bot.hwnd))
show("窗口矩形", (bot.L, bot.T, bot.R, bot.B))
fg = win32gui.GetForegroundWindow()
show("当前前台窗口", f"{hex(fg)} / {win32gui.GetWindowText(fg)!r}")
show("企业微信是否在前台", fg == bot.hwnd)
show("_ensure_visible()", bot._ensure_visible())
show("_window_ready", bot._window_ready)
show("安全验证页锁死", bot.security_verification_required)
show("_security_gate_visible()", bot._security_gate_visible())
print()
print("=" * 70)
print("3) 页面与几何标定")
print("=" * 70)
full = bot._capture_full_window()
show("全窗口截图", None if full is None else full.shape)
nav_ok = bot._message_nav_selected(full)
show("_message_nav_selected", nav_ok)
show("_refresh_message_geometry", bot._refresh_message_geometry(full))
show("_session_geometry_valid", getattr(bot, "_session_geometry_valid", None))
show("_input_geometry_valid", getattr(bot, "_input_geometry_valid", None))
show("_composer_geometry_valid", getattr(bot, "_composer_geometry_valid", None))
show("_chat_geometry_valid", getattr(bot, "_chat_geometry_valid", None))
identity = bot._chat_identity_signature()
surface = bot._chat_surface_signature()
show("聊天标题指纹", identity.hex() or "<空>")
show("消息布局指纹", surface.hex() or "<空>")
time.sleep(1.0)
show("消息布局指纹(1秒后)", bot._chat_surface_signature().hex() or "<空>")
try:
raw_sel = bot._raw_selected_session_fingerprint()
except Exception as exc:
raw_sel = f"异常 {exc}"
show("选中行指纹(raw)", raw_sel.hex() if isinstance(raw_sel, bytes) else raw_sel)
print()
print("=" * 70)
print("4) 发送与人机共存闸门")
print("=" * 70)
show("_send_gate_open()", bot._send_gate_open())
show("剩余限流秒数", f"{bot._send_gate_remaining():.1f}")
show("_mouse_is_idle_now()", bot._mouse_is_idle_now())
print()
print("=" * 70)
print("5) 待回复任务")
print("=" * 70)
pending = bot._pending_reply_sessions
show("任务数", len(pending))
now = time.time()
for key, state in pending.items():
print("-" * 70)
show("会话键", key)
show("键长度(字节)", len(bytes.fromhex(key)))
show("batch_ready", state.get("batch_ready"))
show("confirmed_unread", state.get("confirmed_unread"))
show("requires_visual_proof", state.get("requires_visual_proof"))
show("send_state", repr(state.get("send_state")))
show("视觉否决次数", state.get("visual_rejection_count"))
show("创建于(秒前)", f"{now - float(state.get('created_at') or 0):.0f}")
show("更新于(秒前)", f"{now - float(state.get('updated_at') or 0):.0f}")
deadline = float(state.get("batch_deadline_at") or 0.0)
show(
"合并窗口截止",
"未开始" if not deadline else f"{deadline - now:+.0f} 秒",
)
stored_identity = state.get("identity_signature") or b""
show(
"任务标题指纹",
bytes(stored_identity).hex() if stored_identity else "<空>",
)
show("与当前标题一致", bytes(stored_identity) == identity)
try:
matched = bot._chat_target_matches(
bytes.fromhex(key),
bytes(stored_identity or identity),
current_identity=identity,
)
except Exception as exc:
matched = f"异常 {exc}"
show("_chat_target_matches", matched)
show("render_identities", state.get("render_identities"))
show("被拉黑(纯色头像否决)", key in getattr(bot, "_flat_rejected_session_fps", set()))
chat_text = str(state.get("chat_text") or "")
show("已缓存待回复文本长度", len(chat_text))
if chat_text:
print(" 最后 120 字: " + repr(chat_text[-120:]))
print()
print("=" * 70)
print("6) 未读红点扫描(只读,不点击)")
print("=" * 70)
preview = bot.capture_session_list()
show("会话列表截图", None if preview is None else preview.shape)
try:
found = bot._find_next_unread_session(set(), set(), full=bot._capture_full_window())
except Exception as exc:
found = f"异常 {exc}"
if isinstance(found, tuple):
_img, rel_y, target_fp = found
show("发现未读", f"rel_y={rel_y} fp={target_fp.hex()}")
else:
show("发现未读", found)
if __name__ == "__main__":
main()