检察与公安数字档案馆数据交换与OFD归档实战指南

一、开发环境与核心依赖构建

在构建跨部门数字档案馆数据交换中间件时，为了保证数据的完整性与可追溯性，我们选择Python作为核心开发语言，配合FastAPI框架提供高性能的API接口。此方案能够无缝处理公安与检察系统之间的元数据映射及OFD版式文档的校验工作。

请在Linux服务器（推荐CentOS 7.9或Ubuntu 20.04）上执行以下命令，建立隔离的虚拟环境并安装必要的依赖库。这些库涵盖了Web服务、数据校验、加密算法及OFD处理工具。

```bash 创建项目目录 mkdir police_procuratorate_archive cd police_procuratorate_archive 创建虚拟环境 python3.9 -m venv venv source venv/bin/activate 安装核心依赖（务必指定版本以防止兼容性问题） pip install fastapi==0.95.0 uvicorn==0.21.1 python-multipart==0.0.6 pydantic==1.10.7 cryptography==39.0.1 Pillow==9.5.0 ofdparser==2.3.0 ```

二、公安与检察元数据映射模型设计

公安数字档案馆与检察数字档案馆遵循不同的元数据标准。公安侧通常使用警综平台标准，而检察侧遵循检察业务应用系统的标准。为了实现数据自动流转，必须在中间件层建立一个双向映射字典。

请在项目根目录下创建 models.py 文件，并写入以下代码。这段代码定义了标准化的数据模型，并实现了字段名的自动转换逻辑。

```python from pydantic import BaseModel, Field from typing import Optional, List from datetime import datetime 定义通用的档案交换标准模型 class ArchiveExchangeModel(BaseModel): file_name: str = Field(..., description="文件名") file_hash: str = Field(..., description="文件SHA256哈希值") business_id: str = Field(..., description="业务唯一标识") org_code: str = Field(..., description="单位编码") create_time: datetime = Field(default_factory=datetime.now, description="归档时间") 扩展字段，用于存储未映射的原始数据 extended_data: dict = {} 字段映射配置表：Key为公安字段，Value为检察字段 MAPPING_RULES = { "gong_an_aj_bh": "jian_cha_aj_bh", 案件编号 "gong_an_xgr_mc": "jian_cha_xgr_mc", 相关人名称 "gong_an_aj_lb": "jian_cha_aj_lb", 案件类别 "gong_an_sahj": "jian_cha_sahj" 涉案环节 } def transform_metadata(police_data: dict) -> dict: """ 将公安侧元数据转换为检察侧元数据 """ procuratorate_data = {} 处理标准映射字段 for police_key, value in police_data.items(): if police_key in MAPPING_RULES: procuratorate_data[MAPPING_RULES[police_key]] = value else: 无法映射的字段放入扩展区 procuratorate_data.setdefault("extended_data", {})[police_key] = value return procuratorate_data ```

三、OFD文件完整性校验与转换工具

数字档案馆的核心是电子文件的长期保存，OFD（Open Fixed-layout Document）是国家标准版式文档格式。在交换过程中，必须严格校验文件格式是否合法，以及文件内容是否被篡改。

创建 utils.py 文件，实现文件的哈希计算与OFD结构解析功能。这里使用 ofdparser 库进行解析，确保上传的文件是真正的OFD文档而非改了后缀的伪造文件。

```python import hashlib import ofdparser from PIL import Image import io def calculate_sha256(file_path: str) -> str: """ 计算文件的SHA256值，用于确保文件传输完整性 """ sha256_hash = hashlib.sha256() with open(file_path, "rb") as f: for byte_block in iter(lambda: f.read(4096), b""): sha256_hash.update(byte_block) return sha256_hash.hexdigest() def validate_ofd_structure(file_path: str) -> bool: """ 校验OFD文件结构是否完整 尝试解析OFD文件，如果抛出异常则视为无效文件 """ try: 使用ofdparser加载文件，这会解析OFD的物理结构和逻辑结构 ofd = ofdparser.parse_ofd(file_path) 进一步校验：尝试获取第一页的图片数据，确保内容可渲染 如果OFD内部损坏，这一步通常会失败 pages = ofd.get_pages() if not pages: return False 简单的渲染测试（不保存到磁盘，仅测试可读性） first_page = pages[0] img_data = first_page.get_image() img = Image.open(io.BytesIO(img_data)) img.verify() 验证图片完整性 return True except Exception as e: print(f"OFD校验失败: {str(e)}") return False ```

四、国密算法安全传输实现

由于涉及公安与检察敏感数据，数据传输必须符合国密标准（SM4）。本节实现一个基于SM4的ECB模式加密工具，用于对传输的元数据进行加密处理。

在 utils.py 中追加以下代码。注意：实际生产环境中，密钥应从专门的密钥管理服务（KMS）获取，此处仅做演示。

```python from Crypto.Cipher import AES from Crypto.Util.Padding import pad, unpad import base64 模拟国密SM4算法（注：SM4分组长度与AES一致，此处使用AES库作为SM4的接口替代演示 在生产环境中请替换为 `gmssl` 库调用真正的SM4算法） SM4_KEY = b'1q2w3e4r5t6y7u8i' 16字节密钥 def sm4_encrypt(plaintext: str) -> str: """ SM4加密 """ try: cipher = AES.new(SM4_KEY, AES.MODE_ECB) padded_data = pad(plaintext.encode('utf-8'), AES.block_size) ciphertext = cipher.encrypt(padded_data) return base64.b64encode(ciphertext).decode('utf-8') except Exception as e: raise ValueError(f"加密失败: {str(e)}") def sm4_decrypt(ciphertext_b64: str) -> str: """ SM4解密 """ try: ciphertext = base64.b64decode(ciphertext_b64) cipher = AES.new(SM4_KEY, AES.MODE_ECB) decrypted_padded = cipher.decrypt(ciphertext) decrypted = unpad(decrypted_padded, AES.block_size) return decrypted.decode('utf-8') except Exception as e: raise ValueError(f"解密失败: {str(e)}") ```

五、API接口封装与业务逻辑集成

我们将上述模块整合到FastAPI服务中。我们需要创建两个接口：/upload/police 用于接收公安侧数据，处理后存入检察数字档案馆前置库。

创建 main.py 文件：

```python from fastapi import FastAPI, File, UploadFile, Form, HTTPException from fastapi.responses import JSONResponse import shutil import os import json from models import ArchiveExchangeModel, transform_metadata from utils import calculate_sha256, validate_ofd_structure, sm4_encrypt app = FastAPI(title="检察公安数字档案交换中间件") 配置存储路径 UPLOAD_DIR = "./archive_storage" os.makedirs(UPLOAD_DIR, exist_ok=True) @app.post("/api/v1/exchange/police_to_procuratorate") async def exchange_archive( file: UploadFile = File(..., description="OFD电子文件"), metadata: str = Form(..., description="公安侧元数据JSON字符串"), token: str = Form(..., description="安全认证Token") ): """ 接收公安侧档案，校验后转换为检察标准 """ 1. 基础鉴权（示例：简单Token校验） if token != "SECURE_TOKEN_2024": raise HTTPException(status_code=403, detail="无权访问") 2. 保存临时文件 temp_file_path = os.path.join(UPLOAD_DIR, file.filename) try: with open(temp_file_path, "wb") as buffer: shutil.copyfileobj(file.file, buffer) 3. OFD格式与完整性校验 is_valid_ofd = validate_ofd_structure(temp_file_path) if not is_valid_ofd: os.remove(temp_file_path) raise HTTPException(status_code=400, detail="上传文件不是有效的OFD文档或已损坏") 4. 计算文件哈希 file_hash = calculate_sha256(temp_file_path) 5. 解析并转换元数据 try: police_meta = json.loads(metadata) except json.JSONDecodeError: raise HTTPException(status_code=400, detail="元数据JSON格式错误") 执行字段映射 procuratorate_meta = transform_metadata(police_meta) 补充文件校验信息到元数据 procuratorate_meta['file_hash'] = file_hash procuratorate_meta['file_name'] = file.filename 6. 数据加密准备（模拟传输给检察系统前的加密） 将转换后的元数据转为JSON字符串并加密 encrypted_payload = sm4_encrypt(json.dumps(procuratorate_meta, ensure_ascii=False)) 7. 构建响应结果 response_data = { "status": "success", "message": "档案接收并转换成功", "original_business_id": police_meta.get("gong_an_aj_bh"), "target_business_id": procuratorate_meta.get("jian_cha_aj_bh"), "encrypted_metadata_preview": encrypted_payload[:50] + "...", 仅展示部分 "file_saved_at": temp_file_path } return JSONResponse(content=response_data) except Exception as e: 发生任何异常确保清理文件 if os.path.exists(temp_file_path): os.remove(temp_file_path) raise HTTPException(status_code=500, detail=f"服务器内部错误: {str(e)}") if __name__ == "__main__": import uvicorn 启动服务，监听所有IP，端口8000 uvicorn.run(app, host="0.0.0.0", port=8000) ```

六、服务启动与接口测试

代码编写完成后，直接启动服务进行验证。不要使用简单的编辑器运行，而是使用生产级的命令行启动。

```bash 启动API服务 python main.py ```

服务启动后，使用 curl 命令模拟公安系统发送一个OFD文件和对应的元数据。请准备一个名为 test.ofd 的测试文件放在当前目录下。

```bash curl -X POST "http://127.0.0.1:8000/api/v1/exchange/police_to_procuratorate" \ -F "file=@test.ofd" \ -F "metadata={\"gong_an_aj_bh\": \"GA202304001\", \"gong_an_xgr_mc\": \"张三\", \"gong_an_aj_lb\": \"刑事案件\"}" \ -F "token=SECURE_TOKEN_2024" ```

如果返回结果中包含 "status": "success" 并且 target_business_id 有值，说明数据已成功从公安标准映射到检察标准，且OFD文件校验通过。此时检查 ./archive_storage 目录，即可看到已归档的文件。这套流程直接打通了两个异构系统间的数据壁垒，实现了纯技术层面的自动化归档。