如何使用 Python 稳健高效地下载大量大文件（含重试与流式处理）-Python教程-PHP中文网

如何使用 Python 稳健高效地下载大量大文件（含重试与流式处理）

本文介绍如何用 python 可靠下载数千个、单个最大达 1gb 的异构文件（pdf/ppt/zip/iso 等），通过流式响应 + 自动重试 + 连接复用机制，彻底解决“只下 26kb”“连接被远程强制关闭（errno 10054）”等常见失败问题。

本文介绍如何用 python 可靠下载数千个、单个最大达 1gb 的异构文件（pdf/ppt/zip/iso 等），通过流式响应 + 自动重试 + 连接复用机制，彻底解决“只下 26kb”“连接被远程强制关闭（errno 10054）”等常见失败问题。

在批量下载大规模文件（如 16,000+ 个、单文件最大 1GB）的生产场景中，直接调用 requests.get(url).content 极易失败：不仅会因内存溢出导致小文件截断（例如仅写入 26KB），还常因网络抖动、服务端主动断连（[Errno 10054]）、超时或 HTTP 5xx 错误而中断任务。根本原因在于未启用流式传输（stream=True）、缺乏连接复用、缺少错误恢复能力。

✅ 正确做法是组合三项关键技术：

流式下载（Streaming）：避免将整个文件加载进内存，逐块写入磁盘；
会话复用与连接池（Session + Adapter）：复用 TCP 连接，降低握手开销，提升并发稳定性；
智能重试策略（Retry with backoff）：对临时性错误（如 502/503/超时）自动重试，指数退避避免雪崩。

以下为完整可运行的工业级下载方案：

python-flask-vue-creator

创建 Flask + Vue.js 全栈项目，采用单端口架构、uv 依赖管理、结构化日志和生产级前端设计。当用户要求以下操作时触发：创建新的 Flask+Vue 项目、搭建 Web 应用、构建全栈 Python+Vue 应用、初始化 Flask Web 项目、设置单端口 Flask 前后端项目。触发词：新建项目、创建项目、搭建项目、Flask项目、Vue项目。

下载

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import os
from openpyxl import load_workbook

def create_session_with_retries(
    max_retries: int = 5,
    backoff_factor: float = 0.5,
    pool_connections: int = 20,
    pool_maxsize: int = 20
) -> requests.Session:
    """创建带重试、连接池和超时配置的 requests Session"""
    session = requests.Session()
    # 配置重试策略：对 5xx 和连接类错误重试
    retries = Retry(
        total=max_retries,
        backoff_factor=backoff_factor,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods={"GET", "HEAD", "POST"},
        raise_on_status=False  # 防止 raise_for_status 提前中断重试
    )
    adapter = HTTPAdapter(
        max_retries=retries,
        pool_connections=pool_connections,
        pool_maxsize=pool_maxsize
    )
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session

def download_file(
    session: requests.Session,
    url: str,
    local_path: str,
    timeout: tuple = (10, 60),  # (connect, read)
    chunk_size: int = 8192  # 8KB 每次读取，平衡 I/O 与内存
) -> bool:
    """安全下载单个文件，支持断点续传语义（需服务端支持 Range）"""
    try:
        # 确保目录存在
        os.makedirs(os.path.dirname(local_path), exist_ok=True)

        with session.get(url, stream=True, timeout=timeout) as response:
            response.raise_for_status()  # 触发 4xx/5xx 异常

            # 写入文件（流式）
            with open(local_path, "wb") as f:
                for chunk in response.iter_content(chunk_size=chunk_size):
                    if chunk:  # 过滤 keep-alive 空块
                        f.write(chunk)

        print(f"✓ Success: {os.path.basename(local_path)} ({response.headers.get('Content-Length', 'unknown')} bytes)")
        return True

    except requests.exceptions.RequestException as e:
        print(f"✗ Failed {os.path.basename(local_path)}: {type(e).__name__} - {e}")
        return False
    except OSError as e:
        print(f"✗ IO Error for {local_path}: {e}")
        return False

# 主流程：从 Excel 批量读取 URL 并下载
if __name__ == "__main__":
    # ✅ 配置项（请按需修改）
    EXCEL_PATH = "urls.xlsx"         # 含 URL 列的 Excel 文件
    URL_COLUMN_INDEX = 6             # Excel 中 URL 所在列（0-based，即第7列）
    SAVE_ROOT = "./downloads"        # 本地保存根目录
    LOGIN_URL = "https://example.com/login"
    LOGIN_DATA = {"username": "user", "password": "pass"}

    # 创建带重试的会话
    s = create_session_with_retries()

    # 可选：登录认证（若需 Cookie 或 Token）
    try:
        login_resp = s.post(LOGIN_URL, data=LOGIN_DATA, timeout=(10, 30))
        login_resp.raise_for_status()
        print("Login successful.")
    except Exception as e:
        print(f"Login failed: {e}")
        exit(1)

    # 读取 Excel 并逐行下载
    wb = load_workbook(EXCEL_PATH)
    sheet = wb.active

    success_count = 0
    total_count = 0

    for row in sheet.iter_rows(min_row=2):  # 跳过表头
        total_count += 1
        url = row[URL_COLUMN_INDEX].value
        if not isinstance(url, str) or not url.strip():
            continue

        # 生成唯一本地路径（建议基于 URL 或 Excel 行号）
        filename = os.path.basename(url) or f"file_{total_count}"
        # 安全清理文件名（防止路径遍历）
        safe_filename = "".join(c for c in filename if c.isalnum() or c in "._- ")
        local_path = os.path.join(SAVE_ROOT, safe_filename)

        # 下载
        if download_file(s, url, local_path):
            success_count += 1

    print(f"\n? Summary: {success_count}/{total_count} files downloaded successfully.")

? 关键注意事项：

不要省略 stream=True：否则 response.content 会强制加载全部响应体到内存，1GB 文件极易触发 MemoryError 或 OOM Kill；
务必设置 timeout：timeout=(connect, read) 双参数避免卡死（如服务端迟迟不发响应头）；
chunk_size=8192 是经验最优值：太小（如 1）增加系统调用开销；太大（如 1MB）可能加剧内存压力且无实质收益；
重试 status_forcelist 包含 429：应对限流场景；allowed_methods 显式声明支持方法，避免 urllib3 未来版本警告；
文件名安全处理：Excel 中 URL 可能含非法字符或路径遍历片段（如 ../etc/passwd），务必清洗；
监控与日志：生产环境建议集成 logging 模块替代 print，并记录失败 URL 到 CSV 便于重试；
进阶优化：如需更高吞吐，可结合 concurrent.futures.ThreadPoolExecutor 实现多线程下载（注意服务端是否允许并发请求）。