微信 AI 群聊机器人本地搭建教程

微信 AI 机器人是基于人工智能技术的自动化服务系统，具备自然语言处理能力，可理解用户输入并给出响应或执行任务。它能提供 24 小时实时服务，节省搜索时间，提高社群管理效率。

一、申请 API Key

1. 访问千帆大模型官网，在模型广场搜索 ERNIE-Lite（免费），点击体验。

2. 点击应用接入，填入必填项，点击创建（需实名认证）。

3. 获取 API Key 和 Secret Key。

二、代码实现

注：运行代码前请登录 PC 端微信，否则可能运行失败。

完整代码

import requests
import json
import csv
import os
import schedule
import time
from wxauto import WeChat

class WeChatBot:
    def __init__(self):
        self.wx = WeChat()
        self.list_name = [
            '徐长卿', # 机器人账号添加管理员的备注，ros 为机器人，ros 给管理员的备注就应为徐长卿，可进行修改
            '嘻嘻嘻', # 被监控群聊名称
            'AAA 刘哥百货超市',
            '小超市'
        ]
        # 为每个群聊添加监听
        for group in self.list_name:
            self.wx.AddListenChat(who=group, savepic=True)

    def get_access_token(self):
        """使用 API Key，Secret Key 获取 access_token"""
        url = "https://aip.bce.baidu.com/oauth/2.0/token?grant_type=client_credentials&client_id=[应用 API Key]&client_secret=[应用 Secret Key]"
        payload = json.dumps("")
        headers = {
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        }
        response = requests.post(url, headers=headers, data=payload)
        if response.status_code == 200:
            return response.json().get("access_token")
        else:
            print("Failed to get access token:", response.text)
            return None

    def call_ai_model(self, content):
        """调用大模型接口并返回结果"""
        access_token = self.get_access_token()
        if access_token:
            url = f"https://aip.bce.baidu.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/ernie-lite-8k?access_token={access_token}"
            payload = json.dumps({
                "messages": [
                    { "role": "user", "content": content }
                ]
            })
            headers = {
                'Content-Type': 'application/json'
            }
            response = requests.post(url, headers=headers, data=payload)
            if response.status_code == 200:
                return response.json().get('result')
            else:
                print("Failed to get response from chat API:", response.text)
                return None
        else:
            print("Access token is None, cannot proceed with chat API request.")
            return None

    def send_morning_wishes(self):
        """发送早安祝福消息"""
        good_morning_message = "老板生成一段祝福群友的早安祝福"
        ai_reply = self.call_ai_model(good_morning_message)
        if ai_reply:
            for group in self.list_name:
                self.wx.SendMsg(msg=ai_reply, who=group)

    def send_evening_greetings(self):
        """发送晚上好的消息"""
        evening_greetings = "老板生成一段祝福群友的晚上好祝福"
        for group in self.list_name:
            self.wx.SendMsg(msg=evening_greetings, who=group)

    def listen_messages(self):
        wait = 1 # 设置 1 秒查看一次是否有新消息
        while True:
            msgs = self.wx.GetListenMessage()
            for chat, messages in msgs.items():
                who = chat.who # 获取聊天窗口名（群名）
                if who in self.list_name:
                    for msg in messages:
                        msg_type = msg.type
                        content = msg.content
                        print(f'【{who}】：{content}')
                        sender = msg.sender
                        
                        if who != self.list_name[0]:
                            if content.startswith('购买'):
                                try:
                                    parts = content.split('购买')[1].strip()
                                    commodity_name, quantity = parts.split(' ', 1)
                                    if self.is_commodity_exists(commodity_name):
                                        reply = f"{sender}，已为您将{commodity_name}{quantity}添加至订单中，感谢您的支持"
                                        self.save_to_csv_order(sender, commodity_name, quantity)
                                        self.wx.SendMsg(msg=reply, who=who)
                                    else:
                                        reply = f"{sender}，该商品小店尚未出售"
                                        self.wx.SendMsg(msg=reply, who=who)
                                except ValueError:
                                    reply = f"{sender}，输入格式错误，请按照'购买 商品名称 购买数量'的格式输入。"
                                    self.wx.SendMsg(msg=reply, who=who)
                            elif content.startswith('查询'):
                                parts = content.split('查询')[1].strip()
                                if parts:
                                    commodity_name = parts
                                    if self.is_commodity_exists(commodity_name):
                                        price_info = self.get_price_info(commodity_name)
                                        reply = f"{sender}，{commodity_name} {price_info}"
                                    else:
                                        reply = f"{sender}，该商品小店尚未出售"
                                        self.wx.SendMsg(msg=reply, who=who)
                            if content.startswith('老板'):
                                ai_reply = self.call_ai_model(content[2:])
                                if ai_reply:
                                    self.wx.SendMsg(msg=ai_reply, who=who)
                        elif who == self.list_name[0]:
                            if content.startswith('转发'):
                                new_content = content[2:].strip()
                                self.forward_message(new_content)
                            if content.startswith('增加'):
                                try:
                                    parts = content.split('增加')[1].strip()
                                    if parts:
                                        commodity_name, price_info = parts.split(' ', 1)
                                        if self.is_commodity_exists(commodity_name):
                                            reply = f"{sender}，{commodity_name}已经存在。"
                                        else:
                                            reply = f"{sender}，{commodity_name}增加成功。"
                                            self.save_to_csv(commodity_name, price_info)
                                            self.wx.SendMsg(msg=reply, who=who)
                                except ValueError:
                                    reply = f"{sender}，输入格式错误，请按照'增加 商品名称 价格信息'的格式输入。"
                                    self.wx.SendMsg(msg=reply, who=who)
                            elif content.startswith('删除'):
                                commodity_name = content.split('删除')[1].strip()
                                if self.delete_commodity_from_csv(commodity_name):
                                    reply = f"{sender}，{commodity_name}删除成功。"
                                else:
                                    reply = f"{sender}，{commodity_name}不存在。"
                                self.wx.SendMsg(msg=reply, who=who)
            schedule.run_pending()
            time.sleep(wait)

    def forward_message(self, content):
        """将消息转发到所有监控的群聊"""
        for group in self.list_name:
            if group != self.list_name[0]:
                self.wx.SendMsg(msg=content, who=group)

    def save_to_csv_order(self, username, commodity_name, quantity):
        with open('order.csv', 'a', encoding='utf-8') as file:
            writer = csv.writer(file)
            writer.writerow([username, commodity_name, quantity])
            print(f"订单 '{commodity_name}' 已添加到 order.csv")

    def get_price_info(self, commodity_name):
        with open('commodity.csv', 'r', encoding='utf-8') as file:
            reader = csv.reader(file)
            for row in reader:
                if row and row[0] == commodity_name:
                    return row[1]
        return "未知价格"

    def save_to_csv(self, commodity_name, price_info):
        with open('commodity.csv', 'a', encoding='utf-8') as file:
            writer = csv.writer(file)
            writer.writerow([commodity_name, price_info])
            print(f"商品 '{commodity_name}' 已添加到 commodity.csv")

    def is_commodity_exists(self, commodity_name):
        if not os.path.exists('commodity.csv'):
            return False
        with open('commodity.csv', 'r', encoding='utf-8') as file:
            reader = csv.reader(file)
            for row in reader:
                if row and row[0] == commodity_name:
                    return True
        return False

    def delete_commodity_from_csv(self, commodity_name):
        lines = []
        found = False
        with open('commodity.csv', 'r', encoding='utf-8') as file:
            reader = csv.reader(file)
            for row in reader:
                if row and row[0] != commodity_name:
                    lines.append(row)
                elif row:
                    found = True
        with open('commodity.csv', 'w', encoding='utf-8') as file:
            writer = csv.writer(file)
            writer.writerows(lines)
        return found

    def start(self):
        schedule.every().day.at("08:30").do(self.send_morning_wishes)
        schedule.every().day.at("19:30").do(self.send_evening_greetings)
        self.listen_messages()

if __name__ == "__main__":
    bot = WeChatBot()
    bot.start()

代码说明

1. __init__ 函数中第一个元素为管理员，其他可为群聊或个人用户。
2. get_access_token 函数中需替换 URL 中的 [应用 API Key] 和 [应用 Secret Key]，注意删除方括号且不要留空格。
3. 程序通过 wxauto 库对微信进行实时监控，支持管理员和需要管理的群聊互不干扰，可并行检查多个群聊。可根据需求自行设置时间发送早安、晚安祝福。wxauto 库开源，建议配合官方文档查看修改。

上述功能旨在实现自动化管理，正常使用不会导致微信封号。