简单的消息收发:
被@之后进行的简单回复示例:
# -*- coding: utf-8 -*-
import asyncio
import os
import botpy
from botpy import logging
from botpy.ext.cog_yaml import read
from botpy.message import GroupMessage, 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_group_at_message_create(self, message: GroupMessage):
messageResult = await message._api.post_group_message(
group_openid=message.group_openid,
msg_type=0,
msg_id=message.id,
content=f"收到了消息:{message.content}\n)
_log.info(messageResult)
if __name__ == "__main__":
# 通过预设置的类型,设置需要监听的事件通道
# intents = botpy.Intents.none()
# intents.public_messages=True
# 通过kwargs,设置需要监听的事件通道
intents = botpy.Intents(public_messages=True)
client = MyClient(intents=intents)
client.run(appid=test_config["appid"], secret=test_config["secret"])
message的请求参数:
属性 | 类型 | 必填 | 说明 |
content | string | 否 | 文本内容 |
msg_type | int | 是 | 消息类型:0 是文本,2 是 markdown, 3 ark,4 embed,7 media 富媒体 |
markdown | object | 否 | Markdown对象 |
keyboard | object | 否 | Keyboard对象 |
ark | object | 否 | Ark对象 |
media | object | 否 | 富媒体单聊的file_info |
message_reference | object | 否 | 【暂未支持】消息引用 |
event_id | string | 否 | 前置收到的事件 ID,用于发送被动消息,支持事件:"INTERACTION_CREATE"、"C2C_MSG_RECEIVE"、"FRIEND_ADD" |
msg_id | string | 否 | 前置收到的用户发送过来的消息 ID,用于发送被动(回复)消息 |
msg_seq | int | 否 | 回复消息的序号,与 msg_id 联合使用,避免相同消息id回复重复发送,不填默认是1。相同的 msg_id + msg_seq 重复发送会失败。 |
如需发送多段消息需要增加"msg_seq"的参数(不同消息参数值不能一样,否者会被判定并去重)
这里选择用毫秒级时间戳(由于长度有限制[int]值的大小只取后9位)
msg_time = (int(time.time()*1000)+random.randint(0, 999)) % 1000000000
messageResult = await message._api.post_group_message(
group_openid=message.group_openid,
msg_type=0,
msg_id=message.id,
msg_seq=msg_time,#区别消息重复的参数
content=f"{args}")
# 发送消息
_log.info(messageResult)
富媒体(图片)发送的例子:
发送图片(url):
async def url_image(message: GroupMessage):
file_url = "https://pic.png" # 这里需要填写上传的资源Url
uploadMedia = await message._api.post_group_file(
group_openid=message.group_openid,
file_type=1, # 文件类型要对应上,具体支持的类型见下方表格
url=file_url# 文件Url
)
# 资源上传后,会得到Media,用于发送消息
_log.info(await message._api.post_group_message(
group_openid=message.group_openid,
msg_type=7, # 7表示富媒体类型
msg_id=message.id,
media=uploadMedia
))
路径和请求参数:
属性 | 类型 | 必填 | 说明 |
group_openid | string | 是 | 群聊的 openid |
file_type | int | 是 | 媒体类型:1 图片,2 视频,3 语音,4 文件(暂不开放) 资源格式要求 图片:png/jpg,视频:mp4,语音:silk |
url | string | 是 | 需要发送媒体资源的url |
srv_send_msg | bool | 是 | 设置 true 会直接发送消息到目标端,且会占用主动消息频次 |
file_data | 否 | 【暂未支持】 |
图文消息:
增加"content"参数即可
_log.info(await message._api.post_group_message(
group_openid=message.group_openid,
msg_type=7, # 7表示富媒体类型
msg_id=message.id,
media=uploadMedia,
content=f"{args}"#args是存储要发送的文本消息的变量
))
输出结果是图片在文本上方,且仅可包含1张图片!
本地图片的发送:
群聊暂时不支持直接发送本地图片,只能将其url参数上传
但是可以通过对象存储、本地图床或外部图床生成url
主动消息的发送(群聊事件):
class MyClient(botpy.Client):
async def on_group_add_robot(self, event: GroupManageEvent):
_log.info("机器人被添加到群聊:" + str(event))
await self.api.post_group_message(
group_openid=event.group_openid,
msg_type=0,
event_id=event.event_id,
content="hello",
)
async def on_group_del_robot(self, event: GroupManageEvent):
_log.info("机器人被移除群聊:" + str(event))
async def on_group_msg_reject(self, event: GroupManageEvent):
_log.info("群聊关闭机器人主动消息:" + str(event))
async def on_group_msg_receive(self, event: GroupManageEvent):
_log.info("群聊打开机器人主动消息:" + str(event))