1587 lines
74 KiB
Python
1587 lines
74 KiB
Python
"""
|
||
企业微信 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()
|