Python 大文件分片上传的处理方法_石家庄创泽智能科技有限公司

新闻中心 NEWS CENTER

您当前位置：首页 > 新闻中心 > 网络学院

Python 大文件分片上传的处理方法

2025-11-23

浏览次数：次

返回列表

大文件分片上传通过切块避免内存溢出，客户端逐片上传并携带元数据，服务端按序合并；核心是生成器读取、分片传输与完整校验，支持断点续传和进度跟踪，确保GB级文件稳定上传。

python 大文件分片上传的处理方法

大文件上传在Web开发中很常见，直接上传可能因内存占用高或网络中断导致失败。Python中实现大文件分片上传，核心思路是将文件切块、逐个上传、服务端合并。以下是具体处理方法。

1. 文件分片读取

避免一次性加载整个文件到内存，使用生成器按固定大小读取文件块。

设定分片大小（如5MB）
通过循环读取文件流，每次返回一部分数据
适合任意大小的文件，内存占用稳定

示例代码：

def read_file_chunks(file_path, chunk_size=5 * 1024 * 1024):
    with open(file_path, 'rb') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk

2. 客户端上传分片

每上传一个分片，携带必要信息：文件名、分片序号、总分片数等，便于服务端识别和重组。

使用requests库发送POST请求
每个请求附带当前分片数据及元信息（可放在form-data中）
支持断点续传时，可先查询已上传的分片

上传示例：

SuperCms在线订餐系统

模板采用响应式设计，自动适应手机，电脑及平板显示；满足单一店铺外卖需求。功能：1.菜单分类管理2.菜品管理：菜品增加，删除，修改3.订单管理4.友情链接管理5.数据库备份6.文章模块：如：促销活动，帮助中心7.单页模块：如：企业信息，关于我们更强大的功能在开发中……安装方法：上传到网站根目录，运行http://www.***.com/install 自动

0 查看详情 SuperCms在线订餐系统

import requests
<p>for index, chunk in enumerate(read_file_chunks('large<em>file.zip')):
files = {'file': ('chunk</em>%d' % index, chunk)}
data = {
'filename': 'large_file.zip',
'chunk_index': index,
'total_chunks': 10  # 可预先计算
}
response = requests.post('<a href="https://www.php.cn/link/c0dba5809c620f70942856ad09b144d0">https://www.php.cn/link/c0dba5809c620f70942856ad09b144d0</a>', data=data, files=files)
if response.status_code != 200:
print(f"上传失败: 分片 {index}")
break

3. 服务端接收与合并

服务端需保存每个分片，待全部接收后按顺序合并。

根据文件名创建临时目录存放分片
收到分片后以序号命名存储（如 part_0, part_1）
检查是否所有分片已上传，自动触发合并

Flask 示例处理逻辑：

from flask import Flask, request
import os
<p>app = Flask(<strong>name</strong>)
UPLOAD_DIR = '/tmp/uploads'
CHUNKS_DIR = '/tmp/chunks'</p><p>@app.route('/upload', methods=['POST'])
def handle_upload():
file = request.files['file']
filename = request.form['filename']
chunk_index = int(request.form['chunk_index'])
total_chunks = int(request.form['total_chunks'])</p><pre class="brush:php;toolbar:false;">chunk_dir = os.path.join(CHUNKS_DIR, filename + "_parts")
os.makedirs(chunk_dir, exist_ok=True)

chunk_path = os.path.join(chunk_dir, f"part_{chunk_index}")
file.s*e(chunk_path)

# 检查是否全部上传完成
if all(os.path.exists(os.path.join(chunk_dir, f"part_{i}")) for i in range(total_chunks)):
    merge_files(chunk_dir, os.path.join(UPLOAD_DIR, filename))
    cleanup(chunk_dir)  # 删除分片

return "OK", 200

def merge_files(chunk_dir, target_path): with open(target_path, 'wb') as f: for i in sorted(os.listdir(chunk_dir)): part_path = os.path.join(chunk_dir, i) with open(part_path, 'rb') as part: f.write(part.read())