档案系统集成实战：手把手教你搭建中间件实现数据互通

技术方案架构解析

档案管理系统集成困难的核心痛点通常在于：老旧的OA或ERP系统只提供数据库视图或过时的WebService接口，而新档案系统通常采用RESTful API。直接修改源系统风险极高，通过编写轻量级Python中间件进行ETL（抽取、转换、加载）是最低成本、最高效的落地方案。

本方案采用“定时轮询+断点续传”模式。中间件连接源系统数据库，读取增量数据，经过字段映射清洗后，通过HTTP POST推送到档案系统。所有配置外部化，无需重启服务即可调整映射逻辑。

开发环境极速搭建

本指南基于Python 3.8+环境，利用其丰富的生态库实现零门槛开发。请直接复制以下命令在服务器终端执行，完成依赖安装。

1. 安装Python依赖库

我们需要操作数据库、发送HTTP请求、处理定时任务和读取配置文件。执行以下命令：

```bash pip install requests pymysql schedule configparser ```

步骤一：配置源数据库与目标接口

为了实现灵活配置，我们创建一个config.ini文件。请将此文件放在代码同级目录下。这里假设源系统为MySQL，目标档案系统接口为/api/v1/archive/import。

创建 config.ini 文件：

```ini [source] db_host = 192.168.1.100 db_port = 3306 db_user = readonly_user db_pass = secure_password_here db_name = oa_database 增量字段，通常为更新时间或自增ID increment_column = update_time [target] api_url = http://archive-system.internal/api/v1/archive/import api_token = your_archive_system_token [system] 每次拉取的数据量，防止内存溢出 batch_size = 100 轮询间隔（分钟） interval_minutes = 10 ```

步骤二：编写数据抽取模块

此模块负责从源系统安全地获取数据。为了防止数据库连接超时，我们使用上下文管理器。同时，必须支持增量拉取，否则每次全量同步会导致性能灾难。

核心代码逻辑：

```python import pymysql import configparser from datetime import datetime, timedelta class SourceExtractor: def __init__(self, config_path): config = configparser.ConfigParser() config.read(config_path) self.db_conf = config['source'] def get_connection(self): return pymysql.connect( host=self.db_conf['db_host'], port=int(self.db_conf['db_port']), user=self.db_conf['db_user'], password=self.db_conf['db_pass'], database=self.db_conf['db_name'], charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor ) def fetch_incremental_data(self, last_sync_time): """ 根据上次同步时间获取增量数据 """ query = """ SELECT id, doc_title, doc_type, creator, create_time, file_path, status FROM oa_documents WHERE update_time > %s AND status = 1 ORDER BY update_time ASC LIMIT 1000 """ conn = None try: conn = self.get_connection() with conn.cursor() as cursor: cursor.execute(query, (last_sync_time,)) return cursor.fetchall() except Exception as e: print(f"数据库查询失败: {e}") return [] finally: if conn: conn.close() ```

步骤三：编写数据转换与清洗逻辑

源系统字段（如doc_title）通常与档案系统标准字段（如title）不一致。此环节负责字段重命名、日期格式化和必填项校验。

核心代码逻辑：

```python import json class DataTransformer: def transform(self, raw_data_list): transformed_list = [] for item in raw_data_list: try: 字段映射与清洗 payload = { "archiveTitle": item['doc_title'].strip(), 去除首尾空格 "categoryCode": self._map_category(item['doc_type']), "creatorName": item['creator'], "archiveDate": item['create_time'].strftime('%Y-%m-%d %H:%M:%S'), "fileUrl": item['file_path'], "sourceSystemId": str(item['id']), 关键：用于去重 "businessType": "OA_SYNC" } 必填项校验 if not payload['archiveTitle'] or not payload['fileUrl']: print(f"数据缺失，跳过ID: {item['id']}") continue transformed_list.append(payload) except Exception as e: print(f"数据转换失败，原始数据ID: {item.get('id')}, 错误: {e}") return transformed_list def _map_category(self, doc_type): 简单的字典映射，实际可配置化 mapping = { "contract": "1001", "invoice": "1002", "report": "1003" } return mapping.get(doc_type, "9999") 默认分类 ```

步骤四：编写档案系统API推送模块

此模块负责将清洗后的JSON数据推送到目标接口。必须包含重试机制和详细的错误日志，以便排查网络波动或接口报错问题。

核心代码逻辑：

```python import requests import time class TargetLoader: def __init__(self, config_path): config = configparser.ConfigParser() config.read(config_path) self.api_url = config['target']['api_url'] self.token = config['target']['api_token'] def send_data(self, data_list): headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.token}" } success_count = 0 for data in data_list: retry = 0 max_retries = 3 while retry < max_retries: try: response = requests.post( self.api_url, json=data, headers=headers, timeout=10 ) if response.status_code == 200: res_json = response.json() if res_json.get('code') == 0: 假设0表示成功 success_count += 1 print(f"推送成功: SourceID={data['sourceSystemId']}") break else: print(f"业务逻辑错误: {res_json.get('msg')}") break else: raise Exception(f"HTTP {response.status_code}") except Exception as e: retry += 1 print(f"推送失败，重试 {retry}/{max_retries}: {e}") time.sleep(2) 延迟2秒后重试 return success_count ```

步骤五：主程序入口与定时任务

将上述模块串联，并增加状态管理功能。我们需要将“最后一次同步时间”持久化到本地文件state.txt中，确保服务重启后能从断点继续。

完整运行脚本 main.py：

```python import schedule import time import os from datetime import datetime 假设上述类定义在同一文件或已导入 from extractor import SourceExtractor from transformer import DataTransformer from loader import TargetLoader STATE_FILE = "state.txt" def get_last_sync_time(): if not os.path.exists(STATE_FILE): 如果是第一次运行，默认同步当前时间前1小时的数据 return datetime.now() - timedelta(hours=1) with open(STATE_FILE, 'r') as f: time_str = f.read().strip() return datetime.strptime(time_str, '%Y-%m-%d %H:%M:%S') def update_sync_time(current_time): with open(STATE_FILE, 'w') as f: f.write(current_time.strftime('%Y-%m-%d %H:%M:%S')) def job(): print(f"开始执行同步任务: {datetime.now()}") 1. 初始化组件 extractor = SourceExtractor('config.ini') transformer = DataTransformer() loader = TargetLoader('config.ini') 2. 获取断点时间 last_time = get_last_sync_time() print(f"上次同步时间: {last_time}") 3. 抽取数据 raw_data = extractor.fetch_incremental_data(last_time) if not raw_data: print("没有增量数据，任务结束。") return 4. 清洗转换 clean_data = transformer.transform(raw_data) print(f"待推送数据条数: {len(clean_data)}") 5. 推送数据 if clean_data: loader.send_data(clean_data) 6. 更新状态时间（取最后一条数据的时间，防止边界问题） 实际生产中建议取当前服务器时间，避免时钟回拨问题 update_sync_time(datetime.now()) print("任务执行完毕。") if __name__ == "__main__": 立即执行一次 job() 设置定时任务，每10分钟执行一次 schedule.every(10).minutes.do(job) while True: schedule.run_pending() time.sleep(1) ```

部署运行与效果验证

代码编写完成后，通过以下步骤进行部署。建议使用screen或supervisor管理进程，防止终端关闭后程序退出。

1. 验证配置

检查config.ini中的数据库账号密码和API地址是否正确。确保中间件服务器网络能够同时访问源数据库和档案系统API端口。

2. 启动服务

直接运行Python脚本：

```bash python3 main.py ```

观察控制台输出，应能看到“开始执行同步任务”、“推送成功”等日志。如果遇到数据库连接错误，请检查防火墙和MySQL白名单设置。

3. 后台运行（生产环境）

使用nohup命令在后台运行，并将日志输出到sync.log文件：

```bash nohup python3 main.py > sync.log 2>&1 & ```

通过tail -f sync.log实时查看运行状态。至此，一个高可用、可维护的档案系统集成中间件已部署完成，完美解决了异构系统间的数据孤岛问题。