用Python Flask+SQLite快速搭建标准格式电子档案管理系统接收模块附完整校验代码和对接配置手册

一、准备工作

1. 环境搭建

确保电脑已安装Python 3.8+，未安装的直接访问https://www.python.org/downloads/，下载对应系统的安装包，勾选“Add Python to PATH”后一路默认安装。

打开终端（Windows用CMD/PowerShell，Mac/Linux用Terminal），输入以下命令安装依赖：

```bash pip install flask==2.3.3 werkzeug==2.3.7 python-magic-bin==0.4.14 ```

注：python-magic-bin是跨平台文件格式识别库，Windows可直接用bin版，Mac/Linux需先安装libmagic（Mac用brew install libmagic，Linux用sudo apt-get install libmagic1）再换pip install python-magic。

2. 项目目录创建

在桌面新建文件夹archive_receiver，内部创建以下结构：

• archive_receiver/
• static/
• uploads/ （用于存放接收到的电子档案）
• templates/ （暂时不用，留空即可）
• app.py （核心接收代码）
• config.py （对接配置）
• archive.db （启动后自动生成的SQLite数据库）

二、编写核心代码

1. 配置文件config.py

所有参数可直接复制使用，按需修改的地方已标注：

```python import os 项目根路径 BASE_DIR = os.path.abspath(os.path.dirname(__file__)) 允许的电子档案标准格式（可按需扩展，如JPG、PNG需同步添加标准图片类） ALLOWED_EXTENSIONS = {'pdf', 'docx', 'xlsx', 'txt', 'zip', 'rar', 'tar.gz'} 允许的文件大小（单位：字节，此处设为1GB） MAX_CONTENT_LENGTH = 1 1024 1024 1024 上传文件存储路径 UPLOAD_FOLDER = os.path.join(BASE_DIR, 'static', 'uploads') Flask密钥（用于签名请求，自行生成更安全，可通过python -c "import secrets; print(secrets.token_hex(16))"生成） SECRET_KEY = 'your_secure_secret_key_here_change_it' 接收接口的鉴权令牌（前端/调用方必须携带，防止恶意上传） AUTH_TOKEN = 'your_archive_auth_token_here_change_it' 档案接收后保存的命名规则（格式：yyyyMMdd_HHmmss_原文件名） ```

2. 核心接收与校验代码app.py

包含鉴权、格式校验、大小校验、文件名规范、数据库记录、文件存储功能，可直接复制：

```python from flask import Flask, request, jsonify from werkzeug.utils import secure_filename import magic import os import time import sqlite3 from datetime import datetime from config import app = Flask(__name__) app.config.from_pyfile('config.py') 初始化SQLite数据库 def init_db(): conn = sqlite3.connect('archive.db') c = conn.cursor() 创建档案接收记录表 c.execute('''CREATE TABLE IF NOT EXISTS archive_received (id INTEGER PRIMARY KEY AUTOINCREMENT, archive_name TEXT NOT NULL, original_name TEXT NOT NULL, file_path TEXT NOT NULL, file_size INTEGER NOT NULL, file_mime TEXT NOT NULL, receive_time TEXT NOT NULL, uploader_info TEXT, status INTEGER DEFAULT 0)''') conn.commit() conn.close() 检查文件扩展名是否合法 def allowed_file(filename): return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS 检查文件实际MIME类型（防止改扩展名上传） def allowed_mime(file_path): mime = magic.from_file(file_path, mime=True) allowed_mimes = { 'pdf': 'application/pdf', 'docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document', 'xlsx': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', 'txt': 'text/plain', 'zip': 'application/zip', 'rar': 'application/x-rar-compressed', 'tar.gz': 'application/gzip' } ext = os.path.splitext(file_path)[1].lower() if ext == '.gz' and os.path.splitext(os.path.splitext(file_path)[0])[1].lower() == '.tar': ext = '.tar.gz' return mime == allowed_mimes.get(ext.lstrip('.'), '') @app.route('/api/archive/receive', methods=['POST']) def receive_archive(): 鉴权校验 auth_header = request.headers.get('Authorization') if not auth_header or auth_header != f'Bearer {AUTH_TOKEN}': return jsonify({'code': 401, 'msg': '鉴权失败，请检查令牌'}), 401 检查请求中是否有文件 if 'file' not in request.files: return jsonify({'code': 400, 'msg': '请求中未包含档案文件'}), 400 file = request.files['file'] 检查文件名是否为空 if file.filename == '': return jsonify({'code': 400, 'msg': '未选择档案文件'}), 400 检查扩展名 if not allowed_file(file.filename): return jsonify({'code': 400, 'msg': f'不支持的文件格式，仅允许：{",".join(ALLOWED_EXTENSIONS)}'}), 400 生成安全文件名和保存路径 timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') safe_original_name = secure_filename(file.filename) archive_name = f'{timestamp}_{safe_original_name}' temp_path = os.path.join(app.config['UPLOAD_FOLDER'], f'temp_{safe_original_name}') final_path = os.path.join(app.config['UPLOAD_FOLDER'], archive_name) 保存临时文件 file.save(temp_path) 检查实际MIME类型 if not allowed_mime(temp_path): os.remove(temp_path) return jsonify({'code': 400, 'msg': '文件实际格式与扩展名不符'}), 400 检查文件大小（Werkzeug虽有全局限制，但二次校验更稳妥） file_size = os.path.getsize(temp_path) if file_size > app.config['MAX_CONTENT_LENGTH']: os.remove(temp_path) return jsonify({'code': 400, 'msg': f'文件大小超过限制（最大1GB）'}), 400 移动临时文件到最终路径 os.rename(temp_path, final_path) 读取上传方信息（可选，通过请求参数传递） uploader_info = request.form.get('uploader_info', '未指定') receive_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') 写入数据库 conn = sqlite3.connect('archive.db') c = conn.cursor() c.execute('''INSERT INTO archive_received (archive_name, original_name, file_path, file_size, file_mime, receive_time, uploader_info, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?)''', (archive_name, safe_original_name, final_path, file_size, magic.from_file(final_path, mime=True), receive_time, uploader_info, 1)) conn.commit() conn.close() return jsonify({'code': 200, 'msg': '档案接收成功', 'data': {'archive_id': c.lastrowid, 'archive_name': archive_name}}) if __name__ == '__main__': init_db() 启动服务，host设为0.0.0.0允许局域网访问，port设为5001避免冲突 app.run(host='0.0.0.0', port=5001, debug=False) ```

三、启动与测试

1. 启动服务

打开终端，进入archive_receiver目录：

```bash cd 桌面/archive_receiver ```

输入以下命令启动接收服务：

```bash python app.py ```

2. 本地接口测试

使用Postman或curl命令测试，以curl为例（Windows需安装Git Bash或替换curl为PowerShell的Invoke-RestMethod）：

```bash curl -X POST http://127.0.0.1:5001/api/archive/receive \ -H "Authorization: Bearer your_archive_auth_token_here_change_it" \ -F "file=@/Users/YourName/Desktop/test.pdf" \ -F "uploader_info=测试人员张三" ```

注：将路径替换为你本地的真实文件路径，将令牌替换为config.py中设置的AUTH_TOKEN。

3. 验证结果

• 检查static/uploads/目录下是否有生成的规范命名档案文件
• 用SQLite可视化工具（如DB Browser for SQLite，下载地址：https://sqlitebrowser.org/dl/）打开archive.db，查看archive_received表是否有对应的记录

四、常见对接说明

1. 局域网/公网调用

局域网调用只需将请求地址中的127.0.0.1替换为运行服务的电脑IP；公网调用需配置路由器端口映射（将5001端口映射到运行服务的电脑IP的5001端口），或使用内网穿透工具（如花生壳免费版、ngrok）。

2. 前端对接示例（简单HTML+JS）

在templates/目录下新建upload.html：

```html

上传档案

```

然后在app.py中添加访问路由：

```python from flask import render_template @app.route('/') def index(): return render_template('upload.html') ```

重启服务后访问http://127.0.0.1:5001/即可使用。