一、环境准备与依赖安装
在开始构建档案系统之前，需要确保本地环境已安装Python 3.8及以上版本和Docker。我们将使用FastAPI作为后端框架，MinIO作为对象存储服务，SQLite作为元数据库。创建项目目录并安装Python依赖库。
1. 创建项目目录
在终端执行以下命令创建项目文件夹：
```bash
mkdir archive_system
cd archive_system
```
2. 安装Python依赖包
创建一个名为requirements.txt的文件，并填入以下具体依赖版本，以确保环境稳定性：
```text
fastapi==0.104.1
uvicorn==0.24.0
minio==7.2.0
python-multipart==0.0.6
```
保存文件后，在终端执行安装命令：
```bash
pip install -r requirements.txt
```
二、部署MinIO对象存储服务
档案系统的核心是文件的可靠存储与读取。使用Docker部署MinIO是最快的方式，它能提供与Amazon S3兼容的API接口。执行以下命令启动MinIO服务：
```bash
docker run -d \
-p 9000:9000 \
-p 9001:9001 \
--name minio-arch \
-e "MINIO_ROOT_USER=admin" \
-e "MINIO_ROOT_PASSWORD=Admin@123" \
quay.io/minio/minio server /data --console-address ":9001"
```
配置说明：
- -p 9000:9000：映射API端口，后端将通过此端口上传下载文件。
- -p 9001:9001：映射Web管理控制台端口，方便可视化查看文件。
- MINIO_ROOT_USER/PASSWORD：设置初始管理员账号密码，请妥善保管。
服务启动后，打开浏览器访问 http://127.0.0.1:9001，使用上述账号密码登录。登录后手动创建一个名为 archive-bucket 的存储桶，并设置访问权限为 Public（或通过代码自动创建，本指南将在代码中实现自动检查）。
三、后端核心代码开发
在项目根目录下创建main.py文件。该文件将包含数据库初始化、MinIO客户端配置、文件上传、下载及列表查询的核心逻辑。
1. 完整代码实现
直接复制以下代码到main.py中，代码已包含详细的逻辑注释，实现了MD5校验、元数据存储等档案系统必备功能。
```python
import os
import shutil
import sqlite3
import hashlib
import uuid
from datetime import datetime
from fastapi import FastAPI, UploadFile, File, HTTPException, Form
from fastapi.responses import FileResponse, StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from minio import Minio
from minio.error import S3Error
配置常量
MINIO_ENDPOINT = "127.0.0.1:9000"
ACCESS_KEY = "admin"
SECRET_KEY = "Admin@123"
BUCKET_NAME = "archive-bucket"
DB_FILE = "archive.db"
初始化FastAPI应用
app = FastAPI(title="Enterprise Archive System")
添加CORS中间件，允许前端跨域访问
app.add_middleware(
CORSMiddleware,
allow_origins=[""],
allow_credentials=True,
allow_methods=[""],
allow_headers=[""],
)
初始化MinIO客户端
minio_client = Minio(
MINIO_ENDPOINT,
access_key=ACCESS_KEY,
secret_key=SECRET_KEY,
secure=False
)
初始化SQLite数据库
def init_db():
conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS files (
id INTEGER PRIMARY KEY AUTOINCREMENT,
file_id TEXT UNIQUE NOT NULL,
original_name TEXT NOT NULL,
object_name TEXT NOT NULL,
file_size INTEGER,
content_type TEXT,
md5_hash TEXT,
upload_time TEXT,
uploader TEXT
)
''')
conn.commit()
conn.close()
确保存储桶存在
def ensure_bucket():
try:
if not minio_client.bucket_exists(BUCKET_NAME):
minio_client.make_bucket(BUCKET_NAME)
print(f"Bucket '{BUCKET_NAME}' created.")
except S3Error as exc:
print(f"Error occurred: {exc}")
启动时初始化
@app.on_event("startup")
def startup_event():
init_db()
ensure_bucket()
辅助函数：计算文件MD5
def calculate_md5(file_path):
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.post("/upload/")
async def upload_file(
file: UploadFile = File(...),
uploader: str = Form("anonymous")
):
生成唯一文件名和档案ID
file_id = str(uuid.uuid4())
file_extension = os.path.splitext(file.filename)[1]
object_name = f"{file_id}{file_extension}"
临时保存上传的文件以计算MD5和上传到MinIO
temp_file_path = f"temp_{object_name}"
try:
写入临时文件
with open(temp_file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
计算MD5
md5_hash = calculate_md5(temp_file_path)
上传到MinIO
minio_client.fput_object(
BUCKET_NAME,
object_name,
temp_file_path,
content_type=file.content_type
)
获取文件大小
file_size = os.path.getsize(temp_file_path)
写入元数据到SQLite
conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()
upload_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
cursor.execute('''
INSERT INTO files (file_id, original_name, object_name, file_size, content_type, md5_hash, upload_time, uploader)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
''', (file_id, file.filename, object_name, file_size, file.content_type, md5_hash, upload_time, uploader))
conn.commit()
conn.close()
return {"status": "success", "file_id": file_id, "message": "档案上传成功"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
finally:
清理临时文件
if os.path.exists(temp_file_path):
`os.remove(temp_file_path)`
接口：获取档案列表
@app.get("/files/")
async def list_files():
conn = sqlite3.connect(DB_FILE)
conn.row_factory = sqlite3.Row 允许通过列名访问
cursor = conn.cursor()
cursor.execute("SELECT FROM files ORDER BY upload_time DESC")
rows = cursor.fetchall()
conn.close()
return [dict(row) for row in rows]
接口：下载档案
@app.get("/download/{file_id}")
async def download_file(file_id: str):
conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()
cursor.execute("SELECT object_name, original_name FROM files WHERE file_id=?", (file_id,))
result = cursor.fetchone()
conn.close()
if not result:
raise HTTPException(status_code=404, detail="档案未找到")
object_name, original_name = result
try:
从MinIO获取文件流
data = minio_client.get_object(BUCKET_NAME, object_name)
return StreamingResponse(data, media_type="application/octet-stream", headers={"Content-Disposition": f"attachment; filename={original_name}"})
except S3Error as exc:
raise HTTPException(status_code=500, detail=str(exc))
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
```
四、前端交互页面开发
为了快速验证功能，我们在项目根目录创建一个简单的index.html。这个页面包含文件上传表单和档案列表展示，使用原生JavaScript调用后端接口。
创建index.html文件并填入以下代码：
```html
企业档案管理系统
```
五、系统启动与功能验证
所有代码已准备就绪，现在启动服务并进行全流程测试。
1. 启动后端服务
在archive_system目录下打开终端，执行：
```bash
python main.py
```
终端显示Uvicorn running on http://0.0.0.0:8000即表示后端启动成功。此时程序会自动创建SQLite数据库文件和MinIO存储桶（如果不存在）。
2. 打开前端页面
直接双击打开项目目录下的index.html文件，或者将其拖入浏览器中。
3. 执行上传测试
- 在“上传者姓名”栏输入测试人名，例如“TechLead”。
- 点击“选择文件”，选取一个本地的测试文档（如PDF或图片）。
- 点击“上传文件”按钮。
- 观察浏览器弹窗提示“上传成功”，并记录下返回的File ID。
4. 验证数据完整性
- 查看页面下方的表格，新上传的文件应立即出现在列表中。
- 检查表格中的MD5校验列，系统已自动计算并存储了文件的哈希值。
- 点击表格右侧的“下载”链接，浏览器应开始下载文件，对比下载文件与源文件，确保内容一致。
5. 检查持久化存储
- 访问 MinIO 控制台
http://127.0.0.1:9001，进入archive-bucket，可以看到刚才上传的物理文件。
- 在项目目录下会生成一个
archive.db文件，使用SQLite查看工具打开，执行SELECT FROM files;，可以看到完整的元数据记录。
至此，一个具备文件上传、下载、列表查询、MD5完整性校验及元数据管理功能的企业级档案系统核心功能已完全落地。该架构可通过更换MinIO为AWS S3或更换SQLite为PostgreSQL轻松扩展至生产环境。
