如何基于Requests会话实现多分片并发大文件下载并显示进度条

针对你的需求，这里给出基于ThreadPoolExecutor的多分片并发下载实现，完美适配你的现有代码逻辑，保留requests.Session会话和enlighten进度条，同时保证下载高效无错误：

完整代码 #

import lxml.html as htmlparser
import regex
import math
import enlighten
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed

def download_chunk(session, url, start, end, file_obj, counter):
    """下载单个分片并写入文件对应位置"""
    headers = {"Range": f"bytes={start}-{end}"}
    # 复制原会话的cookies到当前会话，保持登录状态
    session.cookies.update(original_session.cookies)
    response = session.get(url, stream=True, headers=headers)
    response.raise_for_status()
    
    # 写入分片到文件指定位置
    file_obj.seek(start)
    for chunk in response.iter_content(chunk_size=2**20):
        if chunk:
            file_obj.write(chunk)
            counter.update(len(chunk) // 2**20)  # 按MiB更新进度

if __name__ == "__main__":
    # 初始化原会话，获取必要的认证信息和下载链接
    original_session = requests.Session()
    url = 'https://serverlink.com/vid/XXXYYY'
    URL = "https://serverhomepage.com/"

    CONTENT_ID = regex.compile(r"/vid/([^?#&/]+)")
    
    # 获取csrf-token
    csrf_token = (
        htmlparser.fromstring(original_session.get(url).text)
        .cssselect("meta[name='csrf-token']")[0]
        .get("content")
    )

    # 获取内容ID和下载流链接
    content_id = CONTENT_ID.search(url).group(1)
    stream_response = original_session.post(
        URL + "api/vid/get&user=xxx&pass=yyy",
        headers={
            "x-requested-with": "XMLHttpRequest",
            "x-csrf-token": csrf_token,
        },
        json={"id": content_id},
    )
    stream_response.raise_for_status()
    stream_url = stream_response.json().get("result").get("Link")

    # 获取最终下载链接和文件总大小
    head_response = original_session.head(stream_url)
    head_response.raise_for_status()
    final_url = head_response.headers.get("Location")
    
    # 获取文件总大小
    size_response = original_session.head(final_url)
    size_response.raise_for_status()
    total_size = int(size_response.headers.get("Content-Length", 0))
    assert total_size > 0, "无法获取文件大小"

    fname = 'vid.mp4'
    num_chunks = 10  # 分片数量，可根据网络情况调整（建议5-20）
    chunk_size = total_size // num_chunks

    # 初始化进度条
    MANAGER = enlighten.get_manager()
    total_mib = math.ceil(total_size / 2**20)
    counter = MANAGER.counter(
        color='green',
        total=total_mib,
        unit='MiB',
        leave=False,
        desc="下载进度"
    )

    # 打开文件，准备写入（先创建空文件并设置大小）
    with open(fname, 'wb') as f:
        # 预分配文件空间（可选，提升写入效率）
        f.truncate(total_size)
        
        # 使用线程池下载分片
        with ThreadPoolExecutor(max_workers=num_chunks) as executor:
            futures = []
            for i in range(num_chunks):
                start = i * chunk_size
                # 最后一个分片覆盖到文件末尾
                end = start + chunk_size - 1 if i != num_chunks - 1 else total_size - 1
                # 每个线程创建独立的Session，避免线程安全问题
                session = requests.Session()
                futures.append(executor.submit(download_chunk, session, final_url, start, end, f, counter))
            
            # 等待所有任务完成，处理异常
            for future in as_completed(futures):
                try:
                    future.result()
                except Exception as e:
                    print(f"分片下载失败: {str(e)}")
                    raise  # 抛出异常终止整个下载

    print(f"文件 {fname} 下载完成，总大小: {total_mib} MiB")
    MANAGER.stop()

核心要点说明 #

• 会话安全：requests.Session不是线程安全的，因此每个下载线程创建独立的Session，并复制原Session的cookies，确保认证状态一致。
• 分片下载逻辑：通过HTTP的Range请求头实现分块下载，服务器需支持断点续传（大部分主流服务器都支持）。
• 文件写入：用seek()定位到分片起始位置写入，避免多线程写入冲突，保证文件完整性。
• 进度条同步：enlighten的counter是线程安全的，每个分片下载时按MiB单位更新进度，总进度实时准确。
• 异常处理：捕获每个分片的下载异常，一旦出错立即终止整个下载，避免生成损坏文件。
• 性能优化：预分配文件空间减少磁盘碎片，chunk_size设为1MiB平衡网络IO和磁盘写入效率。

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