基于Python构建行政强制档案系统实操指南

一、开发环境与依赖准备

在开始构建行政强制档案软件之前，需要确保本地环境已安装Python 3.8及以上版本。为了实现快速开发和高性能文件处理，本系统采用FastAPI作为后端框架，SQLite作为数据库，无需额外配置数据库服务即可直接运行。

请在终端中执行以下命令，创建项目目录并安装核心依赖包：

```bash mkdir admin_enforcement_archive cd admin_enforcement_archive pip install fastapi uvicorn python-multipart aiofiles sqlalchemy ```

依赖说明：
fastapi：用于构建高性能Web API。
uvicorn：ASGI服务器，用于运行FastAPI应用。
python-multipart：用于处理表单文件上传。
aiofiles：用于异步文件I/O操作，提升并发处理能力。
sqlalchemy：Python SQL工具包和对象关系映射（ORM）。

二、数据库模型与初始化配置

为了确保档案数据的结构化存储，我们需要设计一个包含案件编号、当事人信息、强制措施类型及文件路径的数据模型。创建一个名为database.py的文件，写入以下完整的数据库连接与配置代码：

```python database.py from sqlalchemy import create_engine, Column, Integer, String, DateTime from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker import datetime 创建SQLite数据库文件 SQLALCHEMY_DATABASE_URL = "sqlite:///./admin_enforcement.db" engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() 定义档案数据模型 class ArchiveRecord(Base): __tablename__ = "archives" id = Column(Integer, primary_key=True, index=True) case_no = Column(String(50), unique=True, index=True, nullable=False) 案件编号 party_name = Column(String(100), nullable=False) 当事人姓名/名称 measure_type = Column(String(100), nullable=False) 强制措施类型 file_path = Column(String(255), nullable=False) 电子档案存储路径 upload_time = Column(DateTime, default=datetime.datetime.now) 归档时间 初始化数据库表 def init_db(): Base.metadata.create_all(bind=engine) ```

保存上述文件后，在终端执行以下Python命令进行数据库表的初始化：

```bash python -c "from database import init_db; init_db(); print('数据库初始化完成')" ```

三、后端核心接口开发

接下来创建main.py文件，实现档案上传、列表查询及下载检索的核心逻辑。该文件将包含所有业务处理代码，确保复制后即可直接运行。

```python main.py from fastapi import FastAPI, UploadFile, File, Form, Depends, HTTPException, status from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles from fastapi.middleware.cors import CORSMiddleware from sqlalchemy.orm import Session from typing import Optional import shutil import os from database import SessionLocal, ArchiveRecord, init_db 初始化数据库 init_db() app = FastAPI(title="行政强制档案管理系统") 配置CORS中间件，允许前端跨域访问 app.add_middleware( CORSMiddleware, allow_origins=[""], allow_credentials=True, allow_methods=[""], allow_headers=[""], ) 挂载静态文件目录，用于存储和访问上传的档案文件 UPLOAD_DIR = "archives_storage" os.makedirs(UPLOAD_DIR, exist_ok=True) app.mount("/static", StaticFiles(directory=UPLOAD_DIR), name="static") 数据库依赖获取 def get_db(): db = SessionLocal() try: yield db finally: db.close() @app.post("/upload_archive/") async def upload_archive( case_no: str = Form(...), party_name: str = Form(...), measure_type: str = Form(...), file: UploadFile = File(...), db: Session = Depends(get_db) ): """ 上传行政强制档案接口 接收案件基本信息及电子档案文件，保存至本地并记录数据库 """ 检查案件编号是否已存在 existing_record = db.query(ArchiveRecord).filter(ArchiveRecord.case_no == case_no).first() if existing_record: raise HTTPException(status_code=400, detail="该案件编号已存在，请勿重复归档") 构建文件存储路径：案件编号_文件名 file_location = f"{UPLOAD_DIR}/{case_no}_{file.filename}" 保存文件到本地 with open(file_location, "wb+") as buffer: shutil.copyfileobj(file.file, buffer) 创建数据库记录 db_record = ArchiveRecord( case_no=case_no, party_name=party_name, measure_type=measure_type, file_path=file_location ) db.add(db_record) db.commit() db.refresh(db_record) return {"message": "档案归档成功", "case_no": case_no, "file_path": file_location} @app.get("/list_archives/") async def list_archives(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)): """ 获取档案列表接口 支持分页查询 """ records = db.query(ArchiveRecord).offset(skip).limit(limit).all() results = [] for record in records: results.append({ "id": record.id, "case_no": record.case_no, "party_name": record.party_name, "measure_type": record.measure_type, "upload_time": record.upload_time.strftime("%Y-%m-%d %H:%M:%S") }) return results @app.get("/search_archive/") async def search_archive(case_no: str, db: Session = Depends(get_db)): """ 根据案件编号检索档案 """ record = db.query(ArchiveRecord).filter(ArchiveRecord.case_no == case_no).first() if not record: raise HTTPException(status_code=404, detail="未找到该案件档案") return { "case_no": record.case_no, "party_name": record.party_name, "measure_type": record.measure_type, "upload_time": record.upload_time.strftime("%Y-%m-%d %H:%M:%S"), "download_url": f"/static/{os.path.basename(record.file_path)}" } @app.get("/download_file/{filename}") async def download_file(filename: str): """ 文件下载接口 """ file_path = os.path.join(UPLOAD_DIR, filename) if not os.path.exists(file_path): raise HTTPException(status_code=404, detail="文件不存在") return FileResponse(file_path) ```

四、前端交互页面实现

为了方便操作，我们创建一个简单的HTML页面index.html，集成档案上传和查询功能。请将此文件保存在项目根目录下的static文件夹中（需手动创建static文件夹）。

首先创建静态文件夹：

```bash mkdir static ```

然后在static目录下创建index.html，内容如下：

```html ```

五、系统启动与功能验证

代码编写完成后，通过以下步骤启动系统并进行实际操作验证。

1. 启动后端服务
在项目根目录下执行以下命令启动FastAPI服务：

```bash uvicorn main:app --reload --host 0.0.0.0 --port 8000 ```

2. 访问前端界面
打开浏览器，直接访问static目录下的index.html文件，或者在浏览器地址栏输入：
file:///你的项目绝对路径/static/index.html

3. 执行档案上传操作
在页面“档案归档上传”区域，输入以下测试数据：
案件编号：(2023)测执001号
当事人名称：测试单位A
强制措施类型：查封场所
文件：选择任意一个PDF或图片文件作为测试档案。
点击上传归档按钮，若显示“归档成功”，则表示后端逻辑正常。

4. 验证查询功能
在页面“档案查询”区域，输入刚才的案件编号(2023)测执001号，点击查询。下方应立即显示该案件的详细信息，并包含一个可点击的下载链接。点击下载链接，浏览器应能下载回刚才上传的原始文件。

5. 检查数据持久化
停止后端服务（Ctrl+C），然后重新执行启动命令。刷新页面，观察“最新归档列表”是否依然保留之前上传的数据。如果数据存在，说明SQLite数据库持久化配置正确。