一、技术架构选型与环境初始化
为了解决档案管理中常见的存储冗余、检索困难及审计缺失问题，本指南采用Python Flask作为轻量级Web框架，配合SQLite数据库实现零配置部署。这套方案通过内容寻址存储优化存储空间，通过元数据索引优化检索效率，通过全链路审计优化合规性。
请在本地创建一个项目目录，并创建虚拟环境以隔离依赖。执行以下命令完成环境搭建：
```bash
mkdir archive_system
cd archive_system
python3 -m venv venv
source venv/bin/activate Windows下使用 venv\Scripts\activate
pip install flask flask-login werkzeug
```
创建项目基础目录结构，确保后续代码文件位置正确：
```bash
mkdir -p static/uploads
touch app.py models.py utils.py
```
二、数据库结构设计与性能优化
档案制度建设的核心在于数据的标准化。我们需要设计三张核心表：archives（档案元数据）、versions（版本控制）、audit_logs（审计日志）。在models.py中编写如下代码，这直接定义了档案的“身份证”和“履历表”。
```python
import sqlite3
import datetime
from flask import g
DATABASE = 'archive.db'
def get_db():
db = getattr(g, '_database', None)
if db is None:
db = g._database = sqlite3.connect(DATABASE)
db.row_factory = sqlite3.Row
return db
def init_db():
db = get_db()
1. 档案主表：存储核心元数据，建立索引优化检索
db.execute('''
CREATE TABLE IF NOT EXISTS archives (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
category TEXT NOT NULL,
file_hash TEXT UNIQUE,
current_path TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
uploader TEXT
)
''')
优化：为高频查询字段创建索引
db.execute('CREATE INDEX IF NOT EXISTS idx_category ON archives(category)')
db.execute('CREATE INDEX IF NOT EXISTS idx_file_hash ON archives(file_hash)')
2. 版本表：实现版本控制，优化历史回溯能力
db.execute('''
CREATE TABLE IF NOT EXISTS versions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
archive_id INTEGER NOT NULL,
version_number INTEGER,
file_path TEXT,
comment TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY(archive_id) REFERENCES archives(id)
)
''')
3. 审计日志表：记录所有敏感操作，优化合规性
db.execute('''
CREATE TABLE IF NOT EXISTS audit_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT,
action TEXT NOT NULL,
resource_id TEXT,
details TEXT,
ip_address TEXT,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
db.commit()
```
三、核心功能代码实现
在app.py中实现业务逻辑。我们将重点实现文件去重上传（存储优化）和元数据强制录入（制度优化）。请确保utils.py中包含哈希计算工具，或在app.py头部直接定义：
```python
import os
import hashlib
from flask import Flask, request, jsonify, render_template_string
from models import init_db, get_db
app = Flask(__name__)
app.config['UPLOAD_FOLDER'] = './static/uploads'
app.config['MAX_CONTENT_LENGTH'] = 16 1024 1024 限制16MB，优化存储压力
辅助函数：计算文件SHA256值，用于去重
def calculate_file_hash(file_path):
sha256_hash = hashlib.sha256()
with open(file_path, "rb") as f:
for byte_block in iter(lambda: f.read(4096), b""):
sha256_hash.update(byte_block)
return sha256_hash.hexdigest()
@app.route('/upload', methods=['POST'])
def upload_file():
1. 强制校验元数据：落实档案制度，分类必填
if 'file' not in request.files:
return jsonify({'error': '无文件部分'}), 400
file = request.files['file']
title = request.form.get('title')
category = request.form.get('category')
if not title or not category:
return jsonify({'error': '档案标题与分类为必填项，符合归档制度'}), 400
if file.filename == '':
return jsonify({'error': '未选择文件'}), 400
2. 保存临时文件并计算哈希
filename = file.filename
save_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
file.save(save_path)
file_hash = calculate_file_hash(save_path)
db = get_db()
3. 存储优化：检查哈希是否存在，实现秒传或去重
existing = db.execute('SELECT id, current_path FROM archives WHERE file_hash = ?', (file_hash,)).fetchone()
if existing:
如果文件已存在，仅记录审计日志，不重复存储
db.execute('INSERT INTO audit_logs (user_id, action, resource_id, details) VALUES (?, ?, ?, ?)',
('system', 'upload_duplicate', str(existing['id']), f'尝试上传重复文件: {filename}'))
db.commit()
os.remove(save_path) 删除刚上传的重复文件
return jsonify({'message': '文件已存在，归档完成（去重优化）', 'archive_id': existing['id']})
4. 插入新档案记录
cursor = db.execute('INSERT INTO archives (title, category, file_hash, current_path, uploader) VALUES (?, ?, ?, ?, ?)',
(title, category, file_hash, save_path, 'admin'))
archive_id = cursor.lastrowid
5. 初始化版本记录
db.execute('INSERT INTO versions (archive_id, version_number, file_path, comment) VALUES (?, ?, ?, ?)',
(archive_id, 1, save_path, '初始版本'))
6. 记录审计日志
db.execute('INSERT INTO audit_logs (user_id, action, resource_id, details) VALUES (?, ?, ?, ?)',
('admin', 'create_archive', str(archive_id), f'创建档案: {title}'))
db.commit()
return jsonify({'message': '归档成功', 'archive_id': archive_id})
if __name__ == '__main__':
with app.app_context():
init_db()
app.run(debug=True, port=5000)
```
四、审计日志与合规性落地
为了满足档案制度对安全性的要求，必须增加一个审计查询接口。这能优化“谁在什么时候动了什么文件”的追踪难题。在app.py中追加以下代码：
```python
@app.route('/audit/logs', methods=['GET'])
def get_audit_logs():
db = get_db()
优化：分页查询，防止数据量过大导致性能崩溃
page = request.args.get('page', 1, type=int)
per_page = 20
offset = (page - 1) per_page
logs = db.execute('''
SELECT FROM audit_logs
ORDER BY timestamp DESC
LIMIT ? OFFSET ?
''', (per_page, offset)).fetchall()
log_list = [dict(row) for row in logs]
return jsonify({'logs': log_list, 'page': page})
```
五、一键部署与运行
为了实现零门槛落地，我们编写一个简单的HTML前端用于测试，并整合启动命令。在app.py中添加一个简单的上传表单路由：
```python
@app.route('/')
def index():
return render_template_string('''
档案系统上传
档案归档上传
查看审计日志
点击查看最新操作记录
''')
```
现在，所有代码已就绪。执行以下步骤启动系统：
- 初始化数据库：首次运行会自动创建
archive.db及表结构。
- 启动服务：在终端执行启动命令。
```bash
python app.py
```
启动后，打开浏览器访问 http://127.0.0.1:5000。你将看到一个上传界面。
实操验证步骤：
- 上传一个文件，标题填“测试合同”，分类填“法务”。
- 再次上传同一个文件，标题填“重复测试”。
- 观察返回结果，系统应提示“文件已存在，归档完成（去重优化）”，且本地
static/uploads目录下不会生成重复文件。
- 访问
http://127.0.0.1:5000/audit/logs 查看JSON格式的操作记录，确认每一次上传尝试都被记录。
通过以上代码，我们直接落地了档案制度中的三项核心优化：存储去重（节省成本）、元数据强制结构化（规范管理）、操作留痕（安全合规）。无需任何复杂配置，即刻拥有一个可用的档案管理系统原型。
