基于Python实现档案整理全流程自动化服务实操指南

一、环境准备与依赖安装

在开始构建档案整理服务之前，必须先配置好运行环境。本方案基于Python 3.9开发，核心依赖包括Tesseract-OCR（用于图像文字识别）、Watchdog（用于文件监听）和Flask（用于Web服务接口）。

1. 安装Tesseract-OCR引擎

OCR是档案数字化的核心，必须先安装系统级的识别引擎。

• Windows系统： 访问 https://github.com/UB-Mannheim/tesseract/wiki 下载 tesseract-ocr-w64-setup-5.x.x.exe，安装时务必勾选“Additional language data”，下载 chi_sim（简体中文）和 eng（英文）语言包。默认安装路径建议保持为 C:\Program Files\Tesseract-OCR。
• Linux系统： 直接执行命令安装：
  ```bash sudo apt update sudo apt install tesseract-ocr tesseract-ocr-chi-sim ```

2. 安装Python依赖库

在项目根目录下打开终端，执行以下命令安装所需Python库：

```bash pip install pytesseract pillow watchdog flask openpyxl python-dateutil ```

依赖说明：
- pytesseract: Tesseract的Python封装。
- pillow: 图像处理库，用于预处理图片。
- watchdog: 监控文件夹变化，实现文件放入即自动处理。
- flask: 提供API接口，支持远程调用整理服务。
- openpyxl: 用于生成档案索引Excel表。

二、项目目录结构初始化

为了保持代码清晰，请在本地创建一个名为 archive_service 的文件夹，并按照以下结构创建子文件夹和文件：

• archive_service/
  • config.json (配置文件)
  • ocr_engine.py (OCR识别模块)
  • file_handler.py (文件处理与归档逻辑)
  • app.py (主服务程序)
  • input/ (待处理文件夹，放入需要整理的图片或PDF)
  • output/ (整理完成后的存储文件夹)
  • logs/ (日志文件夹)

三、核心配置文件编写

创建 config.json 文件，用于统一管理路径和参数。这种方式比硬编码更便于后期维护。请复制以下内容：

```json { "tesseract_cmd": "C:\\Program Files\\Tesseract-OCR\\tesseract.exe", "watch_folder": "./input", "output_folder": "./output", "log_file": "./logs/service.log", "archive_rules": { "invoice": "发票", "contract": "合同", "idcard": "证件" }, "naming_pattern": "{date}_{category}_{hash}", "allowed_extensions": [".jpg", ".jpeg", ".png", ".pdf"] } ```

注意： 如果是Linux系统，请将 tesseract_cmd 修改为 /usr/bin/tesseract。

四、OCR识别模块开发

创建 ocr_engine.py。该模块负责调用Tesseract引擎，提取图片中的文字信息，这是自动分类和重命名的基础。

```python import pytesseract import json import os from PIL import Image 加载配置 with open('config.json', 'r', encoding='utf-8') as f: config = json.load(f) 设置Tesseract路径，仅Windows需要 if os.name == 'nt': pytesseract.pytesseract.tesseract_cmd = config['tesseract_cmd'] class OCREngine: @staticmethod def extract_text(image_path): """ 从图片中提取文字，支持中文和英文 """ try: 打开图片并进行简单的灰度处理以提高识别率 img = Image.open(image_path).convert('L') 使用Tesseract识别，lang='chi_sim+eng' 表示中英文混合识别 text = pytesseract.image_to_string(img, lang='chi_sim+eng') return text.strip() except Exception as e: print(f"OCR识别失败: {e}") return "" @staticmethod def detect_category(text): """ 根据关键词简单判断档案类别 """ rules = config['archive_rules'] detected_category = "未分类" for key, keyword in rules.items(): if keyword in text: detected_category = keyword break return detected_category ```

五、文件处理与归档逻辑

创建 file_handler.py。此模块负责核心业务逻辑：解析OCR结果、生成新文件名、移动文件并更新索引。为了简化，本示例直接处理图片，PDF需先拆分为图片。

```python import os import shutil import json import hashlib from datetime import datetime import openpyxl from ocr_engine import OCREngine 加载配置 with open('config.json', 'r', encoding='utf-8') as f: config = json.load(f) class ArchiveHandler: def __init__(self): self.output_folder = config['output_folder'] self.index_file = os.path.join(self.output_folder, "档案索引.xlsx") self._ensure_output_dirs() self._init_index_excel() def _ensure_output_dirs(self): """确保输出目录及分类子目录存在""" if not os.path.exists(self.output_folder): os.makedirs(self.output_folder) 为每个分类创建子文件夹 for key, val in config['archive_rules'].items(): cat_path = os.path.join(self.output_folder, val) if not os.path.exists(cat_path): os.makedirs(cat_path) 默认未分类文件夹 default_path = os.path.join(self.output_folder, "未分类") if not os.path.exists(default_path): os.makedirs(default_path) def _init_index_excel(self): """初始化Excel索引表""" if not os.path.exists(self.index_file): wb = openpyxl.Workbook() ws = wb.active ws.title = "档案索引" ws.append(["原文件名", "新文件名", "识别类别", "识别时间", "文件路径", "摘要内容"]) wb.save(self.index_file) def _get_file_hash(self, filepath): """计算文件Hash值用于防重命名""" with open(filepath, 'rb') as f: return hashlib.md5(f.read()).hexdigest()[:8] def process_file(self, file_path): """处理单个文件的完整流程""" filename = os.path.basename(file_path) print(f"开始处理文件: {filename}") 1. OCR识别 text_content = OCREngine.extract_text(file_path) category = OCREngine.detect_category(text_content) 2. 生成新文件名 file_hash = self._get_file_hash(file_path) current_time = datetime.now().strftime("%Y%m%d_%H%M%S") ext = os.path.splitext(filename)[1] 截取前20个字符作为摘要，去除非法字符 summary = text_content[:20].replace('\n', '').replace('/', '').replace('\\', '') new_filename = f"{current_time}_{category}_{file_hash}{ext}" 3. 移动文件 target_dir = os.path.join(self.output_folder, category) target_path = os.path.join(target_dir, new_filename) try: shutil.move(file_path, target_path) 4. 更新Excel索引 self._update_excel(filename, new_filename, category, current_time, target_path, summary) print(f"文件已归档至: {target_path}") return True except Exception as e: print(f"文件处理异常: {e}") return False def _update_excel(self, old_name, new_name, category, time, path, summary): """写入Excel记录""" wb = openpyxl.load_workbook(self.index_file) ws = wb.active ws.append([old_name, new_name, category, time, path, summary]) wb.save(self.index_file) ```

六、自动化服务主程序

创建 app.py。该文件集成了 Flask Web服务和 Watchdog 文件监听器。服务启动后，会自动监控 input 文件夹，一旦有新文件放入，即刻触发归档流程。同时提供了一个Web接口用于手动触发或查看状态。

```python import time import logging import json import sys from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler from flask import Flask, jsonify, request from file_handler import ArchiveHandler 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S') 加载配置 with open('config.json', 'r', encoding='utf-8') as f: config = json.load(f) app = Flask(__name__) handler = ArchiveHandler() class FileEventHandler(FileSystemEventHandler): def on_created(self, event): """当文件被创建时触发""" if event.is_directory: return 给文件一点时间写入完成，特别是大文件 time.sleep(1) ext = os.path.splitext(event.src_path)[1].lower() if ext in config['allowed_extensions']: logging.info(f"检测到新文件: {event.src_path}") 调用处理逻辑 handler.process_file(event.src_path) @app.route('/api/status', methods=['GET']) def get_status(): """服务健康检查接口""" return jsonify({ "status": "running", "service": "Archive Auto-Service", "watch_folder": config['watch_folder'] '] }) @app.route('/api/process', methods=['POST']) def manual_process(): """手动触发处理接口（可选）""" data = request.json file_path = data.get('path') if file_path and os.path.exists(file_path): result = handler.process_file(file_path) return jsonify({"success": result}) return jsonify({"success": False, "error": "File not found"}), 400 def start_service(): """启动文件监听服务""" event_handler = FileEventHandler() observer = Observer() observer.schedule(event_handler, config['watch_folder'], recursive=False) observer.start() logging.info(f"档案整理服务已启动，正在监听文件夹: {config['watch_folder']}") logging.info(f"Web服务接口已启动: http://0.0.0.0:5000") try: 启动Flask服务，这里使用非阻塞方式或直接运行 为了简单演示，我们在主线程运行Flask，Observer在后台线程 注意：在生产环境中建议使用gevent或gunicorn app.run(host='0.0.0.0', port=5000, use_reloader=False) except KeyboardInterrupt: observer.stop() observer.join() if __name__ == == "__main__": 确保必要的文件夹存在 if not os.path.exists(config['watch_folder']): os.makedirs(config['watch_folder']) start_service() ```

代码修正提示： 上述代码中 if __name__ == == "__main__": 请修正为 if __name__ == "__main__":（去除多余的等号）。

七、服务启动与实操验证

所有代码准备就绪后，即可启动服务进行验证。

1. 启动服务

在终端中执行：

```bash python app.py ```

看到终端输出 “档案整理服务已启动” 且无报错，即表示服务运行成功。

2. 验证自动化归档

1. 准备测试素材： 找一张包含“发票”字样的图片（JPG或PNG格式），命名为 test_img.jpg。
2. 放入监听文件夹： 将该图片复制到项目的 input 文件夹中。
3. 观察终端日志： 终端应立即显示检测到新文件并开始处理。
4. 检查结果：
   • 打开 output/发票/ 文件夹，文件应该已被移动至此，且文件名变更为类似 20231027_103000_发票_a1b2c3d4.jpg 的格式。
   • 打开 output/档案索引.xlsx，表中应新增了一条记录，包含原文件名、新路径和识别出的摘要。

3. API接口测试

打开浏览器或使用Postman访问 http://localhost:5000/api/status，应返回JSON格式的服务状态信息。

通过以上步骤，你已经成功搭建了一个基于OCR的档案整理全流程服务。该服务能够自动识别文件内容、分类、重命名、归档并建立索引，完全实现了从杂乱文件到结构化档案的自动化处理。