OCR 终极防御：进程验证+最小化检测+垃圾内容过滤

QQ 被杀/最小化后 OCR 截到终端窗口导致死循环回复。三重防御： 1. _validate_qq_window() — IsWindowVisible+IsIconic+进程名(QT/TIM.exe)验证 2. _is_ocr_console_garbage() — 内容级检测：Python模块名/脚本路径/日志关键词/无中文长文本 3. 连续3次无效自动清缓存清PID防死循环新增 _cached_pid 字典，与 _cached_hwnd 同步生命周期
2026-07-25 23:30:37 +08:00
parent 85b790cb81
commit f90e3fbaaf
1 changed files with 218 additions and 38 deletions
@@ -10,8 +10,13 @@ import uiautomation as auto
 import win32gui
 import win32con
 import win32api
+import win32process
 import re
 import ctypes
+from PIL import ImageGrab
+from winrt.windows.media.ocr import OcrEngine
+from winrt.windows.globalization import Language
+from winrt.windows.graphics.imaging import SoftwareBitmap, BitmapPixelFormat

 logging.getLogger("comtypes").setLevel(logging.WARNING)
 logger = logging.getLogger("msgbot.monitor")
@@ -41,7 +46,8 @@ class MessageMonitor:

        # 窗口句柄缓存（存 HWND，不存 Control 引用）
        self._cached_hwnd: dict = {}
-
+        # 窗口对应进程 ID（验证窗口是否仍属同一进程）
+        self._cached_pid: dict = {}
        # 连续相同签名计数（用于冻结检测）
        self._stale_count: dict = {}
        # 连续空读
@@ -61,6 +67,14 @@ class MessageMonitor:
        # 上次回复时间戳（冷却期内跳过该平台 poll）
        self._last_reply_time: dict = {}  # platform → timestamp

+        # OCR 引擎（Windows 原生，lazy init）
+        self._ocr_engine = None
+        # 上次 OCR 原始文本（用于增量检测）
+        self._ocr_last_text: dict = {}
+
+        # OCR 垃圾连续计数（连续垃圾超过阈值清缓存重找）
+        self._ocr_garbage_count: int = 0
+
    def set_poll_interval(self, seconds: float):
        self._poll_interval = max(0.5, seconds)

@@ -85,6 +99,7 @@ class MessageMonitor:
            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()
@@ -110,9 +125,9 @@ class MessageMonitor:
                if self._wechat_enabled and not self._sending_event.is_set():
                    self._poll("wechat")
                if self._qq_enabled:
-                    # QQ 降频：每 3 周期 poll 一次（~6s）
+                    # QQ 降频：每 4 周期 poll 一次（~8s，OCR 计算略重）
                    self._qq_skip_counter += 1
-                    if self._qq_skip_counter >= 3 and not self._sending_event.is_set():
+                    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:
@@ -136,6 +151,7 @@ class MessageMonitor:
            pass

        self._cached_hwnd.pop(platform, None)
+        self._cached_pid.pop(platform, None)

        if platform == "wechat":
            for name in ["微信", "WeChat"]:
@@ -148,6 +164,7 @@ class MessageMonitor:
                            # 确保存顶层窗口句柄（与 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:
@@ -162,6 +179,7 @@ class MessageMonitor:
                            h = w.NativeWindowHandle
                            top = win32gui.GetAncestor(h, win32con.GA_ROOT)
                            self._cached_hwnd[platform] = top
+                            self._cached_pid[platform] = win32process.GetWindowThreadProcessId(top)[1]
                            self._app_variant[platform] = "qq_nt" if name == "QQ" else "tim"
                            logger.info(f"[{platform}] 窗口已定位(名称): {name} top={top}")
                            return top
@@ -180,6 +198,7 @@ class MessageMonitor:
                                cname = str(child.Name or "")
                                self._app_variant[platform] = "tim" if "TIM" in cname.upper() else "qq_nt"
                                self._cached_hwnd[platform] = top
+                                self._cached_pid[platform] = win32process.GetWindowThreadProcessId(top)[1]
                                logger.info(f"[{platform}] 窗口已定位(TXGuiFoundation): {child.Name} top={top}")
                                return top
                    except Exception:
@@ -391,26 +410,189 @@ class MessageMonitor:
            pass
        return texts

-    def _read_via_clipboard(self, platform: str) -> list:
-        """剪贴板法：聚焦窗口→Ctrl+A→Ctrl+C→读剪贴板（适用 Electron 应用）。
+    # ── OCR 读屏（QQ 专用，Windows 原生 OCR） ──────────────────

-        稳妥可靠但会切焦点+清用户剪贴板（自动恢复原内容）。
-        自动重试 1 次若首次结果为空。
+    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 词内空格保留）
        """
-        texts = self._read_via_clipboard_once(platform)
-        if texts:
+        # 所有紧邻 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
+        )
+        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 进程。
+
+        检查项（全过才返回 True）：
+        1. 窗口存在且可见
+        2. 窗口未最小化
+        3. 所属进程名称匹配 QQ.exe / TIM.exe
+        """
+        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', 'qq.exe', 'tim.exe'):
+                    return True
+        except Exception:
+            pass
+        return False
+
+    def _ocr_read_qq(self) -> list:
+        """OCR 截屏 QQ 消息区域 -> Windows OCR 提取文本。
+        包含窗口验证 + 垃圾过滤 + 连续垃圾清缓存。
+        """
+        hwnd = self._cached_hwnd.get("qq")
+        if not hwnd:
+            return []
+        if not self._validate_qq_window(hwnd):
+            self._ocr_garbage_count += 1
+            if self._ocr_garbage_count == 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 self._ocr_garbage_count >= 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 = 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
+            cap_left = int(left + win_w * 0.30)
+            cap_top = top + 35
+            cap_right = int(left + win_w * 0.55)
+            cap_bottom = bottom - 80
+            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 []
+            if im.mode != 'RGBA':
+                im = im.convert('RGBA')
+            sb = SoftwareBitmap(BitmapPixelFormat.RGBA8, im.width, im.height)
+            sb.copy_from_buffer(im.tobytes())
+            result = engine.recognize_async(sb).get()
+            raw_text = result.text.strip()
+            if not raw_text:
+                return []
+            if self._is_ocr_console_garbage(raw_text):
+                self._ocr_garbage_count += 1
+                if self._ocr_garbage_count >= 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 = 0
+                return []
+            self._ocr_garbage_count = 0
+            self._ocr_last_text["qq"] = raw_text
+            ocr_lines = [l.strip() for l in raw_text.split('\n') if l.strip()]
+            texts = []
+            for line in ocr_lines:
+                line = self._normalize_ocr_text(line)
+                if not line:
+                    continue
+                if self._is_system_message(line):
+                    continue
+                if re.match(r'^[一-鿿·A-Za-z0-9_\-]{1,10}$', 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
+                texts.append(line)
+            logger.debug(f"[qq] OCR -> {len(ocr_lines)} raw, {len(texts)} filtered, "
+                         f"last=\"{texts[-1][:50] if texts else ''}\"")
            return texts
-        # 重试一次：再次 SetForegroundWindow + SendKeys
-        logger.debug(f"[{platform}] 剪贴板首次空，重试")
-        time.sleep(0.5)
-        texts = self._read_via_clipboard_once(platform)
-        return texts
+        except Exception as e:
+            logger.debug(f"[qq] OCR失败: {e}")
+            return []

    def _read_via_clipboard_once(self, platform: str) -> list:
        """剪贴板法单次执行（安全键盘导航：Enter + Ctrl+End + Ctrl+A/C，无 Esc）。

-        返回前把所有原始行预塞进 _seen_messages，再过滤只返回"像用户消息"的行。
-        这样 caller 不会把发送者名/时间/历史消息当新消息处理。
+        去重由 _poll 的 _replied_texts + _seen_messages 负责，本函数只过滤并返回。
+        回复回环防御由 _paste_and_send 写入 _replied_texts 承担。
        """
        hwnd = self._cached_hwnd.get(platform)
        if not hwnd or not win32gui.IsWindow(hwnd):
@@ -505,14 +687,6 @@ class MessageMonitor:

            logger.debug(f"[{platform}] 剪贴板读取 → {len(raw_lines)} lines")

-            # ── 关键：所有原始行预塞进 _seen_messages，避免 caller 当新消息处理 ──
-            self._seen_messages.update(raw_lines)
-            # 回复文本的变体也预塞
-            for line in raw_lines:
-                line2 = line.replace('\n', ' ').rstrip('.')
-                if line2 != line and line2:
-                    self._seen_messages.add(line2)
-
            # ── 过滤：只返回"像用户消息"的行 ──
            sender_pattern = re.compile(r'^[\u4e00-\u9fff·A-Za-z0-9_\-]{1,10}$')
            for line in raw_lines:
@@ -821,8 +995,20 @@ class MessageMonitor:
    def _read_messages(self, window, platform: str = "") -> list:
        """每轮重新定位 ListControl。"""
        if platform == "qq":
-            # QQ NT: UIA 优先（但重建窗口绕过缓存）+ 剪贴板兜底
-            # 第一步：用新窗口流式重建 UIA 树（不用 ControlFromHandle 缓存）
+            # QQ NT: OCR 读屏优先（零依赖，不切焦点，不碰剪贴板）
+            texts = self._ocr_read_qq()
+            if texts:
+                return texts
+
+            # OCR 空 → 剪贴板兜底（窗口置前→滚底→全选→复制）
+            texts = self._read_via_clipboard_once(platform)
+            if texts and len(texts) > 5:
+                return texts
+            texts = self._read_via_clipboard(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)
@@ -854,16 +1040,7 @@ class MessageMonitor:
                except Exception:
                    pass

-            # 剪贴板兜底：仅当 UIA 完全没读到内容时启用
-            texts = self._read_via_clipboard_once(platform)
-            # ＜=5 行的剪贴板很可能是输入框内容（非消息列表）→ 丢弃
-            if texts and len(texts) > 5:
-                return texts
-            texts = self._read_via_clipboard(platform)
-            if texts and len(texts) > 5:
-                return texts
-
-            # 最终兜底
+            # Win32 兜底
            texts = self._read_win32_texts(platform)
            if texts:
                return texts
@@ -1109,6 +1286,7 @@ class MessageMonitor:
                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
@@ -1130,8 +1308,8 @@ class MessageMonitor:
                and not self._is_system_message(ts)):
                seen.add(ts)
                new_items.append(ts)
-        # 不再主动截断 _seen_messages—剪贴板全行已预塞，截断会杀掉历史信息引发回环
-        # 只有极端情况（>50万行）才清，但这个场景几乎不会出现
+        # 不截断 _seen_messages — _replied_texts 已拦截回复回环，截断会误放历史消息
+        # 极端情况（>50万行）才手动清，几乎不会出现

        # 单次 poll 最多处理 3 条新消息，防止系统消息海量涌入触发回环
        MAX_NEW_ITEMS = 3
@@ -1168,6 +1346,7 @@ class MessageMonitor:
                    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}，尝试强制刷新")
@@ -1175,6 +1354,7 @@ class MessageMonitor:
                    elif stale == 10:
                        logger.debug(f"[{platform}] 冻结 #{stale}，重建窗口引用")
                        self._cached_hwnd.pop(platform, None)
+                        self._cached_pid.pop(platform, None)


 def _walk_list_controls(ctrl, depth):