基于wxauto的AI微信聊天助手

1. 项目目标

实时监听微信聊天记录，并提取对话内容。
利用AI语言模型技术对对话内容进行分析，理解对话双方的情感、意图等信息。
根据分析结果，自动生成并回复合适的聊天内容。

2. 功能需求

（1）微信聊天记录监听

实时获取当前聊天窗口的消息记录。
支持监听指定联系人或群聊的聊天记录。
过滤掉系统消息、表情包和图片，仅保留文本内容。

（2）对话内容提取

提取最近N条聊天记录，作为分析样本。
支持自定义提取聊天记录的条数。

（3）AI分析（调用智谱API）

对提取的对话内容进行情感分析。
模仿用户语气，生成符合用户风格的回复内容。

（4）自动回复

根据AI分析结果，自动发送回复内容。
支持设置回复频率，如每N条消息回复一次。
支持设置仅在对方发言后回复。

（5）日志记录

保存聊天记录和分析结果，支持查看历史记录。

3. 项目展示

可以看出来AI微信聊天助手监听到了最近5条消息，并给出了回答：

AI给出的回答：两点半挺好的，那你到了给我电话，我可能稍微准备一下。不过载我就不用了，我骑车或者打车过去也方便，不想麻烦你绕路啦。

生成的日志：

2024-08-09 15:58:47,504 - my_logger - INFO:
Self: 可以.
xxx: 两点半怎么样.
xxx: 我开车载你.
Self: 可以.
Self: 你打个电话给我

4. 项目准备工作

（1）环境

1️⃣ 操作系统：Windows 7以上
2️⃣ Python：3.8
3️⃣ IDE：Pycharm或vscode
4️⃣ 第三方库：① wxauto Version: 3.9.11.17.4 ② zhipuai Version: 2.1.4.20230731

pip install wxauto==3.9.11.17.4 zhipuai -i https://pypi.tuna.tsinghua.edu.cn/simple

（2）注册智谱清言API

登录智谱官网(https://open.bigmodel.cn/)，登录后如下图所示，查看自己的**API keys**并进行复制。

接下来就直接使用复制API，将python代码中的

your zhipu api

进行替换。

智谱API调用的参考文档：LLM之调用智谱API实现问答

（3）登录微信

登录微信客户端，选择一个联系人或群聊，默认监听当前窗口

5. 代码

（1）项目结构

（2）AlUtil.py

该类为国内语言大模型智谱清言的API工具类。

from zhipuai import ZhipuAI  # zhipuai Version: 2.1.4.20230731
class ZhipuAIUtil:
    def __init__(self, api_key):
        self.api_key = api_key
        self.client = ZhipuAI(api_key=self.api_key)
    def get_answer(self, question):
        try:
            response = self.client.chat.completions.create(
                model="glm-4",
                messages=[
                    {"role": "user", "content": question},
                ],
            )
            return response.choices[0].message.content
        except Exception as e:
            error_message = str(e)
            if "系统检测到输入或生成内容可能包含不安全或敏感内容" in error_message:
                return "系统检测到输入或生成内容可能包含不安全或敏感内容，请您避免输入易产生敏感内容的提示语，感谢您的配合。"
            else:
                return error_message
# 使用示例
if __name__ == '__main__':
    api_key = "55f67457af44ddaf5f53ab6dcd50b89d.KRnAjz8uOMaiJPbL"
    zhipu_ai = ZhipuAIUtil(api_key=api_key)
    while True:
        question = input("请输入要提问的问题:\n")
        print(zhipu_ai.get_answer(question if len(question) > 0 else '你好，请你直接写出冒泡排序算法，回答得简短一些'))
        print()

（3）Logger.py

该类为聊天记录日志类

import logging
from datetime import datetime
class LoggerClass:
    def __init__(self, log_file='my_log.log'):
        # 创建一个logger
        self.logger = logging.getLogger('my_logger')
        self.logger.setLevel(logging.DEBUG)  # 设置日志级别
        # 创建一个handler，用于写入日志文件，模式为'a'表示追加
        self.file_handler = logging.FileHandler(log_file, mode='a')
        # 创建一个formatter，设置日志格式
        self.formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s:\n%(message)s\n')
        self.file_handler.setFormatter(self.formatter)
        # 给logger添加handler
        self.logger.addHandler(self.file_handler)
    def log_message(self, message, level=logging.INFO):
        # 记录日志消息
        if level == logging.DEBUG:
            self.logger.debug(message)
        elif level == logging.INFO:
            self.logger.info(message)
        elif level == logging.WARNING:
            self.logger.warning(message)
        elif level == logging.ERROR:
            self.logger.error(message)
        elif level == logging.CRITICAL:
            self.logger.critical(message)
        else:
            self.logger.error(f"Invalid logging level: {level}")
    def close(self):
        # 关闭文件处理器
        self.file_handler.close()
# 使用示例
if __name__ == '__main__':
    # 获取今天的日期
    today = datetime.now().date()
    formatted_today = today.strftime('%Y-%m-%d') + ".log"
    logger_class = LoggerClass(log_file=formatted_today)
    logger_class.log_message('This is a log message at INFO level.')

（4）WeChatListener.py

该类为微信监听类，打开微信客户端后，pycharm中点击运行该文件即可执行程序

import time
from wxauto import WeChat  # wxauto Version: 3.9.11.17.4
from AIUtil import ZhipuAIUtil
from Logger import LoggerClass
from datetime import datetime
"""
自动监听并用AI分析两人的对话
"""
class WeChatWindow:
    def __init__(self, n=1, lens=5, api_key="55f67457af44ddaf5f53ab6dcd50b89d.KRnAjz8uOMaiJPbL", question="",
                 wait_for_oppo=1):
        self.lens = lens
        self.wx = WeChat()
        self.currentWindow = []
        self.msg5 = []
        self.loggerClass = LoggerClass(datetime.now().date().strftime('%Y-%m-%d') + ".log")
        self.aiUtil = ZhipuAIUtil(api_key=api_key)
        self.wait_for_oppo = wait_for_oppo
        self.question = question
        self.cycle = n
        self.update_current_window()  # 启动时获取当前聊天记录
    def update_current_window(self):
        # 获取当前聊天窗口消息
        msgs = self.wx.GetAllMessage(savevoice=True)
        self.currentWindow = []
        for msg in msgs:
            if msg.type == 'friend' or msg.type == 'self':
                self.currentWindow.append(msg.sender + ": " + msg.content)
        self.msg5 = self.currentWindow[-self.lens:]
        self.log(".\n".join(self.currentWindow))
    # def chat_with(self, who): #默认监听当前窗口，不指定对象
    #     self.wx.ChatWith(who)
    #     self.update_current_window()
    def print_last_messages(self):
        print("=" * 100)
        print("捕捉到当前5条消息: ")
        for i in self.msg5:
            print(i.encode('gbk', 'ignore').decode('gbk'))
        print()
    def print_AI_messages(self):
        strs = ' '.join(self.msg5)
        ans = self.aiUtil.get_answer(strs + self.question)
        print("智谱AI:")
        print(ans.encode('gbk', 'ignore').decode('gbk'))
    def listen_for_new_messages(self):
        i = 0
        lastMsg5 = self.msg5
        while True:
            # 获取下一条新消息
            msgs = self.wx.GetNextNewMessage(savevoice=True)
            for msgList in msgs.values():
                for item in msgList:
                    # if item[0] != 'SYS':
                    if item[0] != 'SYS' and item[1] != '[动画表情]' and item[1] != '[图片]':  # 不计入表情包和图片
                        self.currentWindow.append(item[0] + ": " + item[1])
            self.msg5 = self.currentWindow[-self.lens:]
            if self.msg5 != lastMsg5:
                self.log(self.currentWindow[-1])
                lastMsg5 = self.msg5
                self.print_last_messages()
                i += 1
                if i % self.cycle == 0:  # 每隔cycle个消息AI分析一次
                    if self.wait_for_oppo == 1:
                        if not self.currentWindow[-1].startswith('Self'):  # 只要最后一条消息不是自己的(是别人的)，就调用AI回答
                            self.print_AI_messages()
                    else:
                        self.print_AI_messages()
            time.sleep(0.5)
    def log(self, msg):
        self.loggerClass.log_message(msg.encode('gbk', 'ignore').decode('gbk'))
if __name__ == '__main__':
    chat_window = WeChatWindow(n=1,  # 每发1次监听一次
                               lens=5,  # 监听最近5个对话
                               api_key="55f67457af44ddaf5f53ab6dcd50b89d.KRnAjz8uOMaiJPbL",  # 连接国内大模型智谱清言密钥
                               question="以上是一段对话，请你站在Self的角度并且模仿Self的语气(Self是年轻人)，替他回复，给出一个参考回答就可以了，请你反驳对面",
                               wait_for_oppo=1)
    chat_window.print_last_messages()
    chat_window.print_AI_messages()
    chat_window.listen_for_new_messages()

（5）项目启动

方法一：pycharm或vscode中运行WeChatListener.py

方法二：或在命令行中输入以下代码（见3.项目展示）

python WeChatListener.py

标签： python 微信智谱API

本文转载自: https://blog.csdn.net/m0_53140426/article/details/141063243
版权归原作者 安晴晚风 所有，如有侵权，请联系我们删除。