"""消息监控 — 全列表指纹对比 + 窗口冻结检测 + ControlFromHandle 强制刷新""" import time import logging import threading import traceback import unicodedata import faulthandler from typing import Optional, Callable, List faulthandler.enable() # 捕获 C 扩展段错误，输出 traceback 到 stderr logging.getLogger("comtypes").setLevel(logging.WARNING) logger = logging.getLogger("msgbot.monitor") # ⚠ 顺序重要: torch/easyocr 必须在 COM 模块 (uiautomation/winrt) 之前导入, # 否则 c10.dll 会因 COM 初始化冲突抛出 WinError 1114。 _HAS_EASYOCR = False _easyocr = None _np = None try: import easyocr as _easyocr import numpy as _np _HAS_EASYOCR = True except (ImportError, OSError) as _e: import traceback as _tb _easyocr = None _np = None logger.debug(f"[ocr] EasyOCR 导入失败: {type(_e).__name__}: {_e}") logger.debug(f"[ocr] Traceback: {_tb.format_exc()}") import uiautomation as auto import win32gui import win32con import win32api import win32process import re import ctypes import difflib import psutil import os from datetime import datetime import signal from PIL import Image, ImageGrab, ImageEnhance, ImageFilter from winrt.windows.media.ocr import OcrEngine from winrt.windows.globalization import Language from winrt.windows.graphics.imaging import SoftwareBitmap, BitmapPixelFormat if not _HAS_EASYOCR: logger.debug("[ocr] EasyOCR 未就绪，仅使用 Windows OCR") class MessageMonitor: """多线程消息监控器""" def __init__(self, on_message: Optional[Callable[[str, str], None]] = None): self._running = False self._thread: Optional[threading.Thread] = None self._poll_interval = 2.0 self._on_message = on_message self._wechat_enabled = True self._qq_enabled = False # 全列表签名 & 完整文本列表 (用于新消息比对) self._last_sig: dict = {} self._last_texts: dict = {} # 最新消息文本（供 GUI 展示） self._last_msg: dict = {} # 轮询计数 self._poll_count: dict = {} # QQ 轮询跳过计数（每 3 周期 poll 一次） self._qq_skip_counter = 0 # 窗口句柄缓存（存 HWND，不存 Control 引用） self._cached_hwnd: dict = {} # 窗口对应进程 ID（验证窗口是否仍属同一进程） self._cached_pid: dict = {} # 连续相同签名计数（用于冻结检测） self._stale_count: dict = {} # 连续空读 self._empty_reads: dict = {} # 剪贴板原始文本（用于自动回复提取发送者） self._raw_clipboard: dict = {} # 发送中事件锁（防止自动回复时监控轮询干扰） self._sending_event = threading.Event() # 应用变体：qq→"qq_nt"或"tim" self._app_variant: dict = {} # 已处理消息的全局持久化集合（跨平台、跨轮次，防回环） self._seen_messages: set = set() # 已回复文本的永久集—永不裁切（剪贴板读到自己回复时必拦） self._replied_texts: set = set() # 上次回复时间戳（冷却期内跳过该平台 poll） self._last_reply_time: dict = {} # platform → timestamp # 本轮选中的 OCR 引擎（用于日志标记） self._last_chosen_engine: dict = {} # platform → "windows"|"easyocr" # OCR 引擎（Windows 原生，lazy init） self._ocr_engine = None # EasyOCR 备用引擎（lazy init） self._easyocr_reader = None # 引擎选择计数器 self._ocr_engine_wins: dict = {"windows": 0, "easyocr": 0} # 上次 OCR 原始文本（用于增量检测） self._ocr_last_text: dict = {} # OCR 垃圾连续计数（连续垃圾超过阈值清缓存重找） self._ocr_garbage_count: dict = {} # 截图 hash（像素级，截图不变→OCR 必然无新消息） self._last_screenshot_hash: dict = {} # platform → str self._ocr_unchanged: dict = {} # platform → bool（本轮因截图不变跳过） def set_poll_interval(self, seconds: float): self._poll_interval = max(0.5, seconds) def set_enabled_platforms(self, wechat: bool = True, qq: bool = False): self._wechat_enabled = wechat self._qq_enabled = qq def set_callback(self, cb: Callable[[str, str], None]): self._on_message = cb def start(self): if self._running: return self._running = True # 在主线程预加载 EasyOCR，避免 daemon 线程 torch 段错误 if _HAS_EASYOCR: import torch torch.set_num_threads(1) self._init_easyocr() self._thread = threading.Thread(target=self._run_loop, daemon=True, name="monitor") self._thread.start() logger.info("消息监控已启动") def stop(self): self._running = False if self._thread and self._thread.is_alive(): self._thread.join(timeout=3) self._thread = None self._cached_hwnd.clear() self._cached_pid.clear() self._last_sig.clear() self._last_msg.clear() self._stale_count.clear() self._empty_reads.clear() self._poll_count.clear() logger.info("消息监控已停止") @property def is_running(self) -> bool: return self._running # ── 轮询主循环 ────────────────────────────────────────── def _run_loop(self): # 创建 Windows 消息队列（使 AttachThreadInput / SetForegroundWindow 在 daemon 线程生效） try: ctypes.windll.user32.PeekMessageW(None, 0, 0, 0, 1) # PM_NOREMOVE except Exception: pass heartbeat = 0 logger.debug("_run_loop 进入 while 循环") while self._running: try: if self._wechat_enabled and not self._sending_event.is_set(): self._poll("wechat") if self._qq_enabled: # QQ 降频：每 4 周期 poll 一次（~8s，OCR 计算略重） self._qq_skip_counter += 1 if self._qq_skip_counter >= 4 and not self._sending_event.is_set(): self._qq_skip_counter = 0 self._poll("qq") except Exception as e: logger.error(f"轮询异常: {e}\n{traceback.format_exc()}") heartbeat += 1 if heartbeat % 10 == 0: logger.info(f"[♡] 心跳 #{heartbeat}") time.sleep(self._poll_interval) # ── 窗口定位 ───────────────────────────────────────────── 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_hwnd.pop(platform, None) self._cached_pid.pop(platform, None) if platform == "wechat": 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: h = w.NativeWindowHandle # 确保存顶层窗口句柄（与 GetForegroundWindow 返回值一致） top = win32gui.GetAncestor(h, win32con.GA_ROOT) self._cached_hwnd[platform] = top self._cached_pid[platform] = win32process.GetWindowThreadProcessId(top)[1] logger.info(f"[{platform}] 窗口已定位: {name} handle={top}") return top except Exception: pass elif platform == "qq": # 只按进程名找窗口，不管标题是什么。 # TIM/QQ 标题是动态的（跟随会话显示），任何标题都合法。 for proc_name in ["TIM.exe", "QQ.exe"]: try: for proc in psutil.process_iter(['name', 'pid']): if proc.info['name'] and proc.info['name'].lower() == proc_name.lower(): pid = proc.info['pid'] pid_windows = [] def _pid_enum_cb(hwnd, results, _pid=pid): try: _, found_pid = win32process.GetWindowThreadProcessId(hwnd) if found_pid == _pid: top = win32gui.GetAncestor(hwnd, win32con.GA_ROOT) if top not in results: results.append(top) except Exception: pass return True win32gui.EnumWindows(_pid_enum_cb, pid_windows) # 所有属于该进程的顶层窗口中，取第一个尺寸>=聊天区的、可见的 for top in pid_windows: if not (win32gui.IsWindow(top) and win32gui.IsWindowVisible(top)): continue rect = win32gui.GetWindowRect(top) if (rect[2] - rect[0]) > 200 and (rect[3] - rect[1]) > 100: variant = "tim" if "TIM" in proc_name.upper() else "qq_nt" self._app_variant[platform] = variant self._cached_hwnd[platform] = top self._cached_pid[platform] = pid title = win32gui.GetWindowText(top) logger.info(f"[{platform}] 窗口已定位: {proc_name} pid={pid} title={title!r} top={top}") return top except Exception: continue # 兜底：UIA 搜索 TXGuiFoundation 类名窗口（进程已启动但窗口尚未就绪时） try: root = auto.GetRootControl() for child in root.GetChildren(): try: if child.ClassName == "TXGuiFoundation": rect = child.BoundingRectangle if rect.width() > 200 and rect.height() > 100: h = child.NativeWindowHandle top = win32gui.GetAncestor(h, win32con.GA_ROOT) pid = win32process.GetWindowThreadProcessId(top)[1] self._app_variant[platform] = "qq_nt" self._cached_hwnd[platform] = top self._cached_pid[platform] = pid logger.info(f"[{platform}] 窗口已定位(UIA TXGuiFoundation): top={top}") return top except Exception: continue except Exception: pass return None def _ensure_hwnd_with_timeout(self, platform: str, timeout: float = 12.0) -> Optional[int]: """_ensure_hwnd 包装 daemon 线程超时保护。 ThreadPoolExecutor 创建非 daemon 线程 → Ctrl+C 进程不退出。 改用 daemon thread + Event.wait(timeout)，daemon 线程不阻塞进程退出。 """ result = [None] done = threading.Event() def worker(): result[0] = self._ensure_hwnd(platform) done.set() t = threading.Thread(target=worker, daemon=True) t.start() if not done.wait(timeout=timeout): logger.warning(f"[{platform}] _ensure_hwnd 超时({timeout:.0f}s)，跳过本轮") return None return result[0] # ── 强制刷新 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 _find_qq_window_fresh(self): """通过类名重新查找 QQ/TIM 窗口，绕过 UIA 缓存（比 ControlFromHandle 更彻底）。""" try: # 先按窗口名称查找 for name in ["QQ", "TIM"]: try: w = auto.WindowControl(Name=name) if w.Exists(maxSearchSeconds=0.2): rect = w.BoundingRectangle if rect.width() > 200 and rect.height() > 100: return w except Exception: pass # 按 TXGuiFoundation 类名深度查找 root = auto.GetRootControl() for child in root.GetChildren(): try: if child.ClassName == "TXGuiFoundation": rect = child.BoundingRectangle if rect.width() > 200 and rect.height() > 100: return child except Exception: pass except Exception: pass return None def _force_uia_refresh(self, platform: str): """通过交互强制窗口更新 UIA 控件树。""" hwnd = self._cached_hwnd.get(platform) 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 _ensure_foreground(self, hwnd, max_retries=5) -> bool: """确保窗口在前台，重试多次。失败返回 False，不强行发送。""" root_hwnd = win32gui.GetAncestor(hwnd, win32con.GA_ROOT) if not root_hwnd or not win32gui.IsWindow(root_hwnd): logger.warning(f"[前台] 无效句柄 {hwnd}") return False # 前置条件：最小化则恢复，隐藏则显示 if win32gui.IsIconic(root_hwnd): win32gui.ShowWindow(root_hwnd, win32con.SW_RESTORE) time.sleep(0.2) if not win32gui.IsWindowVisible(root_hwnd): win32gui.ShowWindow(root_hwnd, win32con.SW_SHOWNORMAL) time.sleep(0.2) # 允许任意进程 SetForegroundWindow（移除系统限制，与发送路径一致） try: ctypes.windll.user32.AllowSetForegroundWindow(-1) # ASFW_ANY except Exception: pass for i in range(max_retries): # Method 1: AttachThreadInput + SetForegroundWindow self._attach_set_foreground(root_hwnd) time.sleep(0.25) if self._check_foreground(root_hwnd): return True # Method 2: SwitchToThisWindow（权限要求低） try: ctypes.windll.user32.SwitchToThisWindow(root_hwnd, True) time.sleep(0.2) if self._check_foreground(root_hwnd): return True except Exception: pass # Method 3: SetWindowPos 置顶再取消（最后手段） if i >= 2: try: HWND_TOPMOST = -1 HWND_NOTOPMOST = -2 SWP_NOSIZE = 1 SWP_NOMOVE = 2 ctypes.windll.user32.SetWindowPos( root_hwnd, HWND_TOPMOST, 0, 0, 0, 0, SWP_NOMOVE | SWP_NOSIZE ) time.sleep(0.1) ctypes.windll.user32.SetWindowPos( root_hwnd, HWND_NOTOPMOST, 0, 0, 0, 0, SWP_NOMOVE | SWP_NOSIZE ) time.sleep(0.15) if self._check_foreground(root_hwnd): return True except Exception: pass time.sleep(0.15) logger.warning(f"无法将窗口 {hwnd} 带到前台，放弃操作") return False @staticmethod def _check_foreground(root_hwnd) -> bool: """检查窗口是否在前台（或其子窗口）。""" try: fg = win32gui.GetForegroundWindow() if fg == root_hwnd: return True if fg and win32gui.IsChild(root_hwnd, fg): return True except Exception: pass return False def _attach_set_foreground(self, hwnd): """AttachThreadInput + SetForegroundWindow，从 daemon 线程也能成功。""" try: current_tid = ctypes.windll.kernel32.GetCurrentThreadId() target_tid = ctypes.windll.user32.GetWindowThreadProcessId(hwnd, None) if ctypes.windll.user32.AttachThreadInput(current_tid, target_tid, True): try: win32gui.SetForegroundWindow(hwnd) except Exception: pass finally: ctypes.windll.user32.AttachThreadInput(current_tid, target_tid, False) return # AttachThreadInput 失败 → 裸 SetForegroundWindow try: win32gui.SetForegroundWindow(hwnd) except Exception: pass except Exception: try: win32gui.SetForegroundWindow(hwnd) except Exception: pass def _read_win32_texts(self, platform: str) -> list: """Win32 API 读窗口文本，递归枚举（depth 不限）。""" 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()) >= 2: texts.append(text.strip()) except Exception: pass return True try: main_text = win32gui.GetWindowText(hwnd) if main_text and len(main_text.strip()) >= 2: texts.append(main_text.strip()) win32gui.EnumChildWindows(hwnd, enum_callback, None) except Exception: pass return texts # ── OCR 读屏（QQ 专用，Windows 原生 OCR） ────────────────── def _ocr_ensure_engine(self): """惰性初始化 OCR 引擎（中文简体）。""" if self._ocr_engine is None: try: self._ocr_engine = OcrEngine.try_create_from_language( Language('zh-CN') ) if self._ocr_engine: logger.info(f"[ocr] OCR引擎就绪: {self._ocr_engine.recognizer_language.display_name}") else: logger.warning("[ocr] 创建OCR引擎失败") except Exception as e: logger.error(f"[ocr] OCR引擎初始化异常: {e}") return self._ocr_engine @staticmethod def _normalize_ocr_text(text: str) -> str: """Windows OCR 给 CJK 字符间强加空格 → 去除这些空格使文本可去重。 "问 题 1 ： 你 答 ： 你 好" → "问题1：你答：你好" "你 好 he llo" → "你好 he llo"（ASCII 词内空格保留） 此外修复常见 Windows OCR 字符混淆错误（只在非 md 文本上运行， md 文档由分割后逐片段 md 检测过滤，不受影响）。 """ # 所有紧邻 CJK/全角字符的空格全部去掉 # CJK 范围：汉字 (u4e00-) + 扩展 (u3400-) + 全角/兼容 (uff00-) text = re.sub( r'(?<=[\u4e00-\u9fff\u3400-\u4dbf\uff00-\uffef\u3000-\u303f])' r'\s+', '', text ) text = re.sub( r'\s+' r'(?=[\u4e00-\u9fff\u3400-\u4dbf\uff00-\uffef\u3000-\u303f])', '', text ) # ⭐ Windows OCR 字符混淆修复 # 《》→ <>：Minecraft 聊天大量使用 等尖括号，OCR 误读为中文书名号 text = text.replace('《', '<').replace('》', '>') # l→| 管道分隔符：|quiz| / |start| / |info| 误读为 lquizl / lstartl / linfol text = re.sub(r'(?<=\s)l([A-Za-z]\w*?)l(?=\s)', r'|\1|', text) # ．→ . 全角点→半角数字小数点 (0．45 → 0.45) text = re.sub(r'(\d)[．](\d)', r'\1.\2', text) # 清理消息末尾的状态标记（如 "0~" "O~" "C~" "0~O~"），避免污染 KB 匹配 text = re.sub(r'\s*(?:[A-Za-z0-9]~)+\s*$', '', text) return text.strip() @staticmethod def _is_ocr_console_garbage(text: str) -> bool: """OCR 输出的文本是否明显是控制台/Python 垃圾。 当 OCR 截错窗口（如读到终端窗口）时，文本包含： - Python 路径/模块名：comtypes, uiautomation, sklearn, jieba, winrt - 脚本文件名：monitor.py, main.py - 虚拟环境路径：.venv - 日志关键词：DEBUG, INFO, WARNING, msgbot - 控制台序列：←[, [32m, [0m 返回 True = 应当丢弃。 """ garbage_patterns = [ r'comtypes', r'uiautomation', r'sklearn', r'jieba', r'winrt\.', r'msgbot\.', r'monitor\.py', r'main\.py', r'\.venv', r'Loading model', r'Prefix dict', r'WARNING', r'UserWarning', r'Building prefix', r'←\[', r'\[32m', r'\[0m', r'\\n', ] for pat in garbage_patterns: if re.search(pat, text, re.IGNORECASE): return True # 全字母数字+标点无中文 → 肯定是非聊天内容 cjk_count = sum(1 for c in text if '\u4e00' <= c <= '\u9fff') if cjk_count == 0 and len(text.strip()) > 20: return True return False def _validate_qq_window(self, hwnd: int) -> bool: """验证窗口是否确实是 QQ/TIM 进程。 不尝试自动恢复隐藏/最小化窗口——交给剪贴板兜底的 _ensure_foreground 处理。 _ensure_foreground 有多重重试+SetForegroundWindow，比这里抢先 ShowWindow 更能正确弹出主窗口而非托盘预览。 """ if not win32gui.IsWindow(hwnd): return False if not win32gui.IsWindowVisible(hwnd): return False if win32gui.IsIconic(hwnd): return False # 验证进程名 try: _, pid = win32process.GetWindowThreadProcessId(hwnd) hproc = win32api.OpenProcess( 0x0400 | 0x0010, # PROCESS_QUERY_INFORMATION | PROCESS_VM_READ False, pid ) if hproc: exe_path = win32process.GetModuleFileNameEx(hproc, 0) win32api.CloseHandle(hproc) exe_name = exe_path.split('\\')[-1].lower() if exe_path else '' if exe_name in ('qq.exe', 'tim.exe'): return True except Exception: pass return False @staticmethod def _fuzzy_duplicate(new_text: str, seen_set: set, threshold: float = 0.80) -> bool: """检查 new_text 是否与 seen_set 中已有文本相似度超阈值。 OCR 输出有噪声（每次勖彐碎片、行首数字、尾部截断不同），精确匹配无法去重。 用 difflib.SequenceMatcher 的 quick_ratio() 做快速上界剪枝。 """ if not new_text or len(new_text) < 4: return False for existing in seen_set: if abs(len(existing) - len(new_text)) > max(len(existing), len(new_text)) * 0.4: # 长度差距超 40%，不可能相似（quick_ratio 上界 = 2*min/max < 阈值） continue ratio = difflib.SequenceMatcher(None, new_text, existing).quick_ratio() if ratio >= threshold: return True return False # ── 双 OCR 引擎 ────────────────────────────────────────── def _init_easyocr(self): """惰性初始化 EasyOCR（首次约 3-10 秒）。""" if not _HAS_EASYOCR or self._easyocr_reader is not None: return try: logger.info("[ocr] EasyOCR 初始化中（首次约 3-10 秒）...") # 限制 torch 线程数，避免 MKL/OpenMP 在多线程环境段错误 import torch torch.set_num_threads(1) use_gpu = torch.cuda.is_available() self._easyocr_reader = _easyocr.Reader(['ch_sim', 'en'], gpu=use_gpu) if use_gpu: logger.info("[ocr] EasyOCR 就绪（GPU 模式）") else: logger.info("[ocr] EasyOCR 就绪（CPU 模式）") except Exception as e: logger.error(f"[ocr] EasyOCR 初始化失败: {e}") self._easyocr_reader = None @staticmethod def _score_ocr_text(text: str) -> float: """为 OCR 结果质量打分（越高越好）。对比引擎用。""" if not text or len(text.strip()) < 5: return 0.0 score = 0.0 # 1) 有效 CJK 数量（越多 = 识别越好） cjk = sum(1 for c in text if '\u4e00' <= c <= '\u9fff') score += cjk * 0.5 # 2) Unicode 垃圾字符扣分（OCR 常见误识碎片） junk = set('勖彐丬亻俨冉勹刂冫孑亠丷') junk_count = sum(1 for c in text if c in junk) score -= junk_count * 3 # 3) CJK 间空格比例扣分（Windows OCR 特有毛病） # "你 好" vs "你好" — 空格越多说明识别越差 total = len(text) spaces = text.count(' ') if total > 0 and cjk > 0: space_penalty = (spaces / total) * 10 score -= space_penalty # 4) 时间戳加分 = 消息结构更完整 timestamps = len(re.findall(r'\d{1,2}:\d{2}', text)) score += min(timestamps * 2, 6) return max(score, 0.0) def _pick_ocr_text(self, win_text: str, easy_text: str, image: Image.Image) -> tuple: """双引擎对比，选质量更高的结果。日志记录引擎胜负统计。返回 (text, engine_name)。""" if not easy_text: return (win_text, "windows") if not win_text: return (easy_text, "easyocr") win_score = self._score_ocr_text(win_text) easy_score = self._score_ocr_text(easy_text) chosen = "windows" if win_score >= easy_score else "easyocr" self._ocr_engine_wins[chosen] = self._ocr_engine_wins.get(chosen, 0) + 1 total = sum(self._ocr_engine_wins.values()) ratio = self._ocr_engine_wins.get(chosen, 0) / max(total, 1) logger.debug( f"[ocr] 引擎对比: win={win_score:.1f} vs easy={easy_score:.1f} → 选{chosen} " f"(胜率{ratio:.0%}, {total}轮)" ) return (win_text, chosen) if chosen == "windows" else (easy_text, chosen) def _ocr_read_qq(self) -> list: """OCR 截屏 QQ 消息区域 -> 双引擎（Windows OCR + EasyOCR）提取文本。 包含窗口验证 + 垃圾过滤 + 连续垃圾清缓存。 """ hwnd = self._cached_hwnd.get("qq") if not hwnd: return [] if not self._validate_qq_window(hwnd): gc = self._ocr_garbage_count.get("qq", 0) + 1 self._ocr_garbage_count["qq"] = gc if gc == 1: # 首次失败时详细日志 visible = win32gui.IsWindowVisible(hwnd) if win32gui.IsWindow(hwnd) else False iconic = win32gui.IsIconic(hwnd) if win32gui.IsWindow(hwnd) else False logger.debug(f"[qq] OCR: 窗口不可用(visible={visible}, iconic={iconic})") if gc >= 3: logger.info("[qq] OCR: 窗口不合法连续3次，清缓存") self._cached_hwnd.pop("qq", None) self._cached_pid.pop("qq", None) self._app_variant.pop("qq", None) self._ocr_garbage_count["qq"] = 0 return [] engine = self._ocr_ensure_engine() if not engine: return [] try: rect = win32gui.GetWindowRect(hwnd) if not rect: return [] left, top, right, bottom = rect win_w = right - left win_h = bottom - top # QQ NT 窗口布局: ~0-5% 标签栏, ~5-30% 联系人/群列表, ~30-55% 会话列表, ~55-100% 消息面板 # 先切 QQ 到前台再截屏，避免 CMD 窗口遮挡→OCR 读到终端文字 self._ensure_foreground(hwnd) time.sleep(0.15) # 点一下消息面板区域设置焦点（TIM 默认焦点在会话列表，VK_END 只会滚会话列表） try: click_x = int(win_w * 0.72) # 消息面板中心偏右 click_y = int(win_h * 0.50) lparam = win32api.MAKELONG(click_x, click_y) win32gui.PostMessage(hwnd, win32con.WM_LBUTTONDOWN, win32con.MK_LBUTTON, lparam) win32gui.PostMessage(hwnd, win32con.WM_LBUTTONUP, 0, lparam) logger.debug(f"[qq] 点击消息面板({click_x},{click_y})设置焦点") time.sleep(0.15) except Exception as e: logger.debug(f"[qq] 点击消息面板失败: {e}") # Ctrl+End 滚动到底部，只截最新消息（避开历史堆积+bot 旧回复） VK_CONTROL = 0x11 VK_END = 0x23 # 方式A: keybd_event（窗口在前台时生效） ctypes.windll.user32.keybd_event(VK_CONTROL, 0, 0, 0) ctypes.windll.user32.keybd_event(VK_END, 0, 0, 0) time.sleep(0.05) ctypes.windll.user32.keybd_event(VK_END, 0, 2, 0) ctypes.windll.user32.keybd_event(VK_CONTROL, 0, 2, 0) # 方式B: PostMessage 兜底（keybd_event 可能被 TIM 忽略） win32gui.PostMessage(hwnd, win32con.WM_KEYDOWN, win32con.VK_END, 0) win32gui.PostMessage(hwnd, win32con.WM_KEYUP, win32con.VK_END, 0) time.sleep(0.3) # TIM 窗口布局: ~5-30% 左栏, ~30-50% 会话列表(含文档预览), ~50-100% 消息面板 # 退回 c9acd7e 裁剪: 25% 覆盖消息面板两侧(对方+自己消息都可见) # 右边界 78% 排除群成员侧栏(~80%+), md 过滤负责拦截 bot 的 md 格式回复 cap_left = int(left + win_w * 0.25) cap_top = top + max(80, int(win_h * 0.06)) # 跳过标题栏+群名头栏 cap_right = int(left + win_w * 0.78) # 跳过右栏群成员列表(~80%+) 和滚动条 cap_bottom = bottom - max(100, int(win_h * 0.10)) # 输入框上方 if cap_right <= cap_left or cap_bottom <= cap_top: return [] im = ImageGrab.grab(bbox=(cap_left, cap_top, cap_right, cap_bottom)) if im.width < 50 or im.height < 50: return [] # ⭐ 截图 hash：像素级指纹——内容没变就直接跳过整轮（easyocr 同一截图每次输出不同文本） h = self._image_hash(im) if h == self._last_screenshot_hash.get("qq", ""): self._ocr_unchanged["qq"] = True logger.debug(f"[qq] 截图 hash 相同，屏幕无变化，跳过本轮") return [] self._last_screenshot_hash["qq"] = h self._ocr_unchanged.pop("qq", None) # 保存截屏到程序根目录用于调试（保留最近 20 张） try: script_dir = os.path.dirname(os.path.abspath(__file__)) ts = datetime.now().strftime('%Y%m%d_%H%M%S') path = os.path.join(script_dir, f"ocr_debug_{ts}.png") im.save(path) # 清理旧文件：只保留最近 20 张 _clean_old_ocr_debug(script_dir, keep=20) except Exception as e: logger.debug(f"[ocr] 保存截屏失败: {e}") # 图像预处理：提升 Windows OCR 识别准确率 # Windows OCR 对低 DPI/未处理截图识别差——CJK 空格、字符混淆 # 2x LANCZOS 放大 + 锐化 + 对比度增强 try: im = im.resize((im.width * 2, im.height * 2), Image.LANCZOS) im = ImageEnhance.Sharpness(im).enhance(1.8) im = ImageEnhance.Contrast(im).enhance(1.4) logger.debug(f"[ocr] 预处理: resize 2x + sharpen(1.8) + contrast(1.4)") except Exception as e: logger.debug(f"[ocr] 预处理失败(跳过): {e}") # ── 双引擎 OCR ────────────────────────────────────── # Windows OCR if im.mode != 'RGBA': im = im.convert('RGBA') logger.debug("[ocr] Windows OCR 开始...") sb = SoftwareBitmap(BitmapPixelFormat.RGBA8, im.width, im.height) sb.copy_from_buffer(im.tobytes()) win_result = engine.recognize_async(sb).get() win_text = win_result.text.strip() logger.debug(f"[ocr] Windows OCR 完成: {len(win_text)} chars") # EasyOCR 备用 easy_text = "" if _HAS_EASYOCR: logger.debug("[ocr] EasyOCR 检查/初始化...") self._init_easyocr() logger.debug("[ocr] EasyOCR reader 就绪状态") if self._easyocr_reader: try: logger.debug("[ocr] EasyOCR 识别开始...") img_np = _np.array(im) easy_results = self._easyocr_reader.readtext(img_np) easy_text = ' '.join([r[1] for r in easy_results]).strip() logger.debug(f"[ocr] EasyOCR 识别完成: {len(easy_text)} chars") except Exception as e: logger.debug(f"[ocr] EasyOCR 识别失败: {e}") # 引擎原始输出逐行标记日志（用于长期对比两引擎能力） if win_text: for i, l in enumerate(win_text.split('\n')): if l.strip(): logger.info(f"[ocr] [windows/原始] line[{i}]: {l.strip()[:120]!r}") if easy_text: for i, l in enumerate(easy_text.split('\n')): if l.strip(): logger.info(f"[ocr] [easyocr/原始] line[{i}]: {l.strip()[:120]!r}") raw_text, chosen_engine = self._pick_ocr_text(win_text, easy_text, im) if not raw_text: return [] self._last_chosen_engine["qq"] = chosen_engine return self._process_ocr_output(raw_text, "qq") except Exception as e: logger.debug(f"[qq] OCR失败: {e}") return [] def _ocr_read_wechat(self) -> list: """OCR 截屏微信消息区域 -> 双引擎提取文本。 微信 UIA 树不暴露聊天消息内容（新版微信 UIA 架构变化），必须用 OCR。 微信自绘控件，keybd_event 无法滚动，只截屏当前可见区域。 """ hwnd = self._cached_hwnd.get("wechat") if not hwnd: return [] if not (win32gui.IsWindow(hwnd) and win32gui.IsWindowVisible(hwnd)): self._cached_hwnd.pop("wechat", None) self._cached_pid.pop("wechat", None) return [] engine = self._ocr_ensure_engine() if not engine: return [] try: rect = win32gui.GetWindowRect(hwnd) if not rect: return [] left, top, right, bottom = rect win_w = right - left win_h = bottom - top # WeChat 窗口布局: ~0-32% 左栏(会话列表), ~32-98% 消息面板 # 不点击/不滚动（微信自绘控件 keybd_event 无效） cap_left = int(left + win_w * 0.32) cap_top = top + max(40, int(win_h * 0.04)) # 跳过标题栏 cap_right = right - 10 # 跳过右边框 cap_bottom = bottom - max(65, int(win_h * 0.07)) # 跳过输入框 if cap_right <= cap_left or cap_bottom <= cap_top: return [] im = ImageGrab.grab(bbox=(cap_left, cap_top, cap_right, cap_bottom)) if im.width < 50 or im.height < 50: return [] # ⭐ 截图 hash：像素级指纹——内容没变就直接跳过整轮 h = self._image_hash(im) if h == self._last_screenshot_hash.get("wechat", ""): self._ocr_unchanged["wechat"] = True logger.debug(f"[wechat] 截图 hash 相同，屏幕无变化，跳过本轮") return [] self._last_screenshot_hash["wechat"] = h self._ocr_unchanged.pop("wechat", None) # 保存调试截图 try: script_dir = os.path.dirname(os.path.abspath(__file__)) ts = datetime.now().strftime('%Y%m%d_%H%M%S') path = os.path.join(script_dir, f"wechat_ocr_{ts}.png") im.save(path) _clean_old_ocr_debug(script_dir, keep=20) except Exception as e: logger.debug(f"[wechat] OCR 保存截屏失败: {e}") # 图像预处理（同 QQ） try: im = im.resize((im.width * 2, im.height * 2), Image.LANCZOS) im = ImageEnhance.Sharpness(im).enhance(1.8) im = ImageEnhance.Contrast(im).enhance(1.4) except Exception as e: logger.debug(f"[wechat] OCR 预处理失败(跳过): {e}") # ── 双引擎 OCR ── if im.mode != 'RGBA': im = im.convert('RGBA') logger.debug("[wechat] Windows OCR 开始...") sb = SoftwareBitmap(BitmapPixelFormat.RGBA8, im.width, im.height) sb.copy_from_buffer(im.tobytes()) win_result = engine.recognize_async(sb).get() win_text = win_result.text.strip() logger.debug(f"[wechat] Windows OCR 完成: {len(win_text)} chars") easy_text = "" if _HAS_EASYOCR: logger.debug("[wechat] EasyOCR 检查/初始化...") self._init_easyocr() logger.debug("[wechat] EasyOCR reader 就绪状态") if self._easyocr_reader: try: logger.debug("[wechat] EasyOCR 识别开始...") img_np = _np.array(im) easy_results = self._easyocr_reader.readtext(img_np) easy_text = ' '.join([r[1] for r in easy_results]).strip() except Exception as e: logger.debug(f"[wechat] EasyOCR 识别失败: {e}") # 引擎原始输出逐行标记日志（用于长期对比两引擎能力） if win_text: for i, l in enumerate(win_text.split('\n')): if l.strip(): logger.info(f"[wechat] [windows/原始] line[{i}]: {l.strip()[:120]!r}") if easy_text: for i, l in enumerate(easy_text.split('\n')): if l.strip(): logger.info(f"[wechat] [easyocr/原始] line[{i}]: {l.strip()[:120]!r}") raw_text, chosen_engine = self._pick_ocr_text(win_text, easy_text, im) if not raw_text: return [] self._last_chosen_engine["wechat"] = chosen_engine return self._process_ocr_output(raw_text, "wechat") except Exception as e: logger.debug(f"[wechat] OCR失败: {e}") return [] def _process_ocr_output(self, raw_text: str, platform: str) -> list: """OCR 原始文本的通用后处理：垃圾检测 → 时间戳分割 → 过滤清洗。 供 QQ / 微信 OCR 路径共用。 """ # —— 垃圾检测 —— if self._is_ocr_console_garbage(raw_text): gc = self._ocr_garbage_count.get(platform, 0) + 1 self._ocr_garbage_count[platform] = gc if gc >= 3: logger.info(f"[{platform}] OCR: 垃圾连续3次，清缓存") self._cached_hwnd.pop(platform, None) self._cached_pid.pop(platform, None) self._ocr_garbage_count[platform] = 0 return [] self._ocr_garbage_count[platform] = 0 self._ocr_last_text[platform] = raw_text # —— 行切割 + 尾标清理 —— ocr_lines = [] for l in raw_text.split('\n'): l = re.sub(r'\s*(?:[A-Za-z0-9]~)+\s*$', '', l).strip() if l: ocr_lines.append(l) engine_label = self._last_chosen_engine.get(platform, "") for i, rl in enumerate(ocr_lines): logger.info(f"[ocr] [{engine_label}/后处理] line[{i}]: {rl[:120]!r}") # —— 时间戳分割 —— expanded = [] for i, rl in enumerate(ocr_lines): before = rl rl = self._normalize_ocr_text(rl) logger.debug(f"[ocr] stage[{i}] normalize: len={len(before)} -> {rl[:120]!r}") rl = rl.replace('&', '') rl = rl.replace('：', ':') logger.debug(f"[ocr] stage[{i}] clean: len={len(rl)} -> {rl[:120]!r}") rl = re.sub(r'\s+', ' ', rl) parts = re.split(r'(\d{1,2}:\d{1,2}(?::\d{1,2})?)', rl) logger.debug(f"[ocr] stage[{i}] split -> {len(parts)} parts: " f"{[p[:60] for p in parts if p.strip()]}") buf = '' for part in parts: if re.match(r'^\d{1,2}:\d{1,2}', part): if buf.strip(): if self._is_markdown_document(buf.strip()): extracted = self._extract_user_text_from_md_fragment(buf.strip()) for e in extracted: if e: expanded.append(e) if not extracted: logger.debug(f"[ocr] stage[{i}] markdown fragment skipped: {buf.strip()[:80]!r}") else: msg = re.sub(r'\s*(?:[A-Za-z0-9]~)+\s*$', '', buf.strip()) if msg: expanded.append(msg) buf = '' else: buf += part if buf.strip(): if self._is_markdown_document(buf.strip()): extracted = self._extract_user_text_from_md_fragment(buf.strip()) for e in extracted: if e: expanded.append(e) if not extracted: logger.debug(f"[ocr] stage[{i}] markdown fragment skipped: {buf.strip()[:80]!r}") else: msg = re.sub(r'\s*(?:[A-Za-z0-9]~)+\s*$', '', buf.strip()) if msg: expanded.append(msg) # —— 逐行过滤 —— ocr_lines = expanded texts = [] for line in ocr_lines: if not line: continue if line in ("查看更多", "收起", "选择会话开始聊天"): continue if re.match(r'^[✿♡♥❤★☆✦✧❀❁❂❃❄❅❆❇❈❉❊❋➠➡➢➣➤.·•●○◦◎◉◍◐◑◒◓◔◕◖◗◢◣◤◥☺☻☼♠♣♦♤♧♡♢♔♕♖♗♘♙🕐-🕧\s]{1,3}$', line): continue line = re.sub(r'^0+', '', line) line = re.sub(r'[勖彐"\'\s\u2000-\u206f]+$', '', line) # 末尾裸字母：OCR checkbox [V] 碎成 "? O"、" O"、" V" line = re.sub(r'\s+[OV]\s*$', '', line) if not line: continue if re.match(r'^问题\d+\s*[:：]', line): continue if re.match(r'^[\d:\s]+$', line): continue line = re.sub(r'^\d+[\.．]\d+s?\s*\d*[．\.]?\d*', '', line) line = re.sub(r'^[\d．\.\s]{2,20}[一-鿿]{1,5}?[，。、]', '', line) if not line: continue if self._is_system_message(line): continue if re.search(r'群成员|索群成员|搜索成员', line): logger.debug(f"[{platform}] OCR 过滤: 群成员侧栏 line={line[:60]!r}") continue if re.match(r'^[^）】〕」]{2,40}[（【〔「][^）】〕」]{1,15}[）】〕」]', line): logger.debug(f"[{platform}] OCR 过滤: 群名头栏 line={line[:60]!r}") continue if re.match(r'^[一-鿿]{1}$', line): continue # OCR 碎片：≤8 字符且无中文 → 时间戳/checkbox 残片 if len(line) <= 8 and not re.search(r'[\u4e00-\u9fff]', line): continue # checkbox 残片：`[V]` `[X]` `[√]` 开头，明显不是用户提问 if re.match(r'^\[[VX√✗]\s*\](?:\s*[A-Za-z0-9_\-./]+)?\s*$', line): logger.debug(f"[{platform}] OCR 过滤: checkbox残片 line={line!r}") continue # 多 checkbox 碎片：一行含 2+ 个 [V] 标记，md 列表 OCR 残留 if len(re.findall(r'\[\s*[VXvx√✗]\s*\]', line)) >= 2: logger.debug(f"[{platform}] OCR 过滤: 多checkbox碎片 line={line!r}") continue # 全角括号 md 链接碎片：`[text]（url` 或 `]（` — md 链接 OCR 全角退化 if re.search(r'\]（', line): logger.debug(f"[{platform}] OCR 过滤: 全角括号碎片 line={line!r}") continue if re.match(r'^\d{1,2}[:：\s]\d{1,2}', line): continue if re.match(r'^\d{4}[/-]\d{1,2}', line): continue texts.append(line) logger.debug(f"[{platform}] OCR -> {len(ocr_lines)} raw, {len(texts)} filtered, " f"last=\"{texts[-1][:50] if texts else ''}\"") return texts def _read_via_clipboard_once(self, platform: str) -> list: """剪贴板法单次执行（安全键盘导航：Enter + Ctrl+End + Ctrl+A/C，无 Esc）。 去重由 _poll 的 _replied_texts + _seen_messages 负责，本函数只过滤并返回。 回复回环防御由 _paste_and_send 写入 _replied_texts 承担。 """ hwnd = self._cached_hwnd.get(platform) if not hwnd or not win32gui.IsWindow(hwnd): return [] import win32clipboard as wc old = None try: wc.OpenClipboard() if wc.IsClipboardFormatAvailable(wc.CF_UNICODETEXT): old = wc.GetClipboardData(wc.CF_UNICODETEXT) wc.CloseClipboard() except Exception: pass texts = [] raw_lines = [] # 预初始化，防止 content 为空时 UnboundLocalError try: # 确保窗口在前台（失败则返回空，避免发送到错误窗口） # QQ 剪贴板优先路径需要更可靠的前台保证（与发送路径 retries 一致） if not self._ensure_foreground(hwnd, max_retries=5): return [] # PostMessage 点击消息区域（设焦点，不全局，不比鼠标移动） rect = win32gui.GetWindowRect(hwnd) if rect: left, top, right, bottom = rect w = right - left h = bottom - top # QQ NT 消息面板在 ~55-95% 宽度区域 click_x = int(w * 0.72) # 消息面板中部 click_y = int(h * 0.60) # 消息面板中部偏下 lparam = win32api.MAKELONG(click_x, click_y) win32gui.PostMessage(hwnd, win32con.WM_LBUTTONDOWN, win32con.MK_LBUTTON, lparam) win32gui.PostMessage(hwnd, win32con.WM_LBUTTONUP, 0, lparam) time.sleep(0.2) # 滚动到底部（PostMessage 安全，不切焦点，确保新消息可见） self._force_scroll_bottom(platform) time.sleep(0.2) # TIM/QQ NT: 安全键盘导航（无 Esc，不关窗口） # 每步前校验目标窗口仍在焦点，防止 keybd_event 泄露到控制台 VK_CONTROL = 0x11 variant = self._app_variant.get(platform, "") if variant in ("tim", "qq_nt"): VK_RETURN = 0x0D VK_END = 0x23 # Enter 激活当前对话 if self._fg_is_target(hwnd): ctypes.windll.user32.keybd_event(VK_RETURN, 0, 0, 0) time.sleep(0.05) ctypes.windll.user32.keybd_event(VK_RETURN, 0, 2, 0) time.sleep(0.3) # Ctrl+End 滚动到底部（双保险：PostMessage + keybd_event） if self._fg_is_target(hwnd): ctypes.windll.user32.keybd_event(VK_CONTROL, 0, 0, 0) ctypes.windll.user32.keybd_event(VK_END, 0, 0, 0) time.sleep(0.05) ctypes.windll.user32.keybd_event(VK_END, 0, 2, 0) ctypes.windll.user32.keybd_event(VK_CONTROL, 0, 2, 0) time.sleep(0.3) # Ctrl+A (全选) — 仅当目标窗口仍在焦点 VK_A = 0x41 if self._fg_is_target(hwnd): ctypes.windll.user32.keybd_event(VK_CONTROL, 0, 0, 0) ctypes.windll.user32.keybd_event(VK_A, 0, 0, 0) time.sleep(0.05) ctypes.windll.user32.keybd_event(VK_A, 0, 2, 0) ctypes.windll.user32.keybd_event(VK_CONTROL, 0, 2, 0) time.sleep(0.15) # Ctrl+C (复制) VK_C = 0x43 if self._fg_is_target(hwnd): ctypes.windll.user32.keybd_event(VK_CONTROL, 0, 0, 0) ctypes.windll.user32.keybd_event(VK_C, 0, 0, 0) time.sleep(0.05) ctypes.windll.user32.keybd_event(VK_C, 0, 2, 0) ctypes.windll.user32.keybd_event(VK_CONTROL, 0, 2, 0) time.sleep(0.3) wc.OpenClipboard() if wc.IsClipboardFormatAvailable(wc.CF_UNICODETEXT): content = wc.GetClipboardData(wc.CF_UNICODETEXT) if content: raw_lines = [l.strip() for l in content.strip().split('\n') if l.strip()] self._raw_clipboard[platform] = content wc.CloseClipboard() logger.debug(f"[{platform}] 剪贴板读取 → {len(raw_lines)} lines") # ── 过滤：只返回"像用户消息"的行 ── sender_pattern = re.compile(r'^[\u4e00-\u9fff·A-Za-z0-9_\-]{1,10}$') for line in raw_lines: # 清理 QQ 消息末尾的状态标记（如 "O~" "C~" "0~O~"），避免污染 KB 匹配 line = re.sub(r'\s*(?:[A-Za-z0-9]~)+\s*$', '', line).strip() if not line: continue # 跳过：系统消息、纯时间、纯数字、无意义短行 if self._is_system_message(line): continue # 跳过：控制台/Python 垃圾（剪贴板截错窗口内容） if self._is_ocr_console_garbage(line): continue # 跳过：单行发送者名（1-10汉字/字母/数字/下划线） if sender_pattern.match(line): continue # 跳过：纯时间格式（连字符/斜杠分格日期） if re.match(r'^\d{1,2}[:：\s]\d{2}', line): continue if re.match(r'^\d{4}[/-]\d{1,2}', line): continue # 跳过：纯 emoji / 纯标点（短于 6 个字符且只有符号） clean = re.sub('[~!@#$%^&*()_+\\-=\\[\\]{};:\\\'",.<>/?·！@#￥%……&*（）—+。，、；：""\'\'【】｛｝「」『』《》【】 \\t\\n\\r\\u2000-\\u206f\\u2190-\\u2bff\\U0001F000-\\U0001FFFF]', '', line) if len(clean) == 0 and len(line) <= 6: continue # 通过全部检查 → 保留 texts.append(line) logger.debug(f"[{platform}] 剪贴板过滤后 → {len(texts)} 条用户消息") except Exception as e: logger.debug(f"[{platform}] 剪贴板失败: {e}") finally: if old is not None: try: wc.OpenClipboard() wc.EmptyClipboard() wc.SetClipboardText(old, wc.CF_UNICODETEXT) wc.CloseClipboard() except Exception: pass return texts def get_raw_clipboard(self, platform: str) -> str: """获取最近一次剪贴板读取的原始文本。""" return self._raw_clipboard.get(platform, "") @staticmethod def _parse_last_sender(raw_text: str) -> Optional[str]: """从剪贴板原始文本中提取最后一条消息的发送者。 TIM/QQ 剪贴板格式通常是: [发送者] [时间] [消息] (空行) [发送者] [时间] [消息] """ if not raw_text: return None blocks = re.split(r'\n\s*\n', raw_text.strip()) if not blocks: return None last_block = blocks[-1].strip() if not last_block: return None lines = [l.strip() for l in last_block.split('\n') if l.strip()] if not lines: return None candidate = lines[0] # 排除时间格式 if re.match(r'^\d{1,2}[:：]\d{2}', candidate): return None if re.match(r'^\d{4}[/-]\d{1,2}', candidate): return None if len(candidate) > 30: return None return candidate def _search_contact(self, platform: str, contact: str) -> bool: """在 TIM/QQ 搜索框中搜索联系人并切换到对应会话。""" hwnd = self._cached_hwnd.get(platform) if not hwnd: return False try: win = auto.ControlFromHandle(hwnd) search = win.EditControl(Name="搜索", maxSearchSeconds=0.5) if not search.Exists(maxSearchSeconds=0.3): logger.warning(f"[{platform}] 找不到搜索框") return False # 恢复窗口并前置（失败则放弃搜索） if win32gui.IsIconic(hwnd): win32gui.ShowWindow(hwnd, win32con.SW_RESTORE) if not self._ensure_foreground(hwnd): logger.warning(f"[{platform}] 无法前置窗口，放弃搜索联系人") return False time.sleep(0.2) # SetFocus 到搜索框（不移动鼠标） search.SetFocus() time.sleep(0.2) # 全选已有内容并输入联系人名 auto.SendKeys('{Ctrl}a') time.sleep(0.1) auto.SendKeys(contact) time.sleep(0.5) # Enter 打开第一个结果 auto.SendKeys('{Enter}') time.sleep(0.5) logger.info(f"[{platform}] 已搜索并切换到联系人: {contact}") return True except Exception as e: logger.error(f"[{platform}] 搜索联系人失败: {e}") return False # ── SendInput 键盘模拟 ───────────────────────────────── @staticmethod def _key_down(vk: int): """SendInput KEYBDINPUT keydown.""" ctypes.windll.user32.keybd_event(vk, 0, 0, 0) @staticmethod def _key_up(vk: int): """SendInput KEYBDINPUT keyup (KEYEVENTF_KEYUP=2).""" ctypes.windll.user32.keybd_event(vk, 0, 2, 0) @staticmethod def _send_key_press(vk: int, delay: float = 0.05): """SendInput single key press (down + up).""" ctypes.windll.user32.keybd_event(vk, 0, 0, 0) time.sleep(delay) ctypes.windll.user32.keybd_event(vk, 0, 2, 0) @staticmethod def _send_ctrl_combo(vk: int): """Send Ctrl+: Ctrl down → key down → key up → Ctrl up.""" VK_CONTROL = 0x11 ctypes.windll.user32.keybd_event(VK_CONTROL, 0, 0, 0) ctypes.windll.user32.keybd_event(vk, 0, 0, 0) ctypes.windll.user32.keybd_event(vk, 0, 2, 0) ctypes.windll.user32.keybd_event(VK_CONTROL, 0, 2, 0) @staticmethod def _post_key(hwnd, vk, delay=0.05): """PostMessage 键盘到指定窗口（不全局，安全）。""" win32gui.PostMessage(hwnd, win32con.WM_KEYDOWN, vk, 0) time.sleep(delay) win32gui.PostMessage(hwnd, win32con.WM_KEYUP, vk, 0) @staticmethod def _post_ctrl_combo(hwnd, vk, delay=0.05): """PostMessage Ctrl+组合键到指定窗口（不泄露到控制台）。""" VK_CONTROL = 0x11 win32gui.PostMessage(hwnd, win32con.WM_KEYDOWN, VK_CONTROL, 0) win32gui.PostMessage(hwnd, win32con.WM_KEYDOWN, vk, 0) time.sleep(delay) win32gui.PostMessage(hwnd, win32con.WM_KEYUP, vk, 0) win32gui.PostMessage(hwnd, win32con.WM_KEYUP, VK_CONTROL, 0) def _fg_is_target(self, hwnd) -> bool: """校验目标窗口仍在焦点，防止 keybd_event 泄露到控制台。""" try: fg = win32gui.GetForegroundWindow() root = win32gui.GetAncestor(hwnd, win32con.GA_ROOT) return fg == root or (fg and win32gui.IsChild(root, fg)) except Exception: return False @staticmethod def _image_hash(im) -> str: """感知哈希：缩至 8x8 灰阶 → 均值哈希（稳定，抗 1-2px 偏移）。""" small = im.convert('L').resize((8, 8), Image.LANCZOS) pixels = list(small.getdata()) avg = sum(pixels) // len(pixels) return ''.join('1' if p > avg else '0' for p in pixels) @staticmethod def _is_system_message(text: str) -> bool: """过滤机器人/系统消息，不作为用户消息处理。""" if not text: return True # NFKC 标准化：去除 UIA/OCR 夹杂的零宽字符、不可见 Unicode text = unicodedata.normalize('NFKC', text) # 系统前缀/标记 system_markers = [ "✅", "❌", "📝", "🔍", "📄", "➡", "♡", "✿", "[WECHAT]", "[QQ]", "[wechat]", "[qq]", ] for m in system_markers: if text.startswith(m): return True # ⭐ QQ/微信 界面噪声文字（按钮/提示/碎片/导航，非用户消息） qq_ui_noise = [ "查看更多", "可查看在其他设备产生的聊天记录", "收藏", "转发", "多选", "引用", "回复", "选择会话开始聊天", # 微信底部导航 "通讯录", "发现", "我", "消息", "朋友圈", # 微信/QQ 顶部/侧边 "搜索", "添加到表情", "更多", # 微信底部栏 + 附加按钮 "导航", "视频号", # QQ 空间/动态 "空间", "动态", "说说", "日志", "相册", # 通用 UI "发送", "接收", "取消", "确认", "确定", "关闭", ] for n in qq_ui_noise: if text == n: return True # 系统事件模板 — 注意：用 startswith 而非 in，避免混合文本（OCR 多行合一）误杀 system_patterns = [ "已自动回复", "知识库中未找到", "找到相关结果", "监控已启动", "监控已暂停", "设置已保存", "🤖", "找不到联系人", "当前会话回复", "自动回复触发", "匹配度", # KB 文档内容格式（机器人回复格式），OCR 可能把用户消息+KB 回复混成一行 # 只有全行以"答："开头才过滤，内含"答："的混合文本不过滤 "答：", # 纯系统命令反馈 "PS ", "❯ ", "$ ", ] for p in system_patterns: if text.startswith(p): return True # 纯时间格式（不含消息内容） if re.match(r'^\d{1,2}:\d{2}', text): return True return False @staticmethod def _is_markdown_document(text: str) -> bool: """检测是否为 Markdown 格式的文档内容。 Bot 回复用 md 格式（# 标题、[链接](url)、列表等），用户日常对话不用。 借此区分 bot 旧回复和历史堆积 vs 用户新消息。 """ if not text: return False # # 行首标题：短文本也检测（OCR 把 ##协议修 切成 5 字符片段） if re.match(r'^#{1,6}(?:\s|\[|[\u4e00-\u9fff])', text): return True # 全角括号退化 md 链接：短文本也检测 if re.search(r'\]（', text): return True if re.search(r'\[.+?\]（', text): return True if len(text) < 10: return False # ##直接跟汉字（OCR 丢失 ##后空格时常见：##文档导航） if re.match(r'^#{1,6}[\u4e00-\u9fff]', text): return True # 包含 markdown 链接 [text](url) if re.search(r'\[.+?\]\(.+?\)', text): return True # 包含 .md 文件引用（如 "gin.md)" 分割后断开的链接） if re.search(r'\S+\.md\)', text): return True # 行首是 - 或 * 列表项（标准 md 列表） if re.match(r'^[\-\*]\s', text): return True # 包含代码块标记 ``` 或 `code` if re.search(r'`{1,3}', text): return True # 连续 3+ 行包含 markdown 特征：| 表格、--- 分隔线、> 引用 pipes = text.count('|') if pipes >= 3: return True # 内联 # # 标题标记（如 "# # 必装依赖" → 分割后 `# #` 或 `# #`） if re.search(r'#\s*#', text): return True # OCR 退化 md checkbox：`[V]`、`[v]`、`[x]` 等（含空格变体 [V ]） if re.search(r'\[\s*[VXvx√✗]\s*\]', text): return True # OCR 退化 md 链接：`[text](...` 缺闭合括号，QQ 群聊几乎不用此语法 if re.search(r'\[.+?\]\(', text): return True return False @staticmethod def _extract_user_text_from_md_fragment(text: str) -> list: """从混合了 md 文档和用户消息的 fragment 中提取用户文本。 OCR 经常把 bot md 回复和用户新消息合并为一行无换行文本， 如 "用户问题 ##bot标题|表格|数据|...另一个用户问题"。 本方法找到第一个 md 标记位置，提取其前面的用户文本； 再找到末尾连续非 md 尾巴，提取为第二条用户文本。 """ # ── 整段以 ## 开头 → 无用户文本（无空格 ##中文 也捕获）── if re.match(r'^#{1,6}(?:\s|\[|[\u4e00-\u9fff])', text): return [] # 整段以 [text]（ 开头 → 全角括号 md 链接碎片，无用户文本 if re.match(r'^\[.+?\]（', text): return [] # ── 找第一个 markdown 起始标记的位置 ── # 标记：## 标题、[text](url)、.md)、| 表格分隔、``` 代码块 md_start = len(text) # 默认整段都是 md # ## 标题（放宽：允许 ##后直接跟汉字/[，不限于空格） m = re.search(r'(?<=\S)\s*#{1,6}(?:\s|\[|[\u4e00-\u9fff])', text) if m and m.start() < md_start: md_start = m.start() # [text](url) 链接（含全角括号退化 [text]（url）） m = re.search(r'(?<=\S)\s*\[.+?\]\(.+?\)', text) if m and m.start() < md_start: md_start = m.start() m = re.search(r'(?<=\S)\s*\[.+?\]（.+?）', text) if m and m.start() < md_start: md_start = m.start() # .md) 文件引用 m = re.search(r'(?<=\S)\s*\S+\.md\)', text) if m and m.start() < md_start: md_start = m.start() # | 表格 m = re.search(r'(?<=\S)\s*\|', text) if m and m.start() < md_start: md_start = m.start() # ``` 代码块 m = re.search(r'`{3}', text) if m and m.start() < md_start: md_start = m.start() results = [] # 提取 md 标记前的用户文本 if md_start > 2: # 如果 md 标记是 | 且前一个字符非空白 → head 是表格细胞值，不是用户提问 skip_head = ( md_start < len(text) and text[md_start] == '|' and md_start > 0 and not text[md_start-1].isspace() ) if not skip_head: head = text[:md_start].strip() # 去掉尾部状态标记 head = re.sub(r'\s*(?:[A-Za-z0-9]~)+\s*$', '', head).strip() # 过滤纯标点/符号 if head and not re.match(r'^[\d:\s|]+$', head): results.append(head) # 提取末尾可能的用户文本（md 块之后的尾巴） # 找最后一个句号/感叹号/问号后的内容（可能是新消息的开头） tail_pattern = re.search(r'[。！？\?!.]\s*(.{2,50})$', text) if tail_pattern: tail = tail_pattern.group(1).strip() tail = re.sub(r'\s*(?:[A-Za-z0-9]~)+\s*$', '', tail).strip() if tail and not re.match(r'^[\d:\s|]+$', tail): # 不是纯标点、不是表格碎片 if not re.match(r'^[|/\-─━━═]+$', tail): results.append(tail) # 末尾没有句号但有中文问号结尾的（如 "PHP用户端有哪些登录模式?"） if not tail_pattern: tail2 = re.search(r'([\u4e00-\u9fff].{2,50}\?)$', text) if tail2: tail = tail2.group(1).strip() tail = re.sub(r'\s*(?:[A-Za-z0-9]~)+\s*$', '', tail).strip() if tail and len(tail) >= 3: # 避免重复 if tail not in results: results.append(tail) return results def _paste_and_send(self, platform: str, text: str) -> bool: """粘贴文本到输入框并发送（剪贴板 + SendInput）。""" hwnd = self._cached_hwnd.get(platform) if not hwnd: return False import win32clipboard as wc # 保存剪贴板 old = None try: wc.OpenClipboard() if wc.IsClipboardFormatAvailable(wc.CF_UNICODETEXT): old = wc.GetClipboardData(wc.CF_UNICODETEXT) wc.CloseClipboard() except Exception: pass try: # 设置答案到剪贴板 wc.OpenClipboard() wc.EmptyClipboard() wc.SetClipboardText(text, wc.CF_UNICODETEXT) wc.CloseClipboard() # ── 更激进的窗口唤醒 ── # 1) 允许任意进程 SetForegroundWindow（移除系统限制） try: ctypes.windll.user32.AllowSetForegroundWindow(-1) # ASFW_ANY except Exception: pass # 2) 恢复/显示窗口 root_hwnd = win32gui.GetAncestor(hwnd, win32con.GA_ROOT) if win32gui.IsIconic(root_hwnd): win32gui.ShowWindow(root_hwnd, win32con.SW_RESTORE) time.sleep(0.3) if not win32gui.IsWindowVisible(root_hwnd): win32gui.ShowWindow(root_hwnd, win32con.SW_SHOWNORMAL) time.sleep(0.3) # 3) 试图 SetForegroundWindow + Alt 键锁（让当前线程获得用户输入资格） win32gui.SetForegroundWindow(root_hwnd) VK_MENU = 0x12 KEYEVENTF_KEYUP = 0x0002 ctypes.windll.user32.keybd_event(VK_MENU, 0, 0, 0) # Alt down ctypes.windll.user32.keybd_event(VK_MENU, 0, KEYEVENTF_KEYUP, 0) # Alt up time.sleep(0.2) # 4) 再 SetForegroundWindow + SwitchToThisWindow win32gui.SetForegroundWindow(root_hwnd) ctypes.windll.user32.SwitchToThisWindow(root_hwnd, True) time.sleep(0.3) # 5) 验证 foreground — 不成功则用 SendMessage 兜底 fg_now = win32gui.GetForegroundWindow() fg_ok = (fg_now == root_hwnd) or (fg_now and win32gui.IsChild(root_hwnd, fg_now)) if not fg_ok: logger.warning(f"[{platform}] 前台唤醒失败(fg={fg_now}, target={root_hwnd})，SendMessage 兜底") # SendMessage 方式：直接发消息给窗口，不依赖前台 rect = win32gui.GetWindowRect(hwnd) if rect: left, top, right, bottom = rect w = right - left h = bottom - top # QQ NT 输入框在右侧消息面板底部 click_x = int(w * 0.72) click_y = int(h * 0.92) lparam = win32api.MAKELONG(click_x, click_y) win32gui.SendMessage(hwnd, win32con.WM_LBUTTONDOWN, win32con.MK_LBUTTON, lparam) win32gui.SendMessage(hwnd, win32con.WM_LBUTTONUP, 0, lparam) time.sleep(0.1) # Ctrl+A (全选) win32gui.PostMessage(hwnd, win32con.WM_KEYDOWN, 0x41, 0 | (1 << 29)) # Ctrl 修饰 win32gui.PostMessage(hwnd, win32con.WM_KEYUP, 0x41, 0) time.sleep(0.05) # Ctrl+V (粘贴) win32gui.PostMessage(hwnd, win32con.WM_KEYDOWN, 0x56, 0 | (1 << 29)) win32gui.PostMessage(hwnd, win32con.WM_KEYUP, 0x56, 0) time.sleep(0.1) # Enter (发送) win32gui.PostMessage(hwnd, win32con.WM_KEYDOWN, 0x0D, 0) win32gui.PostMessage(hwnd, win32con.WM_KEYUP, 0x0D, 0) time.sleep(0.5) logger.info(f"[{platform}] 已回复(SendMessage): {text[:60]}") else: return False else: # ── 前台成功：PostMessage 点击输入框 + keybd_event 发送 ── rect = win32gui.GetWindowRect(hwnd) if rect: left, top, right, bottom = rect w = right - left h = bottom - top click_x = int(w * 0.72) click_y = int(h * 0.92) lparam = win32api.MAKELONG(click_x, click_y) win32gui.PostMessage(hwnd, win32con.WM_LBUTTONDOWN, win32con.MK_LBUTTON, lparam) win32gui.PostMessage(hwnd, win32con.WM_LBUTTONUP, 0, lparam) time.sleep(0.2) # keybd_event 发送（窗口已在前台） VK_A = 0x41 VK_V = 0x56 VK_BACK = 0x08 VK_RETURN = 0x0D self._send_ctrl_combo(VK_A) time.sleep(0.05) self._key_down(VK_BACK) self._key_up(VK_BACK) time.sleep(0.05) self._send_ctrl_combo(VK_V) time.sleep(0.3) self._key_down(VK_RETURN) self._key_up(VK_RETURN) time.sleep(0.5) logger.info(f"[{platform}] 已回复: {text[:60]}") # 记录回复文本到去重集 # 将回复文本加入 _replied_texts（永久集，永不裁切，必拦）和 seen reply_text = text.strip() self._replied_texts.add(reply_text) self._seen_messages.add(reply_text) # 添加换行符归一化变体（跨频道转发时会替换 \n → 空格） if '\n' in reply_text: self._replied_texts.add(reply_text.replace('\n', ' ')) self._replied_texts.add(reply_text.replace('\n', ' ').rstrip('.')) self._seen_messages.add(reply_text.replace('\n', ' ')) self._seen_messages.add(reply_text.replace('\n', ' ').rstrip('.')) for part in reply_text.split('\n'): part = part.strip() if part and len(part) > 3: self._replied_texts.add(part) self._seen_messages.add(part) # 同时追踪去掉 ## 前缀的标题文本（OCR 常丢失 ## 后的空格） if part.startswith('#'): no_hash = re.sub(r'^#+\s*', '', part).strip() if no_hash and len(no_hash) > 2: self._replied_texts.add(no_hash) self._seen_messages.add(no_hash) self._last_reply_time[platform] = time.time() return True except Exception as e: logger.error(f"[{platform}] 回复失败: {e}\n{traceback.format_exc()}") return False finally: # 恢复原始剪贴板 if old is not None: try: wc.OpenClipboard() wc.EmptyClipboard() wc.SetClipboardText(old, wc.CF_UNICODETEXT) wc.CloseClipboard() except Exception: pass # ── 核心：读消息 ────────────────────────────────────────── def _read_messages(self, window, platform: str = "") -> list: """每轮重新定位 ListControl。""" if platform == "qq": # QQ NT: OCR 读屏优先（零依赖，不切焦点，不碰剪贴板） texts = self._ocr_read_qq() if texts: return texts # 截图未变→跳过所有 fallback（clipboard/UIA 只会读旧数据） if self._ocr_unchanged.get("qq"): return [] # OCR 空 → 剪贴板兜底（窗口置前→滚底→全选→复制，内部自带重试） texts = self._read_via_clipboard_once(platform) if texts and len(texts) > 5: return texts # 剪贴板也空 → UIA 最后兜底（可能返回陈旧数据） fresh_win = self._find_qq_window_fresh() if fresh_win: texts = self._read_message_list(fresh_win) if texts: return texts texts = self._read_conversation_list(fresh_win) if texts: return texts try: rect = fresh_win.BoundingRectangle boundary = rect.left + rect.width() * 0.40 for lst in self._list_controls(fresh_win): try: cr = lst.BoundingRectangle if cr.left >= boundary: texts = self._extract_last_line(lst) if texts: logger.debug(f"[S-C] ListControl(x={cr.left}) → {len(texts)} items") return texts except Exception: pass except Exception: pass try: texts = self._collect_filtered_texts(fresh_win, "qq", max_depth=5) if texts: logger.debug(f"[S-D] 全窗文本 → {len(texts)} items") return texts except Exception: pass # Win32 兜底 texts = self._read_win32_texts(platform) if texts: return texts return [] # 微信: OCR 读屏优先（同 QQ 策略，UIA 树不暴露消息内容） texts = self._ocr_read_wechat() if texts: return texts # 截图未变→跳过 fallback if self._ocr_unchanged.get("wechat"): return [] # 微信: UIA 兜底（可能读到陈旧数据或 UI 文字） texts = self._read_message_list(window) if texts: return texts # 兜底：会话列表 texts = self._read_conversation_list(window) if texts: return texts try: rect = window.BoundingRectangle boundary = rect.left + rect.width() * 0.40 for lst in self._list_controls(window): try: cr = lst.BoundingRectangle if cr.left >= boundary: texts = self._extract_last_line(lst) if texts: logger.debug(f"[S-C] ListControl(x={cr.left}) → {len(texts)} items") return texts except Exception: pass except Exception: pass try: texts = self._collect_filtered_texts(window, "wechat", max_depth=5) if texts: logger.debug(f"[S-D] 全窗文本 → {len(texts)} items, last=\"{texts[-1][:50]}\"") return texts except Exception: pass return [] def _read_message_list(self, window) -> list: # 只接受窗口右半侧（>45%）的 ListControl，避开侧栏/文件列表用同名控件 try: wr = window.BoundingRectangle boundary = wr.left + wr.width() * 0.45 except Exception: boundary = 0 for name in ["消息", "Message"]: try: lst = window.ListControl(Name=name) if lst.Exists(maxSearchSeconds=0.2): try: cr = lst.BoundingRectangle if cr.left < boundary: continue # 左侧的"消息"列表（文件/笔记预览）跳过 except Exception: pass self._scroll_to_bottom(lst) texts = self._extract_last_line(lst) if texts: logger.debug(f"[S-A] ListControl({name}) → {len(texts)} items, last=\"{texts[-1][:50]}\"") return texts except Exception: pass return [] def _read_conversation_list(self, window) -> list: try: wr = window.BoundingRectangle boundary = wr.left + wr.width() * 0.45 except Exception: boundary = 0 for name in ["会话", "聊天", "Chat"]: try: lst = window.ListControl(Name=name) if lst.Exists(maxSearchSeconds=0.2): try: cr = lst.BoundingRectangle if cr.left < boundary: continue except Exception: pass 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 except Exception: pass return [] def _scroll_to_bottom(self, lst): """滚动 ListControl 到底部，触发应用加载新消息。""" try: scroll = lst.GetScrollPattern() if scroll: scroll.SetScrollPercent(100, 0) except Exception: pass @staticmethod def _extract_last_line(lst) -> list: texts = [] try: for item in lst.GetChildren(): try: name = item.Name if name and name.strip(): lines = [l.strip() for l in name.split("\n") if l.strip()] if not lines: continue # NFKC 标准化（去除零宽字符后比较） norm = unicodedata.normalize('NFKC', lines[-1]) # 跳过末尾按钮文字（微信消息项的"关闭"等）和时间戳 ui_known = {"关闭", "发送", "接收", "取消", "确定", "更多"} if len(lines) >= 2 and (norm in ui_known or re.match(r'^\d{1,2}:\d{2}', norm)): texts.append(lines[-2]) else: texts.append(lines[-1]) except Exception: pass except Exception: 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 = [] try: for item in lst.GetChildren(): try: name = item.Name if name and name.strip(): lines = [l.strip() for l in name.split("\n") if l.strip()] if len(lines) >= 2: texts.append(lines[1]) elif lines: texts.append(lines[-1]) except Exception: pass except Exception: pass return texts @staticmethod def _list_controls(window): results = [] try: for ctrl in window.GetChildren(): results.extend(_walk_list_controls(ctrl, 0)) except Exception: pass return results @staticmethod def _collect_filtered_texts(window, platform: str = "", max_depth=14) -> list: """深度遍历 + 位置过滤（排除搜索框/标题栏区域）。""" items = [] # [(text, rect_top, rect_left, depth)] def walk(ctrl, depth): if depth > max_depth: return try: cr = ctrl.BoundingRectangle if cr.width() < 20 or cr.height() < 20: return except Exception: return try: name = ctrl.Name if name and name.strip(): t = name.strip() if len(t) >= 2 and not t.isdigit(): items.append((t, cr.top, cr.left, depth)) except Exception: pass try: for child in ctrl.GetChildren(): walk(child, depth + 1) except Exception: pass try: for child in window.GetChildren(): walk(child, 1) except Exception: pass if not items: return [] try: w_rect = window.BoundingRectangle w_width = w_rect.width() w_height = w_rect.height() except Exception: w_width = 1390 w_height = 953 if platform == "qq": filtered = [ t for t, top, left, d in items if top > w_height * 0.08 ] seen = set() deduped = [] for t in filtered: if t not in seen: seen.add(t) deduped.append(t) return deduped return [t for t, _, _, _ in items] # ── 单次轮询 ───────────────────────────────────────────── def _poll(self, platform: str): # 自动回复发送中跳过轮询（防止窗口焦点被抢） if self._sending_event.is_set(): return # 冷却期：回复后 5 秒内跳过 poll，防止反馈回环 last_reply = self._last_reply_time.get(platform, 0.0) if time.time() - last_reply < 5.0: return counter = self._poll_count.get(platform, 0) + 1 self._poll_count[platform] = counter # 1. 确保窗口句柄 # qq 路径纯 Win32+psutil，直接调用不超时；wechat 用 UIA（可能挂死）才套子线程 if platform == "qq": hwnd = self._ensure_hwnd(platform) else: hwnd = self._ensure_hwnd_with_timeout(platform, timeout=12.0) if not hwnd: if counter % 10 == 0 or (counter == 1 and platform == "qq"): logger.debug(f"[{platform}] 窗口未找到") return # QQ 在托盘隐藏（IsWindowVisible=False）时跳过本轮。 # 任何尝试恢复隐藏窗口的操作（SW_SHOWNORMAL 等）在 QQ NT 上都会触发托盘预览弹窗， # 不会恢复主聊天窗口。用户保持 QQ 窗口最小化到任务栏即可。 if platform == "qq" and hwnd and win32gui.IsWindow(hwnd) and not win32gui.IsWindowVisible(hwnd): return # 2. 从 HWND 创建全新 Control（绕过缓存） win = self._fresh_window(platform) if not win: logger.debug(f"[{platform}] 创建窗口控件失败") return # 3. 读消息 texts = self._read_messages(win, platform) if not texts: # 截图未变→不计入空读，不触发清缓存 if self._ocr_unchanged.get(platform): return empty = self._empty_reads.get(platform, 0) + 1 self._empty_reads[platform] = empty if counter % 5 == 0: logger.debug(f"[{platform}] 空读 #{empty}") if empty >= 5: self._cached_hwnd.pop(platform, None) self._cached_pid.pop(platform, None) self._empty_reads[platform] = 0 logger.info(f"[{platform}] 连续空读，清窗口缓存重试") return self._empty_reads[platform] = 0 # 冻结计数器（不在 early-return 里定义，保证所有路径都能访问） stale = 0 # 4a. 全文本签名快速跳过 — OCR 输出与上轮完全相同 → 无变化，跳过 sig = "||".join([t.strip() for t in texts if t.strip()]) if sig == self._last_sig.get(platform, ""): stale = self._stale_count.get(platform, 0) + 1 self._stale_count[platform] = stale return # 4. 用全局 _seen_messages 去重（跨平台、跨轮次防回环） # 精确匹配 + 模糊匹配（OCR 噪声导致精确匹配失效） seen = self._seen_messages replied = self._replied_texts new_items = [] for t in texts: ts = t.strip() if not ts: continue # _replied_texts 优先检查（永不裁切，100% 拦住自己的回复） if ts in replied: continue # _replied_texts 模糊匹配：OCR 噪声导致精确匹配失效（长文本尤甚） if self._fuzzy_duplicate(ts, replied, threshold=0.85): logger.debug(f"[{platform}] 回复模糊跳过: {ts[:50]}") continue # _replied_texts 子串包含：OCR 提取的 md 表格细胞值（如"依赖插件可选/必选"） # 是 bot 回复表的一部分，不是用户提问。长度>5才检查避免误杀短消息。 if len(ts) > 5 and any(ts in r for r in replied): logger.debug(f"[{platform}] 回复子串包含跳过({len(ts)}c): {ts[:50]}") continue if ts in seen: continue # 模糊匹配：OCR 输出每次微小差异（勖彐碎片、行首数字、尾部截断） # 精确匹配命中不了 → 用 difflib 相似度去重 if self._fuzzy_duplicate(ts, seen, threshold=0.80): logger.debug(f"[{platform}] 模糊重复跳过: {ts[:50]}") continue # 超长文本（>200字符）跳过——大概率是堆积历史或 bot 自己的回复，不是单条新消息 if len(ts) > 200: logger.debug(f"[{platform}] 超长文本跳过({len(ts)}c): {ts[:50]}...") continue if self._is_system_message(ts): continue # Markdown 文档过滤：bot 回复以 md 格式为主（#标题、[链接](url)、-列表项）， # 用户日常对话不用 md 语法。拦截 md 格式文本避免 bot 旧回复被重新处理。 if self._is_markdown_document(ts): logger.debug(f"[{platform}] md文档跳过: {ts[:50]}") continue # 纯装饰符号（单字符非汉字非字母数字）跳过 if len(ts) == 1 and not re.match(r'[\u4e00-\u9fff\w]', ts): continue # 只有符号/空格/emoji 的非消息文本（<3 字符且长度等于标点去掉后长度） clean = re.sub(r'[\s\u2000-\u206f\u2190-\u2bff\U0001F000-\U0001FFFF✿♡♥❤★☆]', '', ts) if len(ts) <= 3 and len(clean) == 0: continue seen.add(ts) new_items.append(ts) # _seen_messages 上限 500 条，超了裁到 300 # 防止集合无限膨胀 → 新消息模糊匹配时被历史噪声误杀 if len(seen) > 500: # 保留后 300 条（最新） self._seen_messages = set(list(seen)[-300:]) # 单次 poll 最多处理 3 条新消息，防止系统消息海量涌入触发回环 MAX_NEW_ITEMS = 3 new_items = new_items[:MAX_NEW_ITEMS] # 过滤纯时间/日期文本 → 不应进入 KB 匹配（OCR 时间戳分割遗留的噪音片段） # 这些文本在 KB 子串匹配阶段会命中文件内任何含相同数字的内容 new_items = [ it for it in new_items if not re.match(r'^\d{1,2}[:：]\d{2}(?:[:：]\d{2})?$', it.strip()) and not re.match(r'^\d{4}[/-]\d{1,2}', it.strip()) ] # 更新 last_texts 与签名 self._last_texts[platform] = list(texts) self._last_sig[platform] = sig self._stale_count[platform] = 0 if new_items: # 有纯粹的新文本 → 逐一报告 for item in new_items: self._last_msg[platform] = item engine_tag = self._last_chosen_engine.get(platform, "") logger.info(f"[{platform}] 新消息[{engine_tag}]: {item[:60]}") if self._on_message: self._on_message(platform, item) if platform == "qq": 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) self._cached_pid.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) self._cached_pid.pop(platform, None) def _clean_old_ocr_debug(dir_path: str, keep: int = 20): """清理 ocr_debug_*.png 文件，只保留最近 keep 张。""" try: files = [f for f in os.listdir(dir_path) if f.startswith("ocr_debug_") and f.endswith(".png")] if len(files) <= keep: return files.sort() # 按字母序 = 时间序 for f in files[:-keep]: os.remove(os.path.join(dir_path, f)) except Exception: pass def _walk_list_controls(ctrl, depth): if depth > 6: return [] results = [] try: cname = ctrl.ControlTypeName except Exception: return results if cname == "ListControl": results.append(ctrl) return results try: for child in ctrl.GetChildren(): results.extend(_walk_list_controls(child, depth + 1)) except Exception: pass return results