注册 QQ 开放平台账号
账号注册
QQ 机器人:一个机器人可以被添加到 群聊/频道 内对话,QQ 用户也可以直接跟机器人 单独对话。
开发者账号主体要求
单聊对话:【定向邀请】
群聊场景:仅支持企业主体【个人主体暂不支持】
频道场景:企业主体与个人主体均可申请
注册地址:QQ 开放平台(opens new window)
注册创建机器人:获得的开发机器人接入票据 AppID AppSecret Token
创建机器人:https://q.qq.com/#/app/create-bot
开发设置:https://q.qq.com/qqbot/#/developer/developer-setting
查询官方文档
新版机器人官方文档:https://bot.q.qq.com/wiki/develop/api-v2/
GitHub 地址
安装 qq-botpy 环境
更新包的话需要添加 --upgrade 兼容版本:python3.8+
pip install qq-botpy
使用
需要使用的地方import botpy
import botpy
兼容提示
原机器人的老版本qq-bot仍然可以使用,但新接口的支持上会逐渐暂停,此次升级不会影响线上使用的机器人。
基础案例演示
通过继承实现bot.Client, 实现自己的机器人Client
实现机器人相关事件的处理方法,如 on_at_message_create, 详细的事件监听列表,请参考 事件监听.md
定义机器人被@的后自动回复
import botpy
from botpy.message import Message
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
await message.reply(content=f"机器人{self.robot.name}收到你的@消息了: {message.content}")
注意:每个事件会下发具体的数据对象,如message
相关事件是message.Message
的对象 (部分事件透传了后台数据,暂未实现对象缓存)
设置机器人需要监听的事件通道并启动client
import botpy
from botpy.message import Message
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
await self.api.post_message(channel_id=message.channel_id, content="content")
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid="12345", token="xxxx")
其他案例
examples 目录下存放示例机器人 具体使用可参考 Readme.md:https://github.com/tencent-connect/botpy/blob/master/examples/README.md
examples/
.
├── README.md
├── config.example.yaml # 示例配置文件(需要修改为config.yaml)
├── demo_announce.py # 机器人公告API使用示例
├── demo_api_permission.py # 机器人授权查询API使用示例
├── demo_at_reply.py # 机器人at被动回复async示例
├── demo_at_reply_ark.py # 机器人at被动回复ark消息示例
├── demo_at_reply_embed.py # 机器人at被动回复embed消息示例
├── demo_at_reply_command.py # 机器人at被动使用Command指令装饰器回复消息示例
├── demo_at_reply_file_data.py # 机器人at被动回复本地图片消息示例
├── demo_at_reply_keyboard.py # 机器人at被动回复md带内嵌键盘的示例
├── demo_at_reply_markdown.py # 机器人at被动回复md消息示例
├── demo_at_reply_reference.py # 机器人at被动回复消息引用示例
├── demo_dms_reply.py # 机器人私信被动回复示例
├── demo_get_reaction_users.py # 机器人获取表情表态成员列表示例
├── demo_guild_member_event.py # 机器人频道成员变化事件示例
├── demo_interaction.py # 机器人互动事件示例(未启用)
├── demo_pins_message.py # 机器人消息置顶示例
├── demo_recall.py # 机器人消息撤回示例
├── demo_schedule.py # 机器人日程相关示例
配置 IP 白名单
部分之前就注册的机器人可能不需要进行配置,但是新注册的机器人在进行腾讯开放平台的提审之前,需要进行 IP 白名单的配置。
注意:只有白名单内的IP可以在沙箱环境外成功调用openAPI接口,机器人上线提审前IP白名单不可为空。
配置 Sandbox QQ 群
在QQ群配置
如需开发在QQ群使用的机器人功能请完成此项配置
管理员QQ在沙箱群需为群主/管理员,且群成员数不大于20人。建议沙箱群包含「测试」相关字眼,且群头像与其他群区分开。配置完成后,群主可从沙箱群“设置-群机器人”打开机器人列表页添加测试机器人进行开发调试。
案例:OpenAi 问答
常用镜像源
清华:https://pypi.tuna.tsinghua.edu.cn/simple
阿里云:http://mirrors.aliyun.com/pypi/simple/
中国科技大学 https://pypi.mirrors.ustc.edu.cn/simple/
华中理工大学:http://pypi.hustunique.com/
山东理工大学:http://pypi.sdutlinux.org/
豆瓣:http://pypi.douban.com/simple/
配置镜像源
pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple
Writing to C:\Users\Administrator\AppData\Roaming\pip\pip.ini
然后我们把这个 C:\Users\Administrator\AppData\Roaming\pip\pip.ini
路径添加到系统环境变量就好了(怎么添加环境变量?跟给 pip 添加环境变量一样操作,只是路径不一样)
安装 openai 库
pip install openai==1.33.0 -i 镜像源
项目结构
bot.py 机器人主函数
plugins/chat_api.py 调用查询 OpenAi 接口插件
config.yaml 机器人基本信息
.env OpenAi token 和 中转路径配置
配置 Token 和 中转地址
在项目同级目录创建 .env 文件
OPENAI_API_KEY=sk-······
OPENAI_BASE_URL=https://api.······
./plugins/chat_api.py
调用查询 OpenAi 接口
from openai import OpenAI
import dotenv
dotenv.load_dotenv(".env")
client = OpenAI()
def chat_answer(text):
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": text}
]
)
return response.choices[0].message.content
查询 openai bot.py
源码
import os
import botpy
from botpy import logging
from botpy.ext.cog_yaml import read
from botpy.message import GroupMessage
from plugins import chat_api
test_config = read(os.path.join(os.path.dirname(__file__), "config.yaml"))
_log = logging.get_logger()
class MyClient(botpy.Client):
async def on_ready(self):
_log.info(f"robot 「{self.robot.name}」 on_ready!")
async def on_group_at_message_create(self, message: GroupMessage):
openid = message.author.member_openid
_log.info(f"robot {openid} on_ready!")
msg = message.content.strip()
result = chat_api.chat_answer(text=msg)
await message._api.post_group_message(
group_openid=message.group_openid,
msg_type=0,
msg_id=message.id,
content=f"{result}")
return
if __name__ == "__main__":
intents = botpy.Intents(public_messages=True)
client = MyClient(intents=intents)
client.run(appid=test_config["appid"], secret=test_config["secret"])
运行效果
案例:查询天气信息
项目结构
bot.py 机器人主函数
plugins/weather_api.py 调用查询天气接口插件
config.yaml 机器人基本信息
./plugins/weather_api.py
调用查询天气接口
import requests
def get_weather(city_name):
# 设置请求的URL和参数
url = f'https://apis.juhe.cn/simpleWeather/query?key=50a3bd415158e186903d6e6994157589&city={city_name}'
# 发送GET请求
response = requests.get(url)
# 检查请求是否成功
if response.status_code == 200:
# 解析返回的JSON数据
data = response.json()
# 检查是否查询成功
if data['reason'] == '查询成功!':
# 返回天气数据
return data['result']
else:
return {"error": "查询失败: " + data['reason']}
else:
return {"error": "请求失败,状态码: " + str(response.status_code)}
# 调用函数并处理返回的天气数据
def format_weather(city_name):
# 假设这里你已经有了城市的URL编码,这里用'%E9%87%8D%E5%BA%86'作为示例
city_encoded = city_name # 重庆的URL编码
weather_data = get_weather(city_encoded)
# 检查是否返回了错误
if 'error' in weather_data:
return weather_data['error']
else:
# 实时天气
realtime_weather = weather_data['realtime']
result = f"实时天气:" + "\n" + f"{realtime_weather['info']}, 温度: {realtime_weather['temperature']}℃, 湿度: {realtime_weather['humidity']}%, 风向: {realtime_weather['direct']}, 风力: {realtime_weather['power']}级, AQI: {realtime_weather['aqi']}"
# 未来几天的天气
result = result + "\n" + "未来几天的天气:"
for day in weather_data['future']:
result = result + "\n" + f"日期: {day['date']}, 天气: {day['weather']}, 温度: {day['temperature']}, 风向: {day['direct']}"
return result
查询天气信息 bot.py
源码
import os
import botpy
from botpy import logging
from botpy.ext.cog_yaml import read
from botpy.message import GroupMessage
from plugins import weather_api
config = read(os.path.join(os.path.dirname(__file__), "config.yaml"))
_log = logging.get_logger()
class MyClient(botpy.Client):
async def on_ready(self):
_log.info(f"robot 「{self.robot.name}」 on_ready!")
async def on_group_at_message_create(self, message: GroupMessage):
openid = message.author.member_openid
_log.info(f"robot {openid} on_ready!")
msg = message.content.strip()
if msg.startswith("/天气"):
city_name = msg.replace("/天气", "").strip()
result = weather_api.format_weather(city_name)
await message._api.post_group_message(
group_openid=message.group_openid,
msg_type=0,
msg_id=message.id,
content=f"{result}")
if __name__ == "__main__":
intents = botpy.Intents(public_messages=True)
client = MyClient(intents=intents)
client.run(appid=config["appid"], secret=config["secret"])
运行结果预览
案例:配置关键词回复
项目结构
bot.py 机器人主函数
plugins/match_question.py 查询本地 excel 问答表插件
config.yaml 机器人基本信息
关键词.xlsx 关键词问答表
./plugins/match_question.py
查询本地 excel 问答表
from openpyxl import load_workbook
# 加载工作簿
wb = load_workbook('关键词.xlsx')
# 选择活动工作表,或者通过名称或索引选择其他工作表
sheet = wb.active # 或者 wb['Sheet1'] 或 wb.worksheets[0]
question_list = []
# 遍历行和列(从第二行开始,假设第一行是标题)
for row in sheet.iter_rows(min_row=2, values_only=True): # 添加 values_only=True 以只获取单元格的值
question = {
"question": row[0] if row[0] is not None else '', # 确保如果单元格为空,则使用空字符串
"answer": row[1] if row[1] is not None else '',
}
question_list.append(question) # 添加创建的字典到列表中,而不是原始的 row
def match(word):
"""
匹配关键词
:param word:
:return:
"""
for question in question_list:
if word in question['question']:
return question['answer']
for question in question_list:
if word == question['question']:
return question['answer']
return ""
bot.py
源码
import os
import botpy
from botpy import logging
from botpy.ext.cog_yaml import read
from botpy.message import GroupMessage
from plugins import match_question
test_config = read(os.path.join(os.path.dirname(__file__), "config.yaml"))
_log = logging.get_logger()
class MyClient(botpy.Client):
async def on_ready(self):
_log.info(f"robot 「{self.robot.name}」 on_ready!")
async def on_group_at_message_create(self, message: GroupMessage):
openid = message.author.member_openid
_log.info(f"robot {openid} on_ready!")
msg = message.content.strip()
if match_question.match(msg):
result = match_question.match(msg)
await message._api.post_group_message(
group_openid=message.group_openid,
msg_type=0,
msg_id=message.id,
content=f"{result}")
return
if __name__ == "__main__":
intents = botpy.Intents(public_messages=True)
client = MyClient(intents=intents)
client.run(appid=test_config["appid"], secret=test_config["secret"])
运行效果