import aiohttp import aiofiles import asyncio from tqdm import tqdm import os # 模拟获取视频URL列表的函数 def get_video_urls(): return [ 'https://example.com/video1.mp4', 'https://example.com/video2.mp4', # 添加更多URL ] # 异步下载单个视频文件 async def download_video(session, url, save_path): try: async with session.get(url) as response: response.raise_for_status() async with aiofiles.open(save_path, 'wb') as f: async for chunk in response.content.iter_chunked(1024): await f.write(chunk) print(f"Downloaded: {url}") except Exception as e: print(f"Failed to download {url}: {e}") # 主下载函数 async def main(): video_urls = get_video_urls() save_dir = 'videos' os.makedirs(save_dir, exist_ok=True) async with aiohttp.ClientSession() as session: tasks = [] for url in video_urls: file_name = os.path.join(save_dir, os.path.basename(url)) tasks.append(download_video(session, url, file_name)) for task in tqdm(asyncio.as_completed(tasks), total=len(tasks)): await task # 运行主函数 if __name__ == '__main__': asyncio.run(main())