钉钉、飞书、企微、QQ机器人PDF加密数字证书签名功能全面上线#
背景#
前面已经完成了钉钉、飞书自动化PDF加密数字证书签名，钉钉和飞书代码实现相似。由于企业微信和QQ机器人只能长连接机器人接收文件，回调机器人不能接收文件，代码实现和钉钉、飞书差异较大，所以迟迟没有同步上线功能。
企业微信机器人实现#
不得不吐槽一下腾讯，官方SDK多少年不更新，开源项目就这么不上心，连上传下载文件都没有，幸好有热心开发者提交了合并请求，完善了基础功能。
ws_client.download_filews_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)