手把手教你开发符合新档案法的电子档案系统

系统架构与合规性核心点解析

根据《中华人民共和国档案法》第三十七条及国家档案局发布的《电子档案单套制管理规范》，开发合规系统的核心在于确保电子档案的“四性”（真实性、完整性、可用性、安全性）以及元数据的齐全。本指南将使用 Python + FastAPI + SQLAlchemy 技术栈，从零构建一个满足法律要求的后端核心模块。该架构将包含元数据管理、电子签名校验、格式转换及审计日志四大模块。

开发环境初始化与依赖安装

我们需要准备一个干净的 Python 虚拟环境并安装必要的依赖库。请确保你的系统已安装 Python 3.9 或更高版本。

执行以下命令创建项目目录并安装核心依赖：

 创建项目目录
mkdir compliance_archive_system
cd compliance_archive_system
创建虚拟环境
python -m venv venv
source venv/bin/activate   Windows下使用 venv\Scripts\activate
安装核心依赖：FastAPI(框架)、SQLAlchemy(ORM)、Cryptography(签名)、Aiofiles(异步文件操作)
pip install fastapi uvicorn sqlalchemy cryptography aiofiles python-multipart

安装完成后，创建项目的基础文件结构。在根目录下创建 main.py（入口文件）、models.py（数据库模型）、utils.py（工具类）和 config.py（配置文件）。

构建符合DA/T 46标准的元数据模型

《电子档案管理规范》要求元数据必须包含档号、题名、责任者、日期等核心字段。我们需要设计一个严格映射 DA/T 46-2009 标准的数据库表结构。以下是 models.py 的完整代码，定义了电子档案表和审计日志表：

 models.py
from sqlalchemy import Column, Integer, String, DateTime, Text, LargeBinary
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
Base = declarative_base()
class ElectronicArchive(Base):
__tablename__ = 'electronic_archives'
id = Column(Integer, primary_key=True, index=True)
档号：唯一标识，符合档号编制规则
archive_code = Column(String(100), unique=True, nullable=False, index=True)
题名
title = Column(String(255), nullable=False)
责任者
creator = Column(String(100), nullable=False)
文件存储路径
file_path = Column(String(500), nullable=False)
文件哈希值（SHA-256），用于验证真实性
file_hash = Column(String(64), nullable=False)
电子签名值，确保法律效力
digital_signature = Column(LargeBinary, nullable=True)
元数据XML，存储扩展的DA/T 46元数据
metadata_xml = Column(Text, nullable=True)
创建时间
created_at = Column(DateTime, default=datetime.utcnow)
归档状态
status = Column(String(20), default="pending")
class AuditLog(Base):
__tablename__ = 'audit_logs'
id = Column(Integer, primary_key=True, index=True)
user_id = Column(String(50))
action = Column(String(50))  例如：upload, sign, convert, delete
resource_id = Column(Integer)  关联的档案ID
ip_address = Column(String(45))
details = Column(Text)
timestamp = Column(DateTime, default=datetime.utcnow)

实现电子文件“四性”检测核心逻辑

真实性是《档案法》对电子档案最核心的要求。我们将使用非对称加密（RSA）和哈希算法（SHA-256）来实现文件的防篡改检测。在 utils.py 中添加以下代码：

 utils.py
import hashlib
import os
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa, padding
生成RSA密钥对（仅演示用，生产环境应从HSM或安全配置文件加载）
def generate_keys():
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
public_key = private_key.public_key()
return private_key, public_key
计算文件SHA-256哈希值（真实性校验基础）
def calculate_file_hash(file_path: str) -> str:
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()
对文件进行数字签名
def sign_file(private_key, file_path: str) -> bytes:
file_hash = calculate_file_hash(file_path)
注意：实际应用中应对哈希值签名，而非整个文件，此处演示逻辑
signature = private_key.sign(
file_hash.encode(),
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH
),
hashes.SHA256()
)
return signature
验证签名
def verify_signature(public_key, file_path: str, signature: bytes) -> bool:
file_hash = calculate_file_hash(file_path)
try:
public_key.verify(
signature,
file_hash.encode(),
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH
),
hashes.SHA256()
)
return True
except Exception:
return False

上述代码通过 calculate_file_hash 生成文件的数字指纹，配合 sign_file 生成不可抵赖的数字签名，直接对应《档案法》中关于电子档案真实性的技术要求。

长期保存格式自动化转换实现

为确保长期可用性，系统必须将上传的文档转换为版式文档（如 PDF/A 或 OFD）。我们需要利用 LibreOffice 提供的命令行工具进行转换。确保服务器安装了 LibreOffice：

 Ubuntu/Debian 安装命令
sudo apt-get update
sudo apt-get install libreoffice
CentOS/RHEL 安装命令
sudo yum install libreoffice

在 utils.py 中增加格式转换函数。该函数使用 subprocess 调用 LibreOffice 的 headless 模式进行转换：

 utils.py 续
import subprocess
import shutil
def convert_to_pdfa(input_path: str, output_dir: str) -> str:
"""
将上传文件转换为 PDF/A 格式以满足长期保存要求
"""
filename = os.path.basename(input_path)
name_only = os.path.splitext(filename)[0]
LibreOffice 转换命令
--headless: 无界面模式
--convert-to pdf:writer_pdf_Export: 指定导出过滤器
try:
cmd = [
"soffice",
"--headless",
"--convert-to", "pdf:writer_pdf_Export",
"--outdir", output_dir,
input_path
]
process = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
if process.returncode == 0:
converted_file = os.path.join(output_dir, f"{name_only}.pdf")
return converted_file
else:
raise Exception(f"Conversion failed: {process.stderr.decode()}")
except FileNotFoundError:
raise Exception("LibreOffice not found. Please install it to enable format conversion.")

构建不可篡改的审计日志系统

《档案法》要求对档案管理全过程进行记录，且日志本身不可被轻易修改。我们在 main.py 中集成数据库连接和日志记录功能。首先配置数据库连接字符串：

 config.py
import os
使用 SQLite 作为演示，生产环境请替换为 PostgreSQL 或 MySQL
DATABASE_URL = "sqlite:///./archive_system.db"
SECRET_KEY = "your-secret-key-for-jwt"   用于后续鉴权扩展
UPLOAD_DIR = "uploads"
ARCHIVE_DIR = "archives"
确保目录存在
os.makedirs(UPLOAD_DIR, exist_ok=True)
os.makedirs(ARCHIVE_DIR, exist_ok=True)

接下来编写完整的 API 接口，整合上传、转换、签名和归档流程。

档案归档全流程API接口实战

以下是 main.py 的完整代码。它包含了一个 FastAPI 应用，提供了 /upload 接口，该接口串联了文件接收、格式转换、哈希计算、数字签名及数据库落地的全过程，确保每一步都符合法规要求。

 main.py
from fastapi import FastAPI, UploadFile, File, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy import create_engine, Session
from sqlalchemy.orm import sessionmaker, Session
from models import Base, ElectronicArchive, AuditLog
from utils import generate_keys, calculate_file_hash, sign_file, convert_to_pdfa
from config import DATABASE_URL, UPLOAD_DIR, ARCHIVE_DIR
import shutil
import uuid
import datetime
初始化数据库
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base.metadata.create_all(bind=engine)
生成密钥对（全局单例，实际应用需妥善保管）
private_key, public_key = generate_keys()
app = FastAPI(title="Compliant Archive System")
app.add_middleware(
CORSMiddleware,
allow_origins=[""],
allow_methods=[""],
allow_headers=[""],
)
依赖项：获取数据库会话
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
辅助函数：写入审计日志
def log_audit(db: Session, action: str, resource_id: int, user: str, details: str):
log_entry = AuditLog(
user_id=user,
action=action,
resource_id=resource_id,
ip_address="127.0.0.1",  实际应从 request 中获取
details=details
)
db.add(log_entry)
db.commit()
@app.post("/api/archive/upload")
async def upload_and_archive(
file: UploadFile = File(...),
title: str = None,
creator: str = "admin",
db: Session = Depends(get_db)
):
1. 保存上传文件
file_id = str(uuid.uuid4())
temp_file_path = os.path.join(UPLOAD_DIR, f"{file_id}_{file.filename}")
try:
with open(temp_file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
2. 格式转换 (转为 PDF/A)
log_audit(db, "convert_start", 0, creator, f"Start converting {file.filename}")
pdf_path = convert_to_pdfa(temp_file_path, ARCHIVE_DIR)
final_filename = os.path.basename(pdf_path)
3. 计算哈希与签名 (确保真实性)
file_hash = calculate_file_hash(pdf_path)
signature = sign_file(private_key, pdf_path)
4. 生成档号 (示例：年度-门类-流水号)
year = datetime.datetime.now().year
archive_code = f"{year}-DW-{uuid.uuid4().hex[:8].upper()}"
5. 入库
new_archive = ElectronicArchive(
archive_code=archive_code,
title=title if title else file.filename,
creator=creator,
file_path=pdf_path,
file_hash=file_hash,
digital_signature=signature,
status="archived"
)
db.add(new_archive)
db.commit()
db.refresh(new_archive)
6. 记录成功日志
log_audit(db, "archive_success", new_archive.id, creator, f"File archived as {archive_code}")
清理临时文件
os.remove(temp_file_path)
return {
"message": "Archive created successfully",
"archive_code": archive_code,
"file_hash": file_hash,
"pdf_path": pdf_path
}
except Exception as e:
db.rollback()
log_audit(db, "archive_failed", 0, creator, str(e))
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/archive/{archive_code}")
asyncdef get_archive(archive_code: str, db: Session = Depends(get_db)):
archive = db.query(ElectronicArchive).filter(ElectronicArchive.archive_code == archive_code).first()
if not archive:
raise HTTPException(status_code=404, detail="Archive not found")
验证文件是否被篡改
is_valid = calculate_file_hash(archive.file_path) == archive.file_hash
return {
"archive_code": archive.archive_code,
"title": archive.title,
"creator": archive.creator,
"integrity_check": "PASS" if is_valid else "FAIL",
"created_at": archive.created_at
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

运行上述代码后，你将拥有一个符合《中华人民共和国档案法》核心要求的电子档案管理系统后端。系统自动处理格式转换以保障可用性，使用哈希和签名保障真实性与完整性，并通过审计日志保障安全性。你可以使用 Postman 或 curl 向 http://localhost:8000/api/archive/upload 发送 POST 请求进行测试。