一、协议接口规范定义
在开始编码前，必须严格定义档案数字化数据的接收协议标准。本指南采用 HTTP POST + JSON 方式传输元数据，Multipart/Form-Data 传输实体文件。这是目前行业内通用性最强、兼容性最好的混合协议模式。
1. 请求地址
POST /api/v1/digitization/upload
2. 请求头
Content-Type 必须设置为 multipart/form-data，且包含边界标识。
3. 请求体结构
请求体需包含两部分：JSON 格式的元数据描述和具体的二进制文件流。
- metadata (JSON字符串): 包含批次号、档号、操作员等核心信息。
- file (Binary文件): 具体的数字化成果文件（如 PDF、TIF、JPG）。
4. metadata JSON 字段详细定义
batch_id: 字符串(必填)，批次号，用于关联同一批次的档案，格式如 BATCH_20231027_001。archive_id: 字符串(必填)，档号，唯一标识一份档案，如 2023-JY-001。file_type: 字符串(必填)，文件类型，枚举值：ORIGINAL(原件)、PROCESSING(处理件)。operator: 字符串(可选)，数字化操作员姓名。checksum: 字符串(必填)，文件 MD5 值，用于校验文件完整性。timestamp: 长整型(必填)，Unix 时间戳，防止重放攻击。
5. 响应体标准
- 成功响应 (200 OK):
{"code": 200, "message": "接收成功", "data": {"saved_path": "/srv/archives/..."}}
- 失败响应 (400/500):
{"code": 1001, "message": "MD5校验失败", "data": null}
二、开发环境极速搭建
本指南使用 Python 的 Flask 框架，因为它轻量、部署简单，非常适合构建此类协议接口。请确保你的系统已安装 Python 3.8 或更高版本。
1. 安装依赖库
打开终端，直接执行以下命令安装 Flask 及相关工具库，不要使用虚拟环境外的全局安装，建议在项目目录下操作：
```bash
pip install flask werkzeug
```
2. 创建项目目录结构
在本地创建一个文件夹 archive_api，并在其中建立以下结构，确保存储目录权限可写：
archive_api/app.py (主程序文件)archive_api/storage/ (文件存储目录，需手动创建)
三、服务端核心代码实现
以下是完整的 app.py 代码。该代码实现了协议解析、MD5完整性校验、安全文件名处理以及目录自动归档功能。请直接复制覆盖到你的 app.py 文件中。
```python
import os
import json
import hashlib
import time
from flask import Flask, request, jsonify
from werkzeug.utils import secure_filename
app = Flask(__name__)
配置文件保存目录，请确保该目录已存在且具有写权限
UPLOAD_FOLDER = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'storage')
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
app.config['MAX_CONTENT_LENGTH'] = 1000 1024 1024 限制最大上传1000MB
def calculate_md5(file_path):
"""计算文件的MD5值"""
hash_md5 = hashlib.md5()
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
hash_md5.update(chunk)
return hash_md5.hexdigest()
@app.route('/api/v1/digitization/upload', methods=['POST'])
def upload_file():
1. 检查请求中是否包含文件
if 'file' not in request.files:
return jsonify({"code": 1002, "message": "请求中缺少文件部分", "data": None}), 400
file = request.files['file']
metadata_str = request.form.get('metadata')
2. 基础非空校验
if file.filename == '':
return jsonify({"code": 1003, "message": "未选择文件", "data": None}), 400
if not metadata_str:
return jsonify({"code": 1004, "message": "缺少metadata元数据", "data": None}), 400
try:
3. 解析JSON元数据
metadata = json.loads(metadata_str)
4. 协议关键字段校验
required_fields = ['batch_id', 'archive_id', 'checksum', 'timestamp']
for field in required_fields:
if field not in metadata:
return jsonify({"code": 1005, "message": f"协议字段缺失: {field}", "data": None}), 400
5. 时间戳校验（简单防重放，允许5分钟内的时间误差）
current_ts = int(time.time())
request_ts = int(metadata['timestamp'])
if abs(current_ts - request_ts) > 300:
return jsonify({"code": 1006, "message": "请求时间戳过期", "data": None}), 400
6. 构建存储路径：/storage/batch_id/archive_id/
使用secure_filename防止路径遍历攻击
safe_batch = secure_filename(metadata['batch_id'])
safe_archive = secure_filename(metadata['archive_id'])
batch_dir = os.path.join(app.config['UPLOAD_FOLDER'], safe_batch)
archive_dir = os.path.join(batch_dir, safe_archive)
if not os.path.exists(archive_dir):
os.makedirs(archive_dir)
7. 保存临时文件以进行MD5校验
temp_filename = secure_filename(file.filename)
save_path = os.path.join(archive_dir, temp_filename)
file.save(save_path)
8. MD5完整性校验
server_md5 = calculate_md5(save_path)
client_md5 = metadata['checksum']
if server_md5 != client_md5:
os.remove(save_path) 校验失败删除文件
return jsonify({"code": 1001, "message": f"MD5校验失败: 服务端{server_md5} != 客户端{client_md5}", "data": None}), 400
9. 返回成功响应
return jsonify({
"code": 200,
"message": "接收成功",
"data": {
"saved_path": save_path,
"server_md5": server_md5
}
}), 200
except json.JSONDecodeError:
return jsonify({"code": 1007, "message": "Metadata JSON格式错误", "data": None}), 400
except Exception as e:
return jsonify({"code": 1099, "message": f"服务器内部错误: {str(e)}", "data": None}), 500
if __name__ == '__main__':
启动服务，监听0.0.0.0允许外部访问
print(f"服务启动中... 文件存储目录: {UPLOAD_FOLDER}")
app.run(host='0.0.0.0', port=5000, debug=True)
```
四、客户端测试实操
服务端启动后，我们需要模拟客户端发送请求来验证协议。这里提供两种方式：使用 curl 命令（Linux/Mac/Windows Git Bash）和 Python 脚本。
1. 准备测试数据
在项目目录创建一个测试文件 test.txt，内容随意，例如：
```text
这是档案数字化测试文件内容。
```
2. 启动服务端
在终端执行：
```bash
python app.py
```
3. 使用 Python 发送测试请求
新建一个 test_client.py，复制以下代码。该脚本会自动计算 MD5 并组装符合协议的请求。
```python
import requests
import json
import hashlib
import time
服务端地址
url = "http://127.0.0.1:5000/api/v1/digitization/upload"
file_path = "test.txt"
1. 计算文件MD5
def get_file_md5(path):
with open(path, 'rb') as f:
return hashlib.md5(f.read()).hexdigest()
2. 构造Metadata元数据
metadata = {
"batch_id": "BATCH_2023_TEST",
"archive_id": "TEST-ARCHIVE-001",
"file_type": "ORIGINAL",
"operator": "Admin",
"checksum": get_file_md5(file_path),
"timestamp": int(time.time())
}
print(f"正在发送请求... Metadata: {json.dumps(metadata, indent=2)}")
3. 构造Multipart/Form-Data请求体
files = {
'file': open(file_path, 'rb')
}
data = {
'metadata': json.dumps(metadata)
}
4. 发送POST请求
try:
response = requests.post(url, files=files, data=data)
print(f"状态码: {response.status_code}")
print(f"响应内容: {json.dumps(response.json(), indent=2, ensure_ascii=False)}")
except Exception as e:
print(f"请求失败: {e}")
```
运行客户端脚本：
```bash
python test_client.py
```
4. 验证结果
如果一切正常，你应该看到客户端输出 "code": 200 和 "message": "接收成功"。同时，在服务端的 storage/BATCH_2023_TEST/TEST-ARCHIVE-001/ 目录下，应该能看到 test.txt 文件。
五、异常场景排查指南
在实操过程中，如果遇到报错，请对照以下方案进行排查：
1. 报错：[Errno 13] Permission denied
这通常是因为 storage 目录没有写入权限。
- Linux/Mac: 执行
chmod 777 storage 赋予权限。
- Windows: 检查文件夹属性，确保“只读”未勾选，并以管理员身份运行终端。
2. 报错：413 Request Entity Too Large
上传的文件超过了 Flask 默认限制。在 app.py 中修改配置：
```python
app.config['MAX_CONTENT_LENGTH'] = 100 1024 1024 修改为100MB或更大
```
3. 报错：MD5校验失败
这是最常见的协议错误。请检查：
- 客户端计算 MD5 时，文件流是否被读取过但没有重置指针（seek(0)）。
- 网络传输过程中是否发生了丢包或截断（通常发生在极不稳定的网络下）。
- 确保发送前文件没有被其他进程修改。
4. 报错：JSON格式错误
检查 metadata 字段是否被正确序列化为字符串。在使用 requests 库时，data 字典中的值如果是字典对象，需要手动 json.dumps()，否则 requests 会尝试将其转为表单格式，导致服务端解析失败。
