From 9a27cbe9e81bec74d00e5b263a2b286bc2283b51 Mon Sep 17 00:00:00 2001 From: youpaishidifu Date: Fri, 24 Jul 2026 22:00:27 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=85=A8=E5=88=97=E8=A1=A8=E7=AD=BE?= =?UTF-8?q?=E5=90=8D=E6=8C=87=E7=BA=B9=E2=86=92=E6=A3=80=E6=B5=8B=E5=AF=B9?= =?UTF-8?q?=E8=AF=9D=E6=8E=92=E5=BA=8F=E5=8F=98=E5=8C=96+ControlFromHandle?= =?UTF-8?q?=E7=BB=95=E8=BF=87UIA=E7=BC=93=E5=AD=98+=E5=86=BB=E7=BB=935?= =?UTF-8?q?=E8=BD=AEresize=E5=BC=BA=E5=88=B6=E5=88=B7=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- monitor.py | 213 +++++++++++++++++++++++++++-------------------- requirements.txt | 1 + 2 files changed, 122 insertions(+), 92 deletions(-) diff --git a/monitor.py b/monitor.py index 6a1d194..11f28e4 100644 --- a/monitor.py +++ b/monitor.py @@ -1,21 +1,20 @@ -"""消息监控 — 每轮全路径重查 ListControl，彻底日志化""" +"""消息监控 — 全列表指纹对比 + 窗口冻结检测 + ControlFromHandle 强制刷新""" import time import logging import threading import traceback -from typing import Optional, Callable +from typing import Optional, Callable, List import uiautomation as auto +import win32gui -# ── 关闭 comtypes 的 DEBUG 洪流 ────────────────────────────── logging.getLogger("comtypes").setLevel(logging.WARNING) - logger = logging.getLogger("msgbot.monitor") class MessageMonitor: - """多线程消息监控器 — 不缓存 ListControl，每轮直接搜""" + """多线程消息监控器""" def __init__(self, on_message: Optional[Callable[[str, str], None]] = None): self._running = False @@ -25,14 +24,19 @@ class MessageMonitor: self._wechat_enabled = True self._qq_enabled = False - # 内容指纹 + # 全列表签名（hash of "|".join(all texts)） + self._last_sig: dict = {} + # 最新消息文本（供 GUI 展示） self._last_msg: dict = {} - # 窗口缓存（只缓窗口，不缓存 ListControl） - self._cached_window: dict = {} - - # 计数器 + # 轮询计数 self._poll_count: dict = {} + + # 窗口句柄缓存（存 HWND，不存 Control 引用） + self._cached_hwnd: dict = {} + + # 连续相同签名计数（用于冻结检测） + self._stale_count: dict = {} # 连续空读 self._empty_reads: dict = {} @@ -59,9 +63,11 @@ class MessageMonitor: if self._thread: self._thread.join(timeout=3) self._thread = None - self._cached_window.clear() - self._empty_reads.clear() + self._cached_hwnd.clear() + self._last_sig.clear() self._last_msg.clear() + self._stale_count.clear() + self._empty_reads.clear() self._poll_count.clear() logger.info("消息监控已停止") @@ -87,85 +93,117 @@ class MessageMonitor: logger.info(f"[♡] 心跳 #{heartbeat}") time.sleep(self._poll_interval) - # ── 窗口缓存 ───────────────────────────────────────────── + # ── 窗口定位 ───────────────────────────────────────────── - def _ensure_window(self, platform: str) -> bool: - win = self._cached_window.get(platform) - if win and win.Exists(maxSearchSeconds=0.1): - return True + def _ensure_hwnd(self, platform: str) -> Optional[int]: + """查找窗口并返回 HWND（不缓存 Control）。""" + hwnd = self._cached_hwnd.get(platform) + try: + if hwnd: + if win32gui.IsWindow(hwnd): + return hwnd + except Exception: + pass - self._cached_window.pop(platform, None) + self._cached_hwnd.pop(platform, None) if platform == "wechat": - names = ["微信", "WeChat"] - for name in names: + for name in ["微信", "WeChat"]: try: w = auto.WindowControl(Name=name) if w.Exists(maxSearchSeconds=0.5): rect = w.BoundingRectangle if rect.width() > 200 and rect.height() > 100: - self._cached_window[platform] = w - logger.info(f"[{platform}] 窗口已定位: {name}") - return True + h = w.NativeWindowHandle + self._cached_hwnd[platform] = h + logger.info(f"[{platform}] 窗口已定位: {name} handle={h}") + return h except Exception: pass elif platform == "qq": - # QQ NT 的窗口类名为 TXGuiFoundation，标题是账号名 - # 先按标题搜，再按类名搜 for name in ["QQ", "TIM"]: try: w = auto.WindowControl(Name=name) if w.Exists(maxSearchSeconds=0.3): rect = w.BoundingRectangle if rect.width() > 200 and rect.height() > 100: - self._cached_window[platform] = w + h = w.NativeWindowHandle + self._cached_hwnd[platform] = h logger.info(f"[{platform}] 窗口已定位(名称): {name}") - return True + return h except Exception: pass - # 兜底：按类名 TXGuiFoundation 找 QQ NT 窗口 + # QQ NT → TXGuiFoundation 类名 try: root = auto.GetRootControl() for child in root.GetChildren(): try: - cls = child.ClassName or "" - if cls == "TXGuiFoundation": + if (child.ClassName or "") == "TXGuiFoundation": rect = child.BoundingRectangle if rect.width() > 200 and rect.height() > 100: - self._cached_window[platform] = child - logger.info(f"[{platform}] 窗口已定位(类名TXGuiFoundation): {child.Name}") - return True + h = child.NativeWindowHandle + self._cached_hwnd[platform] = h + logger.info(f"[{platform}] 窗口已定位(TXGuiFoundation): {child.Name}") + return h except Exception: pass except Exception: pass - return False + return None + + # ── 强制刷新 UIA ────────────────────────────────────────── + + def _fresh_window(self, platform: str): + """从缓存 HWND 创建全新的 WindowControl（绕过 UIA 缓存）。""" + hwnd = self._cached_hwnd.get(platform) + if not hwnd: + return None + try: + # 用 NativeWindowHandle 重建，保证 UIA 新缓存 + win = auto.ControlFromHandle(hwnd) + if win is not None: + return win + except Exception: + pass + return None + + def _force_uia_refresh(self, platform: str): + """强制窗口重绘以触发 UIA 更新。""" + hwnd = self._cached_hwnd.get(platform) + if not hwnd: + return + try: + rect = win32gui.GetWindowRect(hwnd) + # 缩小 1px 再恢复 → 强制重绘 + w, h = rect[2] - rect[0], rect[3] - rect[1] + win32gui.SetWindowPos(hwnd, 0, rect[0], rect[1], w + 1, h, 0x0002 | 0x0004) + win32gui.SetWindowPos(hwnd, 0, rect[0], rect[1], w, h, 0x0002 | 0x0004) + logger.debug(f"[{platform}] UIA强制刷新 (resize 1px)") + except Exception as e: + logger.debug(f"[{platform}] 强制刷新失败: {e}") # ── 核心：读消息 ────────────────────────────────────────── def _read_messages(self, window, platform: str = "") -> list: - """每轮重新定位 ListControl，不缓存。""" - # 微信：先读"会话"（左侧摘要，永远最新） + """每轮重新定位 ListControl。""" if platform == "wechat": texts = self._read_conversation_list(window) if texts: return texts - - # QQ：先读"消息"（右侧详情列表） if platform == "qq": + texts = self._read_conversation_list(window) + if texts: + return texts texts = self._read_message_list(window) if texts: return texts - # 兜底：两个都试 texts = self._read_message_list(window) if texts: return texts texts = self._read_conversation_list(window) if texts: return texts - - # 策略 C: 任何右侧 ListControl try: rect = window.BoundingRectangle boundary = rect.left + rect.width() * 0.40 @@ -179,10 +217,8 @@ class MessageMonitor: return texts except Exception: pass - except Exception as e: - logger.debug(f"[S-C] 异常: {e}") - - # 策略 D: 全窗兜底 + except Exception: + pass try: texts = self._collect_all_texts(window, max_depth=5) if texts: @@ -190,11 +226,9 @@ class MessageMonitor: return texts except Exception: pass - return [] def _read_message_list(self, window) -> list: - """读 ListControl '消息'（右侧详情列表）。""" for name in ["消息", "Message"]: try: lst = window.ListControl(Name=name) @@ -203,13 +237,11 @@ class MessageMonitor: if texts: logger.debug(f"[S-A] ListControl({name}) → {len(texts)} items, last=\"{texts[-1][:50]}\"") return texts - logger.debug(f"[S-A] ListControl({name}) 存在但无文本") - except Exception as e: - logger.debug(f"[S-A] ListControl({name}) 异常: {e}") + except Exception: + pass return [] def _read_conversation_list(self, window) -> list: - """读 ListControl '会话'（左侧摘要，总是包含最新消息摘要）。""" for name in ["会话", "聊天", "Chat"]: try: lst = window.ListControl(Name=name) @@ -218,15 +250,12 @@ class MessageMonitor: if texts: logger.debug(f"[S-B] 会话({name}) → {len(texts)} items, last=\"{texts[-1][:50]}\"") return texts - except Exception as e: - logger.debug(f"[S-B] ListControl({name}) 异常: {e}") + except Exception: + pass return [] - # ── 提取方法 ───────────────────────────────────────────── - @staticmethod def _extract_last_line(lst) -> list: - """从 ListItem Name 提取最后一行（消息详情列表用）。""" texts = [] try: for item in lst.GetChildren(): @@ -244,7 +273,6 @@ class MessageMonitor: @staticmethod def _extract_second_line(lst) -> list: - """从 ListItem Name 提取第二行（会话摘要列表用：格式为 名字\\n消息\\n时间）。""" texts = [] try: for item in lst.GetChildren(): @@ -252,8 +280,6 @@ class MessageMonitor: name = item.Name if name and name.strip(): lines = [l.strip() for l in name.split("\n") if l.strip()] - # 会话列表格式：姓名、消息内容、时间、状态... - # 取索引 1（消息内容），不够两行时取最后一行 if len(lines) >= 2: texts.append(lines[1]) elif lines: @@ -264,11 +290,8 @@ class MessageMonitor: pass return texts - # ── 枚举所有 ListControl ──────────────────────────────── - @staticmethod def _list_controls(window): - """返回窗口内所有 ListControl。""" results = [] try: for ctrl in window.GetChildren(): @@ -277,8 +300,6 @@ class MessageMonitor: pass return results - # ── 全窗遍历 ───────────────────────────────────────────── - @staticmethod def _collect_all_texts(window, max_depth=4): texts = [] @@ -315,53 +336,62 @@ class MessageMonitor: counter = self._poll_count.get(platform, 0) + 1 self._poll_count[platform] = counter - # 窗口缓存 - if not self._ensure_window(platform): - if counter % 20 == 0: + # 1. 确保窗口句柄 + hwnd = self._ensure_hwnd(platform) + if not hwnd: + if counter % 10 == 0: logger.debug(f"[{platform}] 窗口未找到") return - win = self._cached_window[platform] + # 2. 从 HWND 创建全新 Control（绕过缓存） + win = self._fresh_window(platform) + if not win: + logger.debug(f"[{platform}] 创建窗口控件失败") + return - # 强制刷新 UIA 控件树（微信/QQ 虚拟列表不自动同步新消息） - try: - win.Refresh() - except Exception: - pass - - # 读消息 + # 3. 读消息 texts = self._read_messages(win, platform) if not texts: empty = self._empty_reads.get(platform, 0) + 1 self._empty_reads[platform] = empty - if counter % 10 == 0: + if counter % 5 == 0: logger.debug(f"[{platform}] 空读 #{empty}") if empty >= 5: - self._cached_window.pop(platform, None) + self._cached_hwnd.pop(platform, None) self._empty_reads[platform] = 0 logger.info(f"[{platform}] 连续空读，清窗口缓存重试") return - # 有文本，清空读 self._empty_reads[platform] = 0 - latest = texts[-1].strip() - if not latest: - return + # 4. 全列表签名比对 + sig = "||".join([t.strip() for t in texts if t.strip()]) + old_sig = self._last_sig.get(platform, "") - old = self._last_msg.get(platform, "") - if latest != old: - self._last_msg[platform] = latest - logger.info(f"[{platform}] 新消息: {latest}") - if self._on_message: - self._on_message(platform, latest) + if sig != old_sig: + self._last_sig[platform] = sig + self._stale_count[platform] = 0 + latest = texts[-1].strip() + if latest: + self._last_msg[platform] = latest + logger.info(f"[{platform}] 新消息: {latest[:60]}") + if self._on_message: + self._on_message(platform, latest) + else: + # 相同签名 → 冻结计数 + 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) -# ── 模块级别辅助 ────────────────────────────────────────────── - def _walk_list_controls(ctrl, depth): - """递归寻找 ListControl，深度限制。""" if depth > 6: return [] results = [] @@ -369,10 +399,9 @@ def _walk_list_controls(ctrl, depth): cname = ctrl.ControlTypeName except Exception: return results - # 用 ControlTypeName 判断，避免 isinstance 不可靠 if cname == "ListControl": results.append(ctrl) - return results # 不深入 ListControl 内部 + return results try: for child in ctrl.GetChildren(): results.extend(_walk_list_controls(child, depth + 1)) diff --git a/requirements.txt b/requirements.txt index 398a38d..e698830 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,3 +3,4 @@ customtkinter>=5.2.2 jieba>=0.42.1 scikit-learn>=1.3.0 Pillow>=10.0.0 +pywin32>=308