На aiogram 2 делал так, через middleware:
class AlbumMiddleware(BaseMiddleware):
"""This middleware is for capturing media groups."""
album_data: dict = {}
def __init__(self, latency = 0.6):
"""
You can provide custom latency to make sure
albums are handled properly in highload.
"""
self.latency = latency
super().__init__()
async def on_process_message(self, message: types.Message, data: dict):
if not message.media_group_id:
if message.photo :
data["one_photo"] = message
elif message.animation or message.video or message.video_note or message.voice or message.audio:
data["album"] = [message]
return
try:
self.album_data[message.media_group_id].append(message)
raise CancelHandler() # Tell aiogram to cancel handler for this group element
except KeyError:
self.album_data[message.media_group_id] = [message]
await asyncio.sleep(self.latency)
message.conf["is_last"] = True
data["album"] = self.album_data[message.media_group_id]
async def on_post_process_message(self, message: types.Message, result: dict, data: dict):
"""Clean up after handling our album."""
if message.media_group_id and message.conf.get("is_last"):
del self.album_data[message.media_group_id]
И потом при обработке сообщения, будет 2 переменные album, one_photo:
async def func(
message: types.Message,
album: Optional[List[types.Message]] = None,
one_photo: Optional[types.Message] = None
):
если в сообщении только одна картинка, она будет в one_photo, в остальном случае все будет в album
