系列已经写了100多篇，覆盖了数据采集、清洗、存储、展示几乎完整的链路。但在实际业务中，有一个高频场景始终没有单独讲过——跨系统数据对账。
电商订单需要在ERP和电商平台之间对账、财务数据需要在银行系统和企业系统之间对账、库存需要在WMS和电商平台之间对账……这些场景的共同特征是：从两个（或多个）系统取数据，按规则比对，标记差异，生成对账报告。
今天专门讲数据对账的自动化方案——比对规则设计、大文件分块处理、差异分类与自动修复。
先说核心：对账的三类差异
| 差异类型 | 说明 | 示例 |
|---------|------|------|
| A有B无 | 系统A有记录，系统B没有 | 淘宝有订单，ERP里没有 |
| A无B有 | 系统B有记录，系统A没有 | ERP有订单，淘宝里没有 |
| 数据不一致 | 两边都有记录，但字段值不同 | 金额不一致、状态不一致 |
第一部分：对账流程架构
【系统A数据】 【系统B数据】
↓ ↓
采集（API/Excel/数据库） 采集（API/Excel/数据库）
↓ ↓
清洗（统一格式） 清洗（统一格式）
↓ ↓
↓ ↓
【数据比对引擎】
↓
【差异分类处理】
↓
【对账报告生成】
↓
【差异人工复核】
第二部分：数据采集与统一格式
2.1 定义统一数据格式
# ===== 统一订单格式 =====
# 无论来自哪个系统，都转换成这个格式
{
"order_id": "O20260701001", # 订单号（唯一键）
"platform": "淘宝", # 来源平台
"order_date": "2026-07-01", # 订单日期
"amount": 299.00, # 订单金额
"status": "已完成", # 订单状态
"buyer": "张三", # 买家
"item_name": "蓝牙耳机", # 商品名
"quantity": 1, # 数量
"source": "淘宝API" # 数据来源
}
2.2 从不同数据源统一采集
# ============================================
# 子流程：采集淘宝订单（通过API）
# ============================================
子流程 Fetch_Taobao_Orders(开始日期, 结束日期):
输出日志("【采集】淘宝订单：" + 开始日期 + " ~ " + 结束日期)
# 调用淘宝API获取订单
HTTP请求(
方法="GET",
URL="https://gw.api.taobao.com/router/rest?method=taobao.trade.fullinfo.get...",
请求头={"Authorization": "Bearer " + TOKEN}
)
订单列表 = 解析订单数据(HTTP响应内容)
# 转换为统一格式
统一格式列表 = []
按列表循环(订单列表, 循环项=订单):
统一格式 = {
"order_id": 订单["tid"],
"platform": "淘宝",
"order_date": 订单["created"][:10],
"amount": float(订单["payment"]),
"status": 映射状态(订单["status"]),
"buyer": 订单["buyer_nick"],
"item_name": 订单["orders"][0]["title"],
"quantity": int(订单["orders"][0]["num"]),
"source": "淘宝API"
}
将 统一格式 加入列表(统一格式列表)
结束循环
返回结果(统一格式列表)
结束子流程
# ============================================
# 子流程：采集ERP订单（通过数据库）
# ============================================
子流程 Fetch_ERP_Orders(开始日期, 结束日期):
输出日志("【采集】ERP订单：" + 开始日期 + " ~ " + 结束日期)
# 连接ERP数据库
执行Python代码(
输入: 开始日期, 结束日期,
输出: 订单列表,
代码="""
import pymysql
import json
conn = pymysql.connect(
host='192.168.1.100',
user='rpa_user',
password='xxx',
database='erp'
)
cursor = conn.cursor()
cursor.execute('''
SELECT order_no, create_date, total_amount, order_status, customer_name, product_name, qty
FROM orders
WHERE create_date BETWEEN %s AND %s
''', (start_date, end_date))
rows = cursor.fetchall()
conn.close()
# 转换为统一格式
result = []
for row in rows:
result.append({
'order_id': row[0],
'platform': 'ERP',
'order_date': row[1].strftime('%Y-%m-%d'),
'amount': float(row[2]),
'status': row[3],
'buyer': row[4],
'item_name': row[5],
'quantity': int(row[6]),
'source': 'ERP数据库'
})
订单列表 = json.dumps(result, ensure_ascii=False)
"""
)
返回结果(订单列表)
结束子流程
第三部分：数据比对引擎
3.1 建立索引加快比对
# ============================================
# 比对引擎（Python）
# ============================================
import pandas as pd
import json
def reconcile(data_a, data_b, key_field='order_id'):
"""
数据对账核心引擎
data_a, data_b: 两个系统的数据列表
key_field: 用于匹配的唯一键
返回：比对结果
"""
# 转换为DataFrame
df_a = pd.DataFrame(data_a)
df_b = pd.DataFrame(data_b)
# 建立索引（加速查找）
df_a.set_index(key_field, inplace=True)
df_b.set_index(key_field, inplace=True)
# 获取所有键
keys_a = set(df_a.index)
keys_b = set(df_b.index)
# 差异分类
only_in_a = keys_a - keys_b # A有B无
only_in_b = keys_b - keys_a # A无B有
both = keys_a & keys_b # 共有
# 数据不一致（字段值不同）
inconsistent = []
for key in both:
row_a = df_a.loc[key]
row_b = df_b.loc[key]
diff_fields = []
# 比较关键字段（排除key本身）
for field in ['amount', 'status', 'quantity']:
if field in row_a and field in row_b:
if row_a[field] != row_b[field]:
diff_fields.append({
'field': field,
'value_a': row_a[field],
'value_b': row_b[field]
})
if diff_fields:
inconsistent.append({
'key': key,
'differences': diff_fields,
'row_a': row_a.to_dict(),
'row_b': row_b.to_dict()
})
# 返回结果
return {
'total_a': len(df_a),
'total_b': len(df_b),
'only_in_a': list(only_in_a),
'only_in_b': list(only_in_b),
'both_count': len(both),
'inconsistent': inconsistent,
'consistent_count': len(both) - len(inconsistent)
}
3.2 影刀中调用比对引擎
# ============================================
# 影刀端：执行对账
# ============================================
# 1. 采集数据
调用子流程(Fetch_Taobao_Orders, 输入: 开始日期, 结束日期, 输出=淘宝订单)
调用子流程(Fetch_ERP_Orders, 输入: 开始日期, 结束日期, 输出=ERP订单)
# 2. 执行对账
执行Python代码(
输入:
data_a=淘宝订单,
data_b=ERP订单,
key_field="order_id",
输出:
reconciliation_result=对账结果,
代码=上面的比对引擎代码
)
解析JSON(对账结果, 存入变量=对账结果对象)
# 3. 输出统计
输出日志("===== 对账结果 =====")
输出日志("A系统（淘宝）总订单数：" + 对账结果["total_a"])
输出日志("B系统（ERP）总订单数：" + 对账结果["total_b"])
输出日志("A有B无：" + 获取列表长度(对账结果["only_in_a"]) + " 条")
输出日志("A无B有：" + 获取列表长度(对账结果["only_in_b"]) + " 条")
输出日志("数据不一致：" + 获取列表长度(对账结果["inconsistent"]) + " 条")
输出日志("数据一致：" + 对账结果["consistent_count"] + " 条")
第四部分：差异分类与处理策略
4.1 差异分类
# ============================================
# 差异分类
# ============================================
# 获取差异列表
A有B无列表 = 对账结果["only_in_a"]
A无B有列表 = 对账结果["only_in_b"]
不一致列表 = 对账结果["inconsistent"]
# ----- 差异分类 -----
设置变量(严重差异 = [])
设置变量(中等差异 = [])
设置变量(轻微差异 = [])
# 1. 金额差异 > 10元 → 严重
按列表循环(不一致列表, 循环项=差异):
金额A = 差异["row_a"]["amount"]
金额B = 差异["row_b"]["amount"]
金额差 = 绝对值(金额A - 金额B)
如果 金额差 > 10:
将 差异 加入列表(严重差异)
否则如果 金额差 > 1:
将 差异 加入列表(中等差异)
否则:
将 差异 加入列表(轻微差异)
结束If
结束循环
# 2. 状态不一致 → 中等
按列表循环(不一致列表, 循环项=差异):
状态A = 差异["row_a"]["status"]
状态B = 差异["row_b"]["status"]
如果 状态A != 状态B:
如果 差异 不在 严重差异 中:
将 差异 加入列表(中等差异)
结束If
结束If
结束循环
# 3. 数量差异 → 中等
按列表循环(不一致列表, 循环项=差异):
数量A = 差异["row_a"]["quantity"]
数量B = 差异["row_b"]["quantity"]
如果 数量A != 数量B:
如果 差异 不在 严重差异 中 且 差异 不在 中等差异 中:
将 差异 加入列表(中等差异)
结束If
结束If
结束循环
4.2 差异自动修复
# ============================================
# 差异修复策略
# ============================================
# 1. 金额差异 → 以淘宝为准（假设电商平台数据更准确）
按列表循环(严重差异, 循环项=差异):
订单号 = 差异["key"]
正确金额 = 差异["row_a"]["amount"]
输出日志("【修复】订单 " + 订单号 + " 金额以淘宝为准：" + 正确金额)
# 更新ERP数据库
执行Python代码(
输入: order_id=订单号, correct_amount=正确金额,
代码=更新ERP金额的SQL
)
结束循环
# 2. A有B无 → 补录到ERP
按列表循环(A有B无列表, 循环项=订单号):
订单详情 = 查找订单(淘宝订单, 订单号)
输出日志("【补录】订单 " + 订单号 + " 补录到ERP")
# 调用ERP创建订单API
调用子流程(Create_ERP_Order, 输入: 订单详情)
结束循环
# 3. A无B有 → 补录到淘宝（或标记为待确认）
按列表循环(A无B有列表, 循环项=订单号):
订单详情 = 查找订单(ERP订单, 订单号)
输出日志("【标记】订单 " + 订单号 + " 在淘宝中不存在，标记待确认")
# 写入待确认列表
将 订单详情 加入列表(待确认订单列表)
结束循环
4.3 差异报告生成
# ============================================
# 生成对账报告
# ============================================
报告 = """
📊 **数据对账报告 - {日期}**
━━━━━━━━━━━━━━━━━━━━━━━━━━━
📋 **总体统计**
• A系统（淘宝）：{total_a} 条
• B系统（ERP）：{total_b} 条
• 一致：{consistent_count} 条
• 不一致：{inconsistent_count} 条
⚠️ **差异明细**
**严重差异（{severe_count}条）**
{severe_detail}
**中等差异（{medium_count}条）**
{medium_detail}
**轻微差异（{minor_count}条）**
{minor_detail}
🔧 **修复建议**
{fix_suggestions}
━━━━━━━━━━━━━━━━━━━━━━━━━━━
🕐 生成时间：{time}
"""
# 填充报告
报告 = 报告.replace("{日期}", 当前日期)
报告 = 报告.replace("{total_a}", str(对账结果["total_a"]))
报告 = 报告.replace("{total_b}", str(对账结果["total_b"]))
报告 = 报告.replace("{consistent_count}", str(对账结果["consistent_count"]))
报告 = 报告.replace("{inconsistent_count}", str(len(不一致列表)))
报告 = 报告.replace("{severe_count}", str(获取列表长度(严重差异)))
# ... 其他替换
# 发送报告
调用子流程(Send_Notification, 输入: 报告)
第五部分：大文件对账优化
当数据量超过1万行时，需要分块处理：
# ============================================
# 分块对账（大数据量）
# ============================================
子流程 Chunk_Reconciliation(系统A数据, 系统B数据, 分块大小=1000):
"""
分批对账，避免内存溢出
"""
A总数 = 获取列表长度(系统A数据)
B总数 = 获取列表长度(系统B数据)
设置变量(所有差异 = [])
# 按分块大小分批
A块数 = (A总数 + 分块大小 - 1) // 分块大小
B块数 = (B总数 + 分块大小 - 1) // 分块大小
# 对A的每个分块，与B的所有分块比对
# 但更高效：对B建立索引后再分块比对
执行Python代码(
输入: data_a=系统A数据, data_b=系统B数据, chunk_size=分块大小,
输出: 差异结果,
代码="""
import pandas as pd
def chunk_reconcile(data_a, data_b, chunk_size):
df_b = pd.DataFrame(data_b)
df_b.set_index('order_id', inplace=True)
all_diffs = []
total_chunks = (len(data_a) + chunk_size - 1) // chunk_size
for i in range(total_chunks):
start = i * chunk_size
end = min(start + chunk_size, len(data_a))
chunk_a = data_a[start:end]
df_chunk = pd.DataFrame(chunk_a)
df_chunk.set_index('order_id', inplace=True)
# 合并对账
merged = df_chunk.join(df_b, how='inner', lsuffix='_a', rsuffix='_b')
# 比较字段...
# （省略具体比对逻辑）
print(f"已处理分块 {i+1}/{total_chunks}")
return all_diffs
"""
)
返回结果(差异结果)
结束子流程
第六部分：完整主流程
# ============================================
# 主流程：Daily_Reconciliation
# 执行频率：每天一次
# ============================================
输出日志("===== 每日对账流程启动 =====")
# ----- 1. 设置对账日期（默认昨天）-----
对账日期 = 获取昨天日期()
输出日志("对账日期：" + 对账日期)
# ----- 2. 采集数据 -----
输出日志("【1/4】采集数据")
调用子流程(Fetch_Taobao_Orders, 输入: 对账日期, 输出=淘宝订单)
调用子流程(Fetch_ERP_Orders, 输入: 对账日期, 输出=ERP订单)
# ----- 3. 执行对账 -----
输出日志("【2/4】执行对账")
调用子流程(Reconcile_Data,
输入: 淘宝订单, ERP订单,
输出: 对账结果
)
# ----- 4. 差异处理 -----
输出日志("【3/4】差异处理")
调用子流程(Process_Differences, 输入: 对账结果)
# ----- 5. 生成报告 -----
输出日志("【4/4】生成报告")
调用子流程(Generate_Reconciliation_Report, 输入: 对账结果)
输出日志("===== 每日对账流程完成 =====")
常见问题速查
| 问题 | 原因 | 解决方法 |
|---|---|---|
| 数据量太大内存不足 | 全部加载到内存 | 用分块对账 |
| 订单号格式不一致 | 不同系统命名规则不同 | 做订单号映射表 |
| 时区导致日期不对 | 系统时区不同 | 统一用UTC时间 |
| 金额小数点位数不同 | 系统精度不同 | 统一四舍五入到两位 |
| 状态码不一致 | 系统映射不同 | 建立状态映射表 |
推荐资源
- 本系列第63篇《SQLite/MySQL数据库读写》
- Python Pandas文档（
merge、join、concat） - 本系列第96篇《数据质量与一致性校验》
#影刀RPA #RPA自动化 #数据对账 #跨系统 #数据一致性
作者：林焱
本文为《影刀RPA学习手册》系列文章之一，内容源于实操经验的整理与分享。