参考:实现群聊本地图片【base64编码】上传及其发送 by SlieFamily ·拉取请求 #199 ·腾讯连接/Botpy (github.com)
首先找到api.py,如果你是通过pip安装的botpy。找到包对应的地址,如果是clone的,也直接找到api.py。
打开后,翻到最下面,有两个函数post_group_file和post_c2c_file。我们在这里添加我们使用base64上传的函数。post_group_base64file和post_c2c_base64file。
库
async def post_group_file(
self,
group_openid: str,
file_type: int,
url: str,
srv_send_msg: bool = False,
) -> message.Media:
"""
上传/发送群聊图片
Args:
group_openid (str): 您要将消息发送到的群的 ID
file_type (int): 媒体类型:1 图片png/jpg,2 视频mp4,3 语音silk,4 文件(暂不开放)
url (str): 需要发送媒体资源的url
srv_send_msg (bool): 设置 true 会直接发送消息到目标端,且会占用主动消息频次
"""
payload = locals()
payload.pop("self", None)
route = Route("POST", "/v2/groups/{group_openid}/files", group_openid=group_openid)
return await self._http.request(route, json=payload)
async def post_group_base64file(
self,
group_openid: str,
file_type: int,
file_data: str,
srv_send_msg: bool = False,
) -> message.Media:
"""
上传/发送群聊图片
Args:
group_openid (str): 您要将消息发送到的群的 ID
file_type (int): 媒体类型:1 图片png/jpg,2 视频mp4,3 语音silk,4 文件(暂不开放)
file_data (str): 二进制文件的base64编码,可用于代替网络url资源,实现本地上传文件
srv_send_msg (bool): 设置 true 会直接发送消息到目标端,且会占用主动消息频次
"""
payload = locals()
payload.pop("self", None)
route = Route("POST", "/v2/groups/{group_openid}/files", group_openid=group_openid)
return await self._http.request(route, json=payload)
async def post_c2c_file(
self,
openid: str,
file_type: int,
url: str,
srv_send_msg: bool = False,
) -> message.Media:
"""
上传/发送c2c图片
Args:
openid (str): 您要将消息发送到的用户的 ID
file_type (int): 媒体类型:1 图片png/jpg,2 视频mp4,3 语音silk,4 文件(暂不开放)
url (str): 需要发送媒体资源的url
srv_send_msg (bool): 设置 true 会直接发送消息到目标端,且会占用主动消息频次
"""
payload = locals()
payload.pop("self", None)
route = Route("POST", "/v2/users/{openid}/files", openid=openid)
return await self._http.request(route, json=payload)
async def post_c2c_base64file(
self,
openid: str,
file_type: int,
file_data: str,
srv_send_msg: bool = False,
) -> message.Media:
"""
上传/发送C2C图片 (基于base64)
Args:
openid (str): 您要将消息发送到的用户的 ID
file_type (int): 媒体类型:1 图片png/jpg,2 视频mp4,3 语音silk,4 文件(暂不开放)
file_data (str): 二进制文件的base64编码,可用于代替网络url资源,实现本地上传文件
srv_send_msg (bool): 设置 true 会直接发送消息到目标端,且会占用主动消息频次
"""
payload = {
"openid": openid,
"file_type": file_type,
"file_data": file_data,
"srv_send_msg": srv_send_msg
}
route = Route("POST", "/v2/users/{openid}/files", openid=openid)
return await self._http.request(route, json=payload)
测试脚本
在群里@机器人即可自动发送图片。在私聊发送任意消息发送图片。
# -*- coding: utf-8 -*-
import asyncio
import os
import base64
import botpy
import httpx
from botpy import logging
from botpy.ext.cog_yaml import read
from botpy.message import GroupMessage, Message, C2CMessage
# 上传图片测试
test_config = read(os.path.join(os.path.dirname(__file__), "config.yaml"))
_log = logging.get_logger()
async def get_photo_base64(image_url):
try:
async with httpx.AsyncClient() as client:
file_response = await client.get(image_url)
if file_response.status_code == 200:
file_content = file_response.content
base64_encoded = base64.b64encode(file_content).decode('utf-8')
return base64_encoded
else:
print(f"下载文件失败: {file_response.status_code}")
return None
except Exception as e:
print(f"获取文件信息出错: {e}")
return None
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):
await self.process_message(message, is_group=True)
async def on_c2c_message_create(self, message: C2CMessage):
await self.process_message(message, is_group=False)
async def process_message(self, message, is_group):
try:
file_url = "https://s2.loli.net/2024/08/22/IosKfwTe9vLkJBH.jpg"
file_data = await get_photo_base64("https://s2.loli.net/2024/08/22/IosKfwTe9vLkJBH.jpg")
await self.send_media_message(message, file_url=file_url, file_data=None, is_group=is_group, msg_seq=1, retry_count=3)
await self.send_media_message(message, file_url=None, file_data=file_data, is_group=is_group, msg_seq=2, retry_count=3)
except Exception as e:
_log.error(f"Error processing message: {e}")
# 发送文本消息
async def send_text_message(self, message, content, is_group, msg_seq):
try:
if is_group:
messageResult = await message._api.post_group_message(
group_openid=message.group_openid,
msg_type=0,
msg_id=message.id,
msg_seq=msg_seq,
content=content
)
else:
messageResult = await message._api.post_c2c_message(
openid=message.author.user_openid,
msg_type=0,
msg_id=message.id,
msg_seq=msg_seq,
content=content
)
_log.info(messageResult)
except Exception as e:
_log.error(f"Error sending message: {e}")
async def send_media_message(self, message, file_url=None, file_data=None, is_group=False, msg_seq=1, retry_count=3):
try:
if file_data: # 如果提供了base64编码的数据
_log.info(f"开始发送base64图片: {message.content}")
file_type = 1 # 文件类型1表示图片
if is_group:
# 传入base64
uploadMedia = await message._api.post_group_base64file(
group_openid=message.group_openid,
file_type=file_type,
file_data=file_data # 使用base64数据
)
await message._api.post_group_message(
group_openid=message.group_openid,
msg_type=7, # 7表示富媒体类型
msg_id=message.id,
msg_seq=msg_seq,
media=uploadMedia
)
else:
uploadMedia = await message._api.post_c2c_base64file(
openid=message.author.user_openid,
file_type=file_type,
file_data=file_data # 使用base64数据
)
await message._api.post_c2c_message(
openid=message.author.user_openid,
msg_type=7, # 7表示富媒体类型
msg_id=message.id,
msg_seq=msg_seq,
media=uploadMedia
)
else: # 如果提供了URL
_log.info(f"开始发送URL图片")
if is_group:
uploadMedia = await message._api.post_group_file(
group_openid=message.group_openid,
file_type=1,
url=file_url
)
await message._api.post_group_message(
group_openid=message.group_openid,
msg_type=7,
msg_id=message.id,
msg_seq=msg_seq,
media=uploadMedia
)
else:
uploadMedia = await message._api.post_c2c_file(
openid=message.author.user_openid,
file_type=1,
url=file_url
)
await message._api.post_c2c_message(
openid=message.author.user_openid,
msg_type=7,
msg_id=message.id,
msg_seq=msg_seq,
media=uploadMedia
)
_log.info(f"发送图片完成: {message.content}")
except Exception as e:
_log.error(f"Error sending media message: {e}")
if retry_count > 0:
_log.info(f"Retrying to send media message, attempts left: {retry_count}")
await asyncio.sleep(2) # 等待一段时间后再重试
await self.send_media_message(message, file_url=file_url, file_data=file_data, is_group=is_group,
msg_seq=msg_seq + 1,
retry_count=retry_count - 1)
else:
await self.send_text_message(message, "发送图片失败", is_group, msg_seq=msg_seq + 1)
if __name__ == "__main__":
intents = botpy.Intents(public_messages=True)
client = MyClient(intents=intents)
client.run(appid=test_config["appid"], secret=test_config["secret"])