零门槛落地人事档案数字化转递系统实操指南

一、技术架构与环境准备

本方案旨在解决人事档案从物理扫描到电子化转递的全流程问题。为了确保系统具备高并发处理能力及数据完整性校验，后端采用 Python FastAPI 框架，数据存储使用轻量级且无需复杂配置的 SQLite，文件存储格式遵循国家档案局标准（PDF/A + XML元数据）。该架构无需安装额外的数据库服务，适合快速落地。

1.1 基础环境安装

首先需要准备 Python 运行环境。请务必使用 Python 3.9 或更高版本，以确保类型注解及异步特性的正常使用。

在终端中执行以下命令安装核心依赖库：

```bash pip install fastapi uvicorn python-multipart aiofiles ```

依赖说明：

• fastapi：用于构建高性能的Web API接口。
• uvicorn：ASGI服务器，用于运行FastAPI应用。
• python-multipart：处理表单数据及文件上传。
• aiofiles：异步文件读写，提升高并发下的IO性能。

1.2 目录结构规划

在项目根目录下，请严格按照以下结构创建文件夹和文件，这直接关系到后续代码的运行：

```text archive_transfer_system/ ├── main.py 主程序入口 ├── storage/ 文件存储根目录 │ ├── uploads/ 临时接收上传的原始文件 │ └── packages/ 封装好的标准转递包 └── archive.db SQLite数据库文件（自动生成） ```

二、数据库模型设计与初始化

我们需要记录每一次转递操作的详细信息，包括转递单号、人员姓名、身份证号、文件哈希值及转递状态。在 main.py 中直接嵌入数据库初始化逻辑，无需编写额外的SQL脚本。

在 main.py 顶部添加以下代码：

```python import sqlite3 import os from datetime import datetime 数据库初始化函数 def init_db(): conn = sqlite3.connect('archive.db') cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS transfer_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, transfer_id TEXT UNIQUE NOT NULL, name TEXT NOT NULL, id_card TEXT NOT NULL, file_path TEXT NOT NULL, file_hash TEXT NOT NULL, status TEXT DEFAULT 'pending', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') conn.commit() conn.close() 启动时自动初始化 if not os.path.exists('archive.db'): init_db() ```

三、核心功能实现：生成标准化转递包

这是系统的核心。根据档案数字化转递规范，一个合格的转递包不能仅是一个PDF文件，必须包含元数据描述文件（XML）和电子原文，并进行整体打包。我们将实现一个函数，将上传的PDF文件自动封装为标准的ZIP转递包。

在 main.py 中继续添加核心处理逻辑：

```python import zipfile import hashlib import uuid from fastapi import FastAPI, UploadFile, File, Form, HTTPException from fastapi.responses import JSONResponse from typing import Optional app = FastAPI() 确保目录存在 os.makedirs("storage/uploads", exist_ok=True) os.makedirs("storage/packages", exist_ok=True) def calculate_md5(file_path): """计算文件的MD5值，用于数据完整性校验""" hash_md5 = hashlib.md5() with open(file_path, "rb") as f: for chunk in iter(lambda: f.read(4096), b""): hash_md5.update(chunk) return hash_md5.hexdigest() def create_transfer_package(name: str, id_card: str, pdf_file_path: str) -> dict: """ 封装标准转递包 1. 生成XML元数据 2. 将PDF与XML打包为ZIP 3. 计算包哈希 """ transfer_id = str(uuid.uuid4()) timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") 1. 生成XML元数据内容 xml_content = f""" {transfer_id} {name} {id_card} {timestamp} archive.pdf application/pdf """ 定义输出ZIP路径 zip_filename = f"{transfer_id}.zip" zip_path = os.path.join("storage/packages", zip_filename) 2. 打包文件 with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zf: 写入元数据XML zf.writestr("metadata.xml", xml_content) 写入PDF原文 zf.write(pdf_file_path, arcname="archive.pdf") 3. 计算最终包的MD5 file_hash = calculate_md5(zip_path) return { "transfer_id": transfer_id, "zip_path": zip_path, "file_hash": file_hash } ```

四、API接口开发：上传与转递服务

我们需要一个接口供前端或客户端调用。该接口接收人员基本信息和PDF文件，执行打包逻辑，并将结果存入数据库。请将以下代码追加到 main.py：

```python @app.post("/api/transfer/upload") async def upload_and_transfer( name: str = Form(...), id_card: str = Form(...), file: UploadFile = File(...) ): 1. 校验文件类型 if not file.filename.endswith('.pdf'): raise HTTPException(status_code=400, detail="仅支持PDF格式文件") 2. 保存临时文件 temp_file_path = os.path.join("storage/uploads", file.filename) try: with open(temp_file_path, "wb") as buffer: content = await file.read() buffer.write(content) 3. 生成转递包 package_info = create_transfer_package(name, id_card, temp_file_path) 4. 写入数据库 conn = sqlite3.connect('archive.db') cursor = conn.cursor() try: cursor.execute( "INSERT INTO transfer_records (transfer_id, name, id_card, file_path, file_hash, status) VALUES (?, ?, ?, ?, ?, ?)", (package_info['transfer_id'], name, id_card, package_info['zip_path'], package_info['file_hash'], 'success') ) conn.commit() except Exception as e: conn.rollback() raise HTTPException(status_code=500, detail=f"数据库写入失败: {str(e)}") finally: conn.close() 5. 清理临时文件 os.remove(temp_file_path) return JSONResponse(content={ "code": 200, "message": "转递包生成成功", "data": { "transfer_id": package_info['transfer_id'], "download_url": f"/api/download/{package_info['transfer_id']}", "md5": package_info['file_hash'] } }) except Exception as e: 发生异常时清理临时文件 if os.path.exists(temp_file_path): os.remove(temp_file_path) raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/download/{transfer_id}") async def download_package(transfer_id: str): """提供转递包下载接口""" conn = sqlite3.connect('archive.db') cursor = conn.cursor() cursor.execute("SELECT file_path FROM transfer_records WHERE transfer_id=?", (transfer_id,)) result = cursor.fetchone() conn.close() if not result: raise HTTPException(status_code=404, detail="转递包不存在") file_path = result[0] from fastapi.responses import FileResponse return FileResponse(file_path, filename=os.path.basename(file_path)) ```

五、前端对接实操（纯HTML版）

为了方便直接测试，我们编写一个简单的HTML页面。在项目根目录创建 index.html，直接复制以下代码即可。该页面包含了文件选择、表单提交及结果展示逻辑。

```html ```

六、全流程测试与验证

代码编写完毕后，必须进行完整的端到端测试，确保每个环节无遗漏。

6.1 启动服务

在终端进入项目目录，执行以下命令启动服务：

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

看到 Uvicorn running on http://0.0.0.0:8000 提示即表示启动成功。

6.2 执行转递操作

1. 打开页面：在浏览器中直接打开刚才创建的 index.html 文件。
2. 填写数据：输入姓名（如：张三）、身份证号。
3. 上传文件：选择一个本地的PDF测试文件。
4. 提交：点击“生成并转递”按钮。

6.3 结果校验

操作成功后，页面会显示绿色的成功提示框。请按以下步骤严格验证数据正确性：

• 下载转递包：点击页面上的“点击下载转递包”链接，下载生成的ZIP文件。
• 解压检查：解压该ZIP文件，内部必须包含两个文件：metadata.xml 和 archive.pdf。
• 元数据核对：打开 metadata.xml，检查 和 是否与刚才输入的一致，且包含唯一的 。
• 数据库验证：使用 SQLite 客户端打开项目目录下的 archive.db，执行 SELECT FROM transfer_records;，确认表中新增了一条记录，且 status 字段为 success。

至此，一套符合标准、具备数据校验能力的人事档案数字化转递微系统已完全落地。该方案可直接集成到现有的OA或HR系统中，只需调用 /api/transfer/upload 接口即可实现自动化转递。