Telegram机器人通过Bot API上传大型文件的分片方法与断点续传

Telegram Bot API为开发者提供了丰富的文件交互能力，但同时也设定了严格的限额：机器人接收文件最大20MB，发送文件最大50MB。面对更大的文件，比如高清视频、数据集或软件安装包，直接调用API往往会导致请求失败。本文将介绍一套基于分片传输和断点续传的实用方案，帮助你在现有API限制下，依然能够安全、高效地传输大文件。

为什么需要分片与断点续传？

Telegram Bot API的限制是硬性的，无法通过修改参数绕过。但我们可以将一个大文件切割成多个不超过限制的小分片，分别上传到机器人端，再由机器人端合并恢复原文件。同时，为每个分片记录传输状态，当网络闪断或任务中断时，只需重传未完成的部分，这就是断点续传的核心思想。

总体设计：基于消息的分片协议

要实现分片传输，需要客户端（发送方）与机器人端（接收方）约定一套简单的“协议”。我们可以利用Telegram消息中的文件名和文本来传递元数据。

• 分片大小：建议小于20MB，考虑到Base64或传输开销，选15MB较稳妥。
• 命名规则：原始文件名.partN.后缀，其中N为从0开始的分片序号。
• 元数据：在发送文件时，通过caption或文件名附带总片数、文件总大小、SHA256哈希等信息。

以下可作为一个示例的元数据JSON，嵌在文件名尾部：

myvideo.mp4.part0?total=5&size=104857600&hash=...

客户端上传分片的Python示例

使用Python的python-telegram-bot库，我们可以编写一个简单的分片上传器：

import hashlib
import math
import os

from telegram import Bot
from telegram.error import TelegramError

BOT_TOKEN = "YOUR_BOT_TOKEN"
CHAT_ID = "RECEIVER_CHAT_ID"

bot = Bot(token=BOT_TOKEN)

# 分片大小（字节），设为15MB
def upload_large_file(file_path):
    total_size = os.path.getsize(file_path)
    chunk_size = 15 * 1024 * 1024  # 15MB
    total_parts = math.ceil(total_size / chunk_size)

    # 计算整体文件SHA256
    sha256 = hashlib.sha256()
    with open(file_path, "rb") as f:
        while True:
            data = f.read(1024 * 1024)
            if not data:
                break
            sha256.update(data)
    file_hash = sha256.hexdigest()

    with open(file_path, "rb") as f:
        part_index = 0
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            # 构造分片文件名
            part_filename = f"{os.path.basename(file_path)}.part"
            meta = f"total=;size=;hash="
            # 通过caption传递元数据（也可以放在文件名中，但文件名长度有限）
            try:
                with open(part_filename, "wb") as part_file:
                    part_file.write(chunk)
                with open(part_filename, "rb") as part_file:
                    bot.send_document(chat_id=CHAT_ID, document=part_file, filename=part_filename, caption=meta)
                os.remove(part_filename)
                print(f"Part  uploaded successfully.")
            except TelegramError as e:
                print(f"Part  failed: . Retrying soon.")
                # 简单重试，可引入指数退避
                continue
            part_index += 1

    print("All parts sent.")

机器人端接收与合并逻辑

机器人端需要监听MessageHandler，识别带有分片标记的文件，并将分片暂存。当所有分片到齐后，按顺序合并。下面是一个核心接收合并逻辑的简化实现：

import hashlib
import os
import re

from telegram import Update
from telegram.ext import ApplicationBuilder, MessageHandler, filters, ContextTypes

# 临时存储目录，实际生产建议用数据库记录状态
part_dict = {}  # key: 原始文件名，value: {expected_parts, size, hash, chunk_ids: [], received: set}

def parse_meta(caption):
    # 解析类似 total=5;size=104857600;hash=... 的字符串
    meta = {}
    for item in caption.split(';'):
        key, _, value = item.partition('=')
        meta[key] = value
    return meta

async def handle_document(update: Update, context: ContextTypes.DEFAULT_TYPE):
    doc = update.message.document
    file_name = doc.file_name
    if not file_name or '.part' not in file_name:
        return

    base_name, part_num_str = re.match(r'^(.+)\.part(\d+)$', file_name).groups()
    meta = parse_meta(update.message.caption)
    total_parts = int(meta['total'])
    total_size = int(meta['size'])
    file_hash = meta['hash']

    # 初始化记录
    if base_name not in part_dict:
        part_dict[base_name] = {
            'total': total_parts,
            'size': total_size,
            'hash': file_hash,
            'chunks': {}
        }
    record = part_dict[base_name]

    # 下载分片到临时文件
    file = await doc.get_file()
    tmp_path = f"/tmp/.part"
    await file.download_to_drive(tmp_path)

    # 记录分片
    record['chunks'][int(part_num_str)] = tmp_path
    record.setdefault('received', set()).add(int(part_num_str))

    # 检查是否收齐
    if len(record['received']) == total_parts:
        # 合并文件
        output_path = f"./downloaded/"
        with open(output_path, 'wb') as out:
            for i in range(total_parts):
                part_path = record['chunks'][i]
                with open(part_path, 'rb') as pf:
                    out.write(pf.read())
                os.remove(part_path)
        # 校验哈希
        sha256 = hashlib.sha256()
        with open(output_path, 'rb') as f:
            while True:
                data = f.read(1024 * 1024)
                if not data:
                    break
                sha256.update(data)
        assert sha256.hexdigest() == record['hash'], "Hash mismatch!"
        print(f"File  merged successfully.")
        # 清理记录
        del part_dict[base_name]

if __name__ == '__main__':
    app = ApplicationBuilder().token("YOUR_BOT_TOKEN").build()
    dp = app.add_handler(MessageHandler(filters.Document.ALL, handle_document))
    app.run_polling()

断点续传策略

断点续传的关键在于状态同步。客户端在重新发送前，应向机器人查询当前已收到的分片序号，仅重传缺失部分。可以在Bot中添加一个命令（如/status 文件名）返回已收到的分片列表。客户端在每次发送前先调用该命令，然后只发送未标记完成的片段。

# 客户端查询状态（示意）
status = await bot.send_message(chat_id=CHAT_ID, text=f"/status ")
# 解析返回的已收分片集合，跳过这些分片

同时，发送方应采用“先查询后上传”的模式，并设置合理的超时重试间隔，避免重复上传造成流量浪费。

注意事项与优化建议

• 为避免触发Telegram的限流（Rate Limit），建议将分片发送间隔控制在1秒左右，或使用sleep。
• 每个分片都建议进行哈希校验，确保传输无误。
• 合并后的文件必须进行整体SHA256校验，防止数据缺失。
• 临时文件与记录应设置过期清理机制，防止占用过多服务器空间。
• 若需要传输超大文件（>2GB），建议直接采用外部云存储，通过Bot发送链接即可。

总结

分片上传与断点续传是绕过Telegram Bot API限制的好方法。通过设计清晰的分片协议、可靠的状态记录和校验机制，我们可以用最少代价实现稳定的大文件传输。当然，这并非官方推荐的方式，对于大型业务场景，仍建议使用专业的文件传输服务。希望本文能为你打开思路，开发出更强大的Telegram机器人。