微信监控信息并自动回复

关键词：微信、监控、自动回复

一、场景描述及代码：

监控微信信息，包括群聊和私聊，通过一个关键词的匹配进行回复判定，后续可接AP以达成智能回复的效果，也可以修改为列表关键字进行匹配。

# 使用提醒:
# 1. xbot包提供软件自动化、数据表格、Excel、日志、AI等功能
# 2. package包提供访问当前应用数据的功能，如获取元素、访问全局变量、获取资源文件等功能
# 3. 当此模块作为流程独立运行时执行main函数
# 4. 可视化流程中可以通过"调用模块"的指令使用此模块

import xbot
from xbot import print, sleep
from .import package
from .package import variables as glv
import time
import win32api
import uiautomation as auto
import re
import json

def main(args):
    pass

# 启动微信
wechat_path = r'D:\Program Files (x86)\Tencent\WeChat\WeChat.exe'
win32api.ShellExecute(0, 'open', wechat_path, '', '', 1)
 
# 等待微信完全启动
time.sleep(5)
 
# 查找微信主窗口
wechat_window = auto.WindowControl(searchDepth=1, ClassName="WeChatMainWndForPC")
# 正则匹配时间格式（xx:xx）
time_pattern = re.compile(r"^\d{2}:\d{2}$")
 
# 需要监控的关键词（例如 "客服"、"人工"）
target_keyword = "测试"

# 后续可以接上AP，进行智能回复
# AP生成回复内容存储为变量，存入下方reply_text变量

# 发送的自动回复文本
reply_text = "发送成功，已完成自动回复！"

def get_chat_data():
    """ 获取当前聊天窗口的聊天数据，并返回控件对象 """
    chat_data = []
    chat_controls = {}  # 存储控件对象，以消息文本为 key
    current_nickname = None
    current_time = None
 
    def process_text_controls(control):
        """ 递归遍历控件，提取 Type: TextControl 的 Name 并解析聊天数据 """
        nonlocal current_nickname, current_time
        control_type = control.ControlTypeName
        name = control.Name.strip()
 
        if control_type == "TextControl" and name:
            if time_pattern.match(name):  # 发现时间
                current_time = name
            elif current_time:  # 如果前面有时间，则当前文本是聊天消息
                chat_entry = {
                    "昵称": current_nickname,
                    "时间": current_time,
                    "消息": name
                }
                chat_data.append(chat_entry)
                chat_controls[name] = control  # 以消息文本作为 key 存储控件对象
                current_time = None  # 处理完当前时间的聊天消息后清空
            else:  # 如果没有时间，可能是昵称
                current_nickname = name
 
        # 继续遍历子控件
        for child in control.GetChildren():
            process_text_controls(child)
 
    if wechat_window.Exists(0, 0):
        process_text_controls(wechat_window)
 
    return chat_data, chat_controls
 
def compare_chat_data(old_data, new_data):
    """ 对比新旧聊天数据，返回新增或变化的数据 """
    old_set = {json.dumps(item, ensure_ascii=False, sort_keys=True) for item in old_data}
    new_set = {json.dumps(item, ensure_ascii=False, sort_keys=True) for item in new_data}
 
    # 找出新增的数据
    changed_data = [json.loads(item) for item in new_set - old_set]
    return changed_data
 
def click_target_keyword(changed_data, chat_controls):
    """ 遍历新增的聊天数据，双击包含目标关键词的消息，并发送自动回复 """
    for chat in changed_data:
        message = chat["消息"]
 
        if target_keyword in message and message in chat_controls:
            print(f"检测到包含关键词 '{target_keyword}' 的消息: {message}，执行双击！")
             
            # 执行双击操作
            chat_controls[message].DoubleClick()
            time.sleep(1)  # 等待窗口分离出来
             
            # 发送自动回复
            send_reply(reply_text)
 
def send_reply(text):
    """ 在分离的窗口输入指定文本并按回车发送 """
    time.sleep(0.5)  # 稍微等待，确保焦点已经在独立的对话框中
    auto.SendKeys(text)  # 直接粘贴文本
    time.sleep(0.5)  # 稍微等待
    auto.SendKeys('{Enter}')  # 模拟按下回车键发送
    print("消息发送成功！")
 
    # 模拟按下 ESC 键关闭独立窗口
    close_with_esc()
 
def close_with_esc():
    """ 使用 ESC 键关闭独立的聊天窗口 """
    time.sleep(0.5)  # 确保消息发送完毕
    auto.SendKeys('{Esc}')  # 模拟按下 ESC 键
    print("已通过 ESC 键关闭窗口。")
 
def close_current_window():
    """ 关闭当前独立的聊天窗口 """
    time.sleep(0.5)  # 确保发送完消息
    try:
        # 查找当前独立窗口
        independent_window = auto.WindowControl(searchDepth=1, ClassName="ChatWnd")
        if independent_window.Exists(0, 0):
            print("正在关闭当前独立聊天窗口...")
            independent_window.Close()  # 关闭窗口
            print("独立聊天窗口已关闭！")
        else:
            print("未找到独立聊天窗口，无法关闭。")
    except Exception as e:
        print(f"关闭窗口时出错: {e}")
 
# 记录上一次的聊天数据
previous_chat_data = []
 
print("开始监控聊天记录，每 2 秒检测一次变化...\n")
 
while True:
    current_chat_data, chat_controls = get_chat_data()
 
    if previous_chat_data:
        # 比较新旧数据，只输出新增或变化的部分
        changed_data = compare_chat_data(previous_chat_data, current_chat_data)
        if changed_data:
            print("检测到新消息:\n", json.dumps(changed_data, ensure_ascii=False, indent=4))
            click_target_keyword(changed_data, chat_controls)  # 进行点击操作并自动回复
 
    # 更新历史聊天数据
    previous_chat_data = current_chat_data
 
    # 每 2 秒执行一次
    time.sleep(2)

二、注意事项：

# 启动微信
wechat_path = r'D:\Program Files (x86)\Tencent\WeChat\WeChat.exe'
win32api.ShellExecute(0, 'open', wechat_path, '', '', 1)

这里需要注意，需要将引号中的微信文件位置换成自己电脑中微信的位置

这部分后期可以切换为API接口调用AP，以达到智能对话

三、运行效果展示：