如何为Aiogram 3编写媒体组转发中间件？实现群组批量转发

核心逻辑 #

用**状态机（FSM）**标记用户是否处于等待媒体组的状态，配合中间件拦截并收集同media_group_id的所有消息，待媒体组发送完成后，统一转发给所有用户。

代码实现步骤 #

1. 基础配置与状态定义

from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.types import InputMediaPhoto, InputMediaVideo
from aiogram import BaseMiddleware
from typing import Dict, Any, Callable, Awaitable
import asyncio

# 用集合模拟存储所有用户ID，实际项目建议用数据库持久化
all_users = set()

# 定义状态类，标记用户是否在等待媒体组
class MediaStates(StatesGroup):
    waiting_media_group = State()

# 初始化Bot和Dispatcher
bot = Bot(token="YOUR_BOT_TOKEN")
dp = Dispatcher()

2. 媒体组收集中间件

Aiogram 3的中间件需继承BaseMiddleware，在消息预处理阶段拦截并收集媒体组消息：

class MediaGroupCollectorMiddleware(BaseMiddleware):
    def __init__(self):
        # 临时存储媒体组消息，key为media_group_id，value为对应消息列表
        self.media_groups: Dict[str, list[types.Message]] = {}

    async def __call__(
        self,
        handler: Callable[[types.Message, Dict[str, Any]], Awaitable[Any]],
        event: types.Message,
        data: Dict[str, Any]
    ) -> Any:
        state = data.get("state")
        if not state:
            return await handler(event, data)
        
        current_state = await state.get_state()
        # 只处理处于等待媒体组状态的消息
        if current_state != MediaStates.waiting_media_group.state:
            return await handler(event, data)
        
        # 处理带media_group_id的媒体组消息
        if event.media_group_id:
            mg_id = event.media_group_id
            if mg_id not in self.media_groups:
                self.media_groups[mg_id] = []
            self.media_groups[mg_id].append(event)

            # 延迟1秒判断是否已收集完整个媒体组（可根据实际场景优化判断逻辑）
            await asyncio.sleep(1)
            # 确认当前列表是完整媒体组后，存入data供后续处理
            if mg_id in self.media_groups:
                data["collected_media"] = self.media_groups.pop(mg_id)
                await state.clear()
                return await handler(event, data)
        else:
            # 用户发送的不是媒体组，提示后重置状态
            await event.answer("请发送完整的媒体组（多张/多条媒体消息）")
            await state.clear()
            return
        
        # 未收集完成时不执行后续handler
        return

3. 命令触发与媒体组转发逻辑

# 注册媒体组收集中间件
dp.message.middleware(MediaGroupCollectorMiddleware())

# 处理/start命令，将用户加入转发列表
@dp.message(Command("start"))
async def cmd_start(message: types.Message):
    all_users.add(message.from_user.id)
    await message.answer("已注册，可发送/share_media触发媒体组转发")

# 处理/share_media命令，触发等待媒体组的状态
@dp.message(Command("share_media"))
async def cmd_share_media(message: types.Message, state: FSMContext):
    await state.set_state(MediaStates.waiting_media_group)
    await message.answer("请发送需要转发的媒体组（多张图片/视频）")

# 处理收集完成的媒体组，批量转发给所有用户
@dp.message(MediaStates.waiting_media_group)
async def process_collected_media(message: types.Message, data: Dict[str, Any]):
    collected_media = data.get("collected_media")
    if not collected_media:
        return
    
    # 将收集到的消息转换为send_media_group所需的InputMedia格式
    media_list = []
    for idx, msg in enumerate(collected_media):
        if msg.photo:
            # 取最高分辨率的照片文件ID
            photo = msg.photo[-1]
            # 仅第一条消息保留caption
            media_list.append(InputMediaPhoto(
                media=photo.file_id,
                caption=msg.caption if idx == 0 else None
            ))
        elif msg.video:
            media_list.append(InputMediaVideo(
                media=msg.video.file_id,
                caption=msg.caption if idx == 0 else None
            ))
        # 可扩展支持音频、文档等其他媒体类型
    
    # 遍历所有用户转发媒体组
    for user_id in all_users:
        try:
            await bot.send_media_group(chat_id=user_id, media=media_list)
        except Exception as e:
            print(f"转发给用户{user_id}失败：{str(e)}")
    
    await message.answer("媒体组已成功转发给所有用户")

4. 启动机器人

async def main():
    await dp.start_polling(bot)

if __name__ == "__main__":
    asyncio.run(main())

关键适配点说明 #

• Aiogram 3的中间件注册方式改为dp.message.middleware()，且必须实现__call__方法，不再兼容v2的旧写法。
• 状态管理改用FSMContext，通过set_state和clear方法控制用户状态。
• 媒体组转发需使用send_media_group方法，且必须将媒体消息转换为InputMedia*格式，仅第一条消息可携带caption。
• 媒体组收集的判断逻辑可根据需求优化，比如通过Telegram API获取媒体组总条数，或用Redis等缓存存储临时消息，避免内存占用过高。

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