youpaishidifu/msgbot
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix QQ消息读取: 跳过ListControl(UIA冻结)直接全窗深度遍历+Win32回退

Browse Source 
- QQ NT TXGuiFoundation UIA树冻结,ListControl始终返回缓存数据
- QQ路径改为: 先force_scroll_bottom, 然后全窗深度遍历(max_depth=8)
- 添加Win32回退(EnumChildWindows+GetWindowText)绕过UIA
- QQ冻结检测更激进(stale=3刷新, stale=6重建引用)
- 微信改动: 标题栏点击(Win10/11安全)->非侵入式刷新
- .gitignore添加*.log
This commit is contained in:
youpaishidifu
2026-07-24 23:10:45 +08:00
parent 4be6ae37c7
commit 2f378f676f
2 changed files with 134 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3
View File
@@ -26,6 +26,9 @@ local_config.json
# Debug dumps
.debug/
# Logs
*.log
# OS
Thumbs.db
.DS_Store
monitor.py
+131 -14
View File
@@ -8,6 +8,7 @@ from typing import Optional, Callable, List
import uiautomation as auto
import win32gui
import win32con
logging.getLogger("comtypes").setLevel(logging.WARNING)
logger = logging.getLogger("msgbot.monitor")
@@ -169,25 +170,89 @@ class MessageMonitor:
def _force_uia_refresh(self, platform: str):
"""通过交互强制窗口更新 UIA 控件树。"""
# 方式1: 向窗口发送非客户区点击（不影响焦点）
hwnd = self._cached_hwnd.get(platform)
if hwnd:
try:
win32gui.PostMessage(hwnd, 0x00A1, 2, 0) # WM_NCLBUTTONDOWN, HTCAPTION
if not hwnd:
return
try:
# 方式1: 窗口重绘
win32gui.InvalidateRect(hwnd, None, True)
win32gui.UpdateWindow(hwnd)
# QQ: 避免标题栏点击（不会干扰用户操作）
if platform == "qq":
win32gui.PostMessage(hwnd, win32con.WM_VSCROLL, win32con.SB_BOTTOM, 0)
win32gui.PostMessage(hwnd, win32con.WM_KEYDOWN, win32con.VK_END, 0)
win32gui.PostMessage(hwnd, win32con.WM_KEYUP, win32con.VK_END, 0)
else:
# 方式2: 向窗口发送非客户区点击（仅微信）
win32gui.PostMessage(hwnd, 0x00A1, 2, 0) # WM_NCLBUTTONDOWN
win32gui.PostMessage(hwnd, 0x00A2, 2, 0) # WM_NCLBUTTONUP
logger.debug(f"[{platform}] UIA强制刷新")
except Exception as e:
logger.debug(f"[{platform}] 强制刷新失败: {e}")
def _force_scroll_bottom(self, platform: str):
"""滚动窗口到底部，触发 UI 更新。"""
hwnd = self._cached_hwnd.get(platform)
if not hwnd:
return
try:
# 1. WM_VSCROLL SB_BOTTOM
win32gui.PostMessage(hwnd, win32con.WM_VSCROLL, win32con.SB_BOTTOM, 0)
# 2. Ctrl+End 快捷键
win32gui.PostMessage(hwnd, win32con.WM_KEYDOWN, win32con.VK_CONTROL, 0)
win32gui.PostMessage(hwnd, win32con.WM_KEYDOWN, win32con.VK_END, 0)
win32gui.PostMessage(hwnd, win32con.WM_KEYUP, win32con.VK_END, 0)
win32gui.PostMessage(hwnd, win32con.WM_KEYUP, win32con.VK_CONTROL, 0)
# 3. InvalidateRect 强制重绘
win32gui.InvalidateRect(hwnd, None, True)
except Exception as e:
logger.debug(f"[{platform}] 滚动到底失败: {e}")
def _read_win32_texts(self, platform: str) -> list:
"""Win32 API 直接读窗口文本（绕过 UIA 缓存）。"""
hwnd = self._cached_hwnd.get(platform)
if not hwnd:
return []
texts = []
def enum_callback(child_hwnd, _):
try:
text = win32gui.GetWindowText(child_hwnd)
if text and len(text.strip()) >= 4:
texts.append(text.strip())
except Exception:
pass
return True
try:
main_text = win32gui.GetWindowText(hwnd)
if main_text and len(main_text.strip()) >= 4:
texts.append(main_text.strip())
win32gui.EnumChildWindows(hwnd, enum_callback, None)
except Exception:
pass
return texts
# ── 核心：读消息 ──────────────────────────────────────────
def _read_messages(self, window, platform: str = "") -> list:
"""每轮重新定位 ListControl。"""
if platform == "qq":
# QQ NT: 消息列表 UIA 不更新，走会话列表（摘要文本实时）
texts = self._read_conversation_list(window)
# QQ NT: UIA 树冻结，ListControl 始终返回缓存数据
# 跳过所有 ListControl 路径，走全窗深度遍历 + Win32 回退
self._force_scroll_bottom(platform)
texts = self._collect_all_texts(window, max_depth=8)
if texts:
logger.debug(f"[QQ] 全窗深度遍历 → {len(texts)} items, last=\"{texts[-1][:50]}\"")
return texts
# 微信/默认: 优先读"消息"（当前选中对话的最新消息）
# Win32 回退：EnumChildWindows 直接读窗口文本（绕过 UIA）
texts = self._read_win32_texts(platform)
if texts:
logger.debug(f"[QQ] Win32 窗口文本 → {len(texts)} items")
return texts
return []
# 微信: 优先读"消息"（ScrollPattern 滚底后实时）
texts = self._read_message_list(window)
if texts:
return texts
@@ -233,12 +298,28 @@ class MessageMonitor:
pass
return []
def _read_message_list_deep(self, window) -> list:
"""QQ专用：从 ListControl('消息') 深度递归提取子控件文本。"""
for name in ["消息", "Message"]:
try:
lst = window.ListControl(Name=name)
if lst.Exists(maxSearchSeconds=0.2):
self._scroll_to_bottom(lst)
texts = self._extract_deep_text(lst)
if texts:
logger.debug(f"[S-Q] ListControl({name}) 深层 → {len(texts)} items")
return texts
except Exception:
pass
return []
def _read_conversation_list(self, window) -> list:
for name in ["会话", "聊天", "Chat"]:
try:
lst = window.ListControl(Name=name)
if lst.Exists(maxSearchSeconds=0.2):
self._scroll_to_bottom(lst)
texts = self._extract_second_line(lst)
if texts:
logger.debug(f"[S-B] 会话({name}) → {len(texts)} items, last=\"{texts[-1][:50]}\"")
return texts
@@ -272,6 +353,33 @@ class MessageMonitor:
pass
return texts
@staticmethod
def _extract_deep_text(lst) -> list:
"""深度递归提取 ListControl 内所有控件的 Name 文本（最多3层）。"""
texts = []
def walk(ctrl, depth):
if depth > 3:
return
try:
name = ctrl.Name
if name and name.strip():
texts.append(name.strip())
except Exception:
pass
try:
for child in ctrl.GetChildren():
walk(child, depth + 1)
except Exception:
pass
try:
for item in lst.GetChildren():
walk(item, 0)
except Exception:
pass
return texts
@staticmethod
def _extract_second_line(lst) -> list:
texts = []
@@ -383,13 +491,22 @@ class MessageMonitor:
# 相同签名 → 冻结计数
stale = self._stale_count.get(platform, 0) + 1
self._stale_count[platform] = stale
# 连续 5 次相同 → 尝试强制刷新 UIA
if stale == 5:
logger.debug(f"[{platform}] 冻结 #{stale}，尝试强制刷新")
self._force_uia_refresh(platform)
elif stale == 10:
logger.debug(f"[{platform}] 冻结 #{stale}，重建窗口引用")
self._cached_hwnd.pop(platform, None)
if platform == "qq":
# QQ: UIA 冻结更快，更激进
if stale == 3:
logger.debug(f"[QQ] 冻结 #{stale}，强制刷新")
self._force_uia_refresh(platform)
elif stale == 6:
logger.debug(f"[QQ] 冻结 #{stale}，重建窗口引用")
self._cached_hwnd.pop(platform, None)
else:
# 微信：标准策略
if stale == 5:
logger.debug(f"[{platform}] 冻结 #{stale}，尝试强制刷新")
self._force_uia_refresh(platform)
elif stale == 10:
logger.debug(f"[{platform}] 冻结 #{stale}，重建窗口引用")
self._cached_hwnd.pop(platform, None)
def _walk_list_controls(ctrl, depth):