对账自动化脚本实现：账单解析、字段归一化与差分算法

对账这件事，流程讲起来十分钟就能说明白：拿账单、比本地流水、处理差异。但真正动手写脚本时，问题全在细节里——账单是压缩包还是纯文本、编码是 GBK 还是 UTF-8、金额带不带元字符、微信给分而支付宝给元、差异怎么分类落库、任务失败了怎么重跑。

这篇给可改造的代码实现。流程与方法论见对账与账单下载操作手册，本文只讲怎么把它写成程序。

一、账单文件的三个坑

在写解析代码之前，先把这三件事确认清楚，否则后面全部白做。

第一个坑：文件不是纯 CSV。 国内通道的账单多为 ZIP 压缩包，里面才是 CSV。而且压缩包内文件名可能带日期，解析时不能写死文件名，要用"取压缩包内第一个 CSV"这种容错逻辑。

第二个坑：编码不一致。 部分账单是 GBK 编码。直接用 UTF-8 读会抛异常或产生乱码，导致金额解析出奇怪的字符。做法是先尝试 UTF-8，失败后回落 GBK。

第三个坑：前后有说明行。 账单开头常有导出时间、商户号之类的摘要行，结尾可能有汇总行。直接 csv.DictReader 会把它们当成数据行，造成笔数与总额全错。

import csv, io, zipfile

def load_bill_rows(path: str) -> list[dict]:
    """读取账单，兼容 zip/csv 与 UTF-8/GBK 编码。"""
    if path.endswith('.zip'):
        with zipfile.ZipFile(path) as zf:
            name = next(n for n in zf.namelist() if n.lower().endswith('.csv'))
            content = zf.read(name)
    else:
        with open(path, 'rb') as fh:
            content = fh.read()

    for enc in ('utf-8-sig', 'gbk'):
        try:
            text = content.decode(enc)
            break
        except UnicodeDecodeError:
            continue
    else:
        raise ValueError(f'无法识别账单编码: {path}')

    # 定位表头：以包含订单号关键字的行作为起点
    lines = text.splitlines()
    start = next(
        i for i, line in enumerate(lines)
        if ('商户订单号' in line) or ('out_trade_no' in line.lower())
    )
    reader = csv.DictReader(io.StringIO('\n'.join(lines[start:])))
    return [row for row in reader if _looks_like_data(row)]

def _looks_like_data(row: dict) -> bool:
    """过滤汇总行与空行：数据行必须有订单号。"""
    first = next(iter(row.values()), '') or ''
    return bool(first.strip()) and not first.startswith('总')

这里有两个刻意的设计：表头定位用关键字而不是固定行号，因为不同日期导出的摘要行数会变；数据行过滤用"必须有订单号"，比判断列数更稳。

二、字段归一化

这是整个脚本里最关键的一步。三个通道的字段名、金额单位、时间格式都不一样，不归一化就没法比。

语义	微信支付	支付宝	银联
商户订单号	商户订单号	商户订单号	商户订单号
通道流水号	微信订单号	支付宝交易号	交易流水号
金额	分	元	元
交易时间	秒级时间串	日期时间	日期时间
交易状态	交易状态	交易状态	交易状态

from decimal import Decimal
from datetime import datetime

def normalize(row: dict, channel: str) -> dict:
    def pick(*keys):
        for k in keys:
            v = row.get(k) or row.get(k.strip())
            if v not in (None, ''):
                return str(v).strip()
        return ''

    raw_amount = pick('订单金额', '总金额', '金额')
    amount = int((Decimal(raw_amount) * 100).quantize(Decimal('1'))) \
        if channel == 'alipay' or channel == 'unionpay' else int(Decimal(raw_amount))

    ts = pick('交易时间', '支付时间', '交易创建时间')
    trade_time = _parse_time(ts)

    return {
        'channel':      channel,
        'order_no':     pick('商户订单号', 'order_no'),
        'channel_txn':  pick('微信订单号', '支付宝交易号', '交易流水号'),
        'amount':       amount,          # 统一为分
        'trade_time':   trade_time,
        'raw_status':   pick('交易状态', '支付状态'),
    }

def _parse_time(value: str) -> datetime:
    for fmt in ('%Y-%m-%d %H:%M:%S', '%Y%m%d%H%M%S'):
        try:
            return datetime.strptime(value, fmt)
        except (ValueError, TypeError):
            continue
    raise ValueError(f'无法解析交易时间: {value!r}')

两个要点值得强调。金额用 Decimal 再乘 100 转整数，不要用 float——0.1 + 0.2 != 0.3 这类浮点误差在对账场景会直接变成"差一分钱"。时间解析要枚举多种格式，微信账单里的时间字段常是紧凑格式，支付宝是带分隔符的，写死一种必然报错。

三、三层差分算法

差分的核心是以订单号为键做集合运算。不要用双层循环，几万行时性能会崩。

def reconcile(local: list[dict], remote: list[dict]) -> dict:
    # 第一层：笔数与总额
    local_total = sum(r['amount'] for r in local)
    remote_total = sum(r['amount'] for r in remote)
    summary = {
        'local_count':  len(local),
        'remote_count': len(remote),
        'local_amount': local_total,
        'remote_amount': remote_total,
        'balanced': len(local) == len(remote) and local_total == remote_total,
    }

    # 第二层：逐笔匹配
    local_map  = {r['order_no']: r for r in local}
    remote_map = {r['order_no']: r for r in remote}
    only_local  = [local_map[k]  for k in local_map.keys()  - remote_map.keys()]
    only_remote = [remote_map[k] for k in remote_map.keys() - local_map.keys()]

    # 第三层：金额与时间校验
    amount_diff = [
        {**local_map[k], 'remote_amount': remote_map[k]['amount']}
        for k in local_map.keys() & remote_map.keys()
        if local_map[k]['amount'] != remote_map[k]['amount']
    ]

    return {
        'summary': summary,
        'only_local': only_local,
        'only_remote': only_remote,
        'amount_diff': amount_diff,
    }

返回的三类差异对应三种处置：

only_remote 是长款，典型原因是回调丢失 → 触发查单补单；
only_local 是短款，可能是伪造回调或误判 → 立即冻结订单并人工核查；
amount_diff 是金额差异，多为单位或优惠分摊错误 → 修正后重跑。

四、差错表与结果输出

差异不能只打日志，要落库形成可追踪的差错台账：

CREATE TABLE `recon_discrepancies` (
  `id` bigint unsigned NOT NULL AUTO_INCREMENT,
  `recon_date` date NOT NULL,
  `channel` varchar(16) NOT NULL,
  `order_no` varchar(32) NOT NULL,
  `diff_type` varchar(24) NOT NULL COMMENT 'overage/shortage/amount_diff',
  `local_amount` bigint DEFAULT NULL,
  `remote_amount` bigint DEFAULT NULL,
  `reason` varchar(255) DEFAULT NULL,
  `status` tinyint NOT NULL DEFAULT 0 COMMENT '0待处理 1处理中 2已闭环',
  `owner` varchar(64) DEFAULT NULL,
  `created_at` datetime NOT NULL,
  `closed_at` datetime DEFAULT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_date_channel_order` (`recon_date`, `channel`, `order_no`, `diff_type`)
);

唯一索引的作用是让任务可以安全重跑：同一天同一通道同一订单的同类型差异只会有一条记录，重跑时用 INSERT ... ON DUPLICATE KEY UPDATE 更新而不是重复插入。这一点在任务失败重试时非常关键。

五、定时调度与告警

对账任务放在 T+1 的早晨执行，通常 06:00–09:00 之间完成，因为账单在凌晨生成。调度上有三条纪律：

前一日的账单可能延迟生成，任务要能识别"账单不存在"并在 30 分钟后重试，而不是直接报失败；
任务必须幂等，重跑不产生重复数据；
结果必须推送，不能只写在日志里。

告警分两级：第一层不平就告警（笔数或总额不一致，属于结构性故障），第二层差异超阈值告警（差异笔数超过日常基线，属于趋势异常）。日常基线需要积累两三周数据才能定，一开始可以只对第一层告警。

六、五个实际会遇到的问题

编码问题：GBK 与 UTF-8 混用，用上面的回落逻辑解决，不要只支持一种。

金额精度：账单里的金额是字符串，直接用 float() 转换会引入误差，必须走 Decimal。

重复订单号：极少数账单里同一订单号会出现多行（比如分账场景）。差分前要先按订单号聚合，否则会被误判为差异。

按时区跨天：账单日期是通道侧的日期，本地订单用 UTC 存储时要注意换算，否则会把跨日订单算成两天的差异。

内存占用：一个月的账单可能有几十万行，chunk 读取或按日期分批是一开始就该做的设计，不要等 OOM 才改。

三个技术篇之间的关系是：订单号与幂等保证"不漏不重"，回调框架保证"状态正确"，对账脚本保证"最终一致"。前两篇见支付订单号设计与幂等表实现与多通道支付回调框架实现，退款与结算运营见退款与清结算操作手册。更多支付方案可参见支付分类。

来源：16IDC 原创，文中涉及的产品能力、费率与准入要求以相关服务商官网文档及合同为准。