pip install aiohttp tqdm

import aiohttp
import asyncio
from tqdm import tqdm

# 異步下載函數(shù)
async def download_file(url, destination):
    """
    異步下載文件并顯示下載進度條

    :param url:          文件下載地址
    :param destination:  本地保存文件路徑
    """
    # 創(chuàng)建 aiohttp 會話
    async with aiohttp.ClientSession() as session:
        # 發(fā)送 GET 請求
        async with session.get(url) as response:
            # 獲取文件總大小（如果服務器支持）
            total_size = int(response.headers.get("Content-Length", 0))

            # 打開本地文件，準備寫入下載內(nèi)容
            with open(destination, "wb") as f:
                # tqdm 進度條
                progress_bar = tqdm(
                    total=total_size,         # 總大小
                    unit="B",                 # 單位：字節(jié)
                    unit_scale=True,          # 自動轉(zhuǎn)換 KB / MB
                    desc=f"下載中: {destination}"  # 顯示的標題
                )

                # 持續(xù)讀取數(shù)據(jù)（每次讀取 1024 字節(jié)）
                while True:
                    chunk = await response.content.read(1024)
                    if not chunk:   # 讀取完畢
                        break
                    f.write(chunk)
                    progress_bar.update(len(chunk))  # 更新進度條

                progress_bar.close()  # 下載完成后關(guān)閉進度條


# 主異步函數(shù)
async def main():
    """
    批量創(chuàng)建多個下載任務，并異步并發(fā)執(zhí)行
    """

    # 要下載的文件列表（可自行擴展）
    download_tasks = [
        {
            "url": "http://example.com/file1.zip",
            "destination": "file1.zip"
        },
        {
            "url": "http://example.com/file2.zip",
            "destination": "file2.zip"
        },
        # 可以繼續(xù)添加更多文件
    ]

    # 為每個任務創(chuàng)建一個 download_file 協(xié)程
    tasks = [
        download_file(task["url"], task["destination"])
        for task in download_tasks
    ]

    # 并發(fā)執(zhí)行所有下載任務
    await asyncio.gather(*tasks)


# 程序入口
if __name__ == "__main__":
    # 運行異步主函數(shù)
    asyncio.run(main())