钉钉、飞书、企微、QQ机器人PDF加密数字证书签名功能全面上线

背景

前面已经完成了钉钉、飞书自动化PDF加密数字证书签名，钉钉和飞书代码实现相似。由于企业微信和QQ机器人只能长连接机器人接收文件，回调机器人不能接收文件，代码实现和钉钉、飞书差异较大，所以迟迟没有同步上线功能。

企业微信机器人实现

不得不吐槽一下腾讯，官方SDK多少年不更新，开源项目就这么不上心，连上传下载文件都没有，幸好有热心开发者提交了合并请求，完善了基础功能。
ws_client.download_file ws_client.upload_media 均为inkOrCloud实现版本。

代码实现

import asyncio
import os
import logging as mylogger
import subprocess

from aibot import WSClient, WSClientOptions,generate_req_id,MediaType

CORPID=""
BOT_ID=os.getenv("BOT_ID")
BOT_SECRET=os.getenv("BOT_SECRET")
# ===============================================
# 设置日志级别
mylogger.basicConfig(
    level=mylogger.DEBUG,   # 可选：DEBUG / INFO / WARNING / ERROR / CRITICAL
    format="%(asctime)s %(levelname)s %(name)s - %(message)s" # ,datefmt="%Y-%m-%d %H:%M:%S" #自定义日期格式
)

# 1. 创建客户端实例
ws_client = WSClient(
    WSClientOptions(
        bot_id=BOT_ID,      # 替换为企业微信后台获取的机器人 ID
        secret=BOT_SECRET,   # 替换为企业微信后台获取的机器人 Secret
        logger=mylogger
    )
)

# 2. 监听认证成功
@ws_client.on('authenticated')
def on_authenticated():
    print('🔐 认证成功')

# 3. 监听进入会话事件（发送欢迎语）
@ws_client.on('event.enter_chat')
async def on_enter_chat(frame):
    await ws_client.reply_welcome(frame, {
        'msgtype': 'text',
        'text': {'content': '您好！我是智能助手，有什么可以帮您的吗？'},
    })

# 4. 监听文本消息并进行回复
@ws_client.on("message.text")
async def on_text(frame):
    content = frame.get('body', {}).get('text', {}).get('content', '')
    print(f'收到文本: {content}')

    stream_id = generate_req_id('stream')

    # 发送流式中间内容
    await ws_client.reply_stream(frame, stream_id, '正在思考中...', False)

    # 发送最终结果
    await asyncio.sleep(1)
    await ws_client.reply_stream(frame, stream_id, f'你好！你说的是: "{content}"', True)

# 5. 监听文件消息并回复文件
@ws_client.on("message.file") 
async def on_file(frame):
    body = frame.get("body", {})
    file_info = body.get("file", {})

    from_user=body.get("from",{}).get("userid",{})
    
    file_url = file_info.get("url")
    aeskey = file_info.get("aeskey")
    # filename = file_info.get("filename", "received_file")
    
    if not file_url or not aeskey:
        print("文件消息缺少 url 或 aeskey")
        return
    
    # 1. 下载并解密文件
    try:
        #filename,file_content = download_and_decrypt(file_url, aeskey)
        file_content,filename = await ws_client.download_file(file_url, aeskey)
        print(f"文件 [{filename}] 解密成功，大小: {len(file_content)} bytes")
        
        # 保存到本地（可选）
        with open(f"./{filename}", 'wb') as f:
            f.write(file_content)
            
    except Exception as e:
        print(f"文件下载解密失败: {e}")
        return

    signpdf(from_user,filename)
    # 2. 回复文件（传入本地路径）
    fname,extname=os.path.splitext(filename)
    signed_filename=fname+"_signed"+extname

    with open("/home/temp/tmpoutput/"+signed_filename,'rb') as reply_file:
        file_bytes=reply_file.read()
        media_id=await ws_client.upload_media(
            file_bytes,
            signed_filename,
            MediaType.File
        )
        reply_msg = {
            "msgtype": "file",
            "file": {"media_id": media_id}
        }
        await ws_client.reply(frame, reply_msg)

        print(f"✅ 文件消息已回复 media_id={media_id}")


def signpdf(from_user:str,filename:str):
    if from_user=='admin':
        # 执行一个简单的 shell 命令
        sign_result = subprocess.run(['/home/pythonenv/bin/python3', '/home/temp/signpdf.py','-f',filename], capture_output=True, text=True)
        # 打印命令的输出
        print(sign_result.stdout)
    else:
        sign_result = subprocess.run(['/home/pythonenv/bin/python3', '/home/temp/signpdf2.py','-f',filename], capture_output=True, text=True)
        # 打印命令的输出
        print(sign_result.stdout)


# 6. 启动（便捷方法，内部管理事件循环）
#ws_client.run()
async def main():
    await ws_client.connect()
    # 保持运行
    await asyncio.Event().wait()

asyncio.run(main())

QQ机器人实现

QQ机器人SDK和企业微信虽然同属一个公司，风格差别挺大。QQ文件下载不需要解密，不支持分片上传文件。文件上传这类的基础功需要自己实现。腾讯耍大牌，文档没有钉钉、飞书详尽。

代码实现

import botpy
import httpx,os,base64
from botpy.message import C2CMessage
import subprocess
import requests
from botpy.robot import Token


# ========== 配置项 ==========
APPID=os.getenv("QQ_APPID")
CLIENT_SECRET=os.getenv("QQ_CLIENT_SECRET")
API_BASE = "https://api.bot.qq.com"
# ===========================

class MyClient(botpy.Client):

    async def on_ready(self):
        print(f"机器人 [{self.robot.name}] 已成功连接并启动！")

    # __私有方法，和普通数相比，在类中定义函数只有一点不同，
    # 第一参数 永远是 类的本身实例变量self，并且调用时，不用传递该参数。
    async def __download_file(self,url: str, save_path: str):
        async with httpx.AsyncClient() as http_client:
            resp = await http_client.get(url)
            if resp.status_code == 200:
                # 确保保存目录存在
                os.makedirs(os.path.dirname(save_path), exist_ok=True)
                with open(save_path, "wb") as f:
                    f.write(resp.content)
                    print(f"文件已成功保存至: {save_path}")
            else:
                print(f"下载失败，状态码: {resp.status_code}")

    async def __upload_file(self, file_path: str, filename: str, openid: str, message: C2CMessage):
        file_url = file_path # 这里需要填写上传的资源Url
        # 1=图片(png/jpg): 软限制 20MB, 硬限制 200MB
        # 2=视频(mp4): 软限制 30MB, 硬限制 200MB
        # 3=语音(silk): 软限制 20MB, 硬限制 200MB
        # 4=文件: 软限制 200MB, 硬限制 200MB 超过软限制会降级为文件类型上传，超过硬限制会报错
        

        upload_result = await self.upload_file_by_base64(file_path=file_url, file_name=filename, openid=openid)
        if upload_result["status_code"] != 200:
            print("文件上传失败！")
        else:
            file_info = upload_result["resp_json"]["file_info"]
            print(file_info)

        # 资源上传后，会得到Media，用于发送消息
        # 发送文件消息 msg_type=7
            await self.api.post_c2c_message(
                openid=openid,
                content="这是你要的PDF文件",
                msg_type=7,
                media={"file_info": file_info},
                msg_id=message.id
            )
        
    async def __file_to_base64(self, file_path: str) -> str:
        with open(file_path, "rb") as f:
            return base64.b64encode(f.read()).decode("utf-8")
    
    async def upload_file_by_base64(self,file_path: str,file_name: str,openid: str) -> dict:
        """调用QQ机器人接口base64上传PDF文件"""
        url = f"{API_BASE}/v2/users/{openid}/files"
        token = Token(APPID, CLIENT_SECRET)
        await token.update_access_token()
        token_str = token.get_string()
        print(f"获取AccessToken成功: {token_str}{token.get_type()}")
        headers = {
            "Authorization": f"{token_str}",
            "Content-Type": "application/json"
        }

        b64_data = await self.__file_to_base64(file_path)

        payload = {
            "file_type": 4,          # 4代表通用文件(PDF/docx/zip等)
            "file_name": file_name,
            "file_data": b64_data,
            "srv_send_msg": False    # 是否自动推送消息，false只获取file_info
        }

        resp = requests.post(url, json=payload, headers=headers)
        # print("响应状态码:", resp.status_code)
        # print("返回内容:", resp.text)
        if resp.status_code != 200:
            print("文件上传失败！")
            return {"status_code": resp.status_code}
        else:
            print("文件上传成功！")
            return {"status_code": resp.status_code, "resp_json": resp.json()}

    async def __signpdf(self, from_user: str, filename: str):
            if from_user=='admin':
                # 执行一个简单的 shell 命令
                sign_result = subprocess.run(['/home/pythonenv/bin/python3', '/home/pdftemp/signpdf.py','-f',filename], capture_output=True, text=True)
                # 打印命令的输出
                print(sign_result.stdout)
            else:
                sign_result = subprocess.run(['/home/pythonenv/bin/python3', '/home/pdftemp/signpdf2.py','-f',filename], capture_output=True, text=True)
                # 打印命令的输出
                print(sign_result.stdout)
    
    async def on_c2c_message_create(self, message: C2CMessage):
        print(f"[诊断] 收到私聊消息: {message.content}")
        openid=message.author.user_openid
        print(f"openid: {openid}")
        # 检查消息中是否包含附件（文件、图片等）
        if message.attachments and len(message.attachments) > 0:
           for attachment in message.attachments:
               file_url = attachment.url
               filename = attachment.filename or "unknown_file"
               fname,extname=os.path.splitext(filename)
               signed_filename=fname+"_signed"+extname
               if extname in [".pdf",".PDF"]:
                    print(f"开始签署文件: {filename}, 开始下载...{file_url}")
                    # 将文件下载到本地的 ./downloads 目录
                    await self.__download_file(file_url, f"./{filename}")

                    ## 处理文件
                    await self.__signpdf(openid, filename)

                    ## 上传文件
                    await self.__upload_file(f"/home/pdftemp/tmpoutput/{signed_filename}", signed_filename, openid, message)

                    # 回复私聊
                    await message._api.post_c2c_message(
                        openid=message.author.user_openid,
                        content="✅文件签署完成！",
                        msg_id=message.id
                    )
               else:
                    await message._api.post_c2c_message(
                        openid=message.author.user_openid,
                        content="⚠️不支持的文件类型！请上传PDF文件！",
                        msg_id=message.id
                    )
                    print(f"不支持的文件类型: {filename}")
                    continue
        else:
            # 否则，回复私聊
            await message._api.post_c2c_message(
                openid=message.author.user_openid,
                content="欢迎使用iBot机器人！更多功能后续开发中...",
                msg_id=message.id
            )

# 1.2.1 版本正确的意图配置：使用 public_messages
intents = botpy.Intents(public_messages=True)
client = MyClient(intents=intents)

client.run(appid=APPID, secret=CLIENT_SECRET)