批量下载:多线程 4 行核心代码

Bulk Download: Multi-threaded in 4 Lines

图片采集的核心是快。单线程一张张下,1 万张图要 5 小时 — 多线程能压到 20 分钟。

1. 单线程版(慢但简单)

import requests


from pathlib import Path





def download(url, save_path):


    resp = requests.get(url, timeout=10)


    save_path.write_bytes(resp.content)





urls = ['https://example.com/1.jpg', 'https://example.com/2.jpg']


for i, url in enumerate(urls):


    download(url, Path(f'downloads/{i}.jpg'))


    if i % 100 == 0: print(f'已下 {i} 张')

2. 多线程版(快 10 倍)

from concurrent.futures import ThreadPoolExecutor, as_completed





def download_one(args):


    url, save_path = args


    try:


        resp = requests.get(url, timeout=10)


        save_path.write_bytes(resp.content)


        return save_path, True


    except Exception as e:


        return save_path, str(e)





tasks = [(url, Path(f'downloads/{i:06d}.jpg')) for i, url in enumerate(urls)]





with ThreadPoolExecutor(max_workers=20) as executor:


    futures = [executor.submit(download_one, t) for t in tasks]


    for i, future in enumerate(as_completed(futures)):


        path, status = future.result()


        if i % 100 == 0:


            print(f'进度: {i}/{len(tasks)}')





print('完成')

3. 进度条 + 失败重试

from tqdm import tqdm


import time





def download_with_retry(url, save_path, max_retries=3):


    for attempt in range(max_retries):


        try:


            resp = requests.get(url, timeout=10)


            resp.raise_for_status()


            save_path.write_bytes(resp.content)


            return True


        except Exception as e:


            if attempt < max_retries - 1:


                time.sleep(2 ** attempt)


            else:


                return False


    return False





for url in tqdm(urls, desc='下载中'):


    filename = url.split('/')[-1]


    download_with_retry(url, Path(f'downloads/{filename}'))

多线程 + 失败重试,1 万张图 5 分钟搞定。下一节我们加上关键词自动分类。