在Telegram机器人开发中,接收用户发送的文档和图片是最常见的需求之一。无论是实现文件转存、图片识别还是自动备份,理解机器人如何获取文件是构建强大功能的基础。本文将从Bot API的更新机制讲起,深入解析Update对象中的文档与图片字段,并给出完整的Python代码示例,帮助你快速掌握从接收到下载的全流程。
一、Telegram Bot如何收到用户文件?
Telegram机器人无法主动接收消息,必须通过两种方式获取用户交互的更新:
- getUpdates轮询:机器人主动调用该API获取新更新,适合开发和简单场景。
- Webhook推送:Telegram服务器将更新以POST请求发送到你的HTTPS服务器,实时性好,适合生产环境。
无论哪种方式,更新的载体都是Update对象,其中包含消息、回调等内容。当用户向机器人发送文档或图片时,对应的Message对象中会包含document或photo字段,这些字段里携带文件的标识信息。
二、解析更新:获取文件元数据
一个典型的Update对象示例如下:
{
"update_id": 10000,
"message": {
"message_id": 1234,
"chat": { "id": 123456789 },
"text": "",
"document": {
"file_name": "report.pdf",
"mime_type": "application/pdf",
"file_id": "BQADAgADCQAD",
"file_unique_id": "AgADAgADCQAD",
"file_size": 102400
},
"photo": [
{ "file_id": "photo1", "file_unique_id": "Af", "width": 100, "height": 100, "file_size": 5000 },
{ "file_id": "photo2", "file_unique_id": "Af2", "width": 320, "height": 320, "file_size": 20000 }
]
}
}
对于文档,Message.document包含file_id、file_unique_id、file_name、mime_type和file_size。file_id是下载文件的关键,但它是可变的,建议使用file_unique_id作为持久化标识。
对于图片,Message.photo是一个数组,从小到大包含多个分辨率,每个元素含file_id和尺寸。通常选择最后一个(最大分辨率)进行下载。
三、通过getUpdates接收文件(轮询模式)
对于简单应用,可以使用轮询。下面是一个使用requests库的Python示例:
import requests
import time
TOKEN = "YOUR_BOT_TOKEN"
API_URL = f"https://api.telegram.org/bot"
def get_updates(offset=None):
url = f"/getUpdates"
params = {"timeout": 30, "offset": offset}
resp = requests.get(url, params=params)
return resp.json()["result"]
while True:
updates = get_updates()
for update in updates:
message = update.get("message")
if not message:
continue
if "document" in message:
file_id = message["document"]["file_id"]
print("收到文档:", message["document"]["file_name"])
elif "photo" in message:
# 取最大尺寸
photo = message["photo"][-1]
file_id = photo["file_id"]
print("收到图片:", file_id)
# 处理完需要更新offset,避免重复
offset = update["update_id"] + 1
time.sleep(1)
注意,在实际项目中要妥善管理offset值,确保每条更新只被处理一次。
四、通过Webhook接收文件(推荐模式)
Webhook模式适合生产环境,因为它能实时推送,无需轮询。以Flask为例:
from flask import Flask, request, jsonify
import requests
TOKEN = "YOUR_BOT_TOKEN"
API_URL = f"https://api.telegram.org/bot"
app = Flask(__name__)
@app.route(f"/", methods=["POST"])
def webhook():
update = request.get_json()
message = update.get("message")
if message:
if "document" in message:
file_id = message["document"]["file_id"]
print("收到文档:", message["document"]["file_name"])
elif "photo" in message:
photo = message["photo"][-1]
file_id = photo["file_id"]
print("收到图片")
return jsonify({"ok": True})
if __name__ == "__main__":
# 设置webhook
requests.post(f"/setWebhook", json={"url": "https://你的域名/" + TOKEN})
app.run(host="0.0.0.0", port=443, ssl_context=("cert.pem", "key.pem"))
Webhook要求你的服务器必须对外可访问且使用HTTPS。
五、下载文件:从file_id到本地保存
获取file_id后,需要通过getFile拿到下载路径,再构造URL下载。步骤:
- 调用
getFile获取file_path。 - 使用
https://api.telegram.org/file/bot/下载文件。
Python实现:
import requests
def download_file(file_id, save_path):
# 1. 获取文件路径
resp = requests.post(f"/getFile", json={"file_id": file_id})
result = resp.json()["result"]
file_path = result["file_path"]
# 2. 下载
file_url = f"/file/" # 注意API_URL不同
file_resp = requests.get(file_url)
with open(save_path, "wb") as f:
f.write(file_resp.content)
return save_path
注意:getFile请求应该发送到https://api.telegram.org/bot,而下载文件则使用https://api.telegram.org/file/bot,两者主机名不同。
六、完整实战:一个接收并保存文件的生产级脚本
下面给出一个结合Webhook和下载的完整Python脚本,它能够接收文档和图片并保存到本地目录:
from flask import Flask, request, jsonify
import requests
import os
TOKEN = "YOUR_BOT_TOKEN"
API_BASE = f"https://api.telegram.org/bot"
FILE_BASE = f"https://api.telegram.org/file/bot"
UPLOAD_DIR = "./uploads"
app = Flask(__name__)
@app.route(f"/", methods=["POST"])
def handle_update():
update = request.get_json()
if "message" not in update:
return jsonify({"ok": True})
message = update["message"]
if "document" in message:
doc = message["document"]
file_id = doc["file_id"]
file_name = doc.get("file_name", "document")
save_file(file_id, os.path.join(UPLOAD_DIR, file_name))
elif "photo" in message:
photo = message["photo"][-1]
file_id = photo["file_id"]
file_name = f"photo_{update['update_id']}.jpg"
save_file(file_id, os.path.join(UPLOAD_DIR, file_name))
return jsonify({"ok": True})
def save_file(file_id, save_path):
# 获取文件路径
r = requests.post(f"/getFile", json={"file_id": file_id})
result = r.json().get("result")
if not result:
print("getFile failed:", r.text)
return
file_path = result["file_path"]
# 下载
file_url = f"/"
file_resp = requests.get(file_url)
os.makedirs(os.path.dirname(save_path), exist_ok=True)
with open(save_path, "wb") as f:
f.write(file_resp.content)
print(f"Saved to ")
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8443, ssl_context=("cert.pem", "key.pem"))
此脚本中,我们使用Flask启动Web服务,并假设已有HTTPS证书。为便于测试,你也可以用ngrok将本地端口映射到公网HTTPS地址。
七、注意事项与最佳实践
- 文件大小限制:Telegram Bot API限制下载的最大文件大小为20MB(或50MB,取决于服务器配置)。如需处理更大文件,可部署本地Bot API服务器。
- 并发处理:Webhook可能会并发推送,建议使用任务队列(如Celery)或异步框架(如FastAPI)来避免阻塞。
- 安全加固:验证请求来源(Telegram服务器IP),并使用令牌作为路径的一部分防止恶意调用。
- 错误处理:网络或Telegram服务器故障时,应记录日志并重试。注意避免重复下载,可使用
file_unique_id作为磁盘文件名。 - 清理旧文件:定期清理临时文件,避免磁盘空间耗尽。
总结
接收文档和图片是Telegram机器人开发中的核心技能。通过解析Update对象中的document和photo字段,结合getFile与文件下载API,你可以轻松实现文件的自动保存、转存或进一步处理。本文提供了从轮询到Webhook,从解析到下载的完整指南,希望你能在此基础上构建更强大的机器人。如果你正在开发文件处理类机器人,强烈推荐使用Webhook模式,并参考官方API文档与最新库版本,确保最佳实践。