Telegram Bot如何获取群组上传图片的全部file_id？（Webhook场景）

这个问题其实是Telegram Webhook的设计特性导致的——当用户发送一组图片（媒体组）时，Telegram不会一次性把所有图片的信息打包成一个请求发送，而是每个图片单独触发一个Webhook请求，但所有请求都会携带同一个media_group_id。你之前只拿到最后一张，大概率是因为每次处理请求时直接覆盖了之前的数据，没有把同一个媒体组的内容聚合起来。

下面是具体的实现方案：

核心思路 #

我们需要通过media_group_id来关联同一个媒体组的所有图片，先临时存储每个请求里的图片信息，等所有图片都接收完毕后，再统一处理整个媒体组的file_id列表。

步骤详解 #

1. 准备临时存储 #

你需要一个临时存储来保存media_group_id到图片列表的映射，可选的方案有：

• 内存字典：适合单进程/单实例的Bot，简单但重启后数据会丢失，多实例环境不适用。
• Redis/内存数据库：适合多实例或需要持久化的场景，能跨进程共享数据，还可以设置自动过期时间。

2. 处理每个Webhook请求 #

每次收到Webhook更新时，先检查是否存在media_group_id：

• 如果存在：把当前图片的file_id添加到该media_group_id对应的列表中。
• 如果不存在（单张图片）：直接处理这张图片即可。

3. 判断媒体组是否接收完成 #

由于Telegram发送同媒体组的请求间隔非常短（一般在1-2秒内），我们可以通过超时机制来判断是否收完所有图片：

• 当第一次收到某个media_group_id的请求时，启动一个定时器（比如3秒）。
• 定时器到期后，从存储中取出该media_group_id对应的所有file_id，进行后续业务处理。
• 处理完成后，删除存储中该media_group_id的数据，避免占用空间。

代码示例（Python + Redis） #

假设你用Python开发Bot，这里给出一个简化的实现：

import redis
from telegram import Update
from telegram.ext import Updater, MessageHandler, Filters, CallbackContext

# 初始化Redis连接
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
MEDIA_GROUP_TIMEOUT = 3  # 等待3秒确认所有图片接收完毕

def handle_photo(update: Update, context: CallbackContext):
    # 获取最高清晰度的图片file_id
    photo = update.message.photo[-1]
    media_group_id = update.message.media_group_id

    if not media_group_id:
        # 单张图片，直接处理
        print(f"单张图片file_id: {photo.file_id}")
        return

    # 将当前图片file_id添加到Redis列表
    r.rpush(f"media_group:{media_group_id}", photo.file_id)

    # 第一次收到该media_group_id时，设置超时任务
    if r.get(f"media_group:{media_group_id}:processing") is None:
        r.setex(f"media_group:{media_group_id}:processing", MEDIA_GROUP_TIMEOUT, "1")
        # 调度定时器任务，超时后处理整个媒体组
        context.job_queue.run_once(process_media_group, MEDIA_GROUP_TIMEOUT, context=media_group_id)

def process_media_group(context: CallbackContext):
    media_group_id = context.job.context
    # 从Redis获取所有file_id
    file_ids = r.lrange(f"media_group:{media_group_id}", 0, -1)
    
    if file_ids:
        print(f"媒体组{media_group_id}的所有file_id: {file_ids}")
        # 这里添加你的业务逻辑，比如保存到数据库、转发等
    
    # 清理Redis中的临时数据
    r.delete(f"media_group:{media_group_id}")
    r.delete(f"media_group:{media_group_id}:processing")

def main():
    # 替换成你的Bot Token
    updater = Updater("YOUR_BOT_TOKEN")
    dp = updater.dispatcher

    dp.add_handler(MessageHandler(Filters.photo, handle_photo))

    # 启动Webhook服务，替换成你的配置
    updater.start_webhook(listen="0.0.0.0", port=8080, url_path="YOUR_WEBHOOK_PATH")
    updater.bot.set_webhook(url="YOUR_WEBHOOK_URL")

    updater.idle()

if __name__ == '__main__':
    main()

注意事项 #

• 如果使用内存字典，在多进程或多线程环境下要注意线程安全，建议用线程锁保护操作。
• Redis的setex命令可以自动过期键，避免存储中残留无效数据。
• 超时时间不要设置太短，避免因网络延迟导致遗漏图片；也不要太长，防止存储占用过高，3-5秒是比较合理的范围。

内容的提问来源于stack exchange，提问作者Alexey Shablowski