档案数字化进度服务：从零搭建实时进度监控系统

一、系统核心架构与工具选型

本系统旨在解决档案数字化过程中进度不透明、管理困难的问题。我们将构建一个轻量级、可实时更新的监控服务，其核心架构分为数据采集、数据处理、数据存储与前端展示四个模块。

你需要准备以下工具，所有工具均为开源免费：

• Python 3.8+：作为后端主要开发语言。
• Flask：轻量级Web框架，用于构建API。
• SQLite：嵌入式数据库，适合轻量级应用，无需安装。
• HTML/CSS/JavaScript (Vanilla JS)：纯原生前端，无需框架。
• 系统命令行：用于执行扫描脚本。

创建项目目录并初始化环境：

``` mkdir archive_digitization_monitor && cd archive_digitization_monitor python -m venv venv Windows: venv\Scripts\activate Linux/Mac: source venv/bin/activate pip install flask ```

二、数据库设计与初始化

在项目根目录下，创建database.py文件，用于定义数据模型和初始化数据库。

``` import sqlite3 import os DB_NAME = 'digitization.db' def init_db(): 连接数据库，如果不存在则自动创建 conn = sqlite3.connect(DB_NAME) cursor = conn.cursor() 创建项目表 cursor.execute(''' CREATE TABLE IF NOT EXISTS projects ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, total_files INTEGER DEFAULT 0, scanned_files INTEGER DEFAULT 0, ocr_files INTEGER DEFAULT 0, archived_files INTEGER DEFAULT 0, last_update TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') 创建任务日志表 cursor.execute(''' CREATE TABLE IF NOT EXISTS task_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, project_id INTEGER, stage TEXT, -- 阶段：scanning, ocr, archiving file_name TEXT, status TEXT, -- 状态：processing, completed, failed timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (project_id) REFERENCES projects (id) ) ''') conn.commit() conn.close() print(f"数据库 {DB_NAME} 初始化完成。") if __name__ == '__main__': init_db() ```

运行此脚本以创建数据库：

``` python database.py ```

执行后，当前目录下会生成一个digitization.db文件。

三、构建后端API服务

创建app.py文件，这是整个服务的核心。我们将实现项目管理和进度更新的API。

``` from flask import Flask, request, jsonify, g import sqlite3 import os from datetime import datetime app = Flask(__name__) DB_NAME = 'digitization.db' 数据库连接辅助函数 def get_db(): if 'db' not in g: g.db = sqlite3.connect(DB_NAME) g.db.row_factory = sqlite3.Row 支持字典方式访问行数据 return g.db @app.teardown_appcontext def close_db(error): if 'db' in g: g.db.close() 1. 创建新项目 @app.route('/api/project', methods=['POST']) def create_project(): data = request.json name = data.get('name') total = data.get('total_files', 0) if not name: return jsonify({'error': '项目名称不能为空'}), 400 db = get_db() cursor = db.cursor() cursor.execute( 'INSERT INTO projects (name, total_files) VALUES (?, ?)', (name, total) ) db.commit() project_id = cursor.lastrowid return jsonify({'id': project_id, 'name': name}), 201 2. 更新项目进度（核心接口） @app.route('/api/progress/', methods=['POST']) def update_progress(project_id): data = request.json stage = data.get('stage') scanning, ocr, archiving increment = data.get('increment', 1) 本次完成的数量，默认为1 file_name = data.get('file_name', '') if stage not in ['scanning', 'ocr', 'archiving']: return jsonify({'error': '无效的阶段类型'}), 400 db = get_db() cursor = db.cursor() 根据阶段更新对应的计数字段 column_map = { 'scanning': 'scanned_files', 'ocr': 'ocr_files', 'archiving': 'archived_files' } column = column_map[stage] 使用原子操作更新进度 cursor.execute(f''' UPDATE projects SET {column} = {column} + ?, last_update = CURRENT_TIMESTAMP WHERE id = ? ''', (increment, project_id)) 插入任务日志 cursor.execute(''' INSERT INTO task_logs (project_id, stage, file_name, status) VALUES (?, ?, ?, 'completed') ''', (project_id, stage, file_name)) db.commit() 返回更新后的项目状态 cursor.execute('SELECT FROM projects WHERE id = ?', (project_id,)) project = cursor.fetchone() return jsonify(dict(project)) 3. 获取所有项目进度概览 @app.route('/api/projects', methods=['GET']) def get_projects(): db = get_db() cursor = db.cursor() cursor.execute('SELECT FROM projects ORDER BY last_update DESC') projects = cursor.fetchall() return jsonify([dict(p) for p in projects]) 4. 获取单个项目详情与日志 @app.route('/api/project/', methods=['GET']) def get_project_detail(project_id): db = get_db() cursor = db.cursor() cursor.execute('SELECT FROM projects WHERE id = ?', (project_id,)) project = cursor.fetchone() if not project: return jsonify({'error': '项目不存在'}), 404 cursor.execute(''' SELECT FROM task_logs WHERE project_id = ? ORDER BY timestamp DESC LIMIT 50 ''', (project_id,)) logs = cursor.fetchall() return jsonify({ 'project': dict(project), 'recent_logs': [dict(log) for log in logs] }) if __name__ == '__main__': 确保数据库已初始化 if not os.path.exists(DB_NAME): from database import init_db init_db() app.run(debug=True, host='0.0.0.0', port=5000) ```

启动后端服务：

``` python app.py ```

服务将在 http://localhost:5000 运行。你可以使用Postman或curl测试API。例如，创建一个新项目：

``` curl -X POST http://localhost:5000/api/project \ -H "Content-Type: application/json" \ -d '{"name":"2023年财务档案", "total_files": 1000}' ```

四、实现进度数据采集脚本

数字化流程通常涉及扫描、OCR、归档等环节。我们需要一个脚本，在每完成一个文件时，调用API更新进度。创建update_progress.py脚本。

``` import requests import sys import argparse API_BASE = 'http://localhost:5000/api' def update_progress(project_id, stage, file_name=None, increment=1): """调用API更新进度""" url = f'{API_BASE}/progress/{project_id}' payload = { 'stage': stage, 'file_name': file_name, 'increment': increment } try: resp = requests.post(url, json=payload) resp.raise_for_status() print(f"进度更新成功: 项目{project_id}, 阶段{stage}, 文件{file_name}") return resp.json() except requests.exceptions.RequestException as e: print(f"进度更新失败: {e}") return None if __name__ == '__main__': 通过命令行参数调用 parser = argparse.ArgumentParser(description='更新数字化进度') parser.add_argument('--project', type=int, required=True, help='项目ID') parser.add_argument('--stage', choices=['scanning', 'ocr', 'archiving'], required=True, help='处理阶段') parser.add_argument('--file', help='处理的文件名') parser.add_argument('--batch', type=int, default=1, help='批量完成的数量，默认为1') args = parser.parse_args() update_progress(args.project, args.stage, args.file, args.batch) ```

如何集成到实际工作流：

• 扫描环节：在扫描仪软件完成一个文件扫描并保存后，调用python update_progress.py --project 1 --stage scanning --file “档案001.pdf”。
• OCR环节：在OCR批处理脚本中，每成功处理一个文件，就调用一次，将--stage参数改为ocr。
• 批量更新：如果一个批处理任务完成了50个文件，使用--batch 50参数一次性更新。

五、开发实时监控前端页面

创建templates目录，并在其中创建index.html文件。这是一个纯原生HTML/JS页面，用于展示实时进度。

``` 档案数字化进度实时监控

正在加载数据...

```

为了让Flask能渲染这个页面，需要在app.py文件末尾（if __name__ == '__main__':之前）添加一个路由：

``` from flask import render_template @app.route('/') def index(): return render_template('index.html') ```

将index.html文件移动到项目根目录下新建的templates文件夹内。重启app.py服务，在浏览器中访问http://localhost:5000即可看到实时监控看板。

六、生产环境部署与优化

1. 使用生产级WSGI服务器

开发服务器不适合生产环境。使用Gunicorn（Linux/Mac）或Waitress（Windows）部署。

``` 安装Gunicorn pip install gunicorn 启动服务（Linux/Mac） gunicorn -w 4 -b 0.0.0.0:8000 app:app 安装Waitress（Windows） pip install waitress 在app.py同级目录创建wsgi.py文件，内容为： from app import app if __name__ == "__main__": from waitress import serve serve(app, host="0.0.0.0", port=8000) 然后运行：python wsgi.py ```

2. 数据库优化与备份

为防止数据丢失，设置一个简单的自动备份脚本backup_db.py：

``` import sqlite3 import shutil from datetime import datetime import os def backup_database(): db_name = 'digitization.db' if not os.path.exists(db_name): print("数据库文件不存在。") return timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') backup_name = f'backup/digitization_backup_{timestamp}.db' os.makedirs('backup', exist