技术架构选型与环境说明
信托业务涉及大量非结构化数据，如合同扫描件、法律意见书、尽职调查报告等。为了实现这些档案的高效存储与全文检索，本指南采用轻量级且高可用的技术栈：MinIO负责对象存储，Elasticsearch负责全文检索，Python FastAPI作为业务网关。这种组合无需依赖昂贵的商业数据库，且在Linux环境下即可一键部署。以下操作均在CentOS 7.9环境下验证通过，假设你已拥有root权限。
第一步：基于Docker Compose编排底层服务
为了避免繁琐的手动安装依赖，我们使用Docker Compose一次性启动MinIO和Elasticsearch。请确保服务器已安装Docker和Docker Compose。创建目录/data/trust_archive，并在该目录下新建docker-compose.yml文件，直接复制以下内容：
docker-compose.yml 完整配置：
```yaml
version: '3.8'
services:
对象存储服务
minio:
image: minio/minio:latest
container_name: trust_minio
ports:
- "9000:9000"
- "9001:9001"
environment:
MINIO_ROOT_USER: minioadmin
MINIO_ROOT_PASSWORD: minioadmin123
command: server /data --console-address ":9001"
volumes:
- ./minio_data:/data
搜索引擎服务
elasticsearch:
image: docker.elastic.co/elasticsearch/elasticsearch:8.11.0
container_name: trust_es
ports:
- "9200:9200"
environment:
- discovery.type=single-node
- xpack.security.enabled=false
- "ES_JAVA_OPTS=-Xms512m -Xmx512m"
volumes:
- ./es_data:/usr/share/elasticsearch/data
```
保存文件后，执行docker-compose up -d。等待约30秒，执行docker ps确认两个容器均处于Up状态。此时，MinIO控制台访问地址为http://服务器IP:9001，Elasticsearch接口地址为http://服务器IP:9200。
第二步：初始化存储桶与索引结构
服务启动后，必须预先创建资源。我们需要在MinIO中创建一个专门存储信托档案的Bucket，并在Elasticsearch中建立索引以支持中文分词检索。
执行以下命令创建名为trust-docs的存储桶：
```bash
curl http://服务器IP:9000/trust-docs/ -X PUT \
--user minioadmin:minioadmin123
```
接着，配置Elasticsearch索引。信托档案检索通常需要精准匹配合同编号，同时对全文进行模糊搜索。执行以下命令创建索引trust_index：
```bash
curl -X PUT "http://服务器IP:9200/trust_index?pretty" -H 'Content-Type: application/json' -d'
{
"settings": {
"analysis": {
"analyzer": {
"ik_smart_analyzer": {
"type": "custom",
"tokenizer": "ik_smart"
}
}
}
},
"mappings": {
"properties": {
"file_name": { "type": "keyword" },
"contract_id": { "type": "keyword" },
"project_name": { "type": "text", "analyzer": "ik_smart_analyzer" },
"content": { "type": "text", "analyzer": "ik_smart_analyzer" },
"upload_time": { "type": "date" },
"file_path": { "type": "keyword" }
}
}
}'
```
注意：上述配置使用了ik_smart分词器，这是处理中文信托合同文本的标准配置。如果Elasticsearch报错提示缺少IK分词插件，请进入ES容器执行elasticsearch-plugin install https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v8.11.0/elasticsearch-analysis-ik-8.11.0.zip并重启容器。
第三步：开发核心业务逻辑代码
现在开始编写核心系统代码。我们需要实现文件上传、文本提取、元数据入库及检索接口。在/data/trust_archive目录下创建app.py。首先安装必要的Python依赖库：
```bash
pip install fastapi uvicorn minio elasticsearch pypdf python-multipart
```
以下是完整的app.py代码，包含了所有必要的逻辑，无需任何修改即可运行：
```python
import os
import uuid
from datetime import datetime
from fastapi import FastAPI, UploadFile, Form, HTTPException
from minio import Minio
from elasticsearch import Elasticsearch
from pypdf import PdfReader
app = FastAPI()
初始化MinIO客户端
minio_client = Minio(
"服务器IP:9000",
access_key="minioadmin",
secret_key="minioadmin123",
secure=False
)
初始化Elasticsearch客户端
es_client = Elasticsearch("http://服务器IP:9200")
BUCKET_NAME = "trust-docs"
def extract_text_from_pdf(file_path: str) -> str:
"""本地提取PDF文本用于检索"""
try:
reader = PdfReader(file_path)
text = ""
for page in reader.pages:
text += page.extract_text() + "\n"
return text[:5000] 限制入库文本长度，避免过大
except Exception as e:
print(f"PDF解析失败: {e}")
return ""
@app.post("/upload")
async def upload_archive(
file: UploadFile,
contract_id: str = Form(...),
project_name: str = Form(...)
):
1. 生成唯一文件名
file_ext = file.filename.split(".")[-1]
unique_name = f"{uuid.uuid4().hex}.{file_ext}"
object_name = f"{datetime.now().strftime('%Y%m%d')}/{unique_name}"
2. 上传文件到MinIO
try:
先将上传的流保存到本地临时文件以便解析文本
temp_path = f"/tmp/{unique_name}"
with open(temp_path, "wb") as buffer:
buffer.write(await file.read())
minio_client.fput_object(BUCKET_NAME, object_name, temp_path)
except Exception as e:
raise HTTPException(status_code=500, detail=f"存储上传失败: {str(e)}")
3. 提取文本并写入ES
content_text = extract_text_from_pdf(temp_path)
doc_body = {
"file_name": file.filename,
"contract_id": contract_id,
"project_name": project_name,
"content": content_text,
"upload_time": datetime.now(),
"file_path": object_name
}
try:
es_client.index(index="trust_index", document=doc_body)
except Exception as e:
如果ES写入失败，建议删除已上传文件或记录日志，此处简化处理
raise HTTPException(status_code=500, detail=f"索引写入失败: {str(e)}")
finally:
if os.path.exists(temp_path):
os.remove(temp_path)
return {"code": 200, "msg": "档案归档成功", "file_id": unique_name}
@app.get("/search")
async def search_archive(keyword: str):
"""基于关键词检索档案"""
query = {
"query": {
"multi_match": {
"query": keyword,
"fields": ["project_name", "content", "contract_id"]
}
}
}
try:
resp = es_client.search(index="trust_index", body=query)
results = []
for hit in resp['hits']['hits']:
source = hit['_source']
results.append({
"project_name": source.get("project_name"),
"contract_id": source.get("contract_id"),
"file_name": source.get("file_name"),
"highlight": source.get("content", "")[:100] + "..."
})
return {"code": 200, "data": results}
except Exception as e:
raise HTTPException(status_code=500, detail=f"检索服务异常: {str(e)}")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
```
代码逻辑解析：上传接口首先将文件存入MinIO的特定日期目录下，确保文件不重名。随后利用pypdf库在本地临时解析PDF内容，截取前5000字符。最后将元数据（合同号、项目名）和文本内容合并存入Elasticsearch。检索接口使用multi_match同时查询项目名、合同号和正文内容。
第四步：系统启动与功能验证
代码编写完成后，启动API服务：
```bash
python3 app.py
```
服务默认运行在8000端口。我们使用curl命令模拟前端进行一次完整的归档和检索流程。
1. 模拟上传一份信托合同：
准备一个测试用的PDF文件test_contract.pdf，内容包含“集合资金信托计划”。执行：
```bash
curl -X POST "http://服务器IP:8000/upload" \
-F "file=@test_contract.pdf" \
-F "contract_id=XT20231001" \
-F "project_name=某基建集合资金信托计划"
```
如果返回{"code": 200, "msg": "档案归档成功"...}，说明文件已成功存入MinIO且索引已建立。
2. 验证全文检索功能：
执行搜索命令，查找关键词“基建”：
```bash
curl "http://服务器IP:8000/search?keyword=基建"
```
系统应返回JSON格式的结果，包含刚才上传的项目名称、合同号以及内容预览。如果检索“XT20231001”，也应能精准定位到该文档，证明了元数据索引的有效性。
