1888 lines
81 KiB
Python
1888 lines
81 KiB
Python
"""消息监控 — 全列表指纹对比 + 窗口冻结检测 + ControlFromHandle 强制刷新"""
|
||
|
||
import time
|
||
import logging
|
||
import threading
|
||
import traceback
|
||
from typing import Optional, Callable, List
|
||
|
||
import uiautomation as auto
|
||
import win32gui
|
||
import win32con
|
||
import win32api
|
||
import win32process
|
||
import re
|
||
import ctypes
|
||
import difflib
|
||
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
|
||
|
||
logging.getLogger("comtypes").setLevel(logging.WARNING)
|
||
logger = logging.getLogger("msgbot.monitor")
|
||
|
||
# EasyOCR 备用引擎（可选依赖）
|
||
try:
|
||
import easyocr as _easyocr
|
||
import numpy as _np
|
||
_HAS_EASYOCR = True
|
||
except (ImportError, OSError):
|
||
_HAS_EASYOCR = False
|
||
_easyocr = None
|
||
_np = None
|
||
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 引擎（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: int = 0
|
||
|
||
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
|
||
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":
|
||
# 1. Win32 FindWindow 直搜 TXGuiFoundation（旧版 QQ 主窗口）
|
||
# ⚠️ 必须检查 IsWindowVisible：QQ NT 在托盘时有一个不可见的 TXGuiFoundation
|
||
# 托盘辅助窗口（空标题, IsWindowVisible=False），FindWindow 可能返回这个。
|
||
# 不可见的窗口被缓存后，_poll 里 IsWindowVisible 检查会跳过所有后续 OCR。
|
||
try:
|
||
hwnd = win32gui.FindWindow("TXGuiFoundation", None)
|
||
if hwnd:
|
||
top = win32gui.GetAncestor(hwnd, win32con.GA_ROOT)
|
||
if win32gui.IsWindow(top) and win32gui.IsWindowVisible(top):
|
||
rect = win32gui.GetWindowRect(top)
|
||
if (rect[2] - rect[0]) > 200 and (rect[3] - rect[1]) > 100:
|
||
title = win32gui.GetWindowText(top)
|
||
self._app_variant[platform] = "tim" if "TIM" in title.upper() else "qq_nt"
|
||
self._cached_hwnd[platform] = top
|
||
self._cached_pid[platform] = win32process.GetWindowThreadProcessId(top)[1]
|
||
logger.info(f"[{platform}] 窗口已定位(Win32 TXGuiFoundation): {title} top={top}")
|
||
return top
|
||
except Exception:
|
||
pass
|
||
# 2. Win32 EnumWindows 兜底 — 类名+标题搜索
|
||
# QQ NT (Electron) 类名 Chrome_WidgetWin_1, 传统 QQ 类名 TXGuiFoundation
|
||
qq_titles = ("QQ", "TIM")
|
||
def _enum_callback(hwnd, results):
|
||
try:
|
||
cls = win32gui.GetClassName(hwnd)
|
||
title = win32gui.GetWindowText(hwnd)
|
||
title_upper = title.upper()
|
||
# 类名匹配（传统 QQ / TIM）
|
||
if cls == "TXGuiFoundation":
|
||
results.append(hwnd)
|
||
return True
|
||
# QQ NT Electron: 类名含 Chrome，标题含 QQ/TIM
|
||
if "Chrome" in cls and ("QQ" in title_upper or "TIM" in title_upper):
|
||
results.append(hwnd)
|
||
return True
|
||
# 标题完全匹配 QQ / TIM
|
||
if title_upper in qq_titles:
|
||
results.append(hwnd)
|
||
return True
|
||
except Exception:
|
||
pass
|
||
return True
|
||
try:
|
||
found = []
|
||
win32gui.EnumWindows(_enum_callback, found)
|
||
logger.debug(f"[{platform}] EnumWindows 候选: {len(found)} windows: "
|
||
f"{[(win32gui.GetWindowText(h)[:30], hex(h)) for h in found]}")
|
||
for hwnd in found:
|
||
top = win32gui.GetAncestor(hwnd, win32con.GA_ROOT)
|
||
if not (win32gui.IsWindow(top) and win32gui.IsWindowVisible(top)):
|
||
cls = win32gui.GetClassName(top)
|
||
title = win32gui.GetWindowText(top)
|
||
logger.debug(f"[{platform}] Enum 候选跳过(不可见): cls={cls} title={title[:30]} vis={win32gui.IsWindowVisible(top)}")
|
||
continue
|
||
rect = win32gui.GetWindowRect(top)
|
||
if not ((rect[2] - rect[0]) > 200 and (rect[3] - rect[1]) > 100):
|
||
cls = win32gui.GetClassName(top)
|
||
title = win32gui.GetWindowText(top)
|
||
logger.debug(f"[{platform}] Enum 候选跳过(尺寸): cls={cls} title={title[:30]} rect={rect}")
|
||
continue
|
||
cls = win32gui.GetClassName(top)
|
||
title = win32gui.GetWindowText(top)
|
||
title_upper = title.upper().strip()
|
||
# _enum_callback 只加了 QQ/TIM 相关窗口，所以 found 里都是有效候选
|
||
# 优先选标题含 QQ/TIM 的（传统模式），其次任何 TXGuiFoundation 或 Chrome 都接受
|
||
if "TIM" in title_upper:
|
||
self._app_variant[platform] = "tim"
|
||
else:
|
||
self._app_variant[platform] = "qq_nt"
|
||
self._cached_hwnd[platform] = top
|
||
self._cached_pid[platform] = win32process.GetWindowThreadProcessId(top)[1]
|
||
if "Chrome" in cls:
|
||
logger.info(f"[{platform}] 窗口已定位(Electron Chrome): {title} top={top}")
|
||
else:
|
||
logger.info(f"[{platform}] 窗口已定位(Win32 Enum): {title} top={top}")
|
||
return top
|
||
except Exception:
|
||
pass
|
||
# 3. UIA 按窗口名搜索（兜底，可能慢但不挂死）
|
||
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:
|
||
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}] 窗口已定位(UIA 名称): {name} top={top}")
|
||
return top
|
||
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 文档由 pre-split check 提前过滤，不受影响）。
|
||
"""
|
||
# 所有紧邻 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 聊天大量使用 <player> 等尖括号，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)
|
||
|
||
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 秒）...")
|
||
self._easyocr_reader = _easyocr.Reader(['ch_sim', 'en'], gpu=False)
|
||
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) -> str:
|
||
"""双引擎对比，选质量更高的结果。日志记录引擎胜负统计。"""
|
||
if not easy_text:
|
||
return win_text
|
||
if not win_text:
|
||
return easy_text
|
||
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 if chosen == "windows" else easy_text
|
||
|
||
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):
|
||
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
|
||
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 []
|
||
# 保存截屏到程序根目录用于调试（保留最近 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')
|
||
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()
|
||
|
||
# EasyOCR 备用
|
||
easy_text = ""
|
||
if _HAS_EASYOCR:
|
||
self._init_easyocr()
|
||
if self._easyocr_reader:
|
||
try:
|
||
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"[ocr] EasyOCR 识别失败: {e}")
|
||
|
||
if win_text and easy_text:
|
||
logger.debug(f"[ocr] Windows OCR: {win_text[:100]!r}...")
|
||
logger.debug(f"[ocr] EasyOCR: {easy_text[:100]!r}...")
|
||
|
||
raw_text = self._pick_ocr_text(win_text, easy_text, im)
|
||
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()]
|
||
# ⭐ debug: 记录 OCR 原始行内容，定位过滤导致空白的原因
|
||
for i, rl in enumerate(ocr_lines):
|
||
logger.info(f"[ocr] raw[{i}] raw_line<{rl!r}>")
|
||
# QQ NT OCR 将所有可见消息合并为一段无换行文本，消息间以时间戳分隔
|
||
# 按时间戳分割恢复独立消息，避免整坨变成一条"新消息"
|
||
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}")
|
||
# 先清理 & 碎片（OCR 把时间戳右边缘的 : 读成 &）
|
||
# 必须在时间戳分割前做，否则 "13：5&07" 无法被时间戳正则捕获
|
||
rl = rl.replace('&', '')
|
||
# 把全角 ：统一成半角 :，让后续正则统一匹配
|
||
rl = rl.replace('：', ':')
|
||
logger.debug(f"[ocr] stage[{i}] clean: len={len(rl)} -> {rl[:120]!r}")
|
||
# 分割"msg1 14:24 msg2 13:27 msg3" → ["msg1", "msg2", "msg3"]
|
||
# 使用更宽松的 \d{1,2}（允许单数字），因为 & 移除后 "5&07"→"507"
|
||
# 打破的时间戳如 "13:5 07" 也尽量捕获
|
||
rl = re.sub(r'\s+', ' ', rl) # 多重空格压为一个
|
||
# ⭐ 分割前 md 预检：rl 中 md 语法完整（链接、标题等），分割后碎片会丢失语法
|
||
if self._is_markdown_document(rl):
|
||
logger.debug(f"[ocr] stage[{i}] markdown document detected pre-split, skip line")
|
||
continue
|
||
parts = re.split(r'(\d{1,2}:\d{1,2}(?::\d{1,2})?)', rl)
|
||
logger.debug(f"[ocr] stage[{i}] split -> {len(parts)} parts: {[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):
|
||
# 遇到时间戳 → 前面的 buf 为一条完整消息
|
||
if buf.strip():
|
||
expanded.append(buf.strip())
|
||
buf = ''
|
||
else:
|
||
buf += part
|
||
if buf.strip():
|
||
expanded.append(buf.strip())
|
||
ocr_lines = expanded
|
||
texts = []
|
||
for line in ocr_lines:
|
||
# 注: OCR 行已在分割时 normalize+全角:转换+&清理，此处不再重复
|
||
if not line:
|
||
continue
|
||
# ⭐ QQ NT 界面固定噪声：整行匹配直接跳过
|
||
if line in ("查看更多", "收起", "选择会话开始聊天"):
|
||
continue
|
||
# 连续纯标点+符号（emoji、装饰字符等）
|
||
if re.match(r'^[✿♡♥❤★☆✦✧❀❁❂❃❄❅❆❇❈❉❊❋➠➡➢➣➤.·•●○◦◎◉◍◐◑◒◓◔◕◖◗◢◣◤◥☺☻☼♠♣♦♤♧♡♢♔♕♖♗♘♙🕐-🕧\s]{1,3}$', line):
|
||
continue
|
||
# 去噪：去掉行首未读红点数字（如 "0"）和行尾删除按钮误识别（勖彐 等）
|
||
# 否则同一消息每次 OCR 输出微小差异 → 精确去重失效 → 重复触发
|
||
line = re.sub(r'^0+', '', line)
|
||
line = re.sub(r'[勖彐"\'\s\u2000-\u206f]+$', '', line)
|
||
if not line:
|
||
continue
|
||
# 过滤 KB 预览文字（行起始）: "问题1：你叫什么名字答：你好我叫123"
|
||
# 注意：行中 KB 预览在分割前 rl 阶段已按整行丢弃，此处只过滤以问题开头的残留
|
||
if re.match(r'^问题\d+\s*[:：]', line):
|
||
continue
|
||
# 过滤纯时间戳垃圾：整段只有数字、冒号、空格（如 "13:5 07 14:2 15"）
|
||
if re.match(r'^[\d:\s]+$', line):
|
||
continue
|
||
# 半角全角数字混杂的界面碎片（如 "1.0s 0．35后监控应用设置"）
|
||
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
|
||
# 群成员侧栏/搜索残留（QQ/TIM 右栏成员列表）
|
||
if re.search(r'群成员|索群成员|搜索成员', line):
|
||
logger.debug(f"[qq] OCR 过滤: 群成员侧栏 line={line[:60]!r}")
|
||
continue
|
||
# 聊天面板顶部群名头栏：群名（角色〕/ 【群名】/ 群名[标签]
|
||
# 特有签名：使用 `〕` `」` `】` 等非日常聊天中出现的闭合括号
|
||
if re.match(r'^[^）】〕」]{2,40}[（【〔「][^）】〕」]{1,15}[）】〕」]', line):
|
||
logger.debug(f"[qq] OCR 过滤: 群名头栏 line={line[:60]!r}")
|
||
continue
|
||
# 纯中文短文本（1 字符）— UI 噪声碎片
|
||
if re.match(r'^[一-鿿]{1}$', line):
|
||
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"[qq] OCR -> {len(ocr_lines)} raw, {len(texts)} filtered, "
|
||
f"last=\"{texts[-1][:50] if texts else ''}\"")
|
||
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）。
|
||
|
||
去重由 _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:
|
||
# 跳过：系统消息、纯时间、纯数字、无意义短行
|
||
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+<key>: 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 _is_system_message(text: str) -> bool:
|
||
"""过滤机器人/系统消息，不作为用户消息处理。"""
|
||
if not text:
|
||
return True
|
||
# 系统前缀/标记
|
||
system_markers = [
|
||
"✅", "❌", "📝", "🔍", "📄", "➡", "♡", "✿",
|
||
"[WECHAT]", "[QQ]", "[wechat]", "[qq]",
|
||
]
|
||
for m in system_markers:
|
||
if text.startswith(m):
|
||
return True
|
||
# ⭐ QQ NT 界面噪声文字（按钮/提示/碎片，非用户消息）
|
||
qq_ui_noise = [
|
||
"查看更多", "可查看在其他设备产生的聊天记录",
|
||
"收藏", "转发", "多选", "引用", "回复",
|
||
"选择会话开始聊天",
|
||
]
|
||
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 or len(text) < 10:
|
||
return False
|
||
# 行首是 # 标题（# ## ### #### 等）
|
||
if re.match(r'^#{1,6}\s', 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
|
||
return False
|
||
|
||
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)
|
||
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
|
||
|
||
# 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 []
|
||
|
||
# 微信: 优先读"消息"（ScrollPattern 滚底后实时）
|
||
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 lines:
|
||
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. 确保窗口句柄（带超时保护，防止 UIA COM 调用挂死不返回）
|
||
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:
|
||
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
|
||
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
|
||
logger.info(f"[{platform}] 新消息: {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
|