Telegram Bot：如何在单次更新中下载同一media_group_id下的多张图片？

问题根源 #

Telegram会将同一media_group_id下的图片拆分为多条独立消息推送，你的当前代码仅处理单条消息中的图片，未对同组消息进行聚合，因此只能下载到一张。

实现步骤 #

• 暂存同组消息：利用bot的上下文存储（context.bot_data），将带有相同media_group_id的消息临时保存，等待同组所有消息接收完成。
• 批量下载同组图片：设置超时机制确保同组消息都被收集，之后遍历所有消息，逐个下载其中的最高清图片。

修改后的代码 #

import asyncio
from telegram import Update
from telegram.ext import ContextTypes
import requests
import logging

logger = logging.getLogger(__name__)

# 设置media_group的超时时间（秒），确保同组消息都能被收集
MEDIA_GROUP_TIMEOUT = 2

async def download_photo(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """下载同一media_group_id下的所有最高清图片"""
    media_paths = []
    message = update.message

    # 初始化bot_data中的media_groups存储
    if 'media_groups' not in context.bot_data:
        context.bot_data['media_groups'] = {}

    media_group_id = message.media_group_id

    if media_group_id:
        # 将当前消息加入对应media_group的集合
        if media_group_id not in context.bot_data['media_groups']:
            context.bot_data['media_groups'][media_group_id] = []
            # 启动超时任务，超时后自动处理该组消息
            asyncio.create_task(process_media_group(media_group_id, context))
        
        context.bot_data['media_groups'][media_group_id].append(message)
        logger.info(f"已将消息 {message.message_id} 加入media_group {media_group_id}")
        return media_paths
    else:
        # 处理单张图片消息
        if message.photo:
            await download_single_photo(message, media_paths)
        logger.info(f'所有下载的媒体路径: {media_paths}')
        return media_paths

async def process_media_group(media_group_id: str, context: ContextTypes.DEFAULT_TYPE):
    """处理暂存的media_group消息，下载所有图片"""
    await asyncio.sleep(MEDIA_GROUP_TIMEOUT)
    media_paths = []
    messages = context.bot_data['media_groups'].pop(media_group_id, [])
    
    if not messages:
        return
    
    logger.info(f"开始处理media_group {media_group_id}，共 {len(messages)} 条消息")
    for msg in messages:
        if msg.photo:
            await download_single_photo(msg, media_paths)
    
    logger.info(f'media_group {media_group_id} 下载完成，路径列表: {media_paths}')
    # 这里可以添加后续逻辑，比如发布到Twitter等
    # 示例：await post_to_twitter(media_paths)

async def download_single_photo(message, media_paths):
    """下载单条消息中的最高清图片"""
    photo = message.photo
    file_id = photo[-1].file_id
    logger.info(f'处理图片，file_id: {file_id}')
    
    try:
        file_info = await message.bot.get_file(file_id)
        download_url = f"https://api.telegram.org/file/bot{message.bot.token}/{file_info.file_path}"
        logger.info(f'下载URL: {download_url}')
        
        file_name = f'photo_{message.message_id}_{photo[-1].file_unique_id}.jpg'
        response = requests.get(download_url)
        response.raise_for_status()
        
        with open(file_name, 'wb') as f:
            f.write(response.content)
        
        media_paths.append(file_name)
        logger.info(f'图片下载成功，file_id: {file_id}')
    except Exception as e:
        logger.error(f'下载图片失败，file_id {file_id}: {e}')

关键说明 #

• MEDIA_GROUP_TIMEOUT：设置为2秒是因为Telegram推送同组消息的间隔通常很短，足够收集所有同组消息；如果网络延迟高，可适当延长。
• context.bot_data：全局存储，用于跨请求保存media_group的消息集合，重启bot后会清空，若需持久化可改用数据库。
• 异步超时任务：确保即使最后一条消息延迟，也能触发下载逻辑，不会卡住。

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