第 1 课、注册 QQ 开放平台账户
QQ开放平台官方地址:https://q.qq.com/#/app/bot
QQ开放平台包含:QQ机器人、QQ小程序、QQ小游戏,我们这边选择QQ机器人。
机器人类型:设置私域机器人或者公域机器人,当然公域机器人对于服务器的要求过高,我们这边选择私域机器人
进行开发。
特别注意在选择沙箱频道的时候,你必须提前创建个测试频道,必须<20人的频道。
例如我这边创建了迷你小唤QQ机器人的测试频道。
创建完成后就能够进入你刚刚创建QQ机器人的后台,展示如下所示:
然后点击发布流程中的代码开发
即可查看API文档,进行编写我们服务器端的程序代码。
点击开发设置能够查看 BotAppID、机器人令牌、机器人密钥,后续服务器端开发需要。
第 2 课、QQ 频道机器人接口说明
QQ频道机器人提供两类接口:
第一种:基于 REST 风格的 OPENAPI。
传统风格 描述形式:http://localhost/user/getById?id=1
Rest 风格 描述形式:http://localhost/user/1
显而易见,Rest风格优点:隐藏资源的访问行为,无法通过地址得知对资源是哪种操作;简化书写。
第二种:基于 WebSocket 提供实时事件通知的事件中心。
这种的话,和野生的nonebot框架采用的类似,都是基于协议端与服务器端的长连接模式。
博客推荐:Python基于cq-http协议端,使用nonebot2框架制作属于自己的智能机器人
第 3 课、SDK 开发基础库安装
QQ 小程序的开发 SDK 官方提供了三种语言,分别是:GO语言、JS语言、Python语言,分别对应如下 3 基础库:
golang: https://github.com/tencent-connect/botgo(opens new window)
nodejs: https://github.com/tencent-connect/bot-node-sdk(opens new window)
python: https://github.com/tencent-connect/botpy
我们选择 Python 语言进行开发:
GITHUB:https://github.com/tencent-connect/botpy
安装 qq-botpy 第三方库
pip install qq-botpy
使用方法:需要使用的地方import botpy
import botpy
兼容提示:原机器人的老版本qq-bot仍然可以使用,但新接口的支持上会逐渐暂停,此次升级不会影响线上使用的机器人
第 4 课、botpy 快速入门案例测试
版本介绍:
首先介绍下版本问题,老版本是 qq-bot,新版本是 botpy,当然随着新版本迭代升级,老版本将会陆续的停用相关api,但是腾讯云的案例是 基于 老版本 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")
也可以通过预设置的类型,设置需要监听的事件通道
import botpy
intents = botpy.Intents.none()
intents.public_guild_messages=True
第 5 课、腾讯云官方模拟实验
环境搭建
安装Python3,推荐使用Python3,实验环境已经预安装,可执行下面命令,进行Python版本验证。
python3 --version
安装机器人SDK
在终端执行下面命令安装机器人PythonSDK:
pip install qq-bot
同时,由于需要读取 yaml 文件的内容,我们也需要安装 pyyaml
pip install pyyaml
创建项目文件
创建一个 demo 项目文件夹
mkdir /home/demo && cd /home/demo
在 demo 文件夹下创建名为 config.yaml 的配置文件
touch config.yaml
接着,在 demo 文件夹下创建一个名为 robot.py 的文件:
touch robot.py
导入Token 和 AppID
请点击打开 config.yaml 文件,并填入自己机器人的 AppID 和 Token ,注意保存:
token:
appid: "自己QQ小程序AppID"
token: "自己QQ小程序token"
导入依赖包
请点击打开 robot.py ,并在文件中复制导入相关依赖包的代码,注意保存:
import asyncio
import json
import os.path
import threading
from typing import Dict, List
import aiohttp
import qqbot
from qqbot.core.util.yaml_util import YamlUtil
from qqbot.model.message import MessageEmbed, MessageEmbedField, MessageEmbedThumbnail, CreateDirectMessageRequest, \
MessageArk, MessageArkKv, MessageArkObj, MessageArkObjKv
test_config = YamlUtil.read(os.path.join(os.path.dirname(__file__), "config.yaml"))
设置机器人自动回复普通消息
在 robot.py 文件中添加如下代码,注意保存:
async def _message_handler(event, message: qqbot.Message):
"""
定义事件回调的处理
:param event: 事件类型
:param message: 事件对象(如监听消息是Message对象)
"""
msg_api = qqbot.AsyncMessageAPI(t_token, False)
# 打印返回信息
qqbot.logger.info("event %s" % event + ",receive message %s" % message.content)
# 发送消息告知用户
message_to_send = qqbot.MessageSendRequest(content="你好", msg_id=message.id)
await msg_api.post_message(message.channel_id, message_to_send)
# async的异步接口的使用示例
if __name__ == "__main__":
t_token = qqbot.Token(test_config["token"]["appid"], test_config["token"]["token"])
# @机器人后推送被动消息
qqbot_handler = qqbot.Handler(
qqbot.HandlerType.AT_MESSAGE_EVENT_HANDLER, _message_handler
)
qqbot.async_listen_events(t_token, False, qqbot_handler)
代码运行
在终端命令行输入并执行下列命令,运行机器人
python3 /home/demo/robot.py
这时在频道内 @机器人 hello 指令就可以收到回复了
第 6 课、腾讯云官方案例:天气
代码说明
上面链接的天气API使用的sign=b59bc3ef6191eb9f747dd4e83c99f2a4可能会过期,如下图指引:
请前往天气API地址查看最新的测试sign并替换,或注册账号申请一个免费sign
天气免费API:https://www.nowapi.com/api/weather.today
修改 _message_handler 方法
在 _message_handler 方法中,加入调用 get_weather 函数并发送天气的代码。完整 _message_handler 的实现如下:
async def _message_handler(event, message: qqbot.Message):
"""
定义事件回调的处理
:param event: 事件类型
:param message: 事件对象(如监听消息是Message对象)
"""
msg_api = qqbot.AsyncMessageAPI(t_token, False)
# 打印返回信息
qqbot.logger.info("event %s" % event + ",receive message %s" % message.content)
# 获取天气数据并发送消息告知用户
weather_dict = await get_weather("深圳")
weather_desc = weather_dict['result']['citynm'] + " " \
+ weather_dict['result']['weather'] + " " \
+ weather_dict['result']['days'] + " " \
+ weather_dict['result']['week']
message_to_send = qqbot.MessageSendRequest(msg_id=message.id, content=weather_desc, image=weather_dict['result']['weather_icon'])
await msg_api.post_message(message.channel_id, message_to_send)
接口函数 get_weather(city_name: str) -> Dict:
async def get_weather(city_name: str) -> Dict:
weather_api_url = "http://api.k780.com/?app=weather.today&appkey=10003&sign=b59bc3ef6191eb9f747dd4e83c99f2a4&format=json&cityNm=" + city_name
async with aiohttp.ClientSession() as session:
async with session.get(
url=weather_api_url,
timeout=5,
) as resp:
content = await resp.text()
content_json_obj = json.loads(content)
return content_json_obj
代码运行
在终端命令行输入并执行下列命令,运行机器人
python3 /home/demo/robot.py
效果图如下:
第 7 课、腾讯云官方案例:主动推送
设置机器人主动推送消息
上面的教程只实现一个简单的获取天气的功能,但是我们做的是天气机器人,希望实现一个报告天气的功能。一般的天气应用都会在一个特定时间给你推送天气通知,在频道机器人中,你可以通过主动消息来实现这个功能。
在 robot.py 中添加定时发送消息的函数,代码如下:
async def send_weather_message_by_time():
"""
任务描述:每天推送一次普通天气消息(演示方便改为100s定时运行)
"""
# 获取天气数据
weather_dict = await get_weather("深圳")
# 获取频道列表都取首个频道的首个子频道推送
user_api = qqbot.AsyncUserAPI(t_token, False)
guilds = await user_api.me_guilds()
guilds_id = guilds[0].id
channel_api = qqbot.AsyncChannelAPI(t_token, False)
channels = await channel_api.get_channels(guilds_id)
channels_id = channels[0].id
qqbot.logger.info("channelid %s" % channel_id)
# 推送消息
weather = "当前天气是:" + weather_dict['result']['weather']
send = qqbot.MessageSendRequest(content=weather)
msg_api = qqbot.AsyncMessageAPI(t_token, False)
await msg_api.post_message(channels_id, send)
# 如果需要每天都执行,加上下面两句
t = threading.Timer(100, await send_weather_message_by_time)
t.start()
在main中添加执行 send_weather_message_by_time() 的语句:
# 定时推送主动消息
send_weather_message_by_time()
编写完毕,注意保存
在终端命令行输入并执行下列命令,运行机器人
python3 /home/demo/robot.py
效果图如下:
第 8 课、腾讯云官方案例:指令回复
设置机器人指令回复ark消息
提供给个人开发者的 Ark 有3种,这里使用 23 号 Ark。其它 Ark 见 消息模板
消息模板:https://bot.q.qq.com/wiki/develop/api/openapi/message/message_template.html
先在 robot.py 中添加发送ark的函数
async def _create_ark_obj_list(weather_dict) -> List[MessageArkObj]:
obj_list = [MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value=weather_dict['result']['citynm'] + " " + weather_dict['result']['weather'] + " " + weather_dict['result']['days'] + " " + weather_dict['result']['week'])]),
MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="当日温度区间:" + weather_dict['result']['temperature'])]),
MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="当前温度:" + weather_dict['result']['temperature_curr'])]),
MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="当前湿度:" + weather_dict['result']['humidity'])])]
return obj_list
async def send_weather_ark_message(weather_dict, channel_id, message_id):
"""
被动回复-子频道推送模版消息
:param channel_id: 回复消息的子频道ID
:param message_id: 回复消息ID
:param weather_dict:天气消息
"""
# 构造消息发送请求数据对象
ark = MessageArk()
# 模版ID=23
ark.template_id = 23
ark.kv = [MessageArkKv(key="#DESC#", value="描述"),
MessageArkKv(key="#PROMPT#", value="提示消息"),
MessageArkKv(key="#LIST#", obj=await _create_ark_obj_list(weather_dict))]
# 通过api发送回复消息
send = qqbot.MessageSendRequest(content="", ark=ark, msg_id=message_id)
msg_api = qqbot.AsyncMessageAPI(t_token, False)
await msg_api.post_message(channel_id, send)
再在被动回复处理函数 _message_handler 添加以下代码,用于发送ark
# 根据指令触发不同的推送消息
content = message.content
if "/天气" in content:
# 通过空格区分城市参数
split = content.split("/天气 ")
weather = await get_weather(split[1])
await send_weather_ark_message(weather, message.channel_id, message.id)
编写完毕,注意保存
代码运行,在终端命令行输入并执行下列命令,运行机器人
python3 /home/demo/robot.py
@机器人后输入“/天气 城市名”,效果如下图:
设置机器人私信
我们希望能提供不同用户不同地方的天气,但是发太多的消息会影响其它的用户。针对这种情况,我们可以通过私信来实现。下面函数中,当我们@机器人hello时收到机器人的私信。
私信中我们不使用ark,而是使用Embed。Embed也是一种结构化消息,它比Ark简单
在 robot.py 添加发送Embed的函数如下:
async def send_weather_embed_direct_message(weather_dict, guild_id, user_id):
"""
被动回复-私信推送天气内嵌消息
:param user_id: 用户ID
:param weather_dict: 天气数据字典
:param guild_id: 发送私信需要的源频道ID
"""
# 构造消息发送请求数据对象
embed = MessageEmbed()
embed.title = weather_dict['result']['citynm'] + " " + weather_dict['result']['weather']
embed.prompt = "天气消息推送"
# 构造内嵌消息缩略图
thumbnail = MessageEmbedThumbnail()
thumbnail.url = weather_dict['result']['weather_icon']
embed.thumbnail = thumbnail
# 构造内嵌消息fields
embed.fields = [MessageEmbedField(name="当日温度区间:" + weather_dict['result']['temperature']),
MessageEmbedField(name="当前温度:" + weather_dict['result']['temperature_curr']),
MessageEmbedField(name="最高温度:" + weather_dict['result']['temp_high']),
MessageEmbedField(name="最低温度:" + weather_dict['result']['temp_low']),
MessageEmbedField(name="当前湿度:" + weather_dict['result']['humidity'])]
# 通过api发送回复消息
send = qqbot.MessageSendRequest(embed=embed, content="")
dms_api = qqbot.AsyncDmsAPI(t_token, False)
direct_message_guild = await dms_api.create_direct_message(CreateDirectMessageRequest(guild_id, user_id))
await dms_api.post_direct_message(direct_message_guild.guild_id, send)
qqbot.logger.info("/私信推送天气内嵌消息 成功")
在_message_handler中调用刚刚添加的函数,使机器人是在私信里给你发送Embed
elif "/私信天气" in content:
# 通过空格区分城市参数
split = content.split("/私信天气 ")
weather = await get_weather(split[1])
await send_weather_embed_direct_message(weather, message.guild_id, message.author.id)
编写完毕,注意保存
在终端命令行输入并执行下列命令,运行机器人
python3 /home/demo/robot.py
在频道中执行下列步骤验证效果:
@机器人后输入“/私信天气 城市名”执行
等待几分钟后,到私信面板看看是否有机器人推送过来的天气消息。
第 9 课、腾讯云官方案例:小程序
当用户想要查看全国或者某个省份的天气情况,一次次@机器人就显得十分麻烦,这个时候你可以使用小程序来解决这个问题。了解具体的小程序开发可以看 QQ小程序开发文档,这里只介绍如何通过机器人打开小程序。
机器人打开小程序非常简单,只需要按照下面配置就可以了,不需要增加额外的代码:
第 10 课、Pycharm SSH 连接服务器
为了方便使用我们也可以,直接使用Python Pycharm SSH 连接我们的的小程序服务器端:这边我们现在宝塔面板下载 id_rsa 密钥文件,当然你也可以直接 host + password 进行连接。
然后连接配置后,选择IDE 版本 和 项目工程的目录(对应服务器的目录)
然后需要等待大概10分钟左右让它安装好环境,然后会自动打开,可见如下所示已经是刚才配置的服务器端的项目目录了。
在服务器端项目目录编写测试代码,其中 appid 和 token 换成你自己小程序的数据:
import botpy
from botpy.types.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={appid}, token={token})
报错提示: from typing import TypedDict ImportError: cannot import name ‘TypedDict’
如果出现如上的问题:说明你的 python 版本和 botpy 版本不匹配:
官方说明:botpy 更新包的话需要添加 --upgrade 兼容版本:python3.8+
第 11 课、Python 读取 YAML 文件
在框架搭建中,一些配置文件(例如:数据库账号、url,日志级别等…)会经常的用到,所以我们需要一个存放这些参数的文件,并从中读取和写入。当然,存放参数除了用yaml,像ini文件、json文件、excel文件也是可以的,看个人选择。本文就介绍yaml文件的操作方法。
安装 pyyaml 模块:
pip install pyyaml
读取YAML代码如下:
import yaml
file = open('test.yaml', 'r', encoding="utf-8")
# 使用文件对象作为参数
data = yaml.load(file)
print(data)
结果运行出现警告,原因是YAML 5.1版本后弃用了 yaml.load(file) 这个用法,因为觉得很不安全,5.1 版本之后就修改了需要指定 Loader,通过默认加载器(FullLoader)禁止执行任意函数,该 load 函数也变得更加安全。建议代码如下:
import yaml
file = open('test.yaml', 'r', encoding="utf-8")
# 读取文件中的所有数据
file_data = file.read()
file.close()
# 指定Loader
data = yaml.load(file_data,Loader=yaml.FullLoader)
print(data)
上述例子和之前的例子我用了两种不同的读取方式,第一种为文件对象,第二种为文件中的所有字符串数据,两种方法都可以。
同时大家也发现了:yaml读取出来的数据结构不是由pyyaml模块读取代码决定的,而是yaml文件中的内容决定了它的结构,比如现在举例的YAML是字典里面嵌套列表,从文件读取出来后,也是这种数据结构形式。
load_all() 方法使用代码如下:
import yaml
file = open('test.yaml', 'r', encoding="utf-8")
#datas为load_all()方法返回的迭代器对象
datas = yaml.load_all(file,Loader=yaml.FullLoader)
for data in datas:
print(data)
本例中的 YAML 文件使用分为两段,load_all()方法会创建一个迭代器对象,分开存放两段数据。
使用 dump() 方法将一个 python 对象生成 yaml 格式的文档
import yaml
data = { 'name':'唤醒手腕', 'age':20 , 'sex':'男' }
yaml_file = open('test.yaml','w')
# yaml_file作为dump()第二个参数,是一个可写状态的文件对象
yaml.dump(data, yaml_file ,default_flow_style=False,encoding='utf-8',allow_unicode=True)
yaml_file.flush()
yaml_file.close()
第 12 课、API 教程:用户对象 User
用户对象中所涉及的 ID 类数据,都仅在机器人场景流通,与真实的 ID 无关。请不要理解为真实的 ID
注意:union_openid 是关联应用的 openid,如果你不是分布式场景,其实是无所谓的。
union_openid 与 union_user_account 只有在单独拉取 member 信息的时候才会提供,在其他的事件中所携带的 user 对象,均无这两个字段的内容。
获取当前用户信息
import botpy
import yaml
from botpy.message import Message
config = yaml.load(open('config.yaml', encoding="utf-8").read(), Loader=yaml.FullLoader)
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
user = await self.api.me()
botpy.logger.info(user)
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid=config['token']['appid'], token=config['token']['token'])
日志打印 QQ 机器人信息
{
"id": "11586990140073229091",
"username": "gitsub",
"avatar": "https://thirdqq.qlogo.cn/g?b=oidb&k=M5TibpXicS7Jt4z89BZxiamAA&s=100&t=1641802698",
"union_openid": "74F138F7F3AF68C4B8E8325013FCA295",
"union_user_account": ""
}
第 13 课、API 教程:发消息 / 回复
单纯回复发送文本消息
消息回复简单调用 message 事件的 reply 方法:
await message.reply(content="content")
官方回复消息案例代码:
# -*- coding: utf-8 -*-
import asyncio
import os
import botpy
from botpy import logging
from botpy.ext.cog_yaml import read
from botpy.message import Message
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_at_message_create(self, message: Message):
_log.info(message.author.avatar)
if "sleep" in message.content:
await asyncio.sleep(10)
_log.info(message.author.username)
await message.reply(content=f"机器人{self.robot.name}收到你的@消息了: {message.content}")
if __name__ == "__main__":
# 通过预设置的类型,设置需要监听的事件通道
# intents = botpy.Intents.none()
# intents.public_guild_messages=True
# 通过kwargs,设置需要监听的事件通道
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid=test_config["appid"], token=test_config["token"])
首先分析下 message 事件对象的信息:
案例代码中 message.content 是获取消息的内容
运行结果:
回复其他类型的消息
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="xxxx", content="xxx", msg_id="xxxx", embed=embed)
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid={appid}, token={token})
官方 api 发送消息:
注意事项:
发送图片案例:回复对方的头像
在第 12 课介绍了,user 对象的相关属性,用户头像就是:message.author.avatar
import os
import botpy
from botpy.ext.cog_yaml import read
from botpy.message import Message
test_config = read(os.path.join(os.path.dirname(__file__), "config.example.yaml"))
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
await message.reply(image=message.author.avatar)
if __name__ == "__main__":
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid=test_config["appid"], token=test_config["token"])
运行结果展示:
内嵌格式:进行 @ 某人 / 全体成员、内嵌表情
功能描述:利用 content 字段发送内嵌格式的消息。
内嵌格式仅在 content 中会生效,在 Ark 和 Embed 中不生效。
表情:https://bot.q.qq.com/wiki/develop/api/openapi/emoji/model.html
await message.reply(content='你在干嘛呀?<emoji:4>')
小补充:事件 id :message.event_id
事件 id 区分不同的事件,每次都是唯一不同的
await message.reply(content='你在干嘛呀?<emoji:4>,{}'.format(message.event_id))
回复消息内容:
async def on_at_message_create(self, message: Message):
# 构造消息发送请求数据对象
message_reference = Reference(message_id=message.id)
# 通过api发送回复消息
await self.api.post_message(
channel_id=message.channel_id,
content="<emoji:4>这是一条引用消息",
msg_id=message.id,
message_reference=message_reference,
)
消息中包含的URL需要进行配置
第 14 课、API 教程:语音房上麦
START 0 开始播放操作、PAUSE 1 暂停播放操作、RESUME 2 继续播放操作、STOP 3 停止播放操作
案例代码测试:
import botpy
from botpy.audio import Audio
class MyClient(botpy.Client):
async def on_audio_start(self, audio: Audio):
audio_control={
"audio_url":"url",
"text":"简单爱-周杰伦",
"status":"START",
}
await self.api.update_audio(audio.channel_id, audio_control=audio_control)
intents = botpy.Intents(audio_action=True)
client = MyClient(intents=intents)
client.run(appid={appid}, token={token})
[botpy] 接口请求异常,请求连接: https://api.sgroup.qq.com/channels/9368829/mic, 错误代码: 401, 返回内容: {‘message’: ‘check app privilege not pass’, ‘code’: 11253}, trace_id:adfcc85b0e21b5c2f4817abe82d9c15d
特别注意:ErrorCheckAppPrivilegeNotPass 检查应用权限不通过,该机器人应用未获得调用该接口的权限,需要向平台申请
第 15 课、API 教程:注册指令 handler
注册指令handler 模式编程
# -*- coding: utf-8 -*-
import os
import botpy
from botpy import logging, BotAPI
from botpy.ext.command_util import Commands
from botpy.message import Message
from botpy.ext.cog_yaml import read
test_config = read(os.path.join(os.path.dirname(__file__), "config.yaml"))
_log = logging.get_logger()
@Commands(name=("你好", "hello"))
async def hello(api: BotAPI, message: Message, params=None):
_log.info(params)
# 第一种用reply发送消息
await message.reply(content=params)
# 第二种用api.post_message发送消息
await api.post_message(channel_id=message.channel_id, content=params, msg_id=message.id)
return True
@Commands("晚安")
async def good_night(api: BotAPI, message: Message, params=None):
_log.info(params)
# 第一种用reply发送消息
await message.reply(content=params)
# 第二种用api.post_message发送消息
await api.post_message(channel_id=message.channel_id, content=params, msg_id=message.id)
return True
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
# 注册指令handler
handlers = [
hello,
good_night,
]
for handler in handlers:
if await handler(api=self.api, message=message):
return
if __name__ == "__main__":
# 通过预设置的类型,设置需要监听的事件通道
# intents = botpy.Intents.none()
# intents.public_guild_messages=True
# 通过kwargs,设置需要监听的事件通道
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid=test_config["appid"], token=test_config["token"])
第 16 课、API 教程:频道对象 guild
获取频道的信息:
import botpy
from botpy.audio import Audio
from botpy.message import Message
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
guild = await self.api.get_guild(guild_id=message.guild_id)
await self.api.post_message(channel_id=message.channel_id, content=str(guild), msg_id=message.event_id)
intents = botpy.Intents(audio_action=True, public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid="", token="")
运行结果:
获取频道的子频道信息:
async def on_at_message_create(self, message: Message):
channels = await self.api.get_channels(guild_id=message.guild_id)
await self.api.post_message(channel_id=message.channel_id, content=str(channels), msg_id=message.event_id)
channel 频道对象信息
第 17 课、API 教程:发送按钮消息
功能描述
通过在 MessageSendRequest 中指定 keyboard 字段发送带按钮的消息,支持 keyboard 模版 和 自定义 keyboard 两种请求格式。
要求操作人在该子频道具有发送消息和 对应消息按钮组件 的权限。请求参数 keyboard 模版 和 自定义 keyboard 只能单一传值。
keyboard 模版
调用前需要先申请消息按钮组件模板,这一步会得到一个模板 id,在请求时填在 keyboard 字段上。申请消息按钮组件模板需要提供响应的 json,具体格式参考 InlineKeyboard。
注意:仅 markdown 消息支持消息按钮。
import botpy
from botpy.message import Message
from botpy.types.message import MarkdownPayload, MessageMarkdownParams
class MyClient(botpy.Client):
async def handle_send_markdown_by_template(self, channel_id, msg_id):
params = [
MessageMarkdownParams(key="title", values=["标题"]),
MessageMarkdownParams(key="content", values=["为了成为一名合格的巫师,请务必阅读频道公告", "藏馆黑色魔法书"]),
]
markdown = MarkdownPayload(template_id=65, params=params)
# 通过api发送回复消息
await self.api.post_message(channel_id, markdown=markdown, msg_id=msg_id)
async def handle_send_markdown_by_content(self, channel_id, msg_id):
markdown = MarkdownPayload(content="# 标题 \n## 简介很开心 \n内容")
# 通过api发送回复消息
await self.api.post_message(channel_id, markdown=markdown, msg_id=msg_id)
async def on_at_message_create(self, message: Message):
await message.reply(content=f"机器人{self.robot.name}收到你的@消息了: {message.content}")
await self.handle_send_markdown_by_template(message.channel_id, message.id)
await self.handle_send_markdown_by_content(message.channel_id, message.id)
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid={appid}, token={token})
第 18 课、API 教程:禁言(非管理员)
import botpy
from botpy.message import Message
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
await self.api.mute_member(guild_id="xxxx", user_id="xxxx", mute_end_timestamp="xxxx", mute_seconds="xxxx")
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid={appid}, token={token})
self.api.mute_member 禁言 非管理员
第 19 课、API 教程:查看错误码
错误码:查看 - 错误码
第 20 课、API 教程:事件监听
通过继承实现 bot.Client, 实现自己的机器人Client,在 MyClient 类中调用函数实现处理事件的功能。
class MyClient(botpy.Client):
监听的事件(公域消息事件的监听)
首先需要订阅事件public_guild_messages
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
监听 @ 机器人 和 频道删除消息
from botpy.message import Message
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
"""
此处为处理该事件的代码
"""
async def on_public_message_delete(self, message: Message):
"""
此处为处理该事件的代码
"""
全部消息监听,而不只是 at 机器人的消息
首先需要订阅事件guild_messages
intents = botpy.Intents(guild_messages=True)
client = MyClient(intents=intents)
from botpy.message import Message
class MyClient(botpy.Client):
async def on_message_create(self, message: Message):
"""
此处为处理该事件的代码
"""
async def on_message_delete(self, message: Message):
"""
此处为处理该事件的代码
"""
频道成员事件的监听
首先需要订阅事件guild_members
intents = botpy.Intents(guild_members=True)
client = MyClient(intents=intents)
首先需要订阅事件guild_members
intents = botpy.Intents(guild_members=True)
client = MyClient(intents=intents)
互动事件的监听
首先需要订阅事件interaction
intents = botpy.Intents(interaction=True)
client = MyClient(intents=intents)
注:需要引入Interaction
from botpy.interaction import Interaction
class MyClient(botpy.Client):
async def on_interaction_create(self, interaction: Interaction):
"""
此处为处理该事件的代码
"""
论坛事件的监听
仅 私域 机器人能够设置此 intents
首先需要订阅事件forums文章来源:https://www.toymoban.com/news/detail-400282.html
intents = botpy.Intents(forums=True)
client = MyClient(intents=intents)
注:需要引入Thread、Post、Reply和AuditResult文章来源地址https://www.toymoban.com/news/detail-400282.html
from botpy.forum import Thread
from botpy.types.forum import Post, Reply, AuditResult
class MyClient(botpy.Client):
async def on_forum_thread_create(self, thread: Thread):
"""
此处为处理该事件的代码
"""
async def on_forum_thread_update(self, thread: Thread):
"""
此处为处理该事件的代码
"""
async def on_forum_thread_delete(self, thread: Thread):
"""
此处为处理该事件的代码
"""
async def on_forum_post_create(self, post: Post):
"""
此处为处理该事件的代码
"""
async def on_forum_post_delete(self, post: Post):
"""
此处为处理该事件的代码
"""
async def on_forum_reply_create(self, reply: Reply):
"""
此处为处理该事件的代码
"""
async def on_forum_reply_delete(self, reply: Reply):
"""
此处为处理该事件的代码
"""
async def on_forum_publish_audit_result(self, auditresult: AuditResult):
"""
此处为处理该事件的代码
"""
到了这里,关于手把手QQ机器人制作教程,根据官方接口进行开发,基于Python语言制作的详细教程(更新中)的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!