使用Python实现带进度条的异步多文件下载器

pip install aiohttp tqdm

import aiohttp
import asyncio
from tqdm import tqdm

# 异步下载函数
async def download_file(url, destination):
    """
    异步下载文件并显示下载进度条

    :param url:          文件下载地址
    :param destination:  本地保存文件路径
    """
    # 创建 aiohttp 会话
    async with aiohttp.ClientSession() as session:
        # 发送 GET 请求
        async with session.get(url) as response:
            # 获取文件总大小（如果服务器支持）
            total_size = int(response.headers.get("Content-Length", 0))

            # 打开本地文件，准备写入下载内容
            with open(destination, "wb") as f:
                # tqdm 进度条
                progress_bar = tqdm(
                    total=total_size,         # 总大小
                    unit="B",                 # 单位：字节
                    unit_scale=True,          # 自动转换 KB / MB
                    desc=f"下载中: {destination}"  # 显示的标题
                )

                # 持续读取数据（每次读取 1024 字节）
                while True:
                    chunk = await response.content.read(1024)
                    if not chunk:   # 读取完毕
                        break
                    f.write(chunk)
                    progress_bar.update(len(chunk))  # 更新进度条

                progress_bar.close()  # 下载完成后关闭进度条


# 主异步函数
async def main():
    """
    批量创建多个下载任务，并异步并发执行
    """

    # 要下载的文件列表（可自行扩展）
    download_tasks = [
        {
            "url": "http://example.com/file1.zip",
            "destination": "file1.zip"
        },
        {
            "url": "http://example.com/file2.zip",
            "destination": "file2.zip"
        },
        # 可以继续添加更多文件
    ]

    # 为每个任务创建一个 download_file 协程
    tasks = [
        download_file(task["url"], task["destination"])
        for task in download_tasks
    ]

    # 并发执行所有下载任务
    await asyncio.gather(*tasks)


# 程序入口
if __name__ == "__main__":
    # 运行异步主函数
    asyncio.run(main())