手把手部署企业级电子档案长期保存系统实战指南

技术选型与架构说明

本文将指导你基于MinIO（高性能对象存储）、PostgreSQL（元数据管理）和Python（业务逻辑）构建一套符合OAIS参考模型的企业级电子档案长期保存系统。该方案具备自包含、高可用、防篡改特性，所有操作均在Linux环境下进行。

第一步：环境准备与Docker安装

在开始部署前，需要确保服务器环境干净。推荐使用CentOS 7.9或Ubuntu 20.04及以上版本。首先安装Docker及Docker Compose，这是整个系统运行的基础。

执行以下命令安装Docker：

curl -fsSL https://get.docker.com | bash -s docker --mirror Aliyun
systemctl start docker
systemctl enable docker

接着安装Docker Compose：

curl -L "https://github.com/docker/compose/releases/download/v2.20.0/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose
chmod +x /usr/local/bin/docker-compose

验证安装是否成功：

docker --version
docker-compose --version

第二步：部署MinIO对象存储服务

MinIO用于存储实际的电子文件（AIP包）。我们将创建一个独立的目录来存放MinIO的数据和配置文件。

创建项目目录：

mkdir -p /data/long-term-archive/minio-data

编写docker-compose.yml文件来启动MinIO服务。请直接复制以下内容覆盖到/data/long-term-archive/docker-compose.yml中：

version: '3.8'
services:
minio:
image: minio/minio:RELEASE.2023-11-20T22-40-07Z
container_name: archive-minio
ports:
- "9000:9000"
- "9001:9001"
environment:
- MINIO_ROOT_USER=admin
- MINIO_ROOT_PASSWORD=ArchiveSecure@2024
volumes:
- /data/long-term-archive/minio-data:/data
command: server /data --console-address ":9001"
restart: always
postgres:
image: postgres:15-alpine
container_name: archive-postgres
ports:
- "5432:5432"
environment:
- POSTGRES_DB=archive_db
- POSTGRES_USER=archive_user
- POSTGRES_PASSWORD=ArchivePass@2024
volumes:
- /data/long-term-archive/pg-data:/var/lib/postgresql/data
restart: always

启动服务：

cd /data/long-term-archive
docker-compose up -d

服务启动后，需要创建一个专用的存储桶（Bucket）用于存放档案。我们可以使用MinIO的客户端工具mc来操作。

安装并配置mc：

wget https://dl.min.io/client/mc/release/linux-amd64/mc
chmod +x mc
mv mc /usr/local/bin/

配置别名并创建Bucket：

mc alias set local http://127.0.0.1:9000 admin ArchiveSecure@2024
mc mb local/erp-archives --ignore-existing

第三步：初始化元数据库

我们需要在PostgreSQL中建立表结构，用于记录档案的元数据、校验信息和存储路径。这步操作对于实现“逻辑保存”至关重要。

连接数据库：

docker exec -it archive-postgres psql -U archive_user -d archive_db

在PostgreSQL命令行中执行以下SQL脚本，创建核心表结构：

CREATE TABLE archive_records (
id SERIAL PRIMARY KEY,
archive_id VARCHAR(64) UNIQUE NOT NULL,
original_filename VARCHAR(255) NOT NULL,
file_size BIGINT NOT NULL,
file_format VARCHAR(50),
checksum_sha256 TEXT NOT NULL,
storage_path TEXT NOT NULL,
ingest_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
preservation_level VARCHAR(20) DEFAULT 'BIT_PRESERVATION'
);
CREATE INDEX idx_archive_id ON archive_records(archive_id);

输入\q退出数据库连接。

第四步：核心业务逻辑代码实现

现在编写Python脚本来实现档案的“摄入”和“校验”功能。这是系统最核心的部分，实现了文件入库、元数据登记和完整性校验。

1. 安装Python依赖

确保服务器已安装Python3，然后安装必要的库：

pip3 install boto3 psycopg2-binary

2. 编写档案入库脚本

创建文件/data/long-term-archive/ingest.py，复制以下完整代码。该脚本会将文件打包为ZIP（AIP包），计算SHA256，上传至MinIO，并记录元数据。

import os
import sys
import hashlib
import zipfile
import uuid
import datetime
import boto3
from botocore.exceptions import NoCredentialsError
import psycopg2
配置信息
MINIO_ENDPOINT = 'http://127.0.0.1:9000'
MINIO_ACCESS_KEY = 'admin'
MINIO_SECRET_KEY = 'ArchiveSecure@2024'
MINIO_BUCKET = 'erp-archives'
DB_HOST = '127.0.0.1'
DB_NAME = 'archive_db'
DB_USER = 'archive_user'
DB_PASS = 'ArchivePass@2024'
def calculate_sha256(file_path):
sha256 = hashlib.sha256()
with open(file_path, 'rb') as f:
for byte_block in iter(lambda: f.read(4096), b""):
sha256.update(byte_block)
return sha256.hexdigest()
def ingest_file(source_file_path):
if not os.path.exists(source_file_path):
print(f"错误：文件 {source_file_path} 不存在")
return
filename = os.path.basename(source_file_path)
archive_id = str(uuid.uuid4())
timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
zip_name = f"{archive_id}_{timestamp}.aip.zip"
1. 创建AIP包 (ZIP)
temp_zip_path = f"/tmp/{zip_name}"
with zipfile.ZipFile(temp_zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
zipf.write(source_file_path, filename)
2. 计算校验码
file_checksum = calculate_sha256(temp_zip_path)
file_size = os.path.getsize(temp_zip_path)
try:
3. 上传到MinIO
s3 = boto3.resource('s3',
endpoint_url=MINIO_ENDPOINT,
aws_access_key_id=MINIO_ACCESS_KEY,
aws_secret_access_key=MINIO_SECRET_KEY,
verify=False)
s3.Bucket(MINIO_BUCKET).upload_file(temp_zip_path, zip_name)
print(f"[MinIO] 上传成功: {zip_name}")
4. 写入PostgreSQL
conn = psycopg2.connect(host=DB_HOST, database=DB_NAME, user=DB_USER, password=DB_PASS)
cursor = conn.cursor()
query = """
INSERT INTO archive_records (archive_id, original_filename, file_size, file_format, checksum_sha256, storage_path)
VALUES (%s, %s, %s, %s, %s, %s)
"""
cursor.execute(query, (archive_id, filename, file_size, 'ZIP', file_checksum, zip_name))
conn.commit()
print(f"[DB] 元数据记录成功, Archive ID: {archive_id}")
cursor.close()
conn.close()
except (NoCredentialsError, Exception) as e:
print(f"操作失败: {e}")
finally:
if os.path.exists(temp_zip_path):
os.remove(temp_zip_path)
if __name__ == "__main__":
if len(sys.argv) < 2:
print("用法: python3 ingest.py <文件路径>")
else:
ingest_file(sys.argv[1])

3. 编写完整性校验脚本

长期保存的核心是定期校验。创建文件/data/long-term-archive/verify.py，用于比对数据库中的SHA256值与存储在MinIO中实际文件的SHA256值。

import boto3
import psycopg2
import hashlib
import io
配置信息 (同上)
MINIO_ENDPOINT = 'http://127.0.0.1:9000'
MINIO_ACCESS_KEY = 'admin'
MINIO_SECRET_KEY = 'ArchiveSecure@2024'
MINIO_BUCKET = 'erp-archives'
DB_HOST = '127.0.0.1'
DB_NAME = 'archive_db'
DB_USER = 'archive_user'
DB_PASS = 'ArchivePass@2024'
def verify_integrity():
s3 = boto3.resource('s3',
endpoint_url=MINIO_ENDPOINT,
aws_access_key_id=MINIO_ACCESS_KEY,
aws_secret_access_key=MINIO_SECRET_KEY,
verify=False)
conn = psycopg2.connect(host=DB_HOST, database=DB_NAME, user=DB_USER, password=DB_PASS)
cursor = conn.cursor()
cursor.execute("SELECT archive_id, storage_path, checksum_sha256 FROM archive_records")
records = cursor.fetchall()
print(f"开始校验，共 {len(records)} 个档案包...")
for record in records:
archive_id, storage_path, db_hash = record
try:
流式下载计算Hash，避免大文件撑爆内存
obj = s3.Object(MINIO_BUCKET, storage_path)
body = obj.get()['Body']
sha256 = hashlib.sha256()
for chunk in body.iter_chunks(chunk_size=8192):
sha256.update(chunk)
current_hash = sha256.hexdigest()
if current_hash == db_hash:
print(f"[OK] {archive_id} 校验通过")
else:
print(f"[FAIL] {archive_id} 校验失败! 数据库记录: {db_hash}, 实际: {current_hash}")
except Exception as e:
print(f"[ERROR] {archive_id} 读取或校验异常: {e}")
cursor.close()
conn.close()
if __name__ == "__main__":
verify_integrity()

第五步：系统验证与实操演示

至此，系统已搭建完毕。现在我们将通过一个实际的操作流程来验证系统是否可用。

1. 模拟生成一份电子档案

创建一个测试文件：

echo "这是一份重要的财务凭证数据，必须长期保存10年以上。" > /tmp/test_doc_001.txt

2. 执行档案入库操作

运行刚才编写的Python脚本，将测试文件存入系统：

python3 /data/long-term-archive/ingest.py /tmp/test_doc_001.txt

如果输出显示“上传成功”和“元数据记录成功”，说明入库流程无误。请记下控制台输出的Archive ID。

3. 执行完整性校验操作

运行校验脚本，验证数据是否一致：

python3 /data/long-term-archive/verify.py

控制台应输出“[OK] ... 校验通过”。

4. 模拟数据篡改测试

为了证明系统的可靠性，我们手动修改MinIO中的文件，再次运行校验，观察系统是否能发现问题。

使用mc命令下载文件：

mc cp local/erp-archives/$(ls -t /data/long-term-archive/minio-data/erp-archives/ | head -1) /tmp/tampered.zip

修改文件内容并重新覆盖上传：

echo "corrupted data" >> /tmp/tampered.zip
mc cp /tmp/tampered.zip local/erp-archives/$(ls -t /data/long-term-archive/minio-data/erp-archives/ | head -1)

再次运行校验脚本：

python3 /data/long-term-archive/verify.py

此时，控制台将明确报错提示“[FAIL] ... 校验失败”，并展示不一致的哈希值。这证明你的长期保存系统具备了完整的数据自检能力。