一、技术架构选型与设计思路
构建一套企业级档案管理系统，核心在于解决海量非结构化数据的存储、全文检索以及元数据管理问题。为了确保方案的高可用性与零门槛落地，本指南采用微服务容器化架构。具体技术栈选型如下：
- 对象存储服务：选用 MinIO。它兼容 S3 API，性能极高，适合存储海量原始档案文件（PDF、OFD、图片等）。
- 元数据数据库：选用 PostgreSQL。用于存储档案的分类、著录项、权限控制等结构化数据。
- 全文检索与OCR引擎：选用 Apache Tika。它能自动提取文档内容（文本）及元数据，支持 PDF、Word、Excel 等数百种格式，解决档案“内容不可见”的痛点。
- 业务应用层：选用 Python (FastAPI)。开发效率高，异步处理能力强，负责串联上述组件。
二、基础环境搭建与容器编排
在开始编码前，必须先通过 Docker Compose 编排好基础设施。请确保服务器已安装 Docker 和 Docker Compose。创建项目目录 arch-solution，并在其中创建 docker-compose.yml 文件。
以下是完整的编排配置，直接复制保存即可。该配置启动了 PostgreSQL、MinIO 和 Tika 服务。
```yaml
version: '3.8'
services:
1. 元数据数据库
db:
image: postgres:15-alpine
container_name: arch_db
environment:
POSTGRES_USER: archuser
POSTGRES_PASSWORD: archpass123
POSTGRES_DB: archdb
volumes:
- pgdata:/var/lib/postgresql/data
ports:
- "5432:5432"
healthcheck:
test: ["CMD-SHELL", "pg_isready -U archuser"]
interval: 10s
timeout: 5s
retries: 5
2. 对象存储服务
minio:
image: minio/minio:latest
container_name: arch_minio
command: server /data --console-address ":9001"
environment:
MINIO_ROOT_USER: minioadmin
MINIO_ROOT_PASSWORD: minioadmin123
volumes:
- miniodata:/data
ports:
- "9000:9000"
- "9001:9001"
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
interval: 30s
timeout: 20s
retries: 3
3. 文档内容提取与OCR服务
tika:
image: apache/tika:latest
container_name: arch_tika
ports:
- "9998:9998"
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:9998/server"]
interval: 30s
timeout: 10s
retries: 5
volumes:
pgdata:
miniodata:
```
保存后，在终端执行以下命令启动基础环境：
docker-compose up -d
注意：请等待约 30 秒，待所有服务状态均为 healthy 后再进行后续操作。可以使用 `docker-compose ps` 查看状态。
三、数据库结构定义
我们需要在 PostgreSQL 中初始化档案表结构。为了方便实操，我们创建一个初始化脚本 init_schema.sql。该脚本包含档案主表、分类表及全文检索触发器。
```sql
-- 档案分类表
CREATE TABLE IF NOT EXISTS archive_categories (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
code VARCHAR(50) UNIQUE NOT NULL
);
-- 档案主表
CREATE TABLE IF NOT EXISTS archives (
id SERIAL PRIMARY KEY,
title VARCHAR(255) NOT NULL,
category_code VARCHAR(50),
file_path VARCHAR(500) NOT NULL, -- MinIO中的对象Key
file_size BIGINT,
mime_type VARCHAR(100),
content_text TEXT, -- 存储提取出的全文，用于检索
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 插入默认分类
INSERT INTO archive_categories (name, code) VALUES
('文书档案', 'WS'),
('科技档案', 'KJ'),
('会计档案', 'KJ')
ON CONFLICT (code) DO NOTHING;
-- 创建全文检索索引
CREATE INDEX idx_archives_content ON archives USING gin(to_tsvector('chinese', content_text));
```
执行该脚本以初始化数据库：
docker exec -i arch_db psql -U archuser -d archdb < init_schema.sql
四、核心业务逻辑实现（采集、OCR、存储）
接下来编写核心应用。我们需要安装 Python 依赖。在项目根目录创建 requirements.txt：
```text
fastapi
uvicorn
sqlalchemy
psycopg2-binary
minio
requests
python-multipart
```
安装依赖：pip install -r requirements.txt
创建 main.py，实现文件上传、自动提取文本（OCR）、存入 MinIO 并将元数据写入 PostgreSQL 的完整闭环。
```python
import os
import uuid
from datetime import datetime
from typing import Optional
from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import JSONResponse
from minio import Minio
from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, BigInteger
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import requests
配置部分
MINIO_ENDPOINT = "localhost:9000"
MINIO_ACCESS_KEY = "minioadmin"
MINIO_SECRET_KEY = "minioadmin123"
MINIO_BUCKET_NAME = "archive-files"
DB_URL = "postgresql://archuser:archpass123@localhost:5432/archdb"
TIKA_SERVER_URL = "http://localhost:9998"
初始化客户端
minio_client = Minio(
MINIO_ENDPOINT,
access_key=MINIO_ACCESS_KEY,
secret_key=MINIO_SECRET_KEY,
secure=False
)
确保Bucket存在
if not minio_client.bucket_exists(MINIO_BUCKET_NAME):
minio_client.make_bucket(MINIO_BUCKET_NAME)
数据库模型
engine = create_engine(DB_URL)
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()
class Archive(Base):
__tablename__ = "archives"
id = Column(Integer, primary_key=True, index=True)
title = Column(String(255))
category_code = Column(String(50))
file_path = Column(String(500))
file_size = Column(BigInteger)
mime_type = Column(String(100))
content_text = Column(Text)
created_at = Column(DateTime, default=datetime.utcnow)
辅助函数：调用Tika提取文本
def extract_text_with_tika(file_path: str, file_content: bytes) -> str:
try:
将文件内容发送给Tika服务器进行解析
headers = {"Content-Type": "application/octet-stream"}
response = requests.put(
f"{TIKA_SERVER_URL}/tika/text",
data=file_content,
headers=headers,
timeout=30
)
if response.status_code == 200:
return response.text
return ""
except Exception as e:
print(f"Tika Error: {e}")
return ""
app = FastAPI(title="Enterprise Archive Solution")
@app.post("/api/v1/upload")
async def upload_archive(
file: UploadFile = File(...),
title: Optional[str] = None,
category_code: Optional[str] = "WS"
):
1. 读取文件内容
file_content = await file.read()
file_size = len(file_content)
如果未指定标题，使用文件名
final_title = title if title else file.filename
2. 存储到 MinIO
object_name = f"{category_code}/{uuid.uuid4()}_{file.filename}"
try:
minio_client.put_object(
MINIO_BUCKET_NAME,
object_name,
data=io.BytesIO(file_content),
length=file_size,
content_type=file.content_type
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"MinIO upload failed: {str(e)}")
3. 调用 Tika 提取全文 (核心步骤)
extracted_text = extract_text_with_tika(object_name, file_content)
4. 写入 PostgreSQL
db = SessionLocal()
try:
new_archive = Archive(
title=final_title,
category_code=category_code,
file_path=object_name,
file_size=file_size,
mime_type=file.content_type,
content_text=extracted_text
)
db.add(new_archive)
db.commit()
db.refresh(new_archive)
return {"message": "Archive uploaded successfully", "id": new_archive.id, "text_length": len(extracted_text)}
except Exception as e:
db.rollback()
raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
finally:
db.close()
import io 用于内存处理
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
```
五、检索与利用接口开发
仅有上传是不够的，档案系统的核心价值在于利用。我们在 main.py 中添加一个基于 PostgreSQL 全文检索的接口。将以下代码追加到 main.py 中 if __name__ == "__main__": 之前。
```python
from sqlalchemy import or_, func
@app.get("/api/v1/search")
async def search_archives(keyword: str):
"""
基于关键词的全文检索接口
"""
db = SessionLocal()
try:
使用PostgreSQL的全文检索功能
to_tsquery('chinese', keyword) 需要数据库已安装 zhparser 插件
为简化演示，这里使用基本的 ilike 模糊匹配，生产环境建议使用 to_tsvector
query = db.query(Archive).filter(
or_(
Archive.title.ilike(f"%{keyword}%"),
Archive.content_text.ilike(f"%{keyword}%")
)
)
results = query.all()
data = []
for item in results:
data.append({
"id": item.id,
"title": item.title,
"category": item.category_code,
"created_at": item.created_at,
"snippet": item.content_text[:100] + "..." if item.content_text else "No content"
})
return {"total": len(data), "items": data}
finally:
db.close()
@app.get("/api/v1/download/{archive_id}")
async def download_archive(archive_id: int):
"""
获取文件的临时下载地址
"""
db = SessionLocal()
try:
item = db.query(Archive).filter(Archive.id == archive_id).first()
if not item:
raise HTTPException(status_code=404, detail="Archive not found")
生成预签名URL，有效期1小时
url = minio_client.presigned_get_object(
MINIO_BUCKET_NAME,
item.file_path,
expires=timedelta(hours=1)
)
return {"download_url": url}
finally:
db.close()
from datetime import timedelta
```
六、系统启动与全流程验证
所有代码已就绪。现在启动 FastAPI 应用：
python main.py
应用启动后，我们可以通过 curl 命令进行全流程测试，模拟真实的业务场景。
步骤 1：上传一份 PDF 档案
假设你当前目录下有一个名为 test_contract.pdf 的文件。执行以下命令上传：
curl -X POST "http://localhost:8000/api/v1/upload" -F "file=@test_contract.pdf" -F "title=采购合同A001" -F "category_code=KJ"
预期返回结果包含 "text_length": > 0，说明 Tika 成功提取了 PDF 中的文字。
步骤 2：全文检索测试
假设 PDF 中包含“采购金额”或“甲方”等词汇。执行搜索：
curl "http://localhost:8000/api/v1/search?keyword=采购"
你将看到刚才上传的档案信息，包括标题和内容片段。
步骤 3：下载测试
获取下载链接（假设 ID 为 1）：
curl "http://localhost:8000/api/v1/download/1"
返回的 JSON 中包含一个 download_url，直接在浏览器中打开该 URL 即可下载存储在 MinIO 中的原始文件。
