如何在aiogram中实现单消息多文件上传功能？

我需要在aiogram中实现单消息内多文件上传的功能，但官方文档里没找到对应的实现方法。目前我已经完成了单次上传单个文件并存储到独立目录的功能，现有代码可以把符合格式的文件保存到指定目录，还能通过计数器记录已保存文件的数量，代码如下：

import os
import asyncio
import logging

from aiogram.types import Message
from aiogram import Bot, Dispatcher, types, Router, F
from aiogram.types import BotCommand, BotCommandScopeDefault
from aiogram.filters.command import Command

router: Router = Router()

logging.basicConfig(level=logging.INFO)
bot = Bot(token="")
dp = Dispatcher()

EXTENSION = '<>'
FILE = ''
counter_files = 0

UPLOAD_FOLDER = "uploads/"
os.makedirs(UPLOAD_FOLDER, exist_ok=True)

@dp.message(Command("start"))
async def cmd_start(message: types.Message):
    await message.answer("Hello!")


@dp.message(Command("check_status"))
async def cmd_start(message: types.Message):
    await message.answer("File: " + FILE)

@dp.message(Command("message"))
async def cmd_start(message: types.Message):
    await message.answer("Message: " + message.text)


@dp.message(F.document)
async def get_doc(message: types.Message, bot: Bot):
    global FILE, counter_files
    file = await bot.get_file(message.document.file_id)
    file_path = file.file_path
    
    if file_path.endswith(EXTENSION):
        destination = os.path.join(UPLOAD_FOLDER, f'file_{counter_files}.{EXTENSION}')
        await bot.download_file(file_path, destination)

        FILE = destination
        counter_files += 1
        await message.answer(f"Your {FILE} is success  downloaded!")
    else:
        await message.answer(f"Format is wrong! Download {EXTENSION}")


async def set_commands():
    commands = [BotCommand(command='start', description='start chating'),
                BotCommand(command='check_status', description='check status'),
                BotCommand(command='message', description='your message'),]
    
    await bot.set_my_commands(commands, BotCommandScopeDefault())


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

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

解决方案 #

要实现单消息内多文件上传，核心是处理消息中的documents列表（该列表包含消息内所有文件对象），对每个文件单独执行校验和下载逻辑。以下是修改后的完整代码：

import os
import asyncio
import logging

from aiogram.types import Message
from aiogram import Bot, Dispatcher, types, F
from aiogram.types import BotCommand, BotCommandScopeDefault
from aiogram.filters.command import Command

logging.basicConfig(level=logging.INFO)
bot = Bot(token="你的BotToken")
dp = Dispatcher()

EXTENSION = "你需要的后缀名"  # 例如"pdf"、"txt"
counter_files = 0

UPLOAD_FOLDER = "uploads/"
os.makedirs(UPLOAD_FOLDER, exist_ok=True)

@dp.message(Command("start"))
async def cmd_start(message: types.Message):
    await message.answer("你好！")

@dp.message(Command("check_status"))
async def cmd_check_status(message: types.Message):
    await message.answer(f"已保存文件总数：{counter_files}")

@dp.message(Command("message"))
async def cmd_echo_message(message: types.Message):
    await message.answer(f"你的消息内容：{message.text}")

@dp.message(F.documents)
async def handle_multiple_docs(message: types.Message, bot: Bot):
    global counter_files
    success_count = 0
    failed_files = []

    # 遍历消息内的所有文件
    for doc in message.documents:
        file = await bot.get_file(doc.file_id)
        file_path = file.file_path
        
        # 校验文件格式
        if file_path.endswith(EXTENSION):
            destination = os.path.join(UPLOAD_FOLDER, f'file_{counter_files}.{EXTENSION}')
            await bot.download_file(file_path, destination)
            counter_files += 1
            success_count += 1
        else:
            failed_files.append(doc.file_name)
    
    # 构建回复反馈
    reply_text = ""
    if success_count > 0:
        reply_text += f"成功保存 {success_count} 个文件！\n"
    if failed_files:
        reply_text += f"以下文件格式不符合要求（需{EXTENSION}格式）：\n" + "\n".join(failed_files)
    
    await message.answer(reply_text)

async def set_commands():
    commands = [
        BotCommand(command='start', description='启动对话'),
        BotCommand(command='check_status', description='查看已保存文件数量'),
        BotCommand(command='message', description='返回你的消息内容'),
    ]
    await bot.set_my_commands(commands, BotCommandScopeDefault())

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

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

关键修改说明 #

• 消息过滤器调整：将F.document改为F.documents，确保能捕获包含单个或多个文件的消息
• 多文件遍历处理：通过for doc in message.documents循环，逐个处理消息内的每个文件
• 反馈优化：增加成功/失败文件统计，给用户明确的上传结果反馈
• 状态命令优化：check_status改为显示已保存文件总数，更符合实际需求

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