Python怎样实现文件断点续传？HTTP请求控制

Range

Range

Range: bytes=1024-

206 Partial Content

Content-Range

200 OK

Range

HEAD

HEAD

GET

HEAD

Accept-Ranges

Range

Range: bytes=0-0

206 Partial Content

200 OK

Range

requests

import requests
import os
import time

def download_file_resumable(url, local_filename, chunk_size=8192):
    """
    实现文件的断点续传下载
    """
    if os.path.exists(local_filename):
        # 检查本地文件大小，作为续传的起点
        downloaded_size = os.path.getsize(local_filename)
        mode = 'ab'  # 追加模式
        headers = {'Range': f'bytes={downloaded_size}-'}
        print(f"检测到本地文件 {local_filename}，大小 {downloaded_size} 字节，尝试从此处续传。")
    else:
        downloaded_size = 0
        mode = 'wb'  # 写入模式
        headers = {}
        print(f"本地无文件 {local_filename}，将从头开始下载。")

    try:
        # 第一次请求，检查服务器是否支持Range
        # 实际下载时，可以先发HEAD请求判断，这里直接用GET并处理200/206
        with requests.get(url, headers=headers, stream=True, timeout=30) as r:
            # 检查服务器是否支持断点续传，以及响应状态码
            if r.status_code == 206:
                print("服务器支持断点续传 (206 Partial Content)。")
                # 确保Content-Range头存在且格式正确
                content_range = r.headers.get('Content-Range')
                if content_range:
                    # Content-Range: bytes 0-1023/10240
                    total_size_str = content_range.split('/')[-1]
                    try:
                        total_size = int(total_size_str)
                    except ValueError:
                        print("警告: 无法解析Content-Range中的总大小。")
                        total_size = None
                else:
                    print("警告: 206响应中缺少Content-Range头部。")
                    total_size = None # 无法确定总大小，可能需要重新下载

                # 如果本地文件大小与服务器响应的起始字节不符，可能需要重新下载
                if total_size is not None and downloaded_size > 0 and downloaded_size != int(content_range.split(' ')[1].split('-')[0]):
                    print("警告: 本地文件大小与服务器续传起始点不符，可能需要重新下载。")
                    # 这里可以加入逻辑，比如删除本地文件，重新下载
                    # os.remove(local_filename)
                    # return download_file_resumable(url, local_filename) # 重新调用
                    # 为了简化，这里选择继续，但用户需注意
                    pass

            elif r.status_code == 200:
                print("服务器不支持断点续传或忽略了Range请求 (200 OK)，将重新下载整个文件。")
                # 如果是200，说明服务器发了整个文件，需要从头开始写入
                mode = 'wb'
                downloaded_size = 0
                total_size = int(r.headers.get('Content-Length', 0))
            else:
                print(f"下载失败: 服务器返回状态码 {r.status_code}")
                return False

            # 获取文件总大小，用于进度显示
            if total_size is None:
                # 尝试从Content-Length获取，如果206没有Content-Range
                total_size = int(r.headers.get('Content-Length', 0)) + downloaded_size # 对于206，Content-Length是剩余部分的大小

            # 写入文件
            with open(local_filename, mode) as f:
                for chunk in r.iter_content(chunk_size=chunk_size):
                    if chunk: # 过滤掉保持连接的空数据块
                        f.write(chunk)
                        downloaded_size += len(chunk)
                        # 简单的进度显示
                        if total_size:
                            progress = (downloaded_size / total_size) * 100
                            print(f"\r下载进度: {progress:.2f}% ({downloaded_size}/{total_size} 字节)", end='')
                        else:
                            print(f"\r已下载: {downloaded_size} 字节", end='')
            print(f"\n文件下载完成: {local_filename}")
            return True

    except requests.exceptions.RequestException as e:
        print(f"网络请求错误: {e}")
        return False
    except Exception as e:
        print(f"发生未知错误: {e}")
        return False

# 示例使用
if __name__ == "__main__":
    # 找一个支持断点续传的URL，比如一些大型文件下载地址
    # 注意：请替换为实际可用的URL
    test_url = "http://speedtest.tele2.net/1MB.zip" # 示例，可能会失效
    # test_url = "https://example.com/some_large_file.zip" # 替换为你的文件URL
    output_file = "downloaded_file.zip"

    print(f"尝试下载: {test_url} 到 {output_file}")
    success = download_file_resumable(test_url, output_file)
    if success:
        print("下载任务成功完成。")
    else:
        print("下载任务失败。")

    # 模拟中断后再次运行，看是否能续传
    # 可以手动中断程序后再次运行，或者在函数内部模拟中断（比如下载一部分后raise Exception）
    # print("\n模拟中断后再次尝试下载...")
    # time.sleep(2) # 模拟中断
    # success_resume = download_file_resumable(test_url, output_file)
    # if success_resume:
    #     print("续传任务成功完成。")
    # else:
    #     print("续传任务失败。")

stream=True

requests

iter_content

Content-Range

206

Content-Length