外资企业数字档案馆系统搭建与安全合规实操

一、系统架构与技术选型

针对外资企业对数据主权、合规性及高并发的严苛要求，本系统采用微服务架构的简化版进行落地。核心组件包括对象存储、关系型数据库、后端API服务及前端交互界面。技术栈选型如下：

• 对象存储：MinIO（开源高性能存储，兼容S3 API，适合私有化部署）
• 数据库：PostgreSQL 15（支持JSONB类型，ACID事务强一致性）
• 后端框架：Python 3.9 + FastAPI（异步高并发，自动生成OpenAPI文档）
• 前端框架：Vue 3 + Element Plus（组件化开发，响应式布局）
• 容器化：Docker & Docker Compose（一键编排，环境隔离）

二、环境准备与目录结构

在开始编码前，需确保服务器已安装Docker环境。执行以下命令安装Docker及Docker Compose：

curl -fsSL https://get.docker.com | sh
sudo usermod -aG docker $USER
sudo curl -L "https://github.com/docker/compose/releases/latest/download/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose
sudo chmod +x /usr/local/bin/docker-compose

创建项目根目录digital_archive，并按以下结构建立文件夹：

digital_archive/
├── backend/
│   ├── main.py
│   ├── models.py
│   ├── database.py
│   └── requirements.txt
├── frontend/
│   ├── src/
│   ├── package.json
│   └── vite.config.js
└── docker-compose.yml

三、基础设施容器编排

在根目录下创建docker-compose.yml文件。该配置定义了PostgreSQL和MinIO服务，是系统的数据底座。请直接复制以下完整配置：

version: '3.8'
services:
db:
image: postgres:15-alpine
container_name: archive_db
environment:
POSTGRES_USER: archive_admin
POSTGRES_PASSWORD: SecurePassword2024!
POSTGRES_DB: digital_archive
volumes:
- postgres_data:/var/lib/postgresql/data
ports:
- "5432:5432"
healthcheck:
test: ["CMD-SHELL", "pg_isready -U archive_admin"]
interval: 10s
timeout: 5s
retries: 5
minio:
image: minio/minio:latest
container_name: archive_minio
command: server /data --console-address ":9001"
environment:
MINIO_ROOT_USER: minioadmin
MINIO_ROOT_PASSWORD: MinioAdmin2024!
volumes:
- minio_data:/data
ports:
- "9000:9000"
- "9001:9001"
volumes:
postgres_data:
minio_data:

执行docker-compose up -d启动基础服务。使用docker ps确认容器状态均为Up。

四、后端核心服务开发

进入backend目录，首先创建requirements.txt：

fastapi==0.104.1
uvicorn[standard]==0.24.0
sqlalchemy==2.0.23
psycopg2-binary==2.9.9
minio==7.2.0
python-multipart==0.0.6
pydantic==2.5.0
pydantic-settings==2.1.0

1. 数据库模型定义

创建database.py，配置数据库连接池。外资企业系统需确保连接池参数合理，防止连接泄露：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
使用docker-compose中的服务名作为主机名
SQLALCHEMY_DATABASE_URL = "postgresql://archive_admin:SecurePassword2024!@db:5432/digital_archive"
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
pool_size=20,
max_overflow=10,
pool_pre_ping=True,
echo=False
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

创建models.py，定义文件档案元数据表及审计日志表。合规性要求必须记录完整的操作链路：

from sqlalchemy import Column, Integer, String, DateTime, Text, ForeignKey
from sqlalchemy.sql import func
from database import Base
class ArchiveFile(Base):
__tablename__ = "archive_files"
id = Column(Integer, primary_key=True, index=True)
filename = Column(String(255), nullable=False)
file_type = Column(String(50))
file_size = Column(Integer)
storage_path = Column(String(500), unique=True)  MinIO object name
department = Column(String(100))  归属部门
uploader = Column(String(100))
upload_time = Column(DateTime(timezone=True), server_default=func.now())
status = Column(String(20), default="active")  active, archived, deleted
class AuditLog(Base):
__tablename__ = "audit_logs"
id = Column(Integer, primary_key=True, index=True)
action_type = Column(String(50))  UPLOAD, DOWNLOAD, DELETE
target_file_id = Column(Integer, nullable=True)
operator = Column(String(100))
operator_ip = Column(String(50))
timestamp = Column(DateTime(timezone=True), server_default=func.now())
details = Column(Text)

2. 业务逻辑实现

创建main.py，实现文件上传、下载及元数据管理。此处需处理MinIO的连接及S3协议交互：

from fastapi import FastAPI, UploadFile, File, Form, Depends, HTTPException, Header
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.orm import Session
from minio import Minio
from minio.error import S3Error
import io
import uuid
from datetime import datetime
from database import SessionLocal, engine, Base
from models import ArchiveFile, AuditLog
初始化数据库表
Base.metadata.create_all(bind=engine)
app = FastAPI(title="Foreign Enterprise Digital Archive API")
配置CORS，允许前端跨域访问
app.add_middleware(
CORSMiddleware,
allow_origins=[""],
allow_credentials=True,
allow_methods=[""],
allow_headers=[""],
)
MinIO 客户端初始化
minio_client = Minio(
"minio:9000",
access_key="minioadmin",
secret_key="MinioAdmin2024!",
secure=False
)
BUCKET_NAME = "enterprise-archives"
启动时检查Bucket是否存在，不存在则创建
try:
if not minio_client.bucket_exists(BUCKET_NAME):
minio_client.make_bucket(BUCKET_NAME)
except S3Error as e:
print(f"MinIO Error: {e}")
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.post("/upload/")
async def upload_file(
file: UploadFile = File(...),
department: str = Form(...),
uploader: str = Form(...),
x_real_ip: str = | Header(None),
db: Session = Depends(get_db)
):
1. 生成唯一文件名防止覆盖
file_ext = file.filename.split(".")[-1]
unique_name = f"{uuid.uuid4()}.{file_ext}"
object_name = f"{datetime.now().strftime('%Y/%m/%d')}/{unique_name}"
2. 读取文件内容并上传至MinIO
file_content = await file.read()
try:
minio_client.put_object(
BUCKET_NAME,
object_name,
io.BytesIO(file_content),
length=len(file_content),
content_type=file.content_type
)
except S3Error as e:
raise HTTPException(status_code=500, detail=f"Storage Error: {e}")
3. 写入数据库元数据
db_file = ArchiveFile(
filename=file.filename,
file_type=file_ext,
file_size=len(file_content),
storage_path=object_name,
department=department,
uploader=uploader
)
db.add(db_file)
4. 记录审计日志
audit = AuditLog(
action_type="UPLOAD",
target_file_id=db_file.id,
operator=uploader,
operator_ip=x_real_ip or "unknown",
details=f"Uploaded {file.filename}"
)
db.add(audit)
db.commit()
db.refresh(db_file)
return {"message": "Upload successful", "file_id": db_file.id, "path": object_name}
@app.get("/files/")
def list_files(department: str = None, db: Session = Depends(get_db)):
query = db.query(ArchiveFile).filter(ArchiveFile.status == "active")
if department:
query = query.filter(ArchiveFile.department == department)
return query.all()
@app.get("/download/{file_id}")
def download_file(file_id: int, db: Session = Depends(get_db)):
db_file = db.query(ArchiveFile).filter(ArchiveFile.id == file_id).first()
if not db_file:
raise HTTPException(status_code=404, detail="File not found")
try:
获取文件临时下载URL（有效期1小时）
presigned_url = minio_client.presigned_get_object(BUCKET_NAME, db_file.storage_path)
return {"download_url": presigned_url}
except S3Error:
raise HTTPException(status_code=500, detail="Download generation failed")

五、前端交互界面开发

进入frontend目录，初始化Vue项目。为了确保零门槛，直接使用CDN方式引入Vue和Element Plus进行单文件演示，实际生产中请使用npm init vue@latest。此处创建一个index.html作为完整的前端入口：

外资企业数字档案馆系统
Secure Digital Archive for Foreign Enterprises















点击选择文件



确认上传归档




已归档文件列表

将此文件保存为frontend/index.html。可以使用Python内置服务器快速启动前端进行测试：cd frontend && python3 -m http.server 8080。

六、安全合规与审计机制

外资企业系统必须具备完善的审计追踪能力。我们在上述代码中已经埋入了审计点，以下是补充的关键安全配置：

1. 启用HTTPS传输加密

在Docker