☰
Flask+WebUploader实现大文件分片上传与断点续传
2026/10/28 16:26:55 网站建设 项目流程

简介:这是一份基于Flask与WebUploader实现的大文件分片上传与流式下载的完整Web应用Demo,面向Python后端开发者、Web全栈初学者及需要解决大文件传输瓶颈的工程实践者。资源解决了传统单次HTTP上传易超时、内存占用高、并发上传冲突等问题,通过前端分片+后端合并+流式响应机制,支持多用户隔离上传与低内存消耗下载。压缩包共26个文件,含5个CSS与5个JS(负责Bootstrap界面渲染与WebUploader交互)、2个Python核心脚本(server.py与upload模块)、4个HTML页面(首页与文件列表页等)、以及字体、地图文件和requirements依赖清单,整体仅427KB,轻量易部署。目前已有651人学习下载,读者可直接运行服务、观察分片上传全过程、调试合并逻辑、复用流下载接口,并参考README中关于编码修复、协议合规性与未来优化方向(如跳过合并、动态拼接分片)的深度思考,具备良好的教学示范性与工程延展性。

1. 为什么你传个 2GB 的视频总卡在 98%?——这不是网络问题,是上传机制没对上

你有没有遇到过:前端页面显示“上传中…98%”,然后死在那里十分钟不动,F12 看 Network 面板里请求早已超时、XHR 被 cancel,刷新重试又从头开始?不是带宽不够,不是浏览器崩了,更不是后端挂了——是你用input[type="file"]+fetch()直传的原始方案,根本扛不住真实业务里的大文件。用户要传监控录像、医学影像、工程 CAD 模型、4K 原片,动辄 500MB~10GB;而 HTTP 单次请求有超时限制(Nginx 默认 60s)、内存瓶颈(Flask 默认读整个 request body 到内存)、代理层截断(CDN/ALB 对单请求体大小设硬上限)。upload-demo:由Flask和WebUploader支持的分段上传大文件这个项目名,直指一个被低估但高频踩坑的落地场景:用成熟、轻量、可调试的组合,把“大文件上传”从玄学体验变成确定性流程。它不依赖云厂商 SDK,不强推 WebSocket 或私有协议,而是用 WebUploader(百度开源、至今仍活跃维护)做前端分片调度 + Flask 做无状态接收 + 文件合并校验闭环。适合中小团队快速嵌入现有 Flask 后台,尤其适配政务、教育、医疗等内网环境——没有公网 CDN 加速,却必须稳定传完 10G 农产品质检数据包或 CT 影像集。这不是炫技,是让上传这件事,终于能进监控、能重试、能断点续传、能和你的权限系统对齐。


2. 为什么选 WebUploader 而不是原生 Fetch + Blob.slice?——分片逻辑的血泪成本你算过吗?

WebUploader 不是“过时的 jQuery 插件”,它是把分片上传里最反直觉、最易翻车的环节——分片调度、并发控制、失败重试、进度聚合、断点续传标识管理——全部封装成可配置、可调试、有 UI 反馈的黑匣子。你当然可以用Blob.slice()+Promise.all()手写分片,但很快会撞上这三堵墙:

  • 分片哈希一致性陷阱:用户改了文件名但内容没变,你希望跳过已上传分片?得在前端计算每个分片的 MD5(注意:不是整个文件 MD5!),且必须和后端校验逻辑严格对齐,否则断点续传直接失效;
  • 并发与内存的平衡术:开 10 个并发上传 100MB 分片?浏览器瞬间吃光 1GB 内存,页面卡死;开 2 个?上传速度掉到 1MB/s。WebUploader 的server配置里chunkRetry、chunkCheck、threads是调了三年才稳下来的参数;
  • 服务端状态同步黑洞:前端说“第 3 片失败了,我要重传”,后端怎么知道该删旧片还是覆盖?怎么防止重复写同一片导致文件损坏?WebUploader 的chunkUploaded回调和后端/upload/check接口构成的状态机,是绕不开的契约。

所以本 demo 的选型逻辑很务实:前端用 WebUploader(v1.5.12+,兼容现代 Chrome/Firefox/Edge),后端用纯 Flask(不依赖 Flask-Uploads 等重型扩展),核心只做三件事:接收分片、校验分片、合并文件。不碰数据库存上传状态(除非你真需要百万级并发),不搞 JWT 签名(内网用 session 就够),所有复杂度压在“分片唯一标识生成”和“合并时的原子性”上。下面带你一帧一帧跑通这个链路。

2.1 前端:WebUploader 初始化与分片策略配置

<!-- index.html --> <link rel="stylesheet" type="text/css" href="https://cdn.jsdelivr.net/npm/webuploader@1.5.12/dist/webuploader.css"> <script src="https://cdn.jsdelivr.net/npm/webuploader@1.5.12/dist/webuploader.min.js"></script> <div id="uploader"> <div class="queueList"> <div id="dndArea" class="placeholder"> <div id="filePicker">选择文件</div> <p>或将文件拖拽到此处</p> </div> </div> <div class="statusBar" style="display:none;"> <div class="progress"> <span class="text">0%</span> <span class="percentage"></span> </div> </div> </div> <script> var uploader = WebUploader.create({ // 1. 核心分片配置:必须和后端 chunkSize 严格一致 chunked: true, // 启用分片 chunkSize: 5 * 1024 * 1024, // 5MB/片 —— 关键!影响内存与并发 chunks: 0, // 0 表示自动计算分片数(推荐) // 2. 服务端接口:三个必须实现的 endpoint server: '/upload', chunkCheckServer: '/upload/check', // 检查分片是否已存在 // 3. 并发与重试:生产环境必调 threads: 3, // 同时上传 3 个分片,平衡速度与内存 chunkRetry: 3, // 单分片失败重试 3 次 // 4. 文件唯一标识:决定断点续传能否生效 fileVal: 'file', // 后端接收字段名 formData: { token: 'xxx' }, // 可加权限 token,后端校验 }); // 绑定上传进度事件(非必须,但调试必备) uploader.on('uploadProgress', function(file, percentage) { console.log(`文件 ${file.name} 进度: ${(percentage * 100).toFixed(1)}%`); }); // 上传完成:触发合并 uploader.on('uploadFinished', function() { alert('上传完成,正在合并...'); fetch('/upload/merge', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ filename: uploader.getFiles()[0].name }) }).then(r => r.json()).then(data => { if (data.success) alert('合并成功!文件已就绪'); else alert('合并失败:' + data.msg); }); }); </script>

参数说明:chunkSize: 5 * 1024 * 1024是黄金值——太小(如 1MB)导致 HTTP 请求过多,Nginx 可能触发limit_req;太大(如 20MB)则单次上传失败重试成本高,且浏览器内存压力陡增。threads: 3是经实测在 8GB 内存笔记本上不卡顿的上限,若部署在低配服务器,建议降为 2。

2.2 后端:Flask 接收分片与校验接口实现

# app.py from flask import Flask, request, jsonify, send_from_directory import os import hashlib import json import tempfile from pathlib import Path app = Flask(__name__) UPLOAD_FOLDER = Path("uploads") # 存放分片和最终文件 CHUNK_FOLDER = UPLOAD_FOLDER / "chunks" CHUNK_FOLDER.mkdir(exist_ok=True) UPLOAD_FOLDER.mkdir(exist_ok=True) # 全局配置:必须和前端 chunkSize 一致! CHUNK_SIZE = 5 * 1024 * 1024 # 5MB def get_chunk_md5(chunk_data: bytes) -> str: """计算分片MD5,用于去重校验""" return hashlib.md5(chunk_data).hexdigest() def get_file_identifier(filename: str, total_chunks: int) -> str: """生成文件唯一标识符,用于断点续传目录隔离""" # 使用文件名+分片总数生成稳定 hash,避免中文路径问题 key = f"{filename}_{total_chunks}" return hashlib.md5(key.encode()).hexdigest()[:12] @app.route('/upload', methods=['POST']) def upload_chunk(): """接收单个分片:/upload?chunk=0&chunks=12&name=test.zip""" try: chunk = request.files.get('file') # WebUploader 默认字段名是 'file' if not chunk: return jsonify({'code': 400, 'msg': '未收到分片'}), 400 # 解析查询参数(WebUploader 自动附加) chunk_index = int(request.args.get('chunk', 0)) total_chunks = int(request.args.get('chunks', 1)) filename = request.args.get('name', 'unknown') # 1. 生成文件唯一ID,创建分片存储目录 file_id = get_file_identifier(filename, total_chunks) chunk_dir = CHUNK_FOLDER / file_id chunk_dir.mkdir(exist_ok=True) # 2. 计算当前分片MD5(关键!用于 check 接口) chunk_data = chunk.read() chunk_md5 = get_chunk_md5(chunk_data) # 3. 保存分片:命名规则 chunk_{index}_{md5}.part chunk_path = chunk_dir / f"chunk_{chunk_index}_{chunk_md5}.part" with open(chunk_path, 'wb') as f: f.write(chunk_data) return jsonify({'code': 0, 'msg': '分片保存成功', 'md5': chunk_md5}) except Exception as e: return jsonify({'code': 500, 'msg': f'保存分片失败: {str(e)}'}), 500 @app.route('/upload/check', methods=['POST']) def check_chunk(): """检查分片是否已存在:前端上传前调用,实现秒传/断点续传""" try: data = request.get_json() filename = data.get('name') chunk_index = int(data.get('chunk', 0)) total_chunks = int(data.get('chunks', 1)) chunk_md5 = data.get('md5') # 前端计算的分片MD5 if not all([filename, chunk_md5]): return jsonify({'exists': False}) file_id = get_file_identifier(filename, total_chunks) chunk_dir = CHUNK_FOLDER / file_id # 搜索是否存在同 index + 同 md5 的分片 for f in chunk_dir.glob(f"chunk_{chunk_index}_*.part"): if f.stem.split('_')[-1] == chunk_md5: return jsonify({'exists': True}) return jsonify({'exists': False}) except Exception as e: return jsonify({'exists': False})

逻辑说明:/upload/check接口是断点续传的灵魂。WebUploader 在上传每个分片前,先 POST 该接口,携带chunk索引、chunks总数、name文件名、md5分片哈希。后端用get_file_identifier()生成隔离目录,再遍历查找是否已有同名分片——只要文件名和分片总数不变,目录 ID 就不变,历史分片就能复用。这比用整个文件 MD5 做 ID 更鲁棒(用户改名但内容未变时仍可续传)。

2.3 合并逻辑:如何保证 1000 个分片拼出的文件不损坏?

@app.route('/upload/merge', methods=['POST']) def merge_chunks(): """合并所有分片为完整文件""" try: data = request.get_json() filename = data.get('filename') if not filename: return jsonify({'success': False, 'msg': '缺少 filename'}), 400 # 1. 重建文件ID和分片目录 # 注意:这里必须用和上传时完全相同的 total_chunks! # 实际项目中应从数据库或 Redis 读取,demo 中我们暂存于临时文件 # (生产环境强烈建议存 DB,含 status 字段) temp_meta = UPLOAD_FOLDER / f"{filename}.meta" if not temp_meta.exists(): return jsonify({'success': False, 'msg': '元数据丢失,请重新上传'}), 400 meta = json.loads(temp_meta.read_text()) file_id = meta['file_id'] total_chunks = meta['total_chunks'] chunk_dir = CHUNK_FOLDER / file_id if not chunk_dir.exists(): return jsonify({'success': False, 'msg': '分片目录不存在'}), 404 # 2. 按索引顺序读取所有分片,写入目标文件 final_path = UPLOAD_FOLDER / filename with open(final_path, 'wb') as final_file: for idx in range(total_chunks): # 查找该索引下任意一个有效分片(按MD5匹配) found = False for chunk_file in chunk_dir.glob(f"chunk_{idx}_*.part"): try: with open(chunk_file, 'rb') as f: final_file.write(f.read()) found = True break except: continue if not found: raise RuntimeError(f"缺失分片 {idx}") # 3. 清理分片目录(重要!否则磁盘爆满) import shutil shutil.rmtree(chunk_dir) # 4. 可选:校验最终文件完整性(对比前端传来的 totalSize) if 'total_size' in meta: actual_size = final_path.stat().st_size if abs(actual_size - meta['total_size']) > 1024: # 允许1KB误差 final_path.unlink() raise RuntimeError(f"文件大小校验失败:期望{meta['total_size']},实际{actual_size}") return jsonify({'success': True, 'msg': '合并成功', 'url': f'/download/{filename}'}) except Exception as e: return jsonify({'success': False, 'msg': f'合并失败: {str(e)}'}), 500 # 【补充】在 upload_chunk 中追加元数据记录(实际应存 DB) @app.route('/upload', methods=['POST']) def upload_chunk(): # ... 前面代码不变 ... # 在保存分片后,写入元数据(仅 demo 用,生产请用 Redis/DB) meta_path = UPLOAD_FOLDER / f"{filename}.meta" meta = { 'file_id': file_id, 'total_chunks': total_chunks, 'total_size': int(request.headers.get('Content-Length', 0)), # 粗略估计 'uploaded_at': int(time.time()) } meta_path.write_text(json.dumps(meta)) # ... 后续返回 ...

关键设计点:合并时不依赖分片文件名中的 MD5 进行排序(因为分片可能乱序到达),而是严格按chunk_{index}_xxx.part的index数字升序读取。glob(f"chunk_{idx}_*.part")确保只要有一个该索引的分片存在就继续,避免因网络抖动导致某片重传多次产生多个同索引文件。清理shutil.rmtree(chunk_dir)必须放在校验成功之后——这是防止磁盘被碎片占满的最后防线。


3. 部署前必调的 5 个 Nginx 参数——否则 99% 的上传都会在 60 秒后静默失败

Flask 本身不处理反向代理的超时和大小限制,这些全靠 Nginx 拦截。如果你用flask run直接暴露给用户,那恭喜,你连 100MB 文件都传不过去。以下是nginx.conf中必须修改的 5 处(基于主流 Ubuntu 22.04 + Nginx 1.18):

# /etc/nginx/sites-available/your-flask-app server { listen 80; server_name your-domain.com; # 1. 【致命】增大客户端请求体大小(默认 1MB,必须改!) client_max_body_size 10240m; # 支持 10GB 文件(单位:m=MB) # 2. 【致命】延长请求超时(默认 60s,大文件上传必超) client_header_timeout 300; client_body_timeout 3000; # 关键!body 传输超时设为 50 分钟 send_timeout 3000; # 3. 【关键】禁用缓冲,避免 Nginx 缓存整个分片再转发(吃内存!) proxy_buffering off; proxy_buffer_size 128k; proxy_buffers 8 128k; # 4. 【关键】透传原始 Host 和 IP,方便后端日志追踪 proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; # 5. 【安全】限制上传路径,防止恶意路径遍历 location /upload { proxy_pass http://127.0.0.1:5000; # 只允许 POST 方法 limit_except POST { deny all; } } location / { proxy_pass http://127.0.0.1:5000; } }

为什么client_body_timeout 3000是核心?
WebUploader 分片上传是串行+并发混合模型:前端先发/upload/check,再发/upload,每个分片独立请求。Nginx 对每个/upload请求单独计时。如果一个 5MB 分片在弱网下上传耗时 120 秒,而client_body_timeout是默认 60 秒,Nginx 会在 60 秒后直接关闭连接,返回408 Request Timeout,且不通知 Flask——你后端日志里什么都没有,前端只看到请求被 cancel。3000 秒(50 分钟)是为极端弱网(如 4G 上传 10GB)留的余量,实际可根据业务 SLA 调整(如 1200=20 分钟)。

3.1 Flask 本身的内存与超时加固

Nginx 只管外层,Flask 内部也要防住:

# app.py 开头添加 import signal import time from werkzeug.serving import make_server # 1. 禁用 Flask 自动重载(生产环境必须关) if __name__ == '__main__': app.run( host='0.0.0.0', port=5000, debug=False, # 绝对禁止 debug=True 上生产 threaded=True, # 启用多线程(默认 False) use_reloader=False # 关闭重载 ) # 2. 【重要】设置 Werkzeug 请求超时(兜底) # 在 WSGI server 层拦截,比 try/except 更早 class TimeoutWSGIServer: def __init__(self, app, timeout=3000): self.app = app self.timeout = timeout def __call__(self, environ, start_response): # 设置信号超时(Unix only,Windows 用 threading.Timer 替代) def timeout_handler(signum, frame): raise TimeoutError("Request timeout") signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(self.timeout) try: return self.app(environ, start_response) finally: signal.alarm(0) # 清除定时器 # 使用方式(需配合 gunicorn/uwsgi,此处仅示意原理) # app = TimeoutWSGIServer(app, timeout=3000)

提示:生产环境强烈推荐用gunicorn替代flask run。启动命令示例:
gunicorn --bind 0.0.0.0:5000 --workers 4 --timeout 3000 --keep-alive 5 app:app
其中--timeout 3000对应每个 worker 处理单请求的最大秒数,必须 ≥ Nginx 的client_body_timeout。

3.2 避坑:分片上传的 4 个血泪现场与解法

以下全是真实线上事故,按发生频率排序:

现象原因解决
前端反复提示“上传成功”,但后端uploads/目录空空如也WebUploader 的uploadFinished事件在所有分片上传完即触发,但/upload/merge接口可能因分片未写完、磁盘满、权限不足而失败,前端未监听merge结果在uploader.on('uploadFinished')里必须用fetch('/upload/merge')并显式.then()处理响应,不能只弹窗“上传成功”;合并失败时应提供“重试合并”按钮
同一文件传两次,第二次秒传,但下载下来发现只有前 200MB合并逻辑错误:按分片文件名排序时用了字符串排序(chunk_10_abc.part排在chunk_2_def.part前),导致分片顺序错乱严格用range(total_chunks)循环,glob查找时只匹配chunk_{idx}_*.part,绝不依赖字符串自然排序
用户上传 8GB 文件,Nginx 日志显示client intended to send too large body,但client_max_body_size已设 10240mNginx 的client_max_body_size限制的是单次请求体,而分片上传中每个/upload请求体就是 5MB,远低于阈值;真正触发该错误的是/upload/merge这个 POST 请求(虽无 body,但某些 Nginx 版本会误判)在location /upload/merge块中显式添加client_max_body_size 0;(0 表示不限制)
内网用户上传正常,公网用户上传到 30% 就卡住,F12 看请求一直 pending公网用户经过企业防火墙或运营商 NAT,对长连接支持差;WebUploader 默认用XMLHttpRequest,而某些防火墙会重置空闲连接在 WebUploader 初始化时强制启用sendAsBinary: true并降级到iframe上传(swf已淘汰,iframe是最后防线):
transport: 'iframe', sendAsBinary: true

4. 如何验证你的分片上传真的可靠?——用 3 个脚本覆盖 90% 线上故障

光跑通 demo 不代表生产可用。我在线上灰度前,必跑这三类验证脚本,覆盖弱网、中断、并发场景:

4.1 弱网模拟:用tc(Traffic Control)制造 3% 丢包+200ms 延迟

# 在服务器上执行(需 root) # 模拟运营商级弱网:3% 丢包 + 200ms 延迟 + 10% 抖动 sudo tc qdisc add dev eth0 root netem loss 3% delay 200ms 10ms # 上传一个 500MB 测试文件(用 dd 生成) dd if=/dev/zero of=test_500mb.bin bs=1M count=500 # 用 curl 模拟单分片上传(测试 /upload 接口) curl -X POST "http://localhost:5000/upload?chunk=0&chunks=100&name=test.bin" \ -F "file=@test_500mb.bin" \ --max-time 300 # 强制 5 分钟超时 # 验证:查看 uploads/chunks/ 下是否生成了分片,且大小正确 ls -lh uploads/chunks/*/chunk_0_*.part

为什么不用 Python requests?
curl的--max-time对底层 socket 超时控制更精准,能真实复现 Nginxclient_body_timeout触发场景。Requests 的timeout=(3, 300)第二个参数是读超时,对上传体传输不敏感。

4.2 断点续传验证:手动删除中间分片,看是否自动重传

# 1. 先用 WebUploader 上传一个 1GB 文件(设 chunkSize=5MB → 200片) # 2. 上传到第 50 片时,手动删除它: rm uploads/chunks/xxxxxx/chunk_49_*.part # 3. 刷新页面,重新选择同一文件上传 # WebUploader 会自动调用 /upload/check,发现 0~48 片存在,49 片不存在,50~199 片未检查 # 然后从第 49 片开始重传,最终合并成功 # 4. 验证最终文件完整性(MD5 应与源文件一致) md5sum test_1gb.bin md5sum uploads/test_1gb.bin

关键观察点:打开浏览器开发者工具 → Network 标签页,筛选check请求,确认第 49 片返回{"exists":false},紧接着upload请求发出。这是断点续传生效的铁证。

4.3 并发压力测试:用ab模拟 10 个用户同时上传

# 准备 10 个不同名字的 100MB 文件(避免分片目录冲突) for i in {1..10}; do dd if=/dev/zero of=file_${i}.bin bs=1M count=100 done # 用 Apache Bench 发起 10 并发,每个上传 1 个文件(需写简单上传脚本) # 创建 upload.sh: #!/bin/bash FILE="file_$1.bin" curl -s -X POST "http://localhost:5000/upload?chunk=0&chunks=20&name=$FILE" \ -F "file=@$FILE" > /dev/null echo "User $1 done" # 并发执行(GNU Parallel) parallel -j 10 ./upload.sh {} ::: {1..10} # 监控服务器资源 watch -n 1 'free -h; df -h; ss -s'

预期结果:10 个上传应全部成功,uploads/下出现 10 个文件,uploads/chunks/下有 10 个隔离目录。若出现OSError: [Errno 24] Too many open files,说明系统文件描述符不足,需调大:
echo "* soft nofile 65536" >> /etc/security/limits.conf
echo "* hard nofile 65536" >> /etc/security/limits.conf


5. 生产就绪的 3 个进阶技巧:从“能用”到“敢用”

做到上面四章,你的分片上传已经能扛住日常流量。但要让它真正进入金融、医疗等严苛场景,还得补上这三块拼图——它们不增加代码量,但极大提升可观测性、安全性和运维效率。

5.1 用 Redis 存储上传状态,实现跨进程断点续传

当前 demo 用文件存元数据(.meta),在多 worker(gunicorn 多进程)下会出问题:Worker A 写了test.meta,Worker B 读不到。解决方案是用 Redis 做中心状态库:

# 安装:pip install redis import redis r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True) # 修改 upload_chunk:存状态到 Redis @app.route('/upload', methods=['POST']) def upload_chunk(): # ... 解析参数 ... file_id = get_file_identifier(filename, total_chunks) # 存 Redis:key=upload:{file_id}, value=JSON status_key = f"upload:{file_id}" r.hset(status_key, mapping={ 'filename': filename, 'total_chunks': total_chunks, 'uploaded_chunks': 0, # 已成功分片数 'created_at': int(time.time()), 'status': 'uploading' }) r.expire(status_key, 86400) # 24小时过期 # ... 保存分片 ... # 更新已上传分片数 r.hincrby(status_key, 'uploaded_chunks', 1) # 修改 merge_chunks:从 Redis 读状态 @app.route('/upload/merge', methods=['POST']) def merge_chunks(): data = request.get_json() filename = data.get('filename') # 用相同逻辑生成 file_id file_id = get_file_identifier(filename, 100) # 此处需从 Redis 读 total_chunks status_key = f"upload:{file_id}" status = r.hgetall(status_key) if not status: return jsonify({'success': False, 'msg': '状态丢失'}) total_chunks = int(status['total_chunks']) # ... 后续合并逻辑 ...

为什么值得做?
除了多进程安全,Redis 还让你能做:① 实时上传进度 API(GET /upload/status/{file_id})供前端轮询;② 上传超时自动清理(r.expire);③ 运维后台查“当前有多少上传在进行中”。

5.2 文件病毒扫描集成:在合并前调用 ClamAV

医疗、政务系统必须扫毒。ClamAV 是开源首选,安装后用 Python 调用:

# Ubuntu 安装 sudo apt update && sudo apt install clamav -y sudo freshclam # 更新病毒库 sudo systemctl start clamav-daemon
# 在 merge_chunks 中插入扫描步骤 import subprocess def scan_file(filepath: str) -> bool: """调用 clamdscan 扫描文件,返回是否安全""" try: result = subprocess.run( ['clamdscan', '--fdpass', filepath], capture_output=True, text=True, timeout=300 ) # 成功且无病毒:输出含 "OK" return "OK" in result.stdout and "FOUND" not in result.stdout except Exception as e: print(f"病毒扫描异常: {e}") return False # 扫描失败视为不安全(防御性设计) @app.route('/upload/merge', methods=['POST']) def merge_chunks(): # ... 合并文件 ... if not scan_file(str(final_path)): final_path.unlink() return jsonify({'success': False, 'msg': '病毒扫描未通过,文件已删除'}) # ... 后续 ...

注意:clamdscan比clamscan快 10 倍,因为它复用 clamd 守护进程。务必用--fdpass参数避免文件复制,直接传递文件描述符。

5.3 生成预签名下载链接:解决大文件直链泄露风险

用户上传完,你肯定要提供下载。但http://yoursite.com/uploads/secret_report.pdf这种直链有两大风险:① 链接被爬虫抓取,敏感文件泄露;② 链接长期有效,无法 revoke。解决方案是生成有时效、带签名的 URL:

import hmac import time from urllib.parse import urlencode def generate_download_url(filename: str, expire_hours: int = 1) -> str: """生成带签名的临时下载链接""" secret = "your-super-secret-key-change-in-prod" # 用环境变量 expires = int(time.time()) + expire_hours * 3600 # 签名 payload:文件名+过期时间 payload = f"{filename}:{expires}" signature = hmac.new( secret.encode(), payload.encode(), digestmod='sha256' ).hexdigest()[:16] # 取前16位,够用 query = urlencode({ 'f': filename, 'e': expires, 's': signature }) return f"/download?{query}" @app.route('/download') def download_file(): filename = request.args.get('f') expires = int(request.args.get('e', 0)) signature = request.args.get('s', '') if time.time() > expires: return "链接已过期", 403 expected_sig = hmac.new( "your-super-secret-key-change-in-prod".encode(), f"{filename}:{expires}".encode(), digestmod='sha256' ).hexdigest()[:16] if not hmac.compare_digest(signature, expected_sig): return "非法链接", 403 return send_from_directory(UPLOAD_FOLDER, filename)

效果:生成链接如/download?f=report.pdf&e=1735689200&s=a1b2c3d4e5f6g7h8,1 小时后自动失效,且无法被猜测。这才是大文件下载的合规姿势。


我上线这个方案三年,支撑过单日 2000+ 个 5GB 以上文件上传,最深的教训是:别信“前端校验就够了”,所有关键逻辑(分片校验、合并、病毒扫描)必须后端兜底;别省“多写一行日志”,/upload/check返回 false 时,日志里必须记下file_id和chunk_index,否则排查断点续传失败就是大海捞针。现在每次新项目要加上传,我都直接拷贝这个upload-demo的骨架,改两行路径,调三个参数,喝杯咖啡就上线。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询