Python 工具类实现大文件断点续传功能详解

简介

在文件上传或下载过程中，如果遇到大文件，如果发生传输失败，需要重新传输的情况会非常不方便，甚至无法实现。为了解决这种问题，我们可以实现大文件断点续传功能，使文件上传或下载可以随时中断和恢复。

实现流程

大文件断点续传的实现流程如下：

1. 设置支持 range 的Range头，支持服务器返回指定范围的数据。

2. 获取已下载或已上传的文件大小，作为Range头的值发送给服务器。

3. 将请求的数据写入到文件中，同时记录已下载或已上传的文件大小，作为下次请求的Range头的值。

4. 如果请求发生中断，下次请求的Range头的值应为已下载或已上传的文件大小。

5. 下一次请求时，从已下载或已上传的文件大小处开始请求即可。

示例说明

下面分两个示例说明大文件断点续传功能的实现。

示例1：大文件上传

在文件上传过程中，如果上传过程中出现中断，可以实现大文件上传的断点续传功能。

import requests

class FileUploader():
    def __init__(self, url):
        self.url = url
        self.chunk_size = 1024*1024*10

    def upload(self, file_path):
        with open(file_path, 'rb') as f:
            size = os.path.getsize(file_path)
            start = 0
            # 读取上次上传的位置
            if os.path.exists(f'{file_path}.cur'):
                with open(f'{file_path}.cur', 'r') as cf:
                    start = int(cf.read())
            f.seek(start)
            while start < size:
                end = min(start + self.chunk_size, size)
                data = f.read(self.chunk_size)
                headers = {'Content-Range': f'bytes {start}-{end-1}/{size}'}
                response = requests.put(self.url, headers=headers, data=data)
                if response.status_code == 200 or response.status_code == 201:
                    # 保存上传位置
                    with open(f'{file_path}.cur', 'w') as cf:
                        cf.write(str(end))
                    start = end
                else:
                    print(f'Error! status_code: {response.status_code}')

if __name__ == '__main__':
    uploader = FileUploader('http://xxx.com/file')
    uploader.upload('bigfile.zip')

上述代码中，我们定义了一个 FileUploader 类，其中 chunk_size 表示单次上传的大小，文件上传的范围使用 Content-Range 头指定，上传前先读取上次上传的位置，然后将已上传的位置写入到文件中。

示例2：大文件下载

在文件下载过程中，如果下载过程中出现中断，可以实现大文件下载的断点续传功能。

import requests
import os

class FileDownloader():
    def __init__(self, url, file_path):
        self.url = url
        self.file_path = file_path
        self.chunk_size = 1024*1024*10

    def download(self):
        headers = {}
        if os.path.exists(self.file_path):
            start = os.path.getsize(self.file_path)
            headers = {'Range': f'bytes={start}-'}
            print(f'Resumable mode: start download from {start}')
        else:
            start = 0
            print('Normal mode: start download from beginning')
        response = requests.get(self.url, headers=headers, stream=True)
        with open(self.file_path, 'ab') as f:
            total_size = int(response.headers.get('content-length', start))
            for chunk in response.iter_content(chunk_size=self.chunk_size):
                if chunk:
                    f.write(chunk)
                    f.flush()
            print('Download success!')

if __name__ == '__main__':
    downloader = FileDownloader('http://xxx.com/bigfile.zip', 'bigfile.zip')
    downloader.download()

上述代码中，我们定义了一个 FileDownloader 类，其中文件下载的范围使用 Range 头指定，下载前先读取已下载的文件大小，然后使用追加模式打开文件，将读取到的数据写入到文件中。

完整代码

将以上两个示例整合起来，我们可以实现大文件的断点续传功能。

import requests
import os

class FileUploader():
    def __init__(self, url):
        self.url = url
        self.chunk_size = 1024*1024*10

    def upload(self, file_path):
        with open(file_path, 'rb') as f:
            size = os.path.getsize(file_path)
            start = 0
            # 读取上次上传的位置
            if os.path.exists(f'{file_path}.cur'):
                with open(f'{file_path}.cur', 'r') as cf:
                    start = int(cf.read())
            f.seek(start)
            while start < size:
                end = min(start + self.chunk_size, size)
                data = f.read(self.chunk_size)
                headers = {'Content-Range': f'bytes {start}-{end-1}/{size}'}
                response = requests.put(self.url, headers=headers, data=data)
                if response.status_code == 200 or response.status_code == 201:
                    # 保存上传位置
                    with open(f'{file_path}.cur', 'w') as cf:
                        cf.write(str(end))
                    start = end
                else:
                    print(f'Error! status_code: {response.status_code}')

class FileDownloader():
    def __init__(self, url, file_path):
        self.url = url
        self.file_path = file_path
        self.chunk_size = 1024*1024*10

    def download(self):
        headers = {}
        if os.path.exists(self.file_path):
            start = os.path.getsize(self.file_path)
            headers = {'Range': f'bytes={start}-'}
            print(f'Resumable mode: start download from {start}')
        else:
            start = 0
            print('Normal mode: start download from beginning')
        response = requests.get(self.url, headers=headers, stream=True)
        with open(self.file_path, 'ab') as f:
            total_size = int(response.headers.get('content-length', start))
            for chunk in response.iter_content(chunk_size=self.chunk_size):
                if chunk:
                    f.write(chunk)
                    f.flush()
            print('Download success!')

if __name__ == '__main__':
    uploader = FileUploader('http://xxx.com/file')
    uploader.upload('bigfile.zip')

    downloader = FileDownloader('http://xxx.com/bigfile.zip', 'bigfile.zip')
    downloader.download()

结论

大文件断点续传的实现思路简单，但需要注意多线程、分块的上传下载原理以及网络请求出现错误的情况，这些都需要人为处理并产生合理的错误反馈。通过使用断点续传技术，我们可以优化文件上传和下载，增强用户体验，提升系统稳定性。