""" 企业微信 PC 端 GUI 自动化机器人 v3.1 ===================================== 【技术路线】 由于企业微信使用完全自定义的 GPU 渲染引擎（UIA 节点=0，Win32子窗口=0）， 所有 UIAutomation / win32gui 内部控件查询均无效。 本脚本采用： 1. win32gui → 查找窗口 HWND、强制还原窗口、获取真实坐标 2. mss.MSS → 高速截图 3. numpy → 红点色彩识别 4. pyautogui → 鼠标点击 + 键盘输入 5. pyperclip → 剪贴板粘贴（防止中文字符乱码） 【安装依赖】 pip install mss Pillow numpy pyautogui pyperclip pywin32 【运行方式】 python wechat_bot.py # 正式监听模式 python wechat_bot.py --calibrate # 标定模式（验证区域划分是否正确） python wechat_bot.py --test-input # 测试输入框点击 """ import sys import time import json import os import re import hashlib import ctypes import win32gui import win32con import win32ui import win32process import win32api import pyautogui import pyperclip import numpy as np from PIL import Image from conversation_store import ConversationStore # ────────────────────────────────────────────────────────────────────────────── # 全局安全设置 # ────────────────────────────────────────────────────────────────────────────── pyautogui.FAILSAFE = True # 鼠标移到屏幕左上角 (0,0) 时强制停止，防止失控 pyautogui.PAUSE = 0.05 # 每次 pyautogui 操作后的基础延时（秒） # ────────────────────────────────────────────────────────────────────────────── # 常量配置（如界面布局变化，只修改这里） # ────────────────────────────────────────────────────────────────────────────── WX_WINDOW_CLASS = 'WeWorkWindow' # 企业微信主窗口类名（已通过 inspect_tree.py 确认） AUTO_REPLY_TEXT = "你好" # 自动回复内容 POLL_INTERVAL = 2.0 # 轮询间隔（秒） # 人机共存：人工移动鼠标后，机器人暂停；鼠标静止满此秒数才继续操作 MOUSE_IDLE_ENABLED = True MOUSE_IDLE_SECONDS = 20 MOUSE_MOVE_THRESHOLD = 8 # 位移超过此像素才视为人工移动（过滤抖动） # 企业微信 UI 布局参数（相对于窗口左上角的像素偏移，适用于大多数 PC 版本） NAV_BAR_W = 68 # 最左侧导航图标栏宽度 SESSION_LIST_W = 230 # 会话列表区域宽度 HEADER_H = 56 # 顶部标题栏高度 SESSION_ITEM_H = 64 # 每个会话条目的高度（用于行号计算） # 输入框位置（从窗口底部量） INPUT_Y_FROM_BOTTOM = 95 # 输入框中心距窗口底部的像素数 INPUT_X_RATIO = 0.62 # 输入框中心在聊天区域的水平比例 RIGHT_SIDEBAR_W = 350 # 右侧工具栏宽度（客户转账/问诊单/会话管理等） # 聊天上下文提取参数（框选 + 剪贴板复制） CHAT_SELECT_TOP_MARGIN = 30 # 框选终点距聊天区域顶部的安全距离（逻辑像素，防止触发翻页加载历史） CHAT_CONTEXT_MAX_LINES = 150 # 提取聊天记录的最大行数 CHAT_CONTEXT_SCREENS = 3 # 向上翻屏复制的屏数（1 = 只复制当前可见一屏） CHAT_SCROLL_CLICKS = 8 # 每向上翻一屏滚动的滚轮格数（不够一屏可调大） CHAT_FULL_SCREEN_LINES = 12 # 一屏少于此行数视为「消息不满一屏」，不再翻屏采集更早的历史 # 红点色彩阈值（企业微信未读徽章颜色 #FA5151 = R250 G81 B81） # mss 截图格式为 BGRA，通道顺序：B=0, G=1, R=2, A=3 # ⚠ 阈值收紧：避免误把会话头像里的红色内容当红点 BADGE_R_MIN = 235 # 红通道最小值（#FA5151 的 R=250，留15点容差） BADGE_R_MAX = 255 # 红通道最大值 BADGE_G_MAX = 90 # 绿通道最大值（#FA5151 的 G=81） BADGE_B_MAX = 90 # 蓝通道最大值（#FA5151 的 B=81） MIN_RED_PIXELS = 15 # 判定为一个红点所需的最少红色像素数（增大，减少噪声） BADGE_MERGE_GAP = 12 # 像素行间距 ≤ 此值视为同一个红点 # ⚠ 空间过滤：企业微信未读数字红点悬浮于头像右上角 # 头像在左侧(x: 0~60)，红点徽章在此区域右上角(x: 35~80)，只扫描此区间避免头像和右侧混淆误判 BADGE_SCAN_X_START = 35 # 扫描起点 BADGE_SCAN_X_END = 80 # 扫描终点 # ────────────────────────────────────────────────────────────────────────────── # 全局路径 # ────────────────────────────────────────────────────────────────────────────── _SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) # ────────────────────────────────────────────────────────────────────────────── # 内存临时黑名单配置 # ────────────────────────────────────────────────────────────────────────────── # 假阳性行号黑名单不再进行文件持久化，而是在每次轮询开始前动态清空， # 配合头像类型检测过滤系统工具。 # ────────────────────────────────────────────────────────────────────────────── # 工具函数 # ────────────────────────────────────────────────────────────────────────────── def safe_set_foreground(hwnd: int) -> bool: """ 安全地将指定窗口设置为前台焦点窗口。 解决 (0, 'SetForegroundWindow', 'No error message is available') 权限限制。 """ try: # 如果当前已经是前台窗口，直接返回成功 if win32gui.GetForegroundWindow() == hwnd: return True # 尝试直接设置 win32gui.SetForegroundWindow(hwnd) return True except Exception: pass # 尝试模拟 ALT 键按下和释放，绕过 Windows 前台限制 try: # VK_MENU = 0x12 win32api.keybd_event(0x12, 0, 0, 0) # ALT 按下 win32gui.SetForegroundWindow(hwnd) win32api.keybd_event(0x12, 0, win32con.KEYEVENTF_KEYUP, 0) # ALT 释放 return True except Exception: pass # 尝试 AttachThreadInput 挂接线程输入 try: fore_hwnd = win32gui.GetForegroundWindow() fore_thread, _ = win32process.GetWindowThreadProcessId(fore_hwnd) curr_thread = win32api.GetCurrentThreadId() if fore_thread != curr_thread: win32process.AttachThreadInput(curr_thread, fore_thread, True) win32gui.ShowWindow(hwnd, win32con.SW_SHOW) win32gui.SetForegroundWindow(hwnd) win32process.AttachThreadInput(curr_thread, fore_thread, False) return True except Exception as e: print(f" [~] safe_set_foreground 彻底失败: {e}") return False def find_wx_hwnd() -> int: """ 用 win32gui 查找企业微信主窗口句柄。 win32gui.FindWindow 只做顶层匹配，比 UIAutomation 更可靠。 """ hwnd = win32gui.FindWindow(WX_WINDOW_CLASS, None) if not hwnd: # 备用：枚举所有顶层窗口，找 ClassName 包含关键词的 result = [] def cb(h, _): cls = win32gui.GetClassName(h) if WX_WINDOW_CLASS in cls: result.append(h) return True win32gui.EnumWindows(cb, None) hwnd = result[0] if result else 0 return hwnd def restore_window(hwnd: int): """如果窗口被最小化，强制还原并置于前台""" placement = win32gui.GetWindowPlacement(hwnd) if placement[1] == win32con.SW_SHOWMINIMIZED: win32gui.ShowWindow(hwnd, win32con.SW_RESTORE) time.sleep(0.4) safe_set_foreground(hwnd) def capture_window_region(hwnd: int, x: int, y: int, w: int, h: int) -> np.ndarray: """ 使用 PrintWindow(PW_RENDERFULLCONTENT=2) 直接从窗口显存截图。 ✅ 不依赖窗口是否可见、是否在前台、是否被其他窗口遮挡。 ✅ 适用于 GPU 渲染（DirectX/OpenGL）的自定义框架应用（如企业微信）。 参数 x,y,w,h 为相对于窗口客户区左上角的偏移。 返回 BGRA numpy 数组，shape = (h, w, 4)。 """ # 获取完整窗口尺寸 rect = win32gui.GetWindowRect(hwnd) win_w = rect[2] - rect[0] win_h = rect[3] - rect[1] if win_w <= 0 or win_h <= 0: raise RuntimeError(f"窗口尺寸异常: {win_w}x{win_h}") # 建立内存 DC 和兼容位图 hwnd_dc = win32gui.GetWindowDC(hwnd) mfc_dc = win32ui.CreateDCFromHandle(hwnd_dc) mem_dc = mfc_dc.CreateCompatibleDC() bmp = win32ui.CreateBitmap() bmp.CreateCompatibleBitmap(mfc_dc, win_w, win_h) mem_dc.SelectObject(bmp) # PW_RENDERFULLCONTENT = 2（Windows 8.1+），专门捕获 GPU 渲染内容 ctypes.windll.user32.PrintWindow(hwnd, mem_dc.GetSafeHdc(), 2) # 读取像素数据（BGRA 32位） raw = bmp.GetBitmapBits(True) full = np.frombuffer(raw, dtype=np.uint8).reshape(win_h, win_w, 4) # 释放 GDI 资源 win32gui.DeleteObject(bmp.GetHandle()) mem_dc.DeleteDC() mfc_dc.DeleteDC() win32gui.ReleaseDC(hwnd, hwnd_dc) # 裁剪到目标区域并返回副本 return full[y: y + h, x: x + w].copy() def save_debug_screenshot(img_np, filename="debug_list.png"): """将 BGRA numpy 数组保存为 PNG""" import os path = os.path.join(os.path.dirname(__file__), filename) Image.fromarray(img_np[:, :, :3][:, :, ::-1]).save(path) # BGRA→RGB return path # ────────────────────────────────────────────────────────────────────────────── # 主类 # ────────────────────────────────────────────────────────────────────────────── class WeChatBot: def __init__(self): self.scale = 1.0 self.hwnd = 0 self.gui_hwnd = 0 self.replied = set() # 内存行号黑名单（每次轮询开始前清空，避免位置改变导致误跳过） self.false_pos_rows = set() # 会话档案：按会话指纹持久化每个会话的完整对话记录 + 上次画面快照。 # 重启不丢失；靠它提供 AI 上下文，每次只需增量提取最新消息。 self.store = ConversationStore(os.path.join(_SCRIPT_DIR, "conversations.json")) # 已知会话指纹集合（用于感知指纹的汉明距离归一化）。 # 从档案键值预热，重启后同一客户仍映射到原档案。 self._known_fps = set() try: for k in list(self.store._data.keys()): if len(k) == 16: # 8 字节感知指纹的 hex self._known_fps.add(bytes.fromhex(k)) except Exception: pass # 置顶状态（自动重连后需要恢复） self._topmost = False # 重连失败计数（用于限流日志，避免每 2s 刷一条） self._reconnect_fails = 0 # 首次轮询时执行一次界面清理（关闭遗留的搜索弹层/取消遗留选中） self._did_initial_cleanup = False # 人机共存：人工操作鼠标时暂停自动回复 self.mouse_idle_enabled = MOUSE_IDLE_ENABLED self.mouse_idle_seconds = MOUSE_IDLE_SECONDS self._bot_controlling = False # 机器人正在操控鼠标时为 True self._last_mouse_pos = None self._last_user_move_ts = 0.0 # 0 = 启动时视为已空闲，可立即开始 self._idle_log_ts = 0.0 self._stop_check = None # 可选 threading.Event，停止时打断等待 self.L = self.T = self.R = self.B = 0 self._list_x = 0 self._list_y = 0 self._list_w = 0 self._list_h = 0 self.list_click_x = 0 self.input_x = 0 self.input_y = 0 self.list_region = {} # 聊天区域坐标（用于 AI 截图） self._chat_region = {} # mss 截图区域 # 动态几何参数（将在 connect() 中根据 DPI 自适应更新） self.session_item_h = SESSION_ITEM_H self.badge_scan_x_start = BADGE_SCAN_X_START self.badge_scan_x_end = BADGE_SCAN_X_END # ── 1. 窗口挂载层 ───────────────────────────────────────────────────────── def connect(self) -> bool: """查找并挂载企业微信主窗口，计算所有关键区域坐标""" print("[*] 正在查找企业微信主窗口...") self.hwnd = find_wx_hwnd() if not self.hwnd: print("[-] 未找到企业微信！请确认已登录且主窗口存在（托盘图标双击打开）。") return False # 获取 DPI 缩放比例 try: dpi = ctypes.windll.user32.GetDpiForWindow(self.hwnd) self.scale = dpi / 96.0 except Exception: try: hdc = win32gui.GetDC(0) dpi_x = ctypes.windll.gdi32.GetDeviceCaps(hdc, 88) # 88 = LOGPIXELSX self.scale = dpi_x / 96.0 win32gui.ReleaseDC(0, hdc) except Exception: self.scale = 1.0 if self.scale != 1.0: print(f"[+] 检测到系统 DPI 缩放比例: {self.scale * 100:.1f}%，启用自适应几何缩放。") # 如果窗口被最小化，先还原它 restore_window(self.hwnd) time.sleep(0.3) # 等待窗口动画完成 # 用 win32gui 读取真实的窗口屏幕坐标（不受最小化影响） left, top, right, bottom = win32gui.GetWindowRect(self.hwnd) self.L, self.T, self.R, self.B = left, top, right, bottom W = self.R - self.L H = self.B - self.T if W <= 0 or H <= 0: print(f"[-] 窗口尺寸异常 ({W}×{H})，请手动将企业微信拖到屏幕上。") return False cls = win32gui.GetClassName(self.hwnd) title = win32gui.GetWindowText(self.hwnd) print(f"[+] 挂载成功: HWND=0x{self.hwnd:08X}, ClassName='{cls}', Title='{title}'") print(f" 窗口坐标: ({self.L},{self.T}) → ({self.R},{self.B})，尺寸: {W}×{H}") # ── 计算各区域坐标 ── nav_bar_w = int(NAV_BAR_W * self.scale) session_list_w = int(SESSION_LIST_W * self.scale) header_h = int(HEADER_H * self.scale) list_left = self.L + nav_bar_w list_top = self.T + header_h list_height = self.B - list_top self.list_region = { "left": list_left, "top": list_top, "width": session_list_w, "height": list_height, } self.list_click_x = list_left + session_list_w // 2 # 会话列表水平中心 # 输入框：聊天区域（会话列表右侧）的水平中间 + 距窗口底部固定偏移 chat_left = list_left + session_list_w chat_right = self.R self.input_x = int(chat_left + (chat_right - chat_left) * INPUT_X_RATIO) self.input_y = self.B - int(INPUT_Y_FROM_BOTTOM * self.scale) # 记录窗口内相对偏移（供 PrintWindow 裁剪使用） self._list_x = nav_bar_w self._list_y = header_h self._list_h = list_height # ★ 截图宽度必须使用 DPI 缩放后的值，否则高分屏下只能截到会话列表左半边， # 导致头像四角采样错位、真实会话被误判为系统工具。 self._list_w = session_list_w # 兼容 calibrate_mode 中对 list_region 的引用 self.list_region = { "left": list_left, "top": list_top, "width": session_list_w, "height": list_height, } print(f" 会话列表区域: left={list_left}, top={list_top}, " f"{session_list_w}×{list_height}px") print(f" 输入框估算坐标: ({self.input_x}, {self.input_y})") # 聊天消息显示区域（用于 AI 截图 + OCR） # 排除左侧会话列表和右侧工具栏，只截取中间的聊天消息气泡区域 # ★ 从 debug_chat_area.png 实测：chat_left 需额外偏移约 300px 跳过残留的会话列表 chat_left_extra = int(300 * self.scale) right_sidebar_w = int(RIGHT_SIDEBAR_W * self.scale) chat_msg_left = chat_left + chat_left_extra chat_msg_top = self.T + header_h + int(60 * self.scale) # 跳过聊天对象名称栏 chat_msg_bottom = self.input_y - int(100 * self.scale) # 排除输入框上方的工具栏图标 chat_msg_width = max(chat_right - chat_msg_left - right_sidebar_w, int(300 * self.scale)) self._chat_region = { "left": chat_msg_left, "top": chat_msg_top, "width": chat_msg_width, "height": max(chat_msg_bottom - chat_msg_top, int(100 * self.scale)), } # 记录窗口内聊天消息区域相对偏移（供 PrintWindow 使用） self._chat_rel_x = nav_bar_w + session_list_w + chat_left_extra self._chat_rel_y = header_h + int(60 * self.scale) self._chat_rel_w = chat_msg_width self._chat_rel_h = max(chat_msg_bottom - chat_msg_top, int(100 * self.scale)) # 更新自适应缩放几何参数 self.session_item_h = int(SESSION_ITEM_H * self.scale) self.badge_scan_x_start = int(BADGE_SCAN_X_START * self.scale) self.badge_scan_x_end = int(BADGE_SCAN_X_END * self.scale) print(f" 聊天区域: {self._chat_region['width']}×{self._chat_region['height']}px") return True # ── 2. 截图层 ───────────────────────────────────────────────────────────── def capture_session_list(self) -> np.ndarray: """ 使用 PrintWindow 直接从窗口显存截取会话列表区域。 不依赖屏幕绝对坐标，完全规避多显示器/高分屏下 mss 截屏为黑色的问题。 """ return capture_window_region( self.hwnd, self._list_x, self._list_y, self._list_w, self._list_h ) def capture_chat_area(self) -> bytes: """使用 PrintWindow 直接从窗口显存截取聊天消息显示区域，返回 PNG bytes。""" import io img_np = capture_window_region( self.hwnd, self._chat_rel_x, self._chat_rel_y, self._chat_rel_w, self._chat_rel_h ) # img_np 格式为 BGRA img = Image.fromarray(img_np[:, :, :3][:, :, ::-1]) # BGRA -> RGB buf = io.BytesIO() img.save(buf, format='PNG') return buf.getvalue() # 剪贴板哨兵：复制前先写入该唯一标记，复制后若剪贴板仍是它，说明本次拖拽未选中任何文字。 _CLIP_SENTINEL = "__WX_RPA_CLIP_EMPTY__" def _drag_select(self, x_start: int, y_start: int, x_end: int, y_end: int, steps: int = 10): """ 手动模拟一次「按下 → 分段移动 → 抬起」的鼠标拖拽，比 pyautogui.dragTo 更易被 企业微信识别为文本框选（dragTo 的补间在自绘控件里经常选不中）。 ★ 拖拽期间临时关闭 pyautogui.PAUSE：全局 PAUSE=0.05 会在补间循环的 每一步 moveTo 后强制延时，一次拖拽白白多耗 1 秒以上。 """ old_pause = pyautogui.PAUSE pyautogui.PAUSE = 0 try: pyautogui.moveTo(x_start, y_start) time.sleep(0.05) pyautogui.mouseDown(button='left') time.sleep(0.08) for i in range(1, steps + 1): ix = int(x_start + (x_end - x_start) * i / steps) iy = int(y_start + (y_end - y_start) * i / steps) pyautogui.moveTo(ix, iy) time.sleep(0.01) time.sleep(0.05) pyautogui.mouseUp(button='left') time.sleep(0.12) finally: pyautogui.PAUSE = old_pause def _copy_selection(self) -> str: """ 用哨兵法可靠判断 Ctrl+C 是否真的复制到了内容。 返回复制到的文本；若未选中任何文字则返回空串。 """ try: pyperclip.copy(self._CLIP_SENTINEL) except Exception: pass time.sleep(0.05) pyautogui.hotkey('ctrl', 'c') time.sleep(0.2) try: data = pyperclip.paste() except Exception: data = '' if not data or data == self._CLIP_SENTINEL: return '' return data def _select_visible_chat(self, left, top, width, height, margin, safe_top) -> list: """ 框选并复制【当前可见一屏】的聊天内容。 返回去掉空行后的文本行列表（空列表 = 本屏复制失败）。 依次尝试两种拖拽方向： 1. 右下 → 左上：适用于消息填满面板的情况（大多数老会话）。 2. 左上 → 右下：适用于【消息不满一屏】的情况——此时消息都靠在顶部， 底部是空白，鼠标在空白处按下无法锚定到任何文字，方向 1 必然选空； 从顶部消息处按下再往下拖就能正常选中。 """ attempts = [ # (起点x, 起点y, 终点x, 终点y) (left + width - margin, top + height - margin, left + margin, safe_top), (left + margin, safe_top, left + width - margin, top + height - margin), ] for x1, y1, x2, y2 in attempts: self._drag_select(x1, y1, x2, y2, steps=12) text = self._copy_selection() if text: return [l.strip() for l in text.splitlines() if l.strip()] return [] @staticmethod def _merge_overlap(older: list, newer: list) -> list: """ 拼接两屏复制到的行列表，自动去掉重叠部分。 翻屏滚动不可能精确一屏，相邻两屏必然有重复消息： 找到 older 尾部与 newer 头部的最大公共子序列，去重后拼接。 """ max_k = min(len(older), len(newer)) for k in range(max_k, 0, -1): if older[-k:] == newer[:k]: return older + newer[k:] return older + newer def extract_chat_text(self, screens: int = None) -> str: """ 通过鼠标框选 + 剪贴板复制，提取当前打开会话的聊天记录。 企业微信为自绘控件，不支持 Ctrl+A 全选，只能靠拖拽框选。 策略：先复制当前可见一屏，再向上滚动翻屏、逐屏复制， 共采集 screens 屏（默认 CHAT_CONTEXT_SCREENS）后按重叠去重拼接， 最后滚回底部。返回最近 CHAT_CONTEXT_MAX_LINES 行文本。 ★ 增量模式（会话已有档案）只需 screens=1，速度最快。 """ region = self._chat_region left = region['left'] top = region['top'] width = region['width'] height = region['height'] margin = 20 center_x = left + width // 2 center_y = top + height // 2 safe_top = top + int(CHAT_SELECT_TOP_MARGIN * self.scale) screens = max(1, screens if screens is not None else CHAT_CONTEXT_SCREENS) if not self.wait_for_mouse_idle(): return '' self._begin_bot_mouse() try: return self._extract_chat_text_locked(screens, left, top, width, height, margin, center_x, center_y, safe_top) finally: self._end_bot_mouse() def _extract_chat_text_locked(self, screens, left, top, width, height, margin, center_x, center_y, safe_top) -> str: """extract_chat_text 的实际实现（调用方已持有 bot_mouse 锁）。""" try: # 1. 备份当前剪贴板内容 old_clipboard = '' try: old_clipboard = pyperclip.paste() except Exception: pass # 2. 确保企业微信在前台并取得焦点 self._activate_wx() # 先在聊天区底部空白处点一下，确保焦点落在聊天面板而非别处 pyautogui.click(center_x, top + height - margin) time.sleep(0.2) # 3. 逐屏采集：blocks[0] = 最新一屏（底部），往后越来越旧 blocks = [] scrolled = 0 # 累计向上滚动的格数（用于最后滚回底部） for i in range(screens): lines = self._select_visible_chat(left, top, width, height, margin, safe_top) # 第一屏就失败 → 区域可能不对，保存调试截图后放弃 if i == 0 and not lines: print(" [剪贴板] 未能复制到聊天内容（两次框选均为空）") try: dbg = capture_window_region( self.hwnd, self._chat_rel_x, self._chat_rel_y, self._chat_rel_w, self._chat_rel_h) p = save_debug_screenshot(dbg, "debug_chat_area.png") print(f" [剪贴板] 已保存聊天区域截图: {p}") except Exception: pass break # 翻屏后内容和上一屏完全一样 → 已到聊天记录顶部，停止 if blocks and lines == blocks[-1]: break if lines: blocks.append(lines) # 本屏行数很少 → 消息不满一屏（整个历史已可见），翻屏是浪费时间 if len(lines) < CHAT_FULL_SCREEN_LINES: break # 还需要更早的消息 → 向上滚动一屏（鼠标须悬停在聊天区内） if i < screens - 1: pyautogui.moveTo(center_x, center_y) pyautogui.scroll(CHAT_SCROLL_CLICKS * 120, center_x, center_y) scrolled += CHAT_SCROLL_CLICKS time.sleep(0.5) # 等待渲染 / 加载更早的历史消息 # 4. 滚回底部（多滚一些确保到底），并点空白处取消选中高亮 if scrolled: pyautogui.moveTo(center_x, center_y) pyautogui.scroll(-(scrolled + CHAT_SCROLL_CLICKS * 2) * 120, center_x, center_y) time.sleep(0.4) pyautogui.click(center_x, top + height - margin) time.sleep(0.1) # 5. 还原之前的剪贴板内容 try: pyperclip.copy(old_clipboard) except Exception: pass if not blocks: return '' # 6. 从最旧一屏开始向新拼接，相邻屏按重叠去重 merged = blocks[-1] for newer in reversed(blocks[:-1]): merged = self._merge_overlap(merged, newer) recent = merged[-CHAT_CONTEXT_MAX_LINES:] result = '\n'.join(recent) print(f" [剪贴板] 成功提取 {len(recent)} 行聊天记录" f"（共采集 {len(blocks)} 屏 / 去重后 {len(merged)} 行）") return result except Exception as e: print(f" [剪贴板] 提取失败: {e}") return '' # ── 会话档案（持久化上下文）────────────────────────────────────────────── def get_session_history(self, fp: bytes) -> list: """获取指定会话的历史消息列表（来自持久化档案）。""" return self.store.history(fp.hex()) def remember_exchange(self, fp: bytes, user_text: str, reply_text: str): """将本轮「客户新消息 + 我方回复」写入该会话的持久化档案。""" fp_hex = fp.hex() self.store.append(fp_hex, "user", user_text) self.store.append(fp_hex, "assistant", reply_text) self.store.save() @staticmethod def _delta_lines(old_lines: list, new_lines: list) -> list: """ 增量比对：old_lines 是档案里上次提取的画面快照，new_lines 是本次画面。 找 old_lines 的尾部片段（最长 20 行）在 new_lines 中的【首次】出现位置， 返回其后的行 = 上次提取之后新增的消息。 （取首次出现而非最后一次：客户重复发相同内容时，取最后一次会把 新消息误判成旧内容而漏掉；取首次最多带上一两行旧内容，AI 可自行忽略。） 找不到重叠（消息刷得太快/首次提取）则整屏都算新增。 """ if not old_lines: return list(new_lines) max_k = min(len(old_lines), len(new_lines), 20) for k in range(max_k, 0, -1): tail = old_lines[-k:] for start in range(len(new_lines) - k + 1): if new_lines[start:start + k] == tail: return new_lines[start + k:] return list(new_lines) def extract_context_for(self, fp: bytes, pre_text: str = None) -> str: """ 智能提取当前打开会话需要发给 AI 的文本： - 该会话无档案（首次遇到）：翻屏提取完整可见历史做建档，全文发给 AI； - 已有档案：只提取最新一屏（快），与档案中上次画面快照做增量比对， 仅返回【新增的消息】——完整上下文由档案 history 提供，不再重复复制。 pre_text 可传入已提取好的一屏文本，避免重复框选。 """ fp_hex = fp.hex() # 首次遇到该会话：完整提取建档 if not self.store.has_record(fp_hex): text = pre_text if pre_text is not None else self.extract_chat_text() lines = [l for l in text.splitlines() if l.strip()] if text else [] if lines: self.store.set_last_lines(fp_hex, lines) self.store.save() print(f" [档案] 首次遇到该会话，已建档（{len(lines)} 行可见历史）") return text or '' # 增量模式：只取最新一屏 text = pre_text if pre_text is not None else self.extract_chat_text(screens=1) if not text: return '' lines = [l for l in text.splitlines() if l.strip()] delta = self._delta_lines(self.store.last_lines(fp_hex), lines) self.store.set_last_lines(fp_hex, lines) self.store.save() if delta: print(f" [档案] 增量提取到 {len(delta)} 行新消息（历史上下文由会话档案提供）") return '\n'.join(delta) # 画面没有新增内容（罕见），退回整屏文本兜底 return text # ── 3. 红点识别层 ───────────────────────────────────────────────────────── def detect_badge_rows(self, img: np.ndarray) -> list: """ 在截图中查找未读红点，返回每个红点中心的 Y 行号列表（相对于截图顶部）。 核心策略： 1. 只扫描截图右侧列（BADGE_SCAN_X_START 之后），头像在左侧不干扰 2. 色彩阈值贴近 #FA5151，减少头像内容误判 3. 最小像素数过滤，排除偶发噪点 """ # 只取截图中的头像右上角区域，过滤其他区域的干扰 scan_region = img[:, self.badge_scan_x_start:self.badge_scan_x_end, :] # shape: (H, W', 4) R = scan_region[:, :, 2].astype(np.int16) # 红通道 G = scan_region[:, :, 1].astype(np.int16) # 绿通道 B = scan_region[:, :, 0].astype(np.int16) # 蓝通道 # 布尔掩码 mask = ( (R >= BADGE_R_MIN) & (R <= BADGE_R_MAX) & (G <= BADGE_G_MAX) & (B <= BADGE_B_MAX) ) row_pixel_counts = mask.sum(axis=1) red_rows = np.where(row_pixel_counts >= 1)[0] if len(red_rows) == 0: return [] # 合并相邻行，取各组中心行号 badge_centers = [] group = [red_rows[0]] for row in red_rows[1:]: if row - group[-1] <= BADGE_MERGE_GAP: group.append(row) else: if row_pixel_counts[group].sum() >= MIN_RED_PIXELS: badge_centers.append(int(np.mean(group))) group = [row] if row_pixel_counts[group].sum() >= MIN_RED_PIXELS: badge_centers.append(int(np.mean(group))) return badge_centers @staticmethod def _patch_color(img: np.ndarray, px: int, py: int, r: int = 2) -> np.ndarray: """ 采样 (px, py) 周围 (2r+1)² 小块的中位数颜色（BGR），降低单像素噪声/抗锯齿边缘的影响。 返回 int16 的 [B, G, R]。 """ x1 = max(0, px - r) x2 = min(img.shape[1], px + r + 1) y1 = max(0, py - r) y2 = min(img.shape[0], py + r + 1) patch = img[y1:y2, x1:x2, :3].reshape(-1, 3) return np.median(patch, axis=0).astype(np.int16) def _is_real_conversation(self, img: np.ndarray, badge_y: int, quiet: bool = False) -> bool: """ 判断该行是否为真实人/群对话，而非系统工具（打卡、行业资讯等）。 quiet=True 时不打印日志、不导出调试截图（用于每轮都会执行的选中行检查）。 原理： 企业微信的真实头像（照片/字母头像/群聊九宫格）多为圆角方形，四角会露出 会话列表背景色；而系统工具图标通常填满整个方块、四角不露背景。 通过对头像四角做「小块中位数采样」并与动态背景色比对来区分，自适应深浅色主题。 ⚠ 为避免「把真实会话误判成系统工具而漏回复」这种最坏情况，判定偏向宽松： 只要 ≥2 个角落露出背景（圆角头像的典型特征）就视为真实会话。 """ avatar_x1 = int(8 * self.scale) avatar_x2 = int(58 * self.scale) avatar_radius = int(25 * self.scale) # 头像宽度从 x: 8~58，半径为 25 # 通过行号计算该会话条目的垂直中心，作为头像的真实垂直中心（避开偏上的未读红点纵坐标） row_idx = badge_y // self.session_item_h y_c = row_idx * self.session_item_h + self.session_item_h // 2 y1 = max(0, y_c - avatar_radius) y2 = min(img.shape[0], y_c + avatar_radius) # 越界保护 if y2 <= y1 or avatar_x2 >= img.shape[1]: return True # 无法判断，默认当作真实对话 # 动态采样背景色（在头像左侧偏边缘处采样 x=3，以 y_c 为垂直高度） bg_x = int(3 * self.scale) if bg_x >= img.shape[1]: bg_x = 0 bg_color = self._patch_color(img, bg_x, y_c) # 取头像区域四角的小块中位数颜色（BGR 通道，截图格式为 BGRA） corner_pts = [ (avatar_x1, y1), (avatar_x2-1, y1), (avatar_x1, y2-1), (avatar_x2-1, y2-1), ] # 统计有多少个角落与背景色一致（三通道色差绝对值之和在容差范围内） match_count = 0 for px, py in corner_pts: c = self._patch_color(img, px, py) if np.sum(np.abs(c - bg_color)) < 45: match_count += 1 # 判定 1：≤ 1 个角落露出背景 → 图标填满方块 → 系统工具 # 判定 2：头像是「大面积高饱和纯色 + 白色图形」→ 系统应用图标 # （客户联系=绿、行业资讯=黄、企小码=蓝等，它们也是圆角方形， # 四角同样露背景，仅靠判定 1 拦不住） is_real = match_count >= 2 and not self._is_flat_icon(img, y_c) if not is_real and not quiet: reason = (f"四角匹配背景数={match_count}" if match_count < 2 else "纯色系统图标（大面积纯色+白色图形）") print(f" [调试] 检测到系统工具图标，已跳过。判定依据: {reason}") # 导出被跳过行的头像截图，便于人工核对/调参（覆盖写入，开销极小） try: crop = img[y1:y2, 0:avatar_x2] save_debug_screenshot(crop, f"debug_skipped_row{row_idx}.png") except Exception: pass return is_real def _is_tool_selected(self, img: np.ndarray, sel_y: int) -> bool: """选中行是否为系统工具页（每轮静默检查用）。""" return not self._is_real_conversation(img, sel_y, quiet=True) def _is_flat_icon(self, img: np.ndarray, y_c: int) -> bool: """ 判断头像是否为「系统应用图标」：大面积高饱和纯色底 + 白色图形 （如 客户联系=绿底、行业资讯=黄底、企小码会话管理=蓝底、微盘/日程等）。 原理（用用户实际截图验证过，区分度非常大）： 将头像中心区域颜色量化到 8 级/通道后，统计「覆盖 95% 像素所需的颜色种数」： 系统图标（纯色底+白色图形）只有 3~5 种有效颜色； 真人照片/群聊九宫格有 18~30 种有效颜色。 再要求主色中存在高饱和彩色（图标底色为亮绿/亮黄/亮蓝）， 避免把低饱和的灰色默认头像误判成图标。 ⚠ 已知取舍：企业微信「姓名文字头像」（纯蓝底+白字）会被误判为系统图标。 客户场景下几乎都是照片头像，此风险可接受；若真遇到，可在 debug_skipped_row*.png 中核对并调大 FLAT_MAX_COLORS 阈值。 """ FLAT_MAX_COLORS = 8 # 有效颜色数 ≤ 此值视为纯色图标（实测：图标 3~5，照片 18+） FLAT_SAT_MIN = 55 # 主色饱和度阈值（max通道-min通道） x1 = int(12 * self.scale) x2 = min(int(54 * self.scale), img.shape[1]) half = int(18 * self.scale) y1 = max(0, y_c - half) y2 = min(img.shape[0], y_c + half) if y2 <= y1 or x2 <= x1: return False # 无法判断时不拦截（宁可误点，不漏真实会话） pix = img[y1:y2, x1:x2, :3].reshape(-1, 3).astype(np.int32) # BGR if len(pix) < 50: return False # 颜色量化到 8 级/通道，统计覆盖 95% 像素所需的颜色种数 codes = (pix[:, 0] // 32) * 64 + (pix[:, 1] // 32) * 8 + (pix[:, 2] // 32) counts = np.bincount(codes) order = np.argsort(counts)[::-1] cum = np.cumsum(counts[order]) / len(codes) n_colors = int(np.searchsorted(cum, 0.95) + 1) if n_colors > FLAT_MAX_COLORS: return False # 颜色丰富 → 照片/九宫格头像 # 前几种主色中需存在高饱和彩色（图标底色）；白色图形/背景饱和度低 for code in order[:min(3, len(order))]: dom = pix[codes == code].mean(axis=0) if dom.max() - dom.min() >= FLAT_SAT_MIN: return True return False def _session_fingerprint(self, img: np.ndarray, rel_y: int) -> bytes: """ 用会话条目的「头像区域像素」生成指纹，唯一标识一个会话。 指纹跟着会话走，不随列表重排 / 行号变化而改变，因此可用于跨重排的去重 和会话档案的隔离（不同客户的上下文绝不互串）。 ★ 感知哈希而非原始像素哈希： 1. 采样区收窄到头像正中心（x 12~42, y ±12），避开圆角处会渗入 悬停/选中背景色的边缘像素； 2. 下采样到 8×8 网格取均值，再把颜色量化到 16 级—— 悬停高亮、抗锯齿、字体渲染等微小差异不会改变指纹， 同一个客户在任何渲染状态下都稳定映射到同一份档案。 （采样头像中心也天然避开右上角的未读红点，红点数字变化不影响指纹。） """ row_idx = rel_y // self.session_item_h y_c = row_idx * self.session_item_h + self.session_item_h // 2 x1 = int(12 * self.scale) x2 = min(int(42 * self.scale), img.shape[1]) half = int(12 * self.scale) y1 = max(0, y_c - half) y2 = min(img.shape[0], y_c + half) if y2 <= y1 or x2 <= x1: return f"row{row_idx}".encode() # 越界兜底 # 灰度块均值 → 与中位数比较得到 64 位二值指纹（经典 pHash 思路） region = img[y1:y2, x1:x2, :3].astype(np.float32) gray = region.mean(axis=2) gh = gw = 8 h, w = gray.shape ys = np.linspace(0, h, gh + 1).astype(int) xs = np.linspace(0, w, gw + 1).astype(int) means = np.zeros((gh, gw), dtype=np.float32) for i in range(gh): for j in range(gw): block = gray[ys[i]:ys[i + 1], xs[j]:xs[j + 1]] if block.size: means[i, j] = block.mean() bits = (means > np.median(means)).flatten() raw = np.packbits(bits).tobytes() # 8 字节 return self._canonical_fp(raw) # 感知指纹的汉明距离容差：≤ 此值视为同一头像（64 位中容 6 位差异） _FP_HAMMING_TOL = 6 def _canonical_fp(self, raw: bytes) -> bytes: """ 指纹归一化：感知哈希对渲染噪声只能做到「几乎不变」，个别位仍可能翻转。 在已知指纹集合中找汉明距离 ≤ _FP_HAMMING_TOL 的最近邻： 找到 → 归一化为已知指纹（同一客户永远映射到同一份档案）； 找不到 → 登记为新会话指纹。 """ if len(raw) != 8: return raw raw_int = int.from_bytes(raw, 'big') best, best_d = None, 999 for known in self._known_fps: d = bin(int.from_bytes(known, 'big') ^ raw_int).count('1') if d < best_d: best, best_d = known, d if best is not None and best_d <= self._FP_HAMMING_TOL: return best self._known_fps.add(raw) return raw # ── 4. 交互动作层 ───────────────────────────────────────────────────────── def set_topmost(self, enable: bool): """ 设置企业微信窗口为「系统级置顶」(HWND_TOPMOST) 或取消置顶。 置顶后窗口永远显示在所有普通窗口之上，鼠标点击也必然落在企业微信上。 """ self._topmost = enable try: flag = win32con.HWND_TOPMOST if enable else win32con.HWND_NOTOPMOST win32gui.SetWindowPos( self.hwnd, flag, 0, 0, 0, 0, win32con.SWP_NOMOVE | win32con.SWP_NOSIZE ) if enable: # 置顶后顺手激活，确保用户看到微信 safe_set_foreground(self.hwnd) except Exception as e: print(f" [~] set_topmost({enable}): {e}") def is_window_alive(self) -> bool: """检查当前持有的窗口句柄是否仍然有效（主面板被关闭后会失效）。""" try: return bool(self.hwnd) and bool(win32gui.IsWindow(self.hwnd)) except Exception: return False def _reconnect(self) -> bool: """ 窗口句柄失效时自动重新挂载企业微信主窗口。 典型场景：主面板被 Esc/X 关闭后重新打开，窗口被销毁重建，旧 HWND 报 1400 错误。 """ if self._reconnect_fails == 0: print("[!] 企业微信窗口句柄已失效（主面板可能被关闭），尝试自动重新挂载...") self.hwnd = 0 try: ok = self.connect() except Exception as e: print(f"[-] 重连异常: {e}") ok = False if not ok: self._reconnect_fails += 1 # 限流：首次失败和之后每 15 次（约 30s）提示一次 if self._reconnect_fails == 1 or self._reconnect_fails % 15 == 0: print("[-] 自动重连失败：未找到企业微信主窗口，请重新打开主面板（双击托盘图标），将自动恢复监听。") return False self._reconnect_fails = 0 if self._topmost: self.set_topmost(True) print("[+] 自动重连成功，恢复监听。") return True # ── 人机共存：鼠标空闲检测 ──────────────────────────────────────────────── def _mouse_pos(self): p = pyautogui.position() return (int(p[0]), int(p[1])) def _sync_user_mouse_activity(self): """非机器人操控期间，若鼠标位置变化则记为人工操作。""" if self._bot_controlling: return try: pos = self._mouse_pos() except Exception: return if self._last_mouse_pos is None: self._last_mouse_pos = pos return dx = abs(pos[0] - self._last_mouse_pos[0]) dy = abs(pos[1] - self._last_mouse_pos[1]) if dx > MOUSE_MOVE_THRESHOLD or dy > MOUSE_MOVE_THRESHOLD: self._last_user_move_ts = time.time() self._last_mouse_pos = pos def _begin_bot_mouse(self): self._bot_controlling = True def _end_bot_mouse(self): try: self._last_mouse_pos = self._mouse_pos() except Exception: pass self._bot_controlling = False def wait_for_mouse_idle(self) -> bool: """ 等待鼠标静止 mouse_idle_seconds 秒后再允许机器人操作。 返回 False 表示监听已停止（被中断），调用方应立即退出本轮。 """ if not self.mouse_idle_enabled: return True while True: if self._stop_check is not None and self._stop_check.is_set(): return False self._sync_user_mouse_activity() idle = time.time() - self._last_user_move_ts if idle >= self.mouse_idle_seconds: return True remain = self.mouse_idle_seconds - idle now = time.time() if now - self._idle_log_ts >= 5: print(f" [人手] 检测到鼠标操作，暂停自动回复，" f"还需静止 {remain:.0f}s…") self._idle_log_ts = now # 短睡并响应停止信号 if self._stop_check is not None: if self._stop_check.wait(0.4): return False else: time.sleep(0.4) def _ensure_visible(self) -> bool: """ 确保企业微信窗口可见且置顶。 - 句柄失效（主面板被关闭重建）→ 自动重连 - 用户手动最小化或切走了 → 自动还原 + 重新置顶 每次轮询截图前必须调用；返回 False 表示窗口当前不可用，应跳过本次轮询。 """ # 句柄失效时先尝试自动重连，避免后续 win32 调用全部报 1400 if not self.is_window_alive(): if not self._reconnect(): return False try: hwnd = self.hwnd placement = win32gui.GetWindowPlacement(hwnd) if placement[1] == win32con.SW_SHOWMINIMIZED: # 窗口被最小化了，先还原 win32gui.ShowWindow(hwnd, win32con.SW_RESTORE) time.sleep(0.3) # 重新设置置顶（最小化后 TOPMOST 标记会丢失） self.set_topmost(True) elif not win32gui.IsWindowVisible(hwnd): win32gui.ShowWindow(hwnd, win32con.SW_SHOW) time.sleep(0.2) self.set_topmost(True) except Exception as e: print(f" [~] _ensure_visible: {e}") return self.is_window_alive() return True def _activate_wx(self): """将企业微信激活到前台并确保可见。""" self._ensure_visible() try: safe_set_foreground(self.hwnd) time.sleep(0.15) except Exception as e: print(f" [~] _activate_wx: {e}") def click_session(self, rel_y: int): """点击会话列表中指定 Y 行的会话条目""" if not self.wait_for_mouse_idle(): return self._begin_bot_mouse() try: screen_y = self.list_region["top"] + rel_y pyautogui.click(self.list_click_x, screen_y) time.sleep(0.6) # 等待右侧聊天面板渲染 finally: self._end_bot_mouse() def send_reply(self, text: str = None): """向当前打开的会话发送回复，发送完后取消选中状态。""" if not self.wait_for_mouse_idle(): return reply_text = text or AUTO_REPLY_TEXT self._begin_bot_mouse() try: pyautogui.click(self.input_x, self.input_y) time.sleep(0.25) pyperclip.copy(reply_text) pyautogui.hotkey('ctrl', 'v') time.sleep(0.2) pyautogui.press('enter') time.sleep(0.3) finally: self._end_bot_mouse() self._deselect_session() def _find_tool_row(self, img: np.ndarray) -> int: """ 在会话列表中找一个【系统工具会话】（打卡/客户联系/行业资讯等纯色图标行）， 返回其相对 Y 中心；找不到返回 -1。 优先返回靠下的工具行（远离列表顶部可能遗留的搜索弹层）。 """ n_rows = img.shape[0] // self.session_item_h best = -1 for r in range(n_rows): y_c = r * self.session_item_h + self.session_item_h // 2 if y_c >= img.shape[0]: break if self._is_flat_icon(img, y_c): best = y_c return best def _deselect_session(self): """ 取消当前客户会话的选中状态。 如果不取消选中，当对方再发新消息时，企业微信会因为会话已打开 而自动标记为已读，导致红点不出现，机器人没法检测到新消息。 实现方式：点击一个【系统工具会话】（打卡/客户联系等纯色图标行）， 把选中焦点「停靠」到无害的工具页上，客户会话随即关闭。 ⚠ 为什么不点搜索框：点击搜索框会弹出「全局搜索/智能搜索」悬浮面板， 遮挡列表顶部的会话红点，且极难可靠关闭—— 点导航栏图标关不掉它；点右侧输入框虽能关掉，但会把客户会话重新 激活选中，形成「取消选中→重新选中」的死循环（机器人反复点搜索框）。 """ try: img = self.capture_session_list() tool_y = self._find_tool_row(img) except Exception: tool_y = -1 if tool_y >= 0: # 点击工具行：选中焦点停到工具页，客户会话关闭，无任何弹层 self.click_session(tool_y) return # 兜底（列表里找不到工具行时）：退回搜索框方案，用消息图标尽力关闭弹层 print(" [~] 未找到系统工具会话，退回搜索框方式取消选中") if not self.wait_for_mouse_idle(): return self._begin_bot_mouse() try: search_x = self.L + int((NAV_BAR_W + SESSION_LIST_W // 2) * self.scale) search_y = self.T + int(30 * self.scale) pyautogui.click(search_x, search_y) time.sleep(0.15) msg_icon_x = self.L + int(NAV_BAR_W * self.scale) // 2 msg_icon_y = self.T + int(105 * self.scale) pyautogui.click(msg_icon_x, msg_icon_y) time.sleep(0.15) finally: self._end_bot_mouse() def _scroll_session_list_top(self): """ 把会话列表滚回最顶端。 企业微信收到新消息时会把对应会话【置顶到列表最上方】； 如果列表被用户翻到中间/下面，置顶的红点不在可视区内， 机器人既检测不到也点不到，新消息就漏回复了。 另外列表停在半格位置时，行号网格（session_item_h）的计算也会错位。 每次轮询前大幅上滚一次即可归位（已在顶部时滚动无副作用）。 """ if not self.wait_for_mouse_idle(): return cx = self.list_region['left'] + self.list_region['width'] // 2 cy = self.list_region['top'] + self.list_region['height'] // 2 old_pause = pyautogui.PAUSE pyautogui.PAUSE = 0 self._begin_bot_mouse() try: pyautogui.moveTo(cx, cy) pyautogui.scroll(50 * 120, cx, cy) # 大幅上滚，确保回到列表顶端 # ★ 滚完把鼠标移出会话列表（停到右侧聊天区顶部）： # 鼠标悬停会让所在行出现悬停高亮，干扰截图取色（头像指纹/红点检测） park_x = self._chat_region['left'] + self._chat_region['width'] // 2 park_y = self._chat_region['top'] - int(10 * self.scale) pyautogui.moveTo(park_x, max(park_y, self.T + 5)) except Exception as e: print(f" [~] _scroll_session_list_top: {e}") finally: pyautogui.PAUSE = old_pause self._end_bot_mouse() time.sleep(0.15) def detect_selected_row(self, img: np.ndarray) -> int: """ 检测会话列表中是否有会话处于「选中」状态（蓝色高亮行）。 返回选中行的相对 Y 中心；没有选中行返回 -1。 采样每行右侧边缘的背景色：选中行为高饱和蓝色，未选中为白色/浅灰(悬停)。 """ x = img.shape[1] - int(6 * self.scale) if x < 0: return -1 n_rows = img.shape[0] // self.session_item_h for r in range(n_rows): y_c = r * self.session_item_h + self.session_item_h // 2 if y_c >= img.shape[0]: break c = self._patch_color(img, x, y_c) # BGR b, g, rr = int(c[0]), int(c[1]), int(c[2]) if b > 170 and b - rr > 60 and b - g > 30: return y_c return -1 def _has_pending_customer_message(self, chat_text: str, fp: bytes) -> bool: """ 判断当前打开会话的【最后一条消息】是否为对方发的、尚未回复的新消息。 判定信号（按可靠性排序）： 1. 最后一条内容与我们最近一次回复相同 → 是我们发的 → 无待回复 2. 复制文本带「说话人 时间」头部时，最后一个说话人是我们 → 无待回复 3. 其余情况 → 视为有待回复（宁可多回一条，不漏客户消息） """ if not chat_text: return False lines = [l.strip() for l in chat_text.splitlines() if l.strip()] # 纯时间戳/日期/系统提示行不算消息内容 meta_pat = re.compile( r'^(\d{1,2}:\d{2}(:\d{2})?)$' r'|^(\d{1,2}月\d{1,2}日.*)$' r'|^(昨天.*|星期.*)$' r'|^以上是打招呼内容$' r'|^你已添加了.*$' ) content = [l for l in lines if not meta_pat.match(l)] if not content: return False last = content[-1] # 信号 1：与我们最近一次回复比对（来自持久化会话档案） history = self.store.history(fp.hex()) for msg in reversed(history): if msg.get('role') == 'assistant': last_reply_line = msg['content'].strip().splitlines()[-1].strip() if last == last_reply_line: return False break # 信号 2：解析「说话人 时间」头部（群聊/部分版本的复制格式） try: from ai_config import AI_AGENT_NAME speaker_pat = re.compile(r'^(.{1,30}?)\s+\d{1,2}:\d{2}(:\d{2})?$') last_speaker = None for l in lines: m = speaker_pat.match(l) if m: last_speaker = m.group(1).strip() if last_speaker and AI_AGENT_NAME in last_speaker: return False except ImportError: pass return True def _check_selected_session(self): """ 处理「会话被手动点开(选中)」导致漏回复的问题。 会话处于选中打开状态时，对方新发的消息会被企业微信【立即标记已读】， 红点不会出现，靠红点检测就永远漏掉这条消息。 对策：每轮轮询发现有选中的会话时—— 1. 提取该会话的聊天内容，若最后一条是对方发的（不是我们刚回复的）， 直接在当前打开的会话里执行 AI 回复； 2. 无论是否回复，最后都取消选中，让后续新消息正常产生红点。 """ try: img = self.capture_session_list() sel_y = self.detect_selected_row(img) except Exception: return if sel_y < 0: return fp = self._session_fingerprint(img, sel_y) # 选中的是系统工具页（打卡/行业资讯等）→ 这正是我们的「停靠」状态， # 工具页不会吞客户消息的红点，保持原样即可（不要再做取消动作，否则会循环折腾） if self._is_tool_selected(img, sel_y): return print("\n[🖱] 检测到有会话处于选中状态（选中期间新消息不出红点），检查是否有未回复消息...") self._activate_wx() # 只提取最新一屏做判断（增量上下文由会话档案提供，无需翻屏） chat_text = self.extract_chat_text(screens=1) if not self._has_pending_customer_message(chat_text, fp): print(" [🖱] 最后一条消息不是对方新发的，仅取消选中。") self._deselect_session() return print(" [🖱] 发现未回复的客户消息，直接在当前会话回复。") reply_text = self._generate_ai_reply(fp, chat_text=chat_text) self._activate_wx() time.sleep(0.2) self.send_reply(reply_text) # send_reply 内部会取消选中 time.sleep(0.5) def _generate_ai_reply(self, fp: bytes, chat_text: str = None) -> str: """ 对【当前已打开】的会话执行 AI 回复流程： 从会话档案取历史上下文 + 增量提取新消息 → 调用 AI → 回写档案。 chat_text 可传入已提取好的一屏文本（避免重复框选），仍会走增量比对。 返回回复文本；AI 未启用或失败时返回 None（调用方会用默认回复兜底）。 """ reply_text = None try: from ai_config import AI_ENABLED, AI_USE_VISION, AI_CONTEXT_ENABLED if not AI_ENABLED: return None from ai_chat import get_ai_reply, call_ai_text ai_reply = None # 该会话的历史上下文（来自持久化档案，按会话指纹隔离，重启不丢） history = self.get_session_history(fp) if AI_CONTEXT_ENABLED else None if history: print(f" [AI] 会话档案提供历史上下文 {len(history)} 条") if AI_USE_VISION: # 视觉模式：截图聊天区域发给多模态 AI time.sleep(0.5) image_bytes = self.capture_chat_area() print(f" [AI] 视觉模式，已截取聊天区域 ({len(image_bytes)} bytes)") ai_reply = get_ai_reply(image_bytes=image_bytes, history=history) chat_text = chat_text or '' else: # 文本模式：增量提取（首次建档全量、之后只取新增消息） time.sleep(0.5) chat_text = self.extract_context_for(fp, pre_text=chat_text) if chat_text: print(f" [AI] 本次发给 AI 的新内容:\n{chat_text[:200]}") ai_reply = call_ai_text(chat_text, history=history) else: # 提取失败，用通用提示词（仍携带档案历史上下文） print(" [AI] 未提取到聊天内容，使用通用提示词") ai_reply = call_ai_text( "客户在企业微信发来了一条新消息。" "请以客服身份生成一条礼貌、简短的问候回复，" "询问对方有什么可以帮到他。", history=history, ) if ai_reply: # 医院名强制甄养堂 + 挂号话术；有挂号需求则写入登记表 try: from registration_store import process_registration_reply, RegistrationStore agent = "" try: from ai_config import AI_AGENT_NAME agent = AI_AGENT_NAME except Exception: pass ai_reply, lead = process_registration_reply( session_id=fp.hex(), user_text=chat_text or "", reply_text=ai_reply, store=RegistrationStore(), agent_name=agent, ) if lead: print( f" [挂号] 已登记 → {lead.get('contact')}｜" f"{lead.get('status')}｜病症:{lead.get('symptom') or '待问清'}" ) except Exception as e: print(f" [挂号] ⚠ 登记处理失败: {e}") reply_text = ai_reply print(f" [AI] 回复内容: {reply_text[:60]}{'...' if len(reply_text) > 60 else ''}") # 记入该会话的上下文记忆，供下一轮回答衔接 if AI_CONTEXT_ENABLED: self.remember_exchange( fp, chat_text or "（客户发来新消息，内容未能提取为文字）", ai_reply, ) else: print(" [AI] ⚠ AI 未返回有效回复，使用默认回复") except ImportError: pass except Exception as e: print(f" [AI] ⚠ AI 调用异常: {e}") return reply_text # ── 5. 主轮询循环 ───────────────────────────────────────────────────────── def loop(self): """持续轮询：截图 → 红点识别 → 点击会话 → 发送回复""" print(f"[*] 监听启动（每 {POLL_INTERVAL}s 轮询一次，Ctrl+C 停止）...") try: while True: self._poll_once() time.sleep(POLL_INTERVAL) except KeyboardInterrupt: print("\n[*] 监听已手动终止。") except Exception as e: print(f"\n[-] 致命异常: {e}") raise def _poll_once(self): """ 执行一次轮询主逻辑。 核心设计：每处理完一个会话后【重新截图】重新定位， 避免因会话列表自动重排（最新消息上移）导致坐标过期，从而遗漏后续红点。 整个过程循环，直到会话列表中没有更多新消息为止。 """ # 人机共存：有人在动鼠标时先等待静止，避免抢鼠标 if not self.wait_for_mouse_idle(): return # 每次轮询开始前，确保企业微信窗口可见且置顶（句柄失效会自动重连） if not self._ensure_visible(): return # 首次轮询先做一次界面清理：关闭上次运行可能遗留的搜索弹层、取消遗留的会话选中 if not self._did_initial_cleanup: self._did_initial_cleanup = True self._activate_wx() self._deselect_session() # ★ 把会话列表滚回顶端：新消息会话被企业微信置顶到列表最上方， # 若列表被翻到下面，红点在可视区外会漏检测（截图区域只有可视部分） self._scroll_session_list_top() # ★ 有会话被手动点开时，选中期间来的新消息会被自动已读（不出红点）， # 先检查该会话有没有未回复的消息并处理，再取消选中恢复红点机制 self._check_selected_session() # 清空内存行号黑名单（会话列表动态重排，行号黑名单仅在单次轮询内的 MAX_PER_ROUND 循环中有效） self.false_pos_rows.clear() # 本轮去重：用「会话头像指纹」标识会话，不随列表重排/行号变化而失效。 # （行号去重在列表重排时会把后续会话误判成"已处理"而漏掉，导致 B 不被处理。） processed_fp = set() # 本轮已回复过的会话指纹 non_conv_fp = set() # 已判定为系统工具/非真实会话的指纹（避免重复判断与刷屏） MAX_PER_ROUND = 30 for _ in range(MAX_PER_ROUND): try: img = self.capture_session_list() badges = self.detect_badge_rows(img) except Exception as e: print(f"[-] 截图/识别异常: {e}") break if not badges: break # 严格串行：每次都重新截图，挑选「最顶部 + 真实会话 + 本轮尚未处理」的 # 一个红点会话来处理。处理完后会重新截图重新挑选下一个， # 从而保证「A 彻底回复完成 → 再读取并处理 B」。 rel_y = None target_fp = None for y in badges: fp = self._session_fingerprint(img, y) if fp in processed_fp or fp in non_conv_fp: continue if not self._is_real_conversation(img, y): non_conv_fp.add(fp) continue rel_y = y target_fp = fp break # 没有需要处理的新会话了（红点要么本轮已处理、要么是系统工具） if rel_y is None: break row_idx = rel_y // self.session_item_h screen_y = self.list_region["top"] + rel_y print(f"\n[🔔] 正在处理 row{row_idx}（坐标: {self.list_click_x}, {screen_y}）") # 企业微信已是置顶窗口，直接激活焦点后打开该会话 self._activate_wx() self.click_session(rel_y) # ── AI 回复流程 ── reply_text = self._generate_ai_reply(target_fp) # ★ AI 请求可能耗时较长，发送前只重新激活窗口焦点（不改变当前打开的会话）。 # 【关键修复·防串聊天】绝对不能再用旧坐标 rel_y 重新点击会话列表： # 在 AI 处理这几秒内，若其他用户发来新消息，企业微信会把对方会话置顶（左侧列表重排）， # 此时 rel_y 对应的位置已经变成了另一个会话，重新点击会把聊天面板切走， # 导致本该发给 A 的回复被发给了 B（串聊天）。 # 而已经打开的聊天面板在收到他人消息时【不会】自动切换，所以无需重新点击， # 直接在当前会话的输入框发送即可。 self._activate_wx() time.sleep(0.2) self.send_reply(reply_text) # 标记该会话本轮已处理（无论回复成功与否，避免红点延迟消失或列表重排导致重复处理） processed_fp.add(target_fp) # 验证：重新截图，按「会话指纹」判断该会话的红点是否已消失 time.sleep(1.5) try: verify_img = self.capture_session_list() after_badges = self.detect_badge_rows(verify_img) after_fps = {self._session_fingerprint(verify_img, y) for y in after_badges} except Exception: after_fps = set() if target_fp in after_fps: print(f" [⚠] row{row_idx} 回复后红点未消失（本轮不再重复处理该会话）") else: print(f" [✓] 回复完成，红点已消失 → row{row_idx}") time.sleep(0.5) # ────────────────────────────────────────────────────────────────────────────── # 标定模式：验证区域划分和输入框坐标是否正确 # ────────────────────────────────────────────────────────────────────────────── def calibrate_mode(): """截取会话列表截图并保存，供肉眼确认区域是否正确""" print("[标定模式] 截取会话列表区域截图...") bot = WeChatBot() if not bot.connect(): return img = bot.capture_session_list() badges = bot.detect_badge_rows(img) path = save_debug_screenshot(img, "calibrate_list.png") print(f"[+] 截图已保存: {path}") print(f"[+] 标定模式检测到未读红点 Y 相对行号: {badges}") print(f" 如果图片内容是企业微信的聊天列表，说明区域划分正确。") print(f" 如果显示的是其他区域，请调整 NAV_BAR_W / SESSION_LIST_W 常量。") print() # 测试输入框：高亮鼠标移动到输入框位置（3秒后移动） print(f"[+] 3 秒后将鼠标移动到估算的输入框位置 ({bot.input_x}, {bot.input_y})，") print(f" 请观察鼠标是否落在企业微信聊天输入框内。") time.sleep(3) pyautogui.moveTo(bot.input_x, bot.input_y, duration=0.5) time.sleep(2) print("[+] 标定完成。如位置不对，请调整 INPUT_Y_FROM_BOTTOM / INPUT_X_RATIO 常量。") def test_input_mode(): """测试模式：直接点击输入框并发送一次测试消息""" print("[测试输入模式] 将在 3 秒后点击输入框并发送测试消息...") bot = WeChatBot() if not bot.connect(): return time.sleep(3) bot.send_reply() print("[+] 测试消息已发送。") # ────────────────────────────────────────────────────────────────────────────── # 入口 # ────────────────────────────────────────────────────────────────────────────── if __name__ == "__main__": args = sys.argv[1:] print("=" * 55) print(" 企业微信 RPA 机器人 v3.1") print(" 技术路线: 截图 → 色彩识别 → 坐标点击") print("=" * 55) if "--calibrate" in args: calibrate_mode() elif "--test-input" in args: test_input_mode() else: bot = WeChatBot() if not bot.connect(): print() print("[提示] 请确认：") print(" 1. 企业微信主窗口已在桌面显示（不是最小化）") print(" 2. 已点击左侧【消息】图标，使会话列表可见") sys.exit(1) print("-" * 55) bot.loop()