INV Manager待提交单据批量脚本提交与对账实践

INV Manager 的待提交列表堆到四位数那天，手工点提交这条路基本就走到头了。这里的 INV Manager 指的是业务侧的单据管理模块，存放所有待落账的库存与物料流转单据；MTI 和 MMTT 是下游两套处理链路，前者负责物料转移指令的校验与派发，后者承担多模式事务的跟踪与最终落账。两者对外都提供提交入口，但报文结构、校验规则、返回码体系并不一致，中间还夹着一层路由判断。用脚本提交要解决的，就是把“查单—组报文—提交—回写状态—失败重试—对账”这一整套循环交给代码跑，人只在异常清单冒出来的时候介入。

这篇内容面向三类人：一是手上压着一批单据、被运营催着要结果的一线执行人员；二是刚接手这套接口、准备写自动化脚本的开发；三是想在动手之前搞清楚有哪些坑、避免返工的技术负责人。下面我会从链路位置、接口摸底、脚本实现、问题排查四个层面把这件事拆开讲，代码和参数都按可复现的方式给出，拿到就能改成自己环境能跑的版本。

1. 场景拆解：脚本提交到底在解决什么问题

在动手写第一行代码之前，得先把这条链路在地图上画出来。很多人一上来就抓接口文档闷头写请求，写了三百行才发现提交完之后状态回写不上、重复提交又被下游挡掉，返工的代价比前期摸清链路大得多。这一节先把位置、痛点、选型三件事说清楚。

1.1 INV Manager 与 MTI/MMTT 在链路里的位置

把整条链路想成一条流水线：INV Manager 是原料仓库，存放着已经录好但还没往下一道工序送的半成品单据；MTI 是第一道加工站，负责校验这批单据的物料编码、批次归属、数量平衡关系，校验通过后派发成正式的转移指令；MMTT 是第二道加工站，它把 MTI 派发出来的指令按不同的业务模式（比如普通流转、跨批次流转、临时调拨）分别跟踪，最终决定这张单据能不能落账。

关键在于，INV Manager 本身并不“知道”单据在 MTI/MMTT 里走到了哪一步。它只有一个状态字段，而这个字段是需要外部回写的。脚本在这里扮演的是“传送带 + 记录员”的双重角色：既要把单据送过去，也要把送过去之后的结果记回来。漏掉回写这一步，单据就会一直挂在“待提交”状态，下一次跑脚本又会被捞出来，形成无限循环提交。

我在模拟项目 A 里见过更隐蔽的情况：有一批单据在 MTI 校验通过、但 MMTT 落账失败，状态停在中间态，脚本只判断了“提交成功与否”，没有判断“落账是否完成”，结果这批单据在两边系统里都是“看起来正常”的，直到月末对账才炸出来。链路设计的第一个原则就是：提交成功不等于处理成功，状态机必须覆盖到最终落账。

1.2 手工提交的痛点与脚本化的真实收益

手工提交在单据量小于五十条的时候是可以忍受的，超过两百条就开始出问题，超过一千条基本就是灾难。我梳理过实际卡点，大致集中在四个地方。

第一是速度上限极低。手工提交单条平均要四十秒左右（切页面、填参数、核对、点提交、等返回、复制单号去查结果），一天满负荷也就七百条上下，还得是人不走神的前提。

第二是错误率随疲劳上升。最常见的错误不是填错字段，而是“重复提交同一张单”和“漏提交某一页的单”。前者会把下游搞出一堆重复指令，后者会让账实差异一直挂着。

第三是异常处理完全靠人记。哪张单失败了、失败在 MTI 还是 MMTT、报的什么码、要不要改数据重提，手工流程里这些信息基本散落在聊天记录和便签上，隔一天就找不回来了。

第四是无法定时执行。很多业务场景要求单据在固定时间窗口内进入下游，比如班次交接前必须清空待提交队列，人不在工位上就只能干等。

脚本化的收益不是“省人力”这么虚的说法，而是把单条处理时间从四十秒压到零点二秒级、把重复率压到零、把异常清单变成结构化数据、把执行时间变成可编排的任务。这四件事叠加起来，才是真正的价值。

1.3 方案选型：脚本直连、批量接口还是走中间件

三条路各有适用边界，选错了会很难受。

我的判断标准比较朴素：如果每天待提交单据在五千条以下，且下游单条接口的响应时间在三百毫秒以内，脚本直连就够了，不要为了“看起来架构合理”去上中间件，运维成本会吃掉所有收益。反过来，如果单据需要跨多个下游系统、且要求“要么全成功要么全回滚”，那脚本直连会很痛苦，因为跨系统的补偿逻辑要自己写，这时候中间件或者统一的事务服务才划算。

还有一个中间选项经常被忽略：脚本直连 + 本地落库。也就是脚本自己维护一张任务表，记录每张单据的提交状态、幂等键、重试次数、最后错误码。这张表成本极低（一个轻量数据库就够），但它的存在把“断点续跑”“失败重投”“对账核对”三件事一次性解决了。我后面给的实现都是按这个思路走的，强烈建议不要省这一步。

2. 提交前的准备：把接口和数据摸清楚

这一节是整件事里最耗时间、也最不能省的部分。接口文档能告诉你的通常只有一半，另一半得靠抓包、试探和问下游同学。摸不清就动手，后面会花三倍时间还债。

2.1 必须问清楚的三件事：端点、报文、返回码

端点要确认的不只是 URL，还包括环境区分（测试环境和生产环境的路径前缀往往不同）、方法（POST 还是 PUT，很多系统两个都支持但语义不同）、超时设置（下游默认超时可能是三十秒，脚本侧如果设成五秒会大量误判为失败）。

报文要确认的是必填字段、字段类型、长度上限、枚举取值。有个细节特别容易翻车：某些字段在文档里写“可选”，但实际上 MMTT 落账时如果缺失会走默认业务模式，导致单据落到了错误的分支上——这种问题不会报错，只会让结果对不上。

返回码是重头戏。必须拿到完整的码表，并且逐条确认“这个码代表什么、要不要重试、重试前要不要改数据”。我在模拟项目 B 里踩过的坑是：下游返回了一个“处理中”的中间码，脚本当成失败去重试，结果同一个幂等键被提交了三次，下游虽然去重了但每次都在流水里留了记录，对账时凭空多出两倍的流水。

下面这张表是常见返回码的归类方式，具体码值以实际文档为准，但分类逻辑是通用的。

2.2 鉴权方式与会话保持

内部系统的鉴权五花八门，常见的有三类：固定令牌放请求头、先换临时凭证再带凭证请求、以及双向证书。前两类在脚本里都好处理，第三类需要额外准备证书文件和路径配置。

用固定令牌的，注意令牌的过期时间和轮换策略。我遇到过令牌有效期是八小时，脚本跑批跑了两小时没问题，但补数据任务在深夜跑的时候令牌早就过期了，报的还是“无权限”这种模糊错误，排查花了半天。解决办法很简单：令牌不要硬编码在脚本里，从环境变量或者配置中心读，并且在初始化时先做一次轻量探测请求验证令牌有效。

用临时凭证的，要注意凭证的复用。每条单据都去换一次凭证是浪费，凭证本身有有效期，通常可以覆盖一整批任务。但反过来，跑批时间如果超过凭证有效期，中途要能自动刷新——这一点必须在脚本里显式处理，不能指望跑批一定在有效期内完成。

还有一种情况是下游做了来源限制，脚本所在机器的出口需要登记。这个在测试环境往往不生效，一到生产就被挡，所以联调阶段一定要用生产的网络路径做一次验证，别等上线才发现。

2.3 字段映射：把业务单据翻译成处理报文

字段映射是脚本的核心逻辑，也是最容易出错的地方。建议单独用一个映射配置文件，而不是散落在代码里，因为映射关系变更的频率远高于代码逻辑变更。

下面是一个典型的映射表样例，左边是 INV Manager 的字段，右边是提交给 MTI 与 MMTT 的字段。

转换规则里有几个坑值得单说。数量精度：INV Manager 存的是四位小数，MTI 要求三位，直接截断会丢精度导致数量不平；正确做法是按单位的最小计量单位四舍五入，并且换算后再校验一次总和是否守恒。类型字典：INV Manager 的单据类型是业务自定义的，MTI 的类型是固定枚举，中间必须有一张对照表，而且这张表要能热更新，不能写死在代码里。时间字段：不同系统的时区处理不一致，统一转 UTC 存、按需展示，能省掉大量“时间差八小时”的争论。

2.4 幂等键：重试不重复提交的底线

幂等键的设计要满足两个条件：同一张单据在同一版本下，无论提交多少次，键值必须一致；单据内容发生实质变化（比如改了数量）后，键值必须变化，否则下游会把新内容当成旧内容挡掉。

我用的构造方式是：单据号 + 版本号 + 业务类型，做一次哈希取前 16 位。

import hashlib def build_idem_key(doc_no: str, version: int, biz_type: str) -> str: raw = f"{doc_no}|{version}|{biz_type}" return hashlib.sha1(raw.encode("utf-8")).hexdigest()[:16]

取 16 位是为了在日志和排查界面里好读，碰撞概率在这个量级下可以忽略。版本号来自 INV Manager 的修改计数，每次单据内容变更会自增，这样天然满足了“内容变了键就变”的要求。

这里有个反面案例：有人用“单据号 + 提交时间戳”做幂等键，结果每次重试键都不一样，下游的幂等机制形同虚设，一次网络抖动就制造了两条重复指令。幂等键的原料里绝对不能包含时间、随机数、机器标识这类易变因素。

3. 脚本实现：从单条打通到批量跑通

到这一步接口和数据都清楚了，可以开始写。我的习惯是先写一个“只跑一条”的最小版本，把鉴权、组报文、提交、解析返回、回写这五步走通，再往上叠批量和重试。一次性写完整版然后调不通，是很难定位的。

3.1 最小可用版本：先把单条跑通

最小版本的代码不超过八十行，但已经把主干逻辑包含了。

import os import json import time import logging import requests BASE = os.environ["INV_BASE"] TOKEN = os.environ["INV_TOKEN"] TIMEOUT = (3, 15) # 连接3秒，读取15秒 logger = logging.getLogger("inv-submit") def fetch_pending(limit=1): resp = requests.get( f"{BASE}/documents", params={"status": "PENDING", "limit": limit}, headers={"Authorization": f"Bearer {TOKEN}"}, timeout=TIMEOUT, ) resp.raise_for_status() return resp.json().get("items", []) def build_payload(doc): return { "transfer_no": doc["doc_no"][:32], "transfer_type": TYPE_MAP[doc["doc_type"]], "material_code": doc["item_code"].zfill(10), "source_location": doc.get("from_loc") or DEFAULT_LOC, "target_location": doc["to_loc"], "quantity": round(float(doc["qty"]), 3), "uom": UOM_MAP[doc["unit"]], "batch": (doc.get("batch_no") or "").upper(), "create_time": to_iso8601(doc["doc_time"]), "idempotency_key": build_idem_key(doc["doc_no"], doc["version"], "MTI"), } def submit_one(doc): payload = build_payload(doc) resp = requests.post( f"{BASE}/mti/submit", json=payload, headers={ "Authorization": f"Bearer {TOKEN}", "X-Idempotency-Key": payload["idempotency_key"], }, timeout=TIMEOUT, ) return resp.status_code, resp.json() def run_once(): docs = fetch_pending(1) if not docs: logger.info("no pending document") return code, body = submit_one(docs[0]) logger.info("submit result code=%s body=%s", code, json.dumps(body, ensure_ascii=False))

这段代码的价值不在功能，而在于它把“外部依赖”暴露干净了：令牌从环境变量走，超时显式设置，幂等键随报文一起带。跑通这一条之后，你手上就有了一个可以对照下游日志的时间戳，后面所有排查都有基准点。

注意：读取超时不要设得太短。下游在高峰期处理一条单据可能耗时八到十秒，超时设五秒会导致大量“请求已发出但被判失败”的情况，重试反而加重下游负担。

3.2 批量提交与并发参数怎么定

批量和并发是最容易拍脑袋的地方，参数定错了要么跑得慢，要么把下游打挂。这里给一套可以照着算的方法。

先确认下游给的配额。假设 MTI 网关给业务侧的配额是 20 TPS（每秒事务数），这个数字必须找下游要到，不能自己猜。

再测单条耗时。串行跑一百条取平均，假设是 180 毫秒。那么单线程的理论吞吐是 1 / 0.18 ≈ 5.5 条/秒。

要跑到 20 TPS 需要 20 / 5.5 ≈ 3.6 个并发。取整到 4 个并发，理论吞吐 22 条/秒，已经略微超过配额，所以实际要加一层令牌桶把速率压到 18 TPS，留 10% 余量。

单批大小的确定要看“批”这一层的意义。如果只是脚本内部的循环分批（不做批量接口），批大小的作用是控制回写频率和断点粒度，一般取 50 到 200 之间。我通常取 100，理由是：一批耗时约 100 / 18 ≈ 5.6 秒，加上组装和回写约 1 秒，整批七秒内结束；这个粒度下即使进程被杀，最多丢一批的进度，重跑成本低。

下面是一个带令牌桶和线程池的实现骨架。

import threading import time from concurrent.futures import ThreadPoolExecutor class TokenBucket: def __init__(self, rate, capacity=None): self.rate = rate self.capacity = capacity or rate self.tokens = self.capacity self.last = time.monotonic() self.lock = threading.Lock() def acquire(self, n=1): with self.lock: now = time.monotonic() self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.rate) self.last = now if self.tokens >= n: self.tokens -= n return 0 wait = (n - self.tokens) / self.rate time.sleep(wait) return wait BUCKET = TokenBucket(rate=18, capacity=18) def worker(doc): BUCKET.acquire() return submit_one(doc) def run_batch(docs, concurrency=4, batch_size=100): for i in range(0, len(docs), batch_size): chunk = docs[i:i + batch_size] with ThreadPoolExecutor(max_workers=concurrency) as pool: results = list(pool.map(worker, chunk)) save_results(results) logger.info("batch done offset=%s size=%s", i, len(chunk))

并发数不是越大越好。我实测过把并发提到 12，吞吐并没有跟着涨，反而因为下游连接池排队，平均耗时从 180 毫秒涨到 420 毫秒，端到端时间反而更长，同时错误率上升。原因是下游单节点的处理能力有上限，压过去只会让它排队。

3.3 重试策略与退避计算

重试必须区分错误类型，无差别重试是灾难。只有系统类失败（超时、连接重置、5xx、限流）才自动重试；参数类和业务类失败重试一万次也是同样的结果，只会把日志刷满。

退避公式用带抖动的指数退避：

等待时间 = min(上限, 基数 × 2^(n-1)) + 随机抖动

参数取值建议：基数 0.5 秒，上限 8 秒，最大重试次数 5 次，抖动范围 0 到 0.3 秒。

按这个公式，五次重试的等待依次约为 0.5、1、2、4、8 秒，加上抖动，总等待时间在 15.5 到 17 秒之间。这个量级能覆盖绝大多数瞬时抖动，又不会让单条单据卡太久。

import random def backoff_delay(attempt, base=0.5, cap=8.0, jitter=0.3): return min(cap, base * (2 ** (attempt - 1))) + random.uniform(0, jitter) def submit_with_retry(doc, max_attempts=5): for attempt in range(1, max_attempts + 1): try: code, body = submit_one(doc) except requests.RequestException as exc: if attempt == max_attempts: return "FAILED", str(exc) time.sleep(backoff_delay(attempt)) continue if code in (200, 201): return "SUBMITTED", body if code == 409: return "DUPLICATE", body if 400 <= code < 500 and code not in (408, 429): return "REJECTED", body if attempt == max_attempts: return "FAILED", body time.sleep(backoff_delay(attempt)) return "FAILED", "exhausted"

抖动这一项看着不起眼，实际很重要。没有抖动的指数退避，在多线程场景下会让所有线程在同一时刻集体重试，形成脉冲，把下游的限流器反复打穿。加了随机抖动之后重试被摊平，成功率明显提升。

注意：409 幂等冲突不要当失败处理。正确做法是拿幂等键去下游查一次已有结果，把结果回写到本地，状态按已有结果走。

3.4 日志、状态回写与断点续跑

日志的价值在出问题那一刻才体现，所以字段要提前设计好。每条单据至少记录：单据号、幂等键、提交时间、耗时、返回码、下游流水号、重试次数、最终状态。有了这七个字段，九成的排查不需要再去翻下游日志。

我在模拟项目 C 里吃过一次亏：日志只记了“提交成功”，没记下游流水号，后来发现有一批单据的落账结果和预期不符，去下游查的时候只能靠时间范围模糊匹配，翻了两个小时。从那以后所有提交日志都必须带下游返回的流水号。

状态回写要放在一个独立的事务里，和提交动作分开。逻辑是：提交拿到结果后，先写本地任务表，再调 INV Manager 的状态更新接口。为什么要先写本地？因为状态回写接口也可能失败，如果先调它失败了，本地就没记录，下次跑批会重复提交。先写本地、再回写、回写失败就标记成“待回写”由另一个任务补偿，这样最稳。

def persist_result(doc_no, status, body, attempt): db.execute( """INSERT INTO submit_task (doc_no, idem_key, status, resp_code, down_ref, attempts, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(doc_no) DO UPDATE SET status=excluded.status, resp_code=excluded.resp_code, down_ref=excluded.down_ref, attempts=excluded.attempts, updated_at=excluded.updated_at""", (doc_no, body.get("idem_key"), status, body.get("code"), body.get("ref_no"), attempt, int(time.time())), ) def sync_back(doc_no, status): try: requests.patch( f"{BASE}/documents/{doc_no}", json={"status": status}, headers={"Authorization": f"Bearer {TOKEN}"}, timeout=TIMEOUT, ) except requests.RequestException: mark_pending_sync(doc_no)

断点续跑靠的就是这张表。每次启动时先捞“状态为空或待重试”的单据，跳过“已完成”的，天然实现续跑，不需要额外的进度文件。

3.5 调度上线与灰度

调度用系统自带的定时任务就够，不需要引入复杂的工作流引擎。建议的配置是：业务高峰期不跑（避免和在线提交抢配额），安排在凌晨和午后两个低峰窗口，每个窗口最多跑三轮（一轮处理完剩余的，第二轮补失败重试的，第三轮做对账）。

灰度上线分三步走。第一步只在测试环境跑，用全量数据的副本，重点看报文组装有没有字段错位。第二步在生产环境跑但只提交一个批次（比如十条），人工逐条核对下游结果。第三步放开批量但把并发压到 1，观察两天吞吐和错误率都平稳之后再调到目标并发。

注意：第一次生产跑批一定要有人在旁边盯着。脚本第一分钟的表现基本能反映后面所有问题，盯着看比事后翻日志快十倍。

4. 常见问题与排查实录

这一节是纯经验，都是实际踩出来的。有些坑看起来很低级，但在赶进度的时候几乎人人都会遇到一次。

4.1 高频报错速查表

这张表建议打印出来贴在显示器边上，前三次跑批会用到其中至少五条。

4.2 数据层面的坑

数据问题比接口问题更难查，因为它不报错，只是结果不对。

精度丢失是最常见的。数量字段在 INV Manager 存四位小数，MTI 要三位，直接round到三位会导致一批单据的合计不等于源合计，下游的平衡校验就会拦住。正确做法是按最小计量单位换算成整数再运算，最后再转回小数。

空值语义不一致。INV Manager 里空字符串和 null 都表示“未填”，但 MTI 把空字符串当“显式清空”，把 null 当“用默认值”。一个字段的差异会导致整批单据落到不同的业务分支上。处理办法是在报文组装层做统一归一化：所有空值统一转 null，需要清空的场景单独走另一个字段。

时区错位是最隐蔽的。INV Manager 存的是本地时间不带时区，MTI 要的是带时区的 ISO8601。如果直接拼个+08:00上去，遇到夏令时切换或者跨时区业务就会错一小时。规范做法是转换时明确指定源时区，不要依赖运行机器的系统时区。

字符集问题在批次号和备注字段上偶尔出现，特别是包含中文或特殊符号的时候。提交前做一次编码检查，非 ASCII 字符统一做转义或者替换，能避免一部分 400 错误。

4.3 对账与巡检

脚本跑通了不等于事情结束了。每天必须有对账，否则差异会累积到无法追溯。

对账逻辑很简单：统计 INV Manager 里“已提交”状态的单据数，和 MMTT 里“已落账”的单据数，按单号做左连接，找出三类差异——本地已提交但下游没有、下游有但本地没标记、两边都有但状态不一致。

SELECT l.doc_no, l.status AS local_status, r.status AS remote_status FROM local_submit_task l LEFT JOIN remote_ledger r ON l.doc_no = r.transaction_ref WHERE r.transaction_ref IS NULL OR l.status <> 'DONE' OR r.status IS NULL;

三类差异的处理方式不同：第一类通常是提交请求发出去了但响应丢了，用幂等键去下游查一次就能确认；第二类一般是下游有别的入口也在提交，需要和业务确认来源；第三类多半是回写失败，跑一次补偿任务就行。

巡检除了对账，还建议加三项健康检查：令牌有效期剩余时长、下游接口的探测请求响应时间、本地任务表的失败堆积量。这三项挂在定时任务上，超过阈值就告警，能在问题变大之前发现苗头。

我个人在实际操作中的体会是，这类脚本的代码量其实不大，三千行顶天了，真正花时间的是摸接口、定参数、建对账。前期把返回码表、映射表、幂等规则这三样东西整理成文档，后面无论是换人维护还是接新下游，成本都会低很多。还有一个小技巧值得分享：把每一轮跑批的关键指标（处理条数、成功率、平均耗时、重试次数）打成一行 JSON 落到日志里，一个月后回头看趋势，比任何监控面板都直观。