污染源普查数字档案馆系统从零到一部署实战指南

一、基础环境准备与依赖安装

在开始构建污染源普查数字档案馆系统之前，我们需要准备好运行环境。本系统采用Python作为后端开发语言，利用MinIO作为对象存储，PostgreSQL作为元数据库，Elasticsearch作为全文检索引擎。所有组件将通过Docker Compose进行编排，确保环境的一致性和易部署性。

请在服务器或本地终端执行以下命令安装Docker及Docker Compose。以下命令适用于CentOS 7/8系统，Ubuntu系统请相应调整包管理器：

 1. 卸载旧版本
sudo yum remove docker docker-client docker-client-latest docker-common docker-latest-containerd docker-latest-logrotate docker-logrotate docker-engine
2. 安装依赖工具
sudo yum install -y yum-utils device-mapper-persistent-data lvm2
3. 添加Docker官方仓库
sudo yum-config-manager --add-repo https://download.docker.com/linux/centos/docker-ce.repo
4. 安装Docker Engine
sudo yum install -y docker-ce docker-ce-cli containerd.io
5. 启动Docker并设置开机自启
sudo systemctl start docker
sudo systemctl enable docker
6. 安装Docker Compose
sudo curl -L "https://github.com/docker/compose/releases/download/v2.20.0/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose
sudo chmod +x /usr/local/bin/docker-compose

安装完成后，输入docker --version和docker-compose --version验证安装是否成功。

二、项目目录结构规划

为了规范管理，请在宿主机创建一个专门的项目目录，并按照以下结构创建子目录和文件：

pollution_archive/
├── docker-compose.yml        容器编排文件
├── backend/                  后端服务目录
│   ├── app/
│   │   ├── main.py          核心业务逻辑
│   │   └── requirements.txt  Python依赖
│   └── Dockerfile           后端镜像构建文件
├── data/                    数据持久化目录
│   ├── pgdata               PostgreSQL数据
│   └── minio_data           对象存储数据
└── esdata/                  Elasticsearch数据目录

执行以下命令快速创建目录结构：

mkdir -p pollution_archive/{backend/app,data/{pgdata,minio_data},esdata}
cd pollution_archive

三、容器化服务编排配置

在pollution_archive根目录下创建docker-compose.yml文件。该文件定义了数据库、存储、搜索引擎和后端服务的启动顺序与网络配置。

请将以下内容完整复制到docker-compose.yml中：

version: '3.8'
services:
1. PostgreSQL 数据库服务
db:
image: postgres:14-alpine
container_name: archive_db
environment:
POSTGRES_USER: admin
POSTGRES_PASSWORD: StrongPassword123
POSTGRES_DB: pollution_archive
volumes:
- ./data/pgdata:/var/lib/postgresql/data
ports:
- "5432:5432"
healthcheck:
test: ["CMD-SHELL", "pg_isready -U admin"]
interval: 10s
timeout: 5s
retries: 5
2. MinIO 对象存储服务
minio:
image: minio/minio:latest
container_name: archive_minio
command: server /data --console-address ":9001"
environment:
MINIO_ROOT_USER: minioadmin
MINIO_ROOT_PASSWORD: MinioAdmin123
volumes:
- ./data/minio_data:/data
ports:
- "9000:9000"
- "9001:9001"
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
interval: 30s
timeout: 20s
retries: 3
3. Elasticsearch 检索服务
elasticsearch:
image: docker.elastic.co/elasticsearch/elasticsearch:8.11.0
container_name: archive_es
environment:
- discovery.type=single-node
- "ES_JAVA_OPTS=-Xms512m -Xmx512m"
- xpack.security.enabled=false
volumes:
- ./esdata:/usr/share/elasticsearch/data
ports:
- "9200:9200"
healthcheck:
test: ["CMD-SHELL", "curl -f http://localhost:9200/_cluster/health || exit 1"]
interval: 30s
timeout: 10s
retries: 5
4. 后端业务服务
backend:
build: ./backend
container_name: archive_backend
command: python app/main.py
volumes:
- ./backend/app:/app
ports:
- "8000:8000"
environment:
- DB_HOST=db
- DB_PORT=5432
- DB_NAME=pollution_archive
- DB_USER=admin
- DB_PASS=StrongPassword123
- MINIO_ENDPOINT=minio:9000
- MINIO_ACCESS_KEY=minioadmin
- MINIO_SECRET_KEY=MinioAdmin123
- ES_HOST=http://elasticsearch:9200
depends_on:
db:
condition: service_healthy
minio:
condition: service_healthy
elasticsearch:
condition: service_healthy

四、后端服务代码实现

接下来编写后端核心代码。首先创建backend/app/requirements.txt，定义Python依赖：

fastapi==0.104.1
uvicorn==0.24.0
sqlalchemy==2.0.23
psycopg2-binary==2.9.9
minio==7.2.0
elasticsearch==8.11.0
python-multipart==0.0.6

创建backend/Dockerfile用于构建后端镜像：

FROM python:3.9-slim
WORKDIR /app
RUN pip install --no-cache-dir -r app/requirements.txt
COPY backend/app /app
CMD ["python", "app/main.py"]

创建核心业务文件backend/app/main.py。该脚本实现了数据库初始化、MinIO文件上传、Elasticsearch索引同步以及文件检索接口。

import os
import time
from fastapi import FastAPI, UploadFile, File, Form, HTTPException
from fastapi.responses import JSONResponse
from sqlalchemy import create_engine, Column, Integer, String, DateTime, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from minio import Minio
from minio.error import S3Error
from elasticsearch import Elasticsearch
from datetime import datetime
配置读取
DB_HOST = os.getenv("DB_HOST", "localhost")
DB_USER = os.getenv("DB_USER", "admin")
DB_PASS = os.getenv("DB_PASS", "StrongPassword123")
DB_NAME = os.getenv("DB_NAME", "pollution_archive")
DB_PORT = os.getenv("DB_PORT", "5432")
MINIO_ENDPOINT = os.getenv("MINIO_ENDPOINT", "minio:9000")
MINIO_ACCESS = os.getenv("MINIO_ACCESS_KEY", "minioadmin")
MINIO_SECRET = os.getenv("MINIO_SECRET_KEY", "MinioAdmin123")
ES_HOST = os.getenv("ES_HOST", "http://elasticsearch:9200")
DATABASE_URL = f"postgresql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
数据库初始化
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
class PollutionRecord(Base):
__tablename__ = "pollution_records"
id = Column(Integer, primary_key=True, index=True)
enterprise_name = Column(String(255), nullable=False)
source_type = Column(String(100))  污染源类型：废水、废气等
district = Column(String(100))     所属区域
file_name = Column(String(255), nullable=False)
object_name = Column(String(500), nullable=False)  MinIO存储路径
upload_time = Column(DateTime, default=datetime.utcnow)
description = Column(Text)
创建表（如果不存在）
try:
Base.metadata.create_all(bind=engine)
except Exception as e:
print(f"DB Init Error: {e}")
time.sleep(5)  等待DB完全启动
Base.metadata.create_all(bind=engine)
MinIO 初始化
minio_client = Minio(
MINIO_ENDPOINT,
access_key=MINIO_ACCESS,
secret_key=MINIO_SECRET,
secure=False
)
BUCKET_NAME = "pollution-files"
try:
if not minio_client.bucket_exists(BUCKET_NAME):
minio_client.make_bucket(BUCKET_NAME)
except S3Error as e:
print(f"MinIO Error: {e}")
Elasticsearch 初始化
es_client = Elasticsearch(ES_HOST)
INDEX_NAME = "pollution_index"
FastAPI 应用
app = FastAPI(title="污染源普查数字档案馆系统")
@app.get("/")
def read_root():
return {"status": "系统运行正常", "message": "污染源数字档案馆API已就绪"}
@app.post("/upload")
async def upload_file(
file: UploadFile = File(...),
enterprise_name: str = Form(...),
source_type: str = Form(...),
district: str = Form(...),
description: str = Form("")
):
"""
上传污染源普查文件并存档
"""
1. 生成唯一对象名
timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
object_name = f"{timestamp}_{file.filename}"
2. 上传至MinIO
try:
data = file.file.read()
minio_client.put_object(
BUCKET_NAME,
object_name,
data,
length=len(data),
content_type=file.content_type
)
except S3Error as e:
raise HTTPException(status_code=500, detail=f"文件上传失败: {str(e)}")
3. 写入PostgreSQL元数据
db = SessionLocal()
try:
record = PollutionRecord(
enterprise_name=enterprise_name,
source_type=source_type,
district=district,
file_name=file.filename,
object_name=object_name,
description=description
)
db.add(record)
db.commit()
db.refresh(record)
record_id = record.id
except Exception as e:
db.rollback()
raise HTTPException(status_code=500, detail=f"数据库写入失败: {str(e)}")
finally:
db.close()
4. 索引到Elasticsearch
try:
doc_body = {
"enterprise_name": enterprise_name,
"source_type": source_type,
"district": district,
"file_name": file.filename,
"description": description,
"upload_time": timestamp,
"record_id": record_id
}
es_client.index(index=INDEX_NAME, id=record_id, document=doc_body)
except Exception as e:
ES索引失败不影响主流程，仅记录日志
print(f"ES Index Error: {e}")
return {"code": 200, "message": "文件归档成功", "data": {"id": record_id, "object_name": object_name}}
@app.get("/search")
def search_records(keyword: str):
"""
根据关键词检索污染源档案
"""
try:
query = {
"query": {
"multi_match": {
"query": keyword,
"fields": ["enterprise_name", "district", "description", "source_type"]
}
}
}
resp = es_client.search(index=INDEX_NAME, body=query)
hits = resp['hits']['hits']
results = [hit["_source"] for hit in hits]
return {"code": 200, "total": len(results), "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)

五、系统启动与功能验证

所有配置文件与代码已准备就绪。现在启动整个系统栈。在pollution_archive目录下执行：

docker-compose up -d --build

注意： 首次启动需要下载镜像并构建Python环境，可能需要3-5分钟。可以通过docker-compose logs -f查看实时日志，确认所有服务均为healthy状态。

1. 验证文件上传归档

使用curl命令模拟前端上传一份污染源普查PDF文件。请确保当前目录下有一个测试文件test_report.pdf，如果没有，请新建一个空文件用于测试：

touch test_report.pdf

执行上传命令：

curl -X POST "http://localhost:8000/upload" \
-F "file=@test_report.pdf" \
-F "enterprise_name=某某化工有限公司" \
-F "source_type=废气" \
-F "district=朝阳区" \
-F "description=2023年度第四季度污染物排放监测报告"

成功返回示例：

{"code":200,"message":"文件归档成功","data":{"id":1,"object_name":"20231115120000_test_report.pdf"}}

2. 验证全文检索功能

上传成功后，立即使用关键词进行检索测试。例如搜索“化工”或“废气”：

curl "http://localhost:8000/search?keyword=化工"

成功返回示例：

{
"code": 200,
"total": 1,
"data": [
{
"enterprise_name": "某某化工有限公司",
"source_type": "废气",
"district": "朝阳区",
"file_name": "test_report.pdf",
"description": "2023年度第四季度污染物排放监测报告",
"upload_time": "20231115120000",
"record_id": 1
}
]
}

3. 验证对象存储

打开浏览器访问 http://localhost:9001，使用账号minioadmin / 密码MinioAdmin123登录。进入pollution-files桶，即可看到刚才上传的物理文件已安全存储。

六、常见运维操作

在日常维护中，您可能需要查看服务状态或重置数据。以下是常用操作命令：

• 查看所有服务状态：
  docker-compose ps
• 查看后端日志：
  docker-compose logs -f backend
• 停止并删除所有容器（保留数据）：
  docker-compose down
• 完全重置系统（警告：将删除所有数据）：
  docker-compose down -v
rm -rf data/ esdata/