QQ官方BOT 机器人Python实现群聊本地图片【base64编码】上传及其发送

news2024/12/23 10:36:09

参考:实现群聊本地图片【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"])

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/2076924.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

io进程----文件io

目录 一丶概念 二丶特点 三丶函数 1.打开文件 open 2.关闭文件 close 3.读取文件 read 4.写入文件 write 5.文件定位操作 标准IO和文件IO区别 四丶获取文件属性 1.stat函数 2.获取文件类型 五丶目录操作 一丶概念 在posix(可移植操作系统接口)中定义的一组输入输出…

STC89C52 定时器浅谈

文章目录 1、定时器1.1 定时器简介1.2 定时器构成1.2.1 系统时钟1.2.2 计数单元1.2.3 中断系统 1.2 定时器0/1的相关寄存器1.2.1 TMOD1.2.2 TCON 1.3 初始化定时器0 1、定时器 1.1 定时器简介 定时器,又称为计数器,是51单片机的内部资源,即…

16、CPU缓存架构

CPU缓存架构 CPU缓存架构详解CPU高速缓存概念CPU多核缓存架构CPU多核缓存架构缓存一致性问题分析 CPU缓存架构缓存一致性的解决方案缓存一致性协议实现原理总线窥探工作原理窥探协议类型 缓存一致性协议MESI协议 伪共享的问题linux下查看Cache Line大小避免伪共享方案 CPU缓存架…

netty编程之实现HTTP服务

写在前面 http是应用层协议,是我们日常开发中直接用到最多的协议了。本文来看下通过netty如何实现。 1:程序 netty不仅仅提供了String相关的编解码器,还贴心的提供了http相关的编码器和解码器,直接拿来用就行了,所以…

【3.3】贪心算法-解分发糖果

一、题目 老师想给孩子们分发糖果,有N 个孩子站成了一条直线,老师会根据每个孩子的表现,预先给他们评分。 你需要按照以下要求,帮助老师给这些孩子分发糖果: 1)每个孩子至少分配到1 个糖果。 2&#xff…

apache经典模型和nginx参数配置

Apache 几个经典的 Web 服务端 ①Apache prefork 模型 ● 预派生模式,有一个主控制进程,然后生成多个子进程,使用select模型,最大并发 1024 ● 每个子进程有一个独立的线程响应用户请求 ● 相对比较占用内存&#xf…

Cyberchef实用功能之-模拟沙箱的文件静态分析能力

本文将介绍如何使用cyberchef 提取文件的重要元数据的信息,根据自己的需求实现沙箱中的静态文件的分析能力。 在网络安全日常的运营,护网行动,重保活动的过程中,样本的分析是一个重要的过程,这些可疑或者恶意的样本的…

Beyond Compare忽略特定格式文本

1 问题背景: 文本对比时忽略某些文本。比如有些生成的文件需要做差异对比,除了内容有差异外,自动生成的ID也不同,想忽略这些ID。特别是文件内容比较多的时候。 如上图,其中UUID“*”的部分我想忽略。 方法&#xff1…

微信开放平台应用签名MD5

可以使用JS转换一下 一、在任一网页 右键 检查 打开调试 二、把字母转换成小写,去除(:) MD5 应用签名示例 "70:71:5F:CA:AE:E5:B1:01:30:11:8F".toLowerCase().replace(/:/g, "") 70715fcaaee5b10130118f 就是要填写…

我们为什么推出数据模型

调用接口之前,要写大量的字段校验代码? 需要关联读写多张表的数据,难以写出复杂的数据库语法? 不仅需要开发核心的业务逻辑,还需要加班搭建CMS和数据管理页面? 数据不止放在云开发,还需要查询…

ctfhub-web-基础认证

HTTP协议:什么是HTTP基本认证_网易订阅 https://zhuanlan.zhihu.com/p/64584734 弹框输入密码账号bp抓包,发现了Authorization:后面有一串BASE64的编码(输入的密码) 账号默认是admin,抓包后是basic 把抓到的包发送到爆破模块i…

MySql【约束】

约束 1.1 约束 是什么? 约束,即限制,就是通过设置约束,可以限制对数据表数据的插入,删除,更新 怎么做? 约束设置的语法,大部分是 create table 表名( 字段 数据类型(长度) 约束, 字段 数据类型(长度) 约束 ); 另外一种是建表后,使用alter语句修改表添加约束 1.1 数据类型 其实…

人社大赛算法赛题解题思路分享+第五名

关联比赛: [国家社保]全国社会保险大数据应用创新大赛 赛题背景分析及理解 本次比赛,“精准社保”的赛题为“基本医疗保险医疗服务智能监控”,由参赛队完成数据算法模型的开发设计,实现对各类医疗保险基金欺诈违规行为的准确识别。 在进行了…

AI工业ros机械臂

1、基本介绍 该产品是一款面向于人工智能、机器人工程等专业的实验平台,能够学习基于人工智能技术的智能机器臂相关知识。主要由工业六轴机械臂、机械臂控制器、边缘计算主机、安全防护工作台四部分构成。该产品可完成的课程:机器视觉、机器人操作系统RO…

Stable Diffusion 使用详解(10)--- 场景立体字

目录 背景 复刻立体雏形 Lora 模型 参数设置与controlnet 出图效果 融入图片 提取 合成 背景 虽然都是字,带场景的立体字和上一节讲的做法不太一样。一般来说,一般这种带字体的场景字现阶段都是Lora模型居多,Lora 属于轻量级模型&…

linux系统中USB模块基本原理分析

大家好,今天主要给大家分享一下,USB设备的发展历程。 第一:USB发展变化 随着时代的发展,USB模块也随之不断的升级。 USB1.1:规范了USB低全速传输; USB2.0:规范了USB高速传输,采用NRZI(反向不归零)编码(NRZI采用8bit编码方式),位填充(在数据进行NRZI编码前…

每日一练-threejs实现三维动态热力图

前言&#xff1a;学习自用Three.js搞个炫酷热力山丘图&#xff0c;作者讲解的十分详细&#xff0c;在这里不再过多赘述&#xff0c;直接上代码&#xff01; <template><div class"map" ref"map"></div> </template><script set…

亿发工单管理系统助力五金行业智造升级:高效生产新篇章

在五金制造行业&#xff0c;效率和质量决定了企业的竞争力。五金厂通过引入先进的工单管理系统&#xff0c;成功实现了从传统制造向智能制造的转型。今天&#xff0c;我们将带您深入了解这场变革背后的力量&#xff0c;揭示工单管理系统如何在五金工厂的各个环节中发挥重要作用…

Redis的哨兵(高可用)

实验环境&#xff1a;用一主两从来实现Redis的高可用架构。 一、Redis哨兵 Sentinel 进程是用于监控redis集群中Master主服务器工作的状态&#xff0c;在Master主服务器发生故障的时候&#xff0c; 可以实现Master和Slave服务器的切换&#xff0c;保证系统的高可用&#xff0c…

java一键生成数据库说明文档html格式

要验收项目了&#xff0c;要写数据库文档&#xff0c;一大堆表太费劲了&#xff0c;直接生成一个吧&#xff0c;本来想用个别人的轮子&#xff0c;网上看了几个&#xff0c;感觉效果不怎么好&#xff0c;自己动手写一个吧。抽空再把字典表补充进去就OK了 先看效果&#xff1a; …