Python批量下载案例如何实现？

wen python案例 2026-06-07 04:03:31 1

Python批量下载案例如何实现？详解高效文件抓取技巧

目录导读

为什么需要Python批量下载？——应用场景与价值
核心原理：HTTP请求与文件流处理机制
实战案例一：批量下载图片（基于requests+多线程）
实战案例二：批量下载PDF文档（处理反爬与动态链接）
进阶优化：断点续传、限速与异常重试
常见问题FAQ（Q&A）
构建稳健的批量下载系统

为什么需要Python批量下载？——应用场景与价值

在日常数据处理、学术研究、数据分析或自动化办公中，我们经常需要从网络上下载大量文件（如:图片、PDF、CSV、压缩包等），手动点击下载不仅效率低下，还容易因重复操作出错，Python凭借其丰富的网络库和并发支持，成为批量下载任务的首选语言。

核心应用场景：

爬取公开数据集（如:气象数据、股票历史数据）
批量保存网页素材（图片、音频、视频）
自动化论文/文档备份
从API接口批量拉取文件

关键优势：

可编程控制：自定义下载策略
并发加速：比单线程快数倍
错误处理：自动重试、记录失败链接

核心原理：HTTP请求与文件流处理

批量下载的本质是：遍历URL列表 → 发送GET请求 → 接收响应流 → 写入本地文件。

关键技术点：

请求库：requests（易用）、urllib（内置）、aiohttp（异步）
文件写入：with open()结合response.iter_content()分块写入，避免内存溢出
并发控制：ThreadPoolExecutor（多线程）、asyncio（协程）
Headers伪装：User-Agent随机化，模拟真实浏览器
超时设置：timeout参数防止卡死

示例代码框架：

import requests
def download_file(url, save_path):
    headers = {'User-Agent': 'Mozilla/5.0'}
    with requests.get(url, headers=headers, stream=True, timeout=10) as r:
        r.raise_for_status()
        with open(save_path, 'wb') as f:
            for chunk in r.iter_content(chunk_size=8192):
                f.write(chunk)

实战案例一：批量下载图片（基于requests+多线程）

场景描述：

需要从一个图片分享网站,下载某个专辑下所有高清图片（约500张），URL格式规范，均为序号递增。

实现步骤：

生成URL列表：通过循环拼接 https://example.com/img/{i}.jpg
编写下载函数：加入随机延迟，避免触发频率限制
多线程加速：使用 concurrent.futures.ThreadPoolExecutor

关键代码：

import os
import time
import random
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
# 生成URL列表
urls = [f"https://example.com/img/{i}.jpg" for i in range(1, 501)]
# 下载函数
def download_img(url, save_dir="images"):
    try:
        filename = url.split("/")[-1]
        path = os.path.join(save_dir, filename)
        if os.path.exists(path):
            return f"已存在: {filename}"
        headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}
        resp = requests.get(url, headers=headers, timeout=10, stream=True)
        resp.raise_for_status()
        with open(path, "wb") as f:
            for chunk in resp.iter_content(1024):
                f.write(chunk)
        time.sleep(random.uniform(0.5, 1.5))  # 随机延迟
        return f"下载成功: {filename}"
    except Exception as e:
        return f"下载失败: {url} - {str(e)}"
# 多线程执行
os.makedirs("images", exist_ok=True)
with ThreadPoolExecutor(max_workers=10) as executor:
    futures = {executor.submit(download_img, url): url for url in urls}
    for future in as_completed(futures):
        print(future.result())

运行效果：

10线程并发,500张图片约2分钟下载完成（取决于网速）
自动跳过已存在的文件,避免重复下载

实战案例二：批量下载PDF文档（处理反爬与动态链接）

场景描述：

某学术网站提供论文下载,但链接需要携带session token，且页面元素通过JavaScript动态加载。

解决思路：

使用requests.Session保持会话：维持cookies
解析真实下载链接：分析网络请求或使用正则提取
添加Referer头：模拟从站内页面跳转

核心代码片段：

session = requests.Session()
# 先登录或获取cookies
login_data = {"username": "user", "password": "pass"}
session.post("https://example.com/login", data=login_data)
# 遍历论文详情页，提取PDF链接
def get_pdf_url(detail_url):
    resp = session.get(detail_url, headers={"Referer": "https://example.com/list"})
    # 假设PDF链接在HTML中为 <a href="/download?file=xxx.pdf">
    import re
    match = re.search(r'/download\?file=.*?\.pdf', resp.text)
    if match:
        return "https://example.com" + match.group()
    return None
# 下载PDF
def download_pdf(pdf_url, save_name):
    headers = {"User-Agent": "Mozilla/5.0", "Referer": "https://example.com/"}
    resp = session.get(pdf_url, headers=headers, stream=True)
    with open(f"{save_name}.pdf", "wb") as f:
        for chunk in resp.iter_content(8192):
            f.write(chunk)

反爬应对技巧：

随机切换User-Agent库（fake_useragent）
添加请求间隔（1-3秒）
使用IP代理池应对封禁

进阶优化：断点续传、限速与异常重试

断点续传

通过Range头实现，避免下载中断后从头开始：

headers = {"Range": f"bytes={existing_size}-"}
resp = requests.get(url, headers=headers, stream=True)
# 追加写入文件
with open(file_path, "ab") as f:
    for chunk in resp.iter_content(1024):
        f.write(chunk)

下载限速

控制每秒钟写入的数据量：

import time
chunk_size = 8192
max_speed = 1024 * 1024  # 1MB/s
for chunk in resp.iter_content(chunk_size):
    f.write(chunk)
    time.sleep(chunk_size / max_speed)

异常重试

使用retry库或自定义重试装饰器：

from tenacity import retry, stop_after_attempt, wait_fixed
@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def robust_download(url, path):
    # 下载逻辑
    pass

常见问题FAQ（Q&A）

Q1: 下载大文件（>1GB）时内存飙升怎么办？

A: 必须使用stream=True，并设置chunk_size（通常8192字节），以流式写入磁盘，避免一次性加载到内存。

Q2: 遇到SSL证书验证错误怎么处理？

A: 临时关闭验证（不推荐生产环境）：

requests.get(url, verify=False)  # 需要加警告忽略

或使用certifi库更新证书。

Q3: 如何下载需要登录才能访问的文件？

A: 使用requests.Session模拟登录过程，保存cookies，并手动添加必要的Headers（如Bearer Token）。

Q4: 批量下载时如何避免被封IP？

降低并发数（建议5-10线程）
增加随机延时（1-5秒）
使用代理IP池（如:免费免费代理网站）
模仿浏览器行为（完整Headers、Referer）

Q5: 多线程下载文件乱序怎么办？

A: 下载完成后用os.rename()原子操作或先使用临时文件名（如: temp_1.pdf），下载完毕再重命名为正确序号。

构建稳健的批量下载系统

Python批量下载的实现核心在于：

结构化URL来源：可以是列表推导式、文件读取或爬虫解析
灵活的错误处理：try-except + 日志记录失败链接
适度的并发：根据目标服务器承受能力调整线程数（通常5-15）
伪装与反反爬：通过User-Agent、Referer、延时模拟真人操作

最终代码模板（可复用）：

下载函数（支持断点续传、限速、重试）
URL生成器（支持从文件/爬虫获取）
主函数（日志记录、并发控制、失败重试）

建议：

运行前先测试3-5个链接，确保格式正确
使用tqdm库添加进度条，提升用户体验
遵守网站robots.txt协议，避免法律风险

通过以上案例和技巧,你已经掌握了从基础到进阶的Python批量下载实现方法，无论是日常工作还是数据处理项目，这套方法论都能帮助你将重复性劳动自动化，大幅提升效率。

标签： Python案例

本文地址： https://www.dfhcn.com/post/1299.html

文章来源： wen