首先找到api.py&#Vff0c;假如你是通过pip拆置的botpy。找到包对应的地址&#Vff0c;假如是clone的&#Vff0c;也间接找到api.py。
翻开后&#Vff0c;翻到最下面&#Vff0c;有两个函数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, srZZZ_send_msg: bool = False, ) -> message.Media: """ 上传/发送群聊图片 Args: group_openid (str): 您要将音讯发送到的群的 ID file_type (int): 媒体类型&#Vff1a;1 图片png/jpg&#Vff0c;2 室频mp4&#Vff0c;3 语音silk&#Vff0c;4 文件&#Vff08;久不开放&#Vff09; url (str): 须要发送媒体资源的url srZZZ_send_msg (bool): 设置 true 会间接发送音讯到目的端&#Vff0c;且会占用自动音讯频率 """ payload = locals() payload.pop("self", None) route = Route("POST", "/ZZZ2/groups/{group_openid}/files", group_openid=group_openid) return await self._ht.request(route, json=payload) async def post_group_base64file( self, group_openid: str, file_type: int, file_data: str, srZZZ_send_msg: bool = False, ) -> message.Media: """ 上传/发送群聊图片 Args: group_openid (str): 您要将音讯发送到的群的 ID file_type (int): 媒体类型&#Vff1a;1 图片png/jpg&#Vff0c;2 室频mp4&#Vff0c;3 语音silk&#Vff0c;4 文件&#Vff08;久不开放&#Vff09; file_data (str): 二进制文件的base64编码&#Vff0c;可用于与代网络url资源&#Vff0c;真现原地上传文件 srZZZ_send_msg (bool): 设置 true 会间接发送音讯到目的端&#Vff0c;且会占用自动音讯频率 """ payload = locals() payload.pop("self", None) route = Route("POST", "/ZZZ2/groups/{group_openid}/files", group_openid=group_openid) return await self._ht.request(route, json=payload) async def post_c2c_file( self, openid: str, file_type: int, url: str, srZZZ_send_msg: bool = False, ) -> message.Media: """ 上传/发送c2c图片 Args: openid (str): 您要将音讯发送到的用户的 ID file_type (int): 媒体类型&#Vff1a;1 图片png/jpg&#Vff0c;2 室频mp4&#Vff0c;3 语音silk&#Vff0c;4 文件&#Vff08;久不开放&#Vff09; url (str): 须要发送媒体资源的url srZZZ_send_msg (bool): 设置 true 会间接发送音讯到目的端&#Vff0c;且会占用自动音讯频率 """ payload = locals() payload.pop("self", None) route = Route("POST", "/ZZZ2/users/{openid}/files", openid=openid) return await self._ht.request(route, json=payload) async def post_c2c_base64file( self, openid: str, file_type: int, file_data: str, srZZZ_send_msg: bool = False, ) -> message.Media: """ 上传/发送C2C图片 (基于base64) Args: openid (str): 您要将音讯发送到的用户的 ID file_type (int): 媒体类型&#Vff1a;1 图片png/jpg&#Vff0c;2 室频mp4&#Vff0c;3 语音silk&#Vff0c;4 文件&#Vff08;久不开放&#Vff09; file_data (str): 二进制文件的base64编码&#Vff0c;可用于与代网络url资源&#Vff0c;真现原地上传文件 srZZZ_send_msg (bool): 设置 true 会间接发送音讯到目的端&#Vff0c;且会占用自动音讯频率 """ payload = { "openid": openid, "file_type": file_type, "file_data": file_data, "srZZZ_send_msg": srZZZ_send_msg } route = Route("POST", "/ZZZ2/users/{openid}/files", openid=openid) return await self._ht.request(route, json=payload) 测试脚原正在群里@呆板人便可主动发送图片。正在私聊发送任意音讯发送图片。
# -*- coding: utf-8 -*- import asyncio import os import base64 import botpy import htV from botpy import logging from botpy.eVt.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 htV.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 eVcept EVception 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 = "hts://s2.loli.net/2024/08/22/IosKfwTe9ZZZLkJBH.jpg" file_data = await get_photo_base64("hts://s2.loli.net/2024/08/22/IosKfwTe9ZZZLkJBH.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) eVcept EVception as e: _log.error(f"Error processing message: {e}") # 发送文原音讯 async def send_teVt_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) eVcept EVception 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}") eVcept EVception 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_teVt_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"])
