diff --git a/.gitignore b/.gitignore index e521f27..0f8b2e7 100644 --- a/.gitignore +++ b/.gitignore @@ -19,6 +19,13 @@ env/ config.json local_config.json +# User data +知识库/ +.knowledge_base/ + +# Debug dumps +.debug/ + # OS Thumbs.db .DS_Store diff --git a/main.py b/main.py index dfecc75..a8795e9 100644 --- a/main.py +++ b/main.py @@ -9,11 +9,17 @@ os.chdir(os.path.dirname(os.path.abspath(__file__))) # 配置日志 logging.basicConfig( - level=logging.INFO, + level=logging.DEBUG, format="%(asctime)s [%(name)s] %(levelname)s: %(message)s", datefmt="%H:%M:%S", ) +# 文件日志（持久化，便于排查） +_fh = logging.FileHandler("msgbot.log", encoding="utf-8", mode="w") +_fh.setLevel(logging.DEBUG) +_fh.setFormatter(logging.Formatter("%(asctime)s [%(name)s] %(levelname)s: %(message)s")) +logging.getLogger().addHandler(_fh) + logger = logging.getLogger("msgbot") diff --git a/monitor.py b/monitor.py index 5cfcdf1..0fa9371 100644 --- a/monitor.py +++ b/monitor.py @@ -1,17 +1,21 @@ -"""消息监控 — 通过 uiautomation 监听微信/QQ 新消息""" +"""消息监控 — 通过 uiautomation 深度遍历微信/QQ 右侧区域文本""" import time import logging import threading +import os from typing import Optional, Callable import uiautomation as auto logger = logging.getLogger("msgbot.monitor") +# 调试日志文件路径 +DEBUG_LOG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".debug") + class MessageMonitor: - """多线程消息监控器 — 右侧消息区加权 + 空读自动重扫""" + """多线程消息监控器 — 深度遍历右侧半区文本 + 一次UI树调试转储""" def __init__(self, on_message: Optional[Callable[[str, str], None]] = None): self._running = False @@ -21,16 +25,18 @@ class MessageMonitor: self._wechat_enabled = True self._qq_enabled = False - # 内容指纹 + # 内容指纹（最近一条消息文本） self._last_msg = {"wechat": "", "qq": ""} # 窗口缓存 self._cached_window: dict = {} - # 消息列表控件缓存 - self._cached_list: dict = {} - # 连续空读计数（超阈值则强制重扫） + + # 连续空读计数 self._empty_reads: dict = {} + # 首次调试转储标记 + self._debug_dumped: set = set() + def set_poll_interval(self, seconds: float): self._poll_interval = max(0.5, seconds) @@ -55,8 +61,9 @@ class MessageMonitor: self._thread.join(timeout=3) self._thread = None self._cached_window.clear() - self._cached_list.clear() self._empty_reads.clear() + self._last_msg.clear() + self._debug_dumped.clear() logger.info("消息监控已停止") @property @@ -69,127 +76,165 @@ class MessageMonitor: while self._running: try: if self._wechat_enabled: - self._ensure_window("wechat") - self._poll("wechat") + self._ensure_window("wechat", ["微信", "WeChat"]) + win = self._cached_window.get("wechat") + if win: + win.SetFocus() # 确保窗口 UIA 树填充 + time.sleep(0.2) + texts = self._collect_right_texts(win) + if texts: + self._dump_debug("wechat", win, texts) + self._check_new("wechat", texts) + else: + self._empty_reads["wechat"] = self._empty_reads.get("wechat", 0) + 1 + if self._qq_enabled: - self._ensure_window("qq") - self._poll("qq") + self._ensure_window("qq", ["QQ", "TIM"]) + win = self._cached_window.get("qq") + if win: + win.SetFocus() + time.sleep(0.2) + texts = self._collect_right_texts(win) + if texts: + self._dump_debug("qq", win, texts) + self._check_new("qq", texts) + else: + self._empty_reads["qq"] = self._empty_reads.get("qq", 0) + 1 + except Exception as e: logger.debug(f"监控轮询异常: {e}") time.sleep(self._poll_interval) - # ── 窗口缓存管理 ──────────────────────────────────────── + # ── 窗口缓存 ───────────────────────────────────────────── - def _ensure_window(self, platform: str): - """缓存窗口，不缓存消息列表（每次 poll 重找，但直取右半侧）""" + def _ensure_window(self, platform: str, names: list): win = self._cached_window.get(platform) if win and win.Exists(maxSearchSeconds=0.1): return self._cached_window.pop(platform, None) - self._cached_list.pop(platform, None) - names = {"wechat": ["微信", "WeChat"], "qq": ["QQ", "TIM"]} - for name in names.get(platform, []): + for name in names: w = auto.WindowControl(Name=name) - if w.Exists(maxSearchSeconds=0.3): - self._cached_window[platform] = w - return + if w.Exists(maxSearchSeconds=0.5): + # 验证窗口尺寸 + try: + rect = w.BoundingRectangle + if rect.width() > 200 and rect.height() > 100: + self._cached_window[platform] = w + return + except Exception: + pass - # ── 定位右侧消息区域（窗口直接子控件，右侧加权 30x） ─ + # ── 核心：深度遍历右侧半区文本 ─────────────────────── - def _find_right_panel(self, window): - """在窗口直接子控件中找最右侧且包含最多文本的控件。""" + def _collect_right_texts(self, window, max_depth=4): + """从窗口右侧半区收集所有可见文本（深度限制 max_depth）。""" + texts = [] try: - win_rect = window.BoundingRectangle - win_cx = win_rect.left + win_rect.width() // 2 - best_ctrl = None - best_score = 0 + rect = window.BoundingRectangle + mid_x = rect.left + rect.width() // 2 + + def walk(ctrl, depth): + if depth > max_depth: + return + try: + cr = ctrl.BoundingRectangle + # 跳过极小控件 + if cr.width() < 20 or cr.height() < 20: + return + # 右侧半区（聊天消息区域必定在右半侧） + if cr.left >= mid_x - 20: + name = ctrl.Name + if name and name.strip(): + texts.append(name.strip()) + for child in ctrl.GetChildren(): + walk(child, depth + 1) + except Exception: + pass for child in window.GetChildren(): - try: - rect = child.BoundingRectangle - - # 跳过太小的控件 - if rect.width() < 80 or rect.height() < 40: - continue - - # 右侧权重：聊天消息区域必定在右半侧 - on_right = rect.left >= win_cx - 30 - right_weight = 30.0 if on_right else 1.0 - - # 只取一层子 Name，不递归全树 - texts = 0 - for sub in child.GetChildren(): - if sub.Name and sub.Name.strip(): - texts += 1 - - score = texts * right_weight - if score > best_score: - best_score = score - best_ctrl = child - except Exception: - continue - - if best_ctrl and best_score >= 2: - return best_ctrl + walk(child, 1) except Exception: pass - return None + return texts - # ── 单次轮询 ──────────────────────────────────────────── - - def _poll(self, platform: str): - win = self._cached_window.get(platform) - if not win: - return - - # 重扫次数限制：最多每 10 秒重扫一次，避免空转 - now = time.time() - last_scan = getattr(self, f"_last_scan_{platform}", 0) - if now - last_scan < 10: - # 用缓存的老控件，但验证有效性 - list_ctrl = self._cached_list.get(platform) - if list_ctrl and list_ctrl.Exists(maxSearchSeconds=0.1): - texts = self._get_texts_fast(list_ctrl) - else: - texts = [] - else: - # 可以做一次重扫 - list_ctrl = self._find_right_panel(win) - self._cached_list[platform] = list_ctrl - setattr(self, f"_last_scan_{platform}", now) - texts = self._get_texts_fast(list_ctrl) if list_ctrl else [] + # ── 当前消息与指纹对比 ──────────────────────────────── + def _check_new(self, platform: str, texts: list): if not texts: - # 空读积累，连续 3 次空读强制重扫下一次 + # 空读计数 empty = self._empty_reads.get(platform, 0) + 1 self._empty_reads[platform] = empty - if empty >= 3: - self._cached_list.pop(platform, None) - setattr(self, f"_last_scan_{platform}", 0) # 允许立即重扫 + logger.debug(f"[{platform}] 空读 #{empty}") return - # 有文本时重置空读计数器 + # 清空读计数 self._empty_reads[platform] = 0 + # 取最新一条文本 latest = texts[-1].strip() if not latest: return - # 内容指纹比较 - if latest != self._last_msg.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) - def _get_texts_fast(self, ctrl) -> list: - """从控件快速提取子文本（只取一层 Name）。""" - texts = [] + # ── 调试：首次运行时转储 UI 树到文件 ─────────────────── + + def _dump_debug(self, platform: str, window, texts: list): + if platform in self._debug_dumped: + return + self._debug_dumped.add(platform) + + os.makedirs(DEBUG_LOG_DIR, exist_ok=True) + path = os.path.join(DEBUG_LOG_DIR, f"ui_dump_{platform}.txt") + + wr = None try: - for child in ctrl.GetChildren(): - if child.Name and child.Name.strip(): - texts.append(child.Name.strip()) + wr = window.BoundingRectangle except Exception: pass - return texts + win_mid_x = (wr.left + wr.width() // 2) if wr else 0 + + lines = [] + lines.append(f"=== UI Tree Dump ({platform}) ===") + lines.append(f"Window: {window.Name}") + lines.append(f"Rect: left={wr.left} top={wr.top} w={wr.width()} h={wr.height()}" if wr else "Rect: N/A") + lines.append(f"Mid-X: {win_mid_x}") + lines.append("") + + def walk(ctrl, depth=0): + indent = " " * depth + try: + cr = ctrl.BoundingRectangle + on_right = cr.left >= win_mid_x - 20 + name = (ctrl.Name or "")[:60] + ctype = ctrl.ControlTypeName + lines.append( + f"{indent}[{ctype}] x={cr.left} y={cr.top} w={cr.width()} h={cr.height()} " + f"right={on_right} name=\"{name}\"" + ) + except Exception: + lines.append(f"{indent}[?] (inaccessible)") + try: + for child in ctrl.GetChildren(): + walk(child, depth + 1) + except Exception: + pass + + for child in window.GetChildren(): + walk(child, 1) + + lines.append("") + lines.append(f"=== Collected Texts ({len(texts)}) ===") + for i, t in enumerate(texts): + lines.append(f" [{i}] {t}") + + with open(path, "w", encoding="utf-8") as f: + f.write("\n".join(lines)) + logger.info(f"[{platform}] UI树已转储到 {path}") +