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

fix: 全列表签名指纹→检测对话排序变化+ControlFromHandle绕过UIA缓存+冻结5轮resize强制刷新

Browse Source
This commit is contained in:
youpaishidifu
2026-07-24 22:00:27 +08:00
parent 93c4a049a9
commit 9a27cbe9e8
2 changed files with 122 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
monitor.py
+118 -89
View File
@@ -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:
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}")
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))
requirements.txt
+1
View File
@@ -3,3 +3,4 @@ customtkinter>=5.2.2
jieba>=0.42.1
scikit-learn>=1.3.0
Pillow>=10.0.0
pywin32>=308