如何向量化非数学类Python数据采集嵌套循环脚本？

首先明确：你的音频采集脚本属于IO密集型任务，NumPy向量化完全不适用。NumPy向量化是用来优化CPU密集的数值计算（比如矩阵运算、统计求和），而你的脚本瓶颈在于网络请求和文件写入，这类操作的延迟远大于CPU计算，提速的核心是并行/异步处理，而非向量化。

下面是针对你脚本的具体优化方案：

核心优化思路：并发处理IO任务 #

你的三层循环本质是重复执行“下载文件→保存文件”的操作，这些操作可以同时进行，不用等待前一个完成再执行下一个。以下是两种高效的实现方式：

1. 多线程实现（简单易上手） #

用concurrent.futures.ThreadPoolExecutor批量处理下载任务，适合快速改造现有脚本：

import os
import requests
from concurrent.futures import ThreadPoolExecutor

# 假设你已定义好chapter_range和bitrates_for_reciters
chapter_range = {"1": [1, 7], "2": [1, 286], ...}
bitrates_for_reciters = {"ar.reciter1": [128, 64], "ar.reciter2": [64, 32], ...}

def generate_all_tasks():
    """预生成所有需要下载的任务（链接+保存路径）"""
    tasks = []
    for chapter in range(1, 115):
        verse_start, verse_end = chapter_range[str(chapter)]
        for verse in range(verse_start, verse_end + 1):
            dir_path = f"data/{chapter}_{verse}"
            # 提前创建所有目录，避免循环内重复判断
            os.makedirs(dir_path, exist_ok=True)
            for reciter in bitrates_for_reciters:
                min_bitrate = min(bitrates_for_reciters[reciter])
                cdn_url = f".../{min_bitrate}/{reciter}/{verse}.mp3"
                save_path = f"{dir_path}/{chapter}_{verse}_{reciter[3:]}.mp3"
                tasks.append((cdn_url, save_path))
    return tasks

def download_single_file(cdn_url, save_path):
    """单个文件的下载与保存逻辑，带错误处理"""
    if os.path.exists(save_path):
        print(f"跳过已存在文件：{save_path}")
        return
    try:
        resp = requests.get(cdn_url, timeout=15)
        resp.raise_for_status()  # 捕获HTTP错误
        with open(save_path, "wb") as f:
            f.write(resp.content)
        print(f"完成：{save_path}")
    except Exception as e:
        print(f"失败 {cdn_url}：{str(e)}")

if __name__ == "__main__":
    all_tasks = generate_all_tasks()
    # 线程数根据CDN限制调整，建议30-50（避免被封禁）
    with ThreadPoolExecutor(max_workers=40) as executor:
        executor.map(download_single_file, *zip(*all_tasks))

2. 异步IO实现（效率更高） #

用aiohttp替代同步的requests，配合asyncio实现异步请求，比多线程开销更低，适合大量IO任务：

import os
import asyncio
import aiohttp

chapter_range = {"1": [1, 7], "2": [1, 286], ...}
bitrates_for_reciters = {"ar.reciter1": [128, 64], "ar.reciter2": [64, 32], ...}

async def download_single_file(session, cdn_url, save_path):
    if os.path.exists(save_path):
        print(f"跳过已存在文件：{save_path}")
        return
    try:
        async with session.get(cdn_url, timeout=15) as resp:
            resp.raise_for_status()
            content = await resp.read()
            with open(save_path, "wb") as f:
                f.write(content)
        print(f"完成：{save_path}")
    except Exception as e:
        print(f"失败 {cdn_url}：{str(e)}")

async def main():
    tasks = []
    # 提前创建所有目录
    for chapter in range(1, 115):
        verse_start, verse_end = chapter_range[str(chapter)]
        for verse in range(verse_start, verse_end + 1):
            dir_path = f"data/{chapter}_{verse}"
            os.makedirs(dir_path, exist_ok=True)
            for reciter in bitrates_for_reciters:
                min_bitrate = min(bitrates_for_reciters[reciter])
                cdn_url = f".../{min_bitrate}/{reciter}/{verse}.mp3"
                save_path = f"{dir_path}/{chapter}_{verse}_{reciter[3:]}.mp3"
                tasks.append((cdn_url, save_path))
    
    # 创建异步会话，批量执行任务
    async with aiohttp.ClientSession() as session:
        await asyncio.gather(*[download_single_file(session, url, path) for url, path in tasks])

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

额外优化建议 #

• 控制并发数：不要设置过大的线程/异步任务数，避免触发CDN的反爬机制导致IP被封禁
• 增加重试机制：对失败的请求自动重试（可以用tenacity库实现）
• 断点续传：如果文件很大，可以支持断点续传，避免网络中断后重新下载整个文件
• 日志记录：把失败的请求记录到文件，方便后续手动处理

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