档案整理外包服务全流程数字化系统搭建指南

技术栈与环境准备

本方案采用Python作为核心开发语言，利用FastAPI构建高性能RESTful接口，结合PaddleOCR实现高精度文字识别，使用SQLite作为轻量级数据存储。这种组合无需复杂的中间件配置，适合快速落地档案整理外包服务的数字化管理。

需要在服务器或本地终端安装Python 3.8及以上版本。为了避免依赖冲突，请务必创建虚拟环境。执行以下命令完成基础环境的搭建：

```bash 创建项目目录 mkdir archive_system && cd archive_system 创建虚拟环境 python3 -m venv venv source venv/bin/activate Windows系统使用 venv\Scripts\activate 安装核心依赖库 pip install fastapi uvicorn python-multipart paddlepaddle paddleocr opencv-python ```

注意：PaddleOCR在首次运行时会自动下载模型文件，请确保服务器能连接外网，或者手动下载模型放置到指定目录。如果服务器没有GPU，PaddlePaddle会自动降级使用CPU运行，速度会稍慢，但功能完全可用。

数据库设计与初始化

档案整理外包的核心在于任务流转与状态管理。我们需要设计三张核心表：projects（项目表）、batches（批次表）和tasks（任务表）。SQLite数据库文件将自动生成在项目根目录下。

在项目根目录创建db_init.py文件，并写入以下完整的数据库初始化代码。该脚本包含建表逻辑及初始索引配置，确保后续查询效率。

```python import sqlite3 def init_db(): conn = sqlite3.connect('archive.db') cursor = conn.cursor() 项目表：存储外包项目的基础信息 cursor.execute(''' CREATE TABLE IF NOT EXISTS projects ( id INTEGER PRIMARY KEY AUTOINCREMENT, project_name TEXT NOT NULL, customer_name TEXT NOT NULL, status TEXT DEFAULT 'active', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') 批次表：管理档案的导入批次 cursor.execute(''' CREATE TABLE IF NOT EXISTS batches ( id INTEGER PRIMARY KEY AUTOINCREMENT, project_id INTEGER, batch_name TEXT NOT NULL, total_files INTEGER DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (project_id) REFERENCES projects (id) ) ''') 任务表：单页档案的处理状态 cursor.execute(''' CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, batch_id INTEGER, file_path TEXT NOT NULL, ocr_result TEXT, status TEXT DEFAULT 'pending', reviewed_by TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (batch_id) REFERENCES batches (id) ) ''') conn.commit() conn.close() print("数据库初始化完成") if __name__ == "__main__": init_db() ```

运行python db_init.py即可生成archive.db文件。这种设计支持多项目并行，每个项目下包含多个批次，每个批次包含具体的档案处理任务，清晰对应外包服务的实际业务流。

OCR识别核心模块开发

这是整个系统的技术核心。我们需要封装一个服务类，专门处理图像上传后的文字提取。PaddleOCR对中文支持极佳，且能自动处理图片旋转和倾斜。在项目根目录创建ocr_service.py。

以下代码实现了图像加载、OCR识别及结果格式化。为了提高实用性，我们增加了简单的正则匹配逻辑，尝试从识别结果中提取日期和文号，这是档案整理中最关键的元数据。

```python from paddleocr import PaddleOCR import cv2 import re class OCRService: def __init__(self): 初始化OCR引擎，use_angle_cls=True开启方向分类，识别率更高 self.ocr = PaddleOCR(use_angle_cls=True, lang="ch", show_log=False) def process_image(self, image_path): try: 读取图像 img = cv2.imread(image_path) if img is None: return {"error": "无法读取图像文件"} 执行识别 result = self.ocr.ocr(img, cls=True) full_text = "" metadata = {"date": None, "doc_no": None} if result and result[0]: 提取所有文本块 lines = [line[1][0] for line in result[0]] full_text = "\n".join(lines) 简单的元数据提取逻辑 匹配日期格式 (YYYY-MM-DD 或 YYYY年MM月DD日) date_pattern = re.compile(r'(\d{4}[-年]\d{1,2}[-月]\d{1,2}[日]?)') 匹配文号 (如：〔2023〕1号) doc_pattern = re.compile(r'[〔\[].?[〕\]].?[号第]') for line in lines: date_match = date_pattern.search(line) if date_match and not metadata["date"]: metadata["date"] = date_match.group(1) doc_match = doc_pattern.search(line) if doc_match and not metadata["doc_no"]: metadata["doc_no"] = doc_match.group(0) return { "text": full_text, "metadata": metadata, "status": "success" } except Exception as e: return {"error": str(e), "status": "failed"} 单例模式，避免重复加载模型导致内存溢出 ocr_instance = OCRService() ```

业务服务接口实现

接下来构建FastAPI主程序，提供文件上传、任务查询和结果审核接口。这些接口将直接对接前端的操作界面或外包人员的上传工具。创建main.py文件。

代码实现了文件流式写入磁盘、异步调用OCR服务更新数据库的逻辑。特别注意文件存储路径的规范化处理，防止路径穿越攻击。

```python from fastapi import FastAPI, UploadFile, File, HTTPException, Form from fastapi.responses import JSONResponse import sqlite3 import os import shutil from datetime import datetime from ocr_service import ocr_instance app = FastAPI(title="Archive Outsourcing System") 配置上传目录 UPLOAD_DIR = "uploads" os.makedirs(UPLOAD_DIR, exist_ok=True) def get_db_connection(): conn = sqlite3.connect('archive.db') conn.row_factory = sqlite3.Row return conn @app.post("/api/upload") async def upload_archive( file: File(...), project_id: int = Form(...), batch_name: str = Form(...) ): """上传档案图片并自动触发OCR识别""" file_location = f"{UPLOAD_DIR}/{project_id}_{batch_name}_{file.filename}" 保存文件 with open(file_location, "wb+") as file_object: shutil.copyfileobj(file.file, file_object) 数据库操作：创建或获取批次，插入任务 conn = get_db_connection() cursor = conn.cursor() 简化逻辑：直接插入批次（实际业务中应先查询是否存在） cursor.execute("INSERT INTO batches (project_id, batch_name) VALUES (?, ?)", (project_id, batch_name)) batch_id = cursor.lastrowid 初始状态为pending cursor.execute("INSERT INTO tasks (batch_id, file_path, status) VALUES (?, ?, 'pending')", (batch_id, file_location)) task_id = cursor.lastrowid conn.commit() 触发OCR处理 ocr_result = ocr_instance.process_image(file_location) if ocr_result.get("status") == "success": 更新任务状态和结果 cursor.execute( "UPDATE tasks SET ocr_result=?, status='completed' WHERE id=?", (ocr_result["text"], task_id) ) conn.commit() conn.close() return {"code": 200, "message": "上传并识别成功", "task_id": task_id, "data": ocr_result["metadata"]} else: conn.close() return JSONResponse(status_code=500, content={"code": 500, "message": "OCR识别失败", "error": ocr_result.get("error")}) @app.get("/api/tasks/{batch_id}") async def get_tasks(batch_id: int): """获取指定批次下的所有任务列表""" conn = get_db_connection() cursor = conn.cursor() cursor.execute("SELECT FROM tasks WHERE batch_id=?", (batch_id,)) tasks = cursor.fetchall() conn.close() result = [] for task in tasks: result.append({ "id": task["id"], "file_path": task["file_path"], "status": task["status"], "ocr_preview": task["ocr_result"][:100] + "..." if task["ocr_result"] else None 仅返回预览 }) return {"code": 200, "data": result} @app.post("/api/review/{task_id}") async def review_task(task_id: int, action: str = Form(...), corrected_text: str = Form(None)): """人工审核接口：通过或驳回""" if action not in ["approve", "reject"]: raise HTTPException(status_code=400, detail="Invalid action") conn = get_db_connection() cursor = conn.cursor() new_status = "approved" if action == "approve" else "rejected" if corrected_text: cursor.execute( "UPDATE tasks SET ocr_result=?, status=?, reviewed_at=? WHERE id=?", (corrected_text, new_status, str(datetime.now()), task_id) ) else: cursor.execute( "UPDATE tasks SET status=?, reviewed_at=? WHERE id=?", (new_status, str(datetime.now()), task_id) ) conn.commit() conn.close() return {"code": 200, "message": f"任务已{new_status}"} ```

系统启动与接口测试

所有代码准备就绪后，使用Uvicorn启动ASGI服务器。执行以下命令启动系统：

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

启动成功后，可以通过命令行工具curl模拟前端操作，验证整个流程是否通畅。假设我们有一张名为contract.jpg的档案图片。

1. 测试上传与OCR接口：该命令会将图片上传到服务器，自动完成识别并返回提取的元数据。

```bash curl -X POST "http://127.0.0.1:8000/api/upload" \ -F "file=@contract.jpg" \ -F "project_id=1" \ -F "batch_name=2023_contracts" ``>

预期返回JSON中包含task_id和提取到的日期、文号等信息。

2. 测试任务列表查询：使用上一步返回的批次ID（此处假设为1）查看任务状态。

```bash curl "http://127.0.0.1:8000/api/tasks/1" ``>

3. 测试人工审核接口：模拟外包人员对识别结果进行修正并审核通过。

```bash curl -X POST "http://127.0.0.1:8000/api/review/1" \ -F "action=approve" \ -F "corrected_text=修正后的准确档案内容文本" ``>

至此，一套完整的、可落地的档案整理外包服务技术方案已完成。该系统具备了文件接收、自动化OCR提取、任务状态管理及人工干预修正的核心闭环能力，可直接作为外包服务的技术底座进行二次开发或接入前端界面。