1006 lines
37 KiB
Python
1006 lines
37 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 re
|
||
import ctypes
|
||
|
||
logging.getLogger("comtypes").setLevel(logging.WARNING)
|
||
logger = logging.getLogger("msgbot.monitor")
|
||
|
||
|
||
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 = {}
|
||
|
||
# 连续相同签名计数（用于冻结检测）
|
||
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()
|
||
# 上次回复时间戳（冷却期内跳过该平台 poll）
|
||
self._last_reply_time: dict = {} # platform → timestamp
|
||
|
||
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._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
|
||
while self._running:
|
||
try:
|
||
if self._wechat_enabled and not self._sending_event.is_set():
|
||
self._poll("wechat")
|
||
if self._qq_enabled:
|
||
# QQ 降频：每 3 周期 poll 一次（~6s）
|
||
self._qq_skip_counter += 1
|
||
if self._qq_skip_counter >= 3 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)
|
||
|
||
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
|
||
logger.info(f"[{platform}] 窗口已定位: {name} handle={top}")
|
||
return top
|
||
except Exception:
|
||
pass
|
||
elif platform == "qq":
|
||
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._app_variant[platform] = "qq_nt" if name == "QQ" else "tim"
|
||
logger.info(f"[{platform}] 窗口已定位(名称): {name} top={top}")
|
||
return top
|
||
except Exception:
|
||
pass
|
||
# QQ NT/TIM → TXGuiFoundation 类名
|
||
try:
|
||
root = auto.GetRootControl()
|
||
for child in root.GetChildren():
|
||
try:
|
||
if (child.ClassName or "") == "TXGuiFoundation":
|
||
rect = child.BoundingRectangle
|
||
if rect.width() > 200 and rect.height() > 100:
|
||
h = child.NativeWindowHandle
|
||
top = win32gui.GetAncestor(h, win32con.GA_ROOT)
|
||
cname = str(child.Name or "")
|
||
self._app_variant[platform] = "tim" if "TIM" in cname.upper() else "qq_nt"
|
||
self._cached_hwnd[platform] = top
|
||
logger.info(f"[{platform}] 窗口已定位(TXGuiFoundation): {child.Name} top={top}")
|
||
return top
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
# ── 强制刷新 UIA ──────────────────────────────────────────
|
||
|
||
def _fresh_window(self, platform: str):
|
||
"""从缓存 HWND 创建全新的 WindowControl（绕过 UIA 缓存）。"""
|
||
hwnd = self._cached_hwnd.get(platform)
|
||
if not hwnd:
|
||
return None
|
||
try:
|
||
# 用 NativeWindowHandle 重建，保证 UIA 新缓存
|
||
win = auto.ControlFromHandle(hwnd)
|
||
if win is not None:
|
||
return win
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
def _force_uia_refresh(self, platform: str):
|
||
"""通过交互强制窗口更新 UIA 控件树。"""
|
||
hwnd = self._cached_hwnd.get(platform)
|
||
if not hwnd:
|
||
return
|
||
try:
|
||
# 方式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)
|
||
|
||
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
|
||
|
||
def _read_via_clipboard(self, platform: str) -> list:
|
||
"""剪贴板法：聚焦窗口→Ctrl+A→Ctrl+C→读剪贴板（适用 Electron 应用）。
|
||
|
||
稳妥可靠但会切焦点+清用户剪贴板（自动恢复原内容）。
|
||
自动重试 1 次若首次结果为空。
|
||
"""
|
||
texts = self._read_via_clipboard_once(platform)
|
||
if texts:
|
||
return texts
|
||
# 重试一次：再次 SetForegroundWindow + SendKeys
|
||
logger.debug(f"[{platform}] 剪贴板首次空，重试")
|
||
time.sleep(0.5)
|
||
texts = self._read_via_clipboard_once(platform)
|
||
return texts
|
||
|
||
def _read_via_clipboard_once(self, platform: str) -> list:
|
||
"""剪贴板法单次执行。"""
|
||
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 = []
|
||
try:
|
||
# 确保窗口在前台（失败则返回空，避免发送到错误窗口）
|
||
if not self._ensure_foreground(hwnd, max_retries=2):
|
||
return []
|
||
|
||
# 注意：不再执行 {Esc}{Esc}{Down}{Enter} 键盘导航，避免每次 poll
|
||
# 切换到不同会话导致读到不同消息集。直接 Ctrl+A/C 读取当前可见对话。
|
||
# 若需滚动到底部，用 PostMessage（不影响焦点）
|
||
hwnd_current = self._cached_hwnd.get(platform)
|
||
if hwnd_current:
|
||
win32gui.PostMessage(hwnd_current, win32con.WM_VSCROLL, win32con.SB_BOTTOM, 0)
|
||
|
||
# 用 SendInput 替代 auto.SendKeys（避免 ord() Unicode 解析 bug）
|
||
VK_A = 0x41
|
||
VK_C = 0x43
|
||
# Ctrl+A (全选)
|
||
self._send_ctrl_combo(VK_A)
|
||
time.sleep(0.15)
|
||
# Ctrl+C (复制)
|
||
self._send_ctrl_combo(VK_C)
|
||
time.sleep(0.3)
|
||
|
||
wc.OpenClipboard()
|
||
if wc.IsClipboardFormatAvailable(wc.CF_UNICODETEXT):
|
||
content = wc.GetClipboardData(wc.CF_UNICODETEXT)
|
||
if content:
|
||
texts = [l.strip() for l in content.strip().split('\n') if l.strip()]
|
||
self._raw_clipboard[platform] = content
|
||
wc.CloseClipboard()
|
||
|
||
logger.debug(f"[{platform}] 剪贴板读取 → {len(texts)} lines")
|
||
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_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 _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
|
||
# 系统事件模板
|
||
system_patterns = [
|
||
"已自动回复", "知识库中未找到", "找到相关结果",
|
||
"监控已启动", "监控已暂停", "设置已保存",
|
||
"🤖", "找不到联系人", "当前会话回复",
|
||
"自动回复触发", "匹配度",
|
||
]
|
||
for p in system_patterns:
|
||
if p in text:
|
||
return True
|
||
# 纯时间格式（不含消息内容）
|
||
if re.match(r'^\d{1,2}:\d{2}', 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()
|
||
|
||
# 确保窗口可见且在前台（处理托盘/最小化状态）
|
||
if not win32gui.IsWindowVisible(hwnd):
|
||
win32gui.ShowWindow(hwnd, win32con.SW_SHOWNORMAL)
|
||
time.sleep(0.2)
|
||
if win32gui.IsIconic(hwnd):
|
||
win32gui.ShowWindow(hwnd, win32con.SW_RESTORE)
|
||
time.sleep(0.2)
|
||
if not self._ensure_foreground(hwnd):
|
||
logger.error(f"[{platform}] 无法带到前台，放弃发送")
|
||
return False
|
||
|
||
# PostMessage 点击输入框区域（底部）— 不移动物理鼠标
|
||
rect = win32gui.GetWindowRect(hwnd)
|
||
if rect:
|
||
left, top, right, bottom = rect
|
||
w = right - left
|
||
h = bottom - top
|
||
click_x = int(w * 0.50)
|
||
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)
|
||
|
||
# 用 SendInput (keybd_event) 替代 auto.SendKeys
|
||
# 避免 uiautomation SendKeys 的 ord() Unicode 解析 bug
|
||
VK_A = 0x41
|
||
VK_V = 0x56
|
||
VK_BACK = 0x08
|
||
VK_RETURN = 0x0D
|
||
|
||
# Ctrl+A (全选)
|
||
self._send_ctrl_combo(VK_A)
|
||
time.sleep(0.05)
|
||
# Backspace (清空)
|
||
self._key_down(VK_BACK)
|
||
self._key_up(VK_BACK)
|
||
time.sleep(0.05)
|
||
# Ctrl+V (粘贴)
|
||
self._send_ctrl_combo(VK_V)
|
||
time.sleep(0.3)
|
||
# Enter (发送)
|
||
self._key_down(VK_RETURN)
|
||
self._key_up(VK_RETURN)
|
||
time.sleep(0.5)
|
||
|
||
logger.info(f"[{platform}] 已回复: {text[:60]}")
|
||
# 将回复文本加入全局 seen 集合，多平台去重
|
||
reply_text = text.strip()
|
||
self._seen_messages.add(reply_text)
|
||
# 添加换行符归一化变体（跨频道转发时会替换 \n → 空格）
|
||
if '\n' in reply_text:
|
||
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._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 UIA ListControl 缓存易过期，跳过 UIA 直接走剪贴板
|
||
# 使用 _once 避免 _read_via_clipboard 的双重重试（减少前台抢占）
|
||
texts = self._read_via_clipboard_once(platform)
|
||
if texts:
|
||
return texts
|
||
|
||
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:
|
||
for name in ["消息", "Message"]:
|
||
try:
|
||
lst = window.ListControl(Name=name)
|
||
if lst.Exists(maxSearchSeconds=0.2):
|
||
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:
|
||
for name in ["会话", "聊天", "Chat"]:
|
||
try:
|
||
lst = window.ListControl(Name=name)
|
||
if lst.Exists(maxSearchSeconds=0.2):
|
||
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. 确保窗口句柄
|
||
hwnd = self._ensure_hwnd(platform)
|
||
if not hwnd:
|
||
if counter % 10 == 0:
|
||
logger.debug(f"[{platform}] 窗口未找到")
|
||
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._empty_reads[platform] = 0
|
||
logger.info(f"[{platform}] 连续空读，清窗口缓存重试")
|
||
return
|
||
|
||
self._empty_reads[platform] = 0
|
||
|
||
# 4. 用全局 _seen_messages 去重（跨平台、跨轮次防回环）
|
||
seen = self._seen_messages
|
||
new_items = []
|
||
for t in texts:
|
||
ts = t.strip()
|
||
if (ts and ts not in seen
|
||
and not self._is_system_message(ts)):
|
||
seen.add(ts)
|
||
new_items.append(ts)
|
||
# 控制集合大小防内存泄漏
|
||
if len(seen) > 500:
|
||
self._seen_messages = set(list(seen)[-300:])
|
||
|
||
# 单次 poll 最多处理 3 条新消息，防止系统消息海量涌入触发回环
|
||
MAX_NEW_ITEMS = 3
|
||
new_items = new_items[:MAX_NEW_ITEMS]
|
||
|
||
# 更新 last_texts 供其他逻辑参考（但去重不依赖它）
|
||
self._last_texts[platform] = list(texts)
|
||
|
||
if new_items:
|
||
# 有纯粹的新文本 → 逐一报告
|
||
sig = "||".join([t.strip() for t in texts if t.strip()])
|
||
self._last_sig[platform] = sig
|
||
self._stale_count[platform] = 0
|
||
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)
|
||
else:
|
||
# 无新文本 → 用全签名判断是否内容变化（emoji 刷新等）
|
||
sig = "||".join([t.strip() for t in texts if t.strip()])
|
||
old_sig = self._last_sig.get(platform, "")
|
||
if sig != old_sig:
|
||
self._last_sig[platform] = sig
|
||
self._stale_count[platform] = 0
|
||
else:
|
||
# 相同签名 → 冻结计数
|
||
stale = self._stale_count.get(platform, 0) + 1
|
||
self._stale_count[platform] = stale
|
||
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)
|
||
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)
|
||
|
||
|
||
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
|