Python后端flask框架接收zip压缩包方法

一、用base64编码发送，以及接收

python">import base64
import io
import zipfile
from flask import request, jsonifydef unzip_and_find_png(zip_data):# 使用 BytesIO 在内存中处理 zip 数据with zipfile.ZipFile(io.BytesIO(zip_data), 'r') as zip_ref:extracted_paths = []for file_name in zip_ref.namelist():if file_name.lower().endswith('.png'):  # 只提取 PNG 文件# 保存文件内容到内存中或其他处理逻辑extracted_paths.append(file_name)return extracted_paths@app.route('/upload_zip', methods=['POST'])
def upload_zip():# 从请求数据中获取 Base64 格式的 zip 文件encoded_zip = request.json.get('imagesZipBase64', '')if encoded_zip == '':res = {"ret": 1, "msg": "No valid images zip's Base64 data", "labels": []}return jsonify(res), 200try:# 解码 Base64 数据为字节数据zip_data = base64.b64decode(encoded_zip)# 解压并查找 PNG 文件image_paths = unzip_and_find_png(zip_data)if len(image_paths) == 0:res = {"ret": 1, "msg": "No valid images in zip", "labels": []}return jsonify(res), 200# 返回找到的 PNG 文件路径列表res = {"ret": 0, "msg": "Images found", "labels": image_paths}return jsonify(res), 200except Exception as e:res = {"ret": 1, "msg": f"Error processing zip file: {str(e)}", "labels": []}return jsonify(res), 200

模拟发送

python">import base64
import requestsdef zip_to_base64(zip_path):"""读取本地 zip 文件并将其转换为 Base64 编码的字符串。:param zip_path: str, 本地 zip 文件路径:return: str, Base64 编码的字符串"""try:with open(zip_path, 'rb') as file:zip_data = file.read()base64_encoded = base64.b64encode(zip_data).decode('utf-8')  # 转换为 Base64 并解码为字符串return base64_encodedexcept Exception as e:print(f"Error reading zip file: {e}")return Nonedef send_base64_to_api(base64_data, api_url):"""将 Base64 编码的 zip 数据发送到指定 API。:param base64_data: str, Base64 编码的 zip 数据:param api_url: str, API 的地址:return: Response, 请求返回的响应"""payload = {"imagesZipBase64": base64_data  # API 要求的字段名}try:headers = {'Content-Type': 'application/json'}response = requests.post(api_url, json=payload, headers=headers)return responseexcept Exception as e:print(f"Error sending request to API: {e}")return Noneif __name__ == "__main__":# 本地 zip 文件路径zip_file_path = "example.zip"# 将 zip 文件转换为 Base64base64_encoded_zip = zip_to_base64(zip_file_path)if not base64_encoded_zip:print("Failed to encode zip file to Base64.")exit()# 目标 API 地址api_endpoint = "http://localhost:5000/get_multi_labels"  # 替换为实际 API 地址# 发送请求response = send_base64_to_api(base64_encoded_zip, api_endpoint)if response is not None:print(f"API Response Status: {response.status_code}")print(f"API Response Data: {response.json()}")else:print("Failed to send data to API.")

二、其他方法

1. 使用 HTTP `multipart/form-data` 上传 -- 推荐

这是最常见的方法，适合中等大小的文件（如几十 MB）。其他后端通过 HTTP POST 请求将文件作为表单数据发送到 Flask。

Flask 后端代码：

python">from flask import Flask, request, jsonifyapp = Flask(__name__)# 配置最大上传文件大小（如 500 MB）
app.config['MAX_CONTENT_LENGTH'] = 500 * 1024 * 1024  # 500 MB@app.route('/upload_zip', methods=['POST'])
def upload_zip():# 检查是否包含文件if 'file' not in request.files:return jsonify({"ret": 1, "msg": "No file uploaded"})zip_file = request.files['file']# 检查文件是否有效if zip_file.filename == '':return jsonify({"ret": 1, "msg": "Empty filename"})# 保存文件到本地save_path = f"./uploads/{zip_file.filename}"zip_file.save(save_path)return jsonify({"ret": 0, "msg": f"File '{zip_file.filename}' uploaded successfully", "path": save_path})if __name__ == "__main__":app.run(debug=True)

其他后端代码示例（Python Requests）：

python">import requests
file_path = "large_file.zip"  # 本地文件路径
url = "http://flask-server-address/upload_zip"
with open(file_path, 'rb') as f:response = requests.post(url, files={'file': f})
print(f"Response: {response.status_code}, Data: {response.json()}")

特点：

优点：实现简单，支持多语言，适合中小型文件传输。
缺点：对于超大文件（>500 MB），容易造成内存和带宽压力。

2. 使用流式传输（Chunked Transfer Encoding）

对于超大文件，流式传输是更优的选择。其他后端将文件逐块读取并发送，Flask 后端逐块接收并处理。

Flask 后端代码：

python">from flask import Flask, request, jsonify
app = Flask(
__name__
)
@app.route('/upload_large_zip', methods=['POST'])
def upload_large_zip():try:save_path = "./uploads/large_file.zip"with open(save_path, 'wb') as f:# 按块读取请求数据并写入文件for chunk in request.stream:f.write(chunk)return jsonify({"ret": 0, "msg": "File uploaded successfully", "path": save_path})except Exception as e:return jsonify({"ret": 1, "msg": f"Error: {str(e)}"})
if 
name== "
__main__
":app.run(debug=True)

其他后端代码示例（Python Requests）：

python">import requests
file_path = "large_file.zip"
url = "http://flask-server-address/upload_large_zip"
def file_chunk_reader(file_path, chunk_size=1024 * 1024):  # 1 MBwith open(file_path, 'rb') as f:while chunk := f.read(chunk_size):yield chunk
response = requests.post(url, data=file_chunk_reader(file_path))
print(f"Response: {response.status_code}, Data: {response.json()}")

特点：

优点：
- 内存占用低。
- 可处理超大文件（如 >1 GB）。
缺点：
- 需要其他后端支持流式上传。

3. 使用文件存储服务（如 S3、OSS）作为中间媒介

如果文件非常大且需要高效传输，可以通过云存储服务作为中转。其他后端将文件上传到云存储，Flask 后端只需下载即可。

流程：

其他后端将 .zip 文件上传到 S3、OSS 或其他文件存储服务。
上传完成后，通知 Flask 后端文件的下载 URL。
Flask 后端通过 URL 下载文件进行处理。

Flask 后端代码：

python">import requests
from flask import Flask, request, jsonify
app = Flask(
__name__
)
@app.route('/fetch_file', methods=['POST'])
def fetch_file():data = request.get_json()file_url = data.get('file_url')if not file_url:return jsonify({"ret": 1, "msg": "No file URL provided"})save_path = "./uploads/remote_file.zip"try:response = requests.get(file_url, stream=True)response.raise_for_status()  # 检查请求是否成功with open(save_path, 'wb') as f:for chunk in response.iter_content(chunk_size=1024 * 1024):  # 1 MBf.write(chunk)return jsonify({"ret": 0, "msg": "File fetched successfully", "path": save_path})except Exception as e:return jsonify({"ret": 1, "msg": f"Error fetching file: {str(e)}"})
if 
name== "
__main__
":app.run(debug=True)

特点：

优点：
- 避免直接传输大文件，节省带宽。
- 利用云存储服务的高效上传下载能力。
缺点：
- 需要配置云存储服务权限和 URL 签名。

4. 使用 WebSocket 或 TCP 直接传输

如果需要更低延迟和更高效的传输，可以考虑使用 WebSocket 或 TCP 传输协议。此方法适合实时性高、带宽充足的场景。

WebSocket 示例（后端使用 Flask-SocketIO）：

python">from flask import Flask
from flask_socketio import SocketIO
app = Flask(
__name__
)
socketio = SocketIO(app)
@socketio.on('upload_chunk')
def handle_upload_chunk(data):# 每次收到一个文件块时处理chunk = data.get('chunk')with open("./uploads/large_file.zip", 'ab') as f:f.write(chunk)
if 
name== "
__main__
":socketio.run(app, debug=True)

选择方案的建议：

文件大小	推荐方案
小文件（<50 MB）	使用 HTTP multipart/form-data 上传
中等文件（50-500 MB）	使用流式传输
大文件（>500 MB）	使用文件存储服务（如 S3/OSS），或者流式传输
实时数据	使用 WebSocket 或 TCP

根据文件大小和后端需求选择合适的方案，同时要考虑文件存储路径、权限管理以及网络稳定性等因素。