☰
旺店通WMS对接实战:签名算法、幂等重试与库存同步全解析
2026/10/8 9:27:31 网站建设 项目流程

简介:这套代码包围绕旺店通WMS系统接口对接展开,面向需要完成WebAPI集成的C#开发者及供应链实施工程师,重点解决销售出库单查询、签名构建、标准定制接口调用等对接难题。资源共32个文件,压缩包约380KB,以cs源码工程、dll依赖库、json配置、txt说明文档为主,搭配py辅助脚本、pdb调试符号及项目配置文件,可支撑完整编译与调试,已有182人学习下载。包内代码示例演示了通过HttpClient发起POST请求、构造URL参数与签名、解析响应数据的具体写法,并配有标准定制接口的调用模板和完整实例,含请求地址、密钥及参数设置等关键信息。开发者参照示例即可快速验证接口连通性,降低文档理解到编码落地的转化成本,适合有一定系统对接经验的开发人员直接复用。

1. 旺店通WMS对接:先搞懂这套接口再写代码

这套旺店通WMS对接流程代码,给我的第一印象是“能跑”比“好看”重要。仓库这边的接口通常不会像电商订单那样有完整沙箱,很多时候你拿到的是文档和一批示例,真正能不能用,得等你把商品、入库单、出库单、库存查询这几条主链路都走通才算数。它解决的是自研ERP、老系统或者一套订单管理系统跟旺店通WMS之间的数据打通问题,适合正在接手对接任务、又不想反复猜文档的开发人员。我觉得拆这份资源最有价值的地方,不在单个接口怎么调,而在于它把单据状态、回调验签、幂等重试这些坑提前踩了一遍,你按流程走能省掉好几个晚上。

2. 对接前准备:先把权限和签名算明白再谈业务

2.1 旺店通WMS开放平台:应用权限与密钥一次理清

对接第一步不是写请求,而是先到旺店通WMS开放平台创建应用。常见做法是管理员账号登录后,在“应用管理”里新建一个应用,系统会给你一对app_key和app_secret。这里要提醒的是,WMS接口是按业务域拆权限的,基础资料、仓库管理、单据中心、库存查询都分属不同分组,并不是每个新应用都能调所有接口。比如业务只需要同步商品和出入库单,那就只申请对应权限,别图省事把权限全选上,审核流程反而更慢,而且生产追责时也不好说清楚。

在资源代码里,config.py文件预留了app_key、app_secret、base_url和seller_nick几个配置项。其中seller_nick是用来区分多仓多店铺的,如果你的场景只对接一个仓库,这个值可以保持空字符串。我一般会把测试环境的base_url指向开放平台提供的沙箱地址,生产环境的地址单独写在部署配置里,避免测试数据污染真实仓库。

以下是我平时习惯维护的配置项清单:

配置项是否必填作用备注
app_key必填应用标识在开放平台创建应用后获得
app_secret必填签名密钥妥善保存,不要提交到Git
base_url必填接口网关地址沙箱和生产分开
seller_nick选填店铺账号多仓多店才需要
warehouse_code选填仓库编码可以从仓库列表接口获取

这种配置方式适合大多数对接项目。很多失败案例都是配置项被写死在内网服务器上,换环境就得改代码。把环境差异收敛到一个配置文件里,后面切换沙箱和正式环境会省很多事。

2.2 sign签名算法:先拼接再加密,顺序错一个就翻车

旺店通WMS的接口签名逻辑不复杂,但顺序要求很严格。我拆包时看到代码里用的是MD5签名,大致流程是:把非空参数按key升序排序,拼接成keyvaluekeyvalue的形式,然后在拼接结果前后各拼上app_secret,最后做MD5并转大写。这段代码可以直接用,不依赖第三方SDK。

import hashlib import time def build_sign(params, secret): # 剔除空值,空字符串不参与签名 items = [(k, str(v)) for k, v in params.items() if v != ""] # 按key升序,排序规则和平台文档保持一致 items.sort(key=lambda x: x[0]) raw = "".join(f"{k}{v}" for k, v in items) raw = secret + raw + secret return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()

这段代码的逻辑其实有两步:第一步把参数字典转成可排序的列表,第二步按升序拼接。最关键的是排序顺序,不是按文档里参数出现的顺序,而是按参数名的字母序。我见过有团队写对了排序,却在拼接业务参数时把biz_params整体又编了一次码,导致签名怎么都对不上。

参数说明里还有一个容易忽略的点:params里的值必须是字符串,数值型参数要提前转成字符串再做拼接。Python 3 下如果直接拼接int类型会报类型错误,这也是常见的“本地能跑,线上报错”原因之一。调试阶段可以先打印raw字符串,跟平台调试工具里显示的内容逐字符比对,差别通常就在空值过滤或者时间戳格式上。

2.3 统一请求模板:把公共参数塞进每个请求

所有旺店通WMS接口都共用一套公共参数,包括app_key、timestamp、v、method、sign_method。我建议你在正式写业务逻辑前,先把请求封装成一个统一函数,后面每个业务接口都通过它发请求。资源里已经有这个函数,但你自己动手时也要注意几个细节。

import requests import time from urllib.parse import urlencode def call_wms(api_name, biz_params, app_key, secret, base_url, seller_nick=""): # 外层是公共参数 params = { "app_key": app_key, "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), "v": "1.0", "method": api_name, "seller_nick": seller_nick, "sign_method": "md5", # 业务参数先做urlencode,形成单一字符串 "biz_params": urlencode(biz_params) } params["sign"] = build_sign(params, secret) return requests.post(base_url, data=params, timeout=10)

这里的核心设计是biz_params先通过urlencode转成一个字符串,再放进外层参数字典。这个做法在不同WMS项目里不一定通用,部分版本要求把业务参数摊平到最外层,也就是每个业务字段都直接参与签名。我拆包时发现代码里保留了SIGN_MODE开关,"nested"表示业务参数整体参与签名,"flatten"表示摊平,你可以按实际文档切换到对应模式。

timeout=10是我在对接过程里刻意加的。有些生产场景里WMS网关处理慢,不设置超时会让请求线程无限挂起。后面第5章会专门说重试和补偿,但基础的请求超时一定要在这里就设好。

3. 核心流程跑通:商品、单据与库存三块代码骨架

3.1 商品资料同步:全量拉取和增量更新

商品同步是所有对接里最先做的一件事。没有基础资料,后面入库单、出库单都没法核验SKU。WMS商品接口一般支持分页查询,我习惯先写一个全量同步函数,跑通后再加增量条件。

def sync_goods(app_key, secret, base_url): page_no = 1 page_size = 100 while True: biz = { "page_no": page_no, "page_size": page_size } resp = call_wms("wms.goods.query", biz, app_key, secret, base_url) data = resp.json()["result"] goods_list = data["goods_list"] if not goods_list: break for g in goods_list: upsert_goods(g) # 已拉完全部数据 if page_no * page_size >= data["total"]: break page_no += 1

这段分页代码里需要注意终止条件:使用page_no * page_size >= total而不是len(goods_list) < page_size。因为最后一页如果恰好取满,后一种判断会导致再请求一次空数据,虽然没有大问题,但白白多一次网络开销。

upsert_goods函数在代码包里对应本地数据库的插入或更新逻辑。如果你之前没做过商品对接,我建议这里只做覆盖更新,不要删除不存在的数据。WMS侧删掉的商品,本地应当保留历史订单引用,否则会导致历史单据找不到商品名称。

增量更新比全量更实用。旺店通WMS的商品查询接口通常支持传入修改时间范围,我一般是记录上一次同步位置,每次只拉modified_begin到modified_end之间的数据。这样凌晨跑批时不会把几万条SKU全量刷一遍。

还有一个容易被忽视的单位问题:WMS返回的商品库存单位可能是基本单位,比如“件”,但业务单据里用的是“箱”。接口里经常会带一个unit_rate或类似字段,表示箱与件的换算关系。我在资源里特意写了单位换算注释,避免把箱数当成件数推给仓库,导致库存直接翻倍。

3.2 入库单/出库单推送:先创建草稿再确认

仓库单据推送是对接中最容易出问题的一环,因为它不是一次请求完成的。旺店通WMS的单据接口常见流程是两步:先创建草稿,拿到order_id,再调用确认接口,仓库作业才会真正生效。很多初次对接的人只调了创建接口,库存却不动,就是这个原因。

def push_stock_in_order(order_no, sku_items, warehouse_code, app_key, secret, base_url): biz = { "order_no": order_no, "warehouse_code": warehouse_code, "order_type": "IN", "items": [ { "sku_no": item["sku"], "qty": item["qty"], "position_no": item.get("position", "") } for item in sku_items ], "remark": "ERP推送入库单", "is_confirm": "false" } resp = call_wms("wms.stockin.create", biz, app_key, secret, base_url) order_id = resp.json()["result"]["order_id"] # 确认入库单,这一步才让仓内真正开始作业 call_wms("wms.stockin.confirm", {"order_id": order_id}, app_key, secret, base_url) return order_id

这里我把is_confirm设置为"false",就是明确告诉接口先不要自动确认,等我拿到单号再手动确认。好处是中间出问题时,可以及时作废草稿,不会在仓库里留一个被确认但实际没货的入库单。

order_no是外部单号,必须保证唯一。我建议在业务系统里用“源单类型+日期+序列号”拼接,比如SO20250612001,不要直接用数据库自增ID。因为后续跟WMS回调核对时要靠这个单号反查,纯自增ID在跨系统排查时不够直观。

出库单的推送逻辑跟入库单基本一致,只是把order_type改成"OUT",并且出库单往往需要传收货人信息。如果你的系统要对接退货,还要确认一下渠道接口是走退货单还是负向出库单。这点我后面在避坑章里会再提到。

3.3 库存查询:按仓过滤,别忘了多仓汇总

库存查询是所有接口里最简单的,但业务口径最容易出问题。WMS返回的字段通常有on_hand_qty、available_qty、frozen_qty等。如果你直接拿on_hand_qty当可售库存,很可能会多出已经占用的预分配库存。

def query_stock(sku_no, warehouse_code, app_key, secret, base_url): biz = { "sku_no": sku_no, "warehouse_code": warehouse_code, "page_size": 100 } resp = call_wms("wms.stock.query", biz, app_key, secret, base_url) result = resp.json()["result"] available = sum(item["available_qty"] for item in result["stock_list"]) return available

这里的available_qty才是仓库确认可以销售的库存。如果业务要做的是“缺货判断”,就按这个字段来。如果是财务要库存金额,那才需要on_hand_qty。多仓时warehouse_code传空串,接口会返回所有仓位的库存记录,再按SKU分组汇总。

库存查询还有一个细节:分页。如果库存记录超过100条,需要同样按分页处理。但绝大多数库存查询是按SKU+仓库组合的,数据量不会太大,所以我更关注口径而不是性能。

4. 避坑:旺店通WMS对接中五个真实翻车点

下面这五个问题来自我实际对接时的报错记录,每一条都按现象、原因、解决三个层面拆开。它们不会同时出现在你手里,但任何一个都足够让你多加班一天。

4.1 签名报错 10001:参数顺序被忽略

现象:测试环境偶尔通过,生产环境频繁返回签名错误,错误码10001。原因:请求参数顺序和平台要求不一致,或者biz_params与公共参数分层方式不对。解决:先把所有参与签名的参数合并成一个字典,剔除空值,按key升序拼接,不要用文档展示的参数顺序;再通过平台调试工具生成一份参考签名,把你自己生成的raw字符串逐字符比对。之后我在build_sign里加了一条断言,签名生成失败就直接抛异常,不发出请求,避免错误请求进入仓库网关。

4.2 时间戳偏差:请求被判定为过期

现象:服务器时间比标准时间快了几分钟,每次请求都提示业务失败。原因:WMS网关会校验timestamp与服务器时间差,超过一定范围直接拒绝。解决:应用服务器开启NTP时间同步;在统一请求模板里加一个time_offset常量,针对网关时间与实际时间差做补偿。我印象最深的是容器内时区是UTC,生成的时间戳比北京时间晚8小时,签名没有问题,但请求就是失败。排查到最后才发现是时区,而不是签名。

4.3 回调报文解密失败:编码或密钥类型不对

现象:WMS主动回调业务系统时,签名校验通过,但报文体解析出来是乱码。原因:回调报文是用私钥解密,私钥格式要求是PKCS1,拷贝过程中可能丢失了换行符,或者用了PKCS8格式而没有对应加载。解决:把私钥转成标准PEM格式,用load_pem_private_key显式加载;不要直接拼接成一行字符串塞给加密库。在资源里我保留了回调解密示例,并加了日志,打印解密前的加密内容和解密后的长度,方便确认哪一步出了问题。

4.4 同一订单重复推送:WMS生成了两个入库单

现象:由于网络超时,业务系统自动重试了一次,同一个order_no被推送了两次,WMS出现两笔入库单。原因:旺店通WMS的部分接口不是天然幂等,如果业务侧没有查重逻辑,重试就会重复创建单子。解决:在本地建一张推送记录表,order_no设为唯一键,推送前先查这个表。如果order_no已存在,直接取出对应的order_id,不再调用创建接口。这样即使业务系统重试,也不会在WMS重复建单。

4.5 接口限流:批量同步时突然被拒绝

现象:凌晨批量同步库存时,跑到一半接口返回“操作过于频繁”或HTTP 429。原因:WMS接口有频控逻辑,按秒或按分钟限制调用次数。解决:在代码包里加一个令牌桶限速器,默认每秒2个请求。批量任务里每个请求之间做短暂暂停,失败请求按指数退避重试3次。如果同步量很大,就拆成队列逐个消费,不要用协程一次性推几千个。

5. 让对接更稳:幂等、补偿重试与日志追踪

5.1 幂等设计:用一张本地表兜住所有重复请求

对接WMS这种外部系统,我最看重的是幂等。业务系统可能因为用户手抖、网络抖动、定时任务重复调度,把同一个外部单号推送两次。WMS侧可能允许创建两个草稿单,但这不是你希望的结果。解决办法是在本地建一张push_log表,把每一次推送记录都留底。

CREATE TABLE push_log ( id INT PRIMARY KEY AUTO_INCREMENT, order_no VARCHAR(64) NOT NULL, order_id VARCHAR(64), push_time DATETIME, status VARCHAR(20), retry_count INT DEFAULT 0, UNIQUE KEY uk_order_no (order_no) );

这里的关键是UNIQUE KEY uk_order_no (order_no)。推送前先尝试插入一条status='processing'的记录,如果插入成功,说明这个单号之前没处理过,可以继续调用WMS;如果插入时唯一键冲突,说明之前已经推过,那就直接查询order_id,不要在WMS重复建单。

我多次强调这个表要放在业务库,不要放在WMS接口模块的本地文件里。因为业务系统可能有多实例部署,文件锁在跨实例时是失效的,而数据库唯一键是全局生效的。另外,不要把成功记录删掉,成功记录是后续对账和排查的凭证。

5.2 补偿与重试:失败单要能重新跑

只做幂等还不够。很多订单因为网络超时、网关限流、参数写错,第一次推送失败后就停在那里。如果不做补偿,这些单子会一直卡在草稿状态,仓库的人就会说“你推的单怎么没下来”。

def retry_failed_orders(): # 只重试失败且未超过上限的记录 rows = db.query( "SELECT * FROM push_log " "WHERE status='failed' AND retry_count < 3" ) for row in rows: try: push_stock_out_order( row.order_no, sku_items_from_order(row.order_no), row.warehouse_code ) mark_success(row.id) except Exception as e: logger.exception("retry order_no=%s failed: %s", row.order_no, e) update_retry_count(row.id)

这段补偿逻辑的核心是“只处理明确失败的单子”。不是说所有失败都马上重试,而是查push_log里的failed状态,并且retry_count < 3的记录。每次失败后给retry_count加1,超过3次的单子需要人工介入,避免一个根本性错误一直打WMS网关。

我在补偿任务里会再加一个延迟条件:第一次失败后5分钟重试,第二次后15分钟,第三次后30分钟,也就是指数退避。同步任务里最怕的就是失败后立刻重试,把已经限流的接口又压一遍。

5.3 日志:把“玄学问题”变可排查问题

对接WMS这类封闭系统,最让人头疼的是“昨天能跑,今天不能跑”之类的问题。我后来养成了一个习惯:每一条外部请求都做结构化日志,记录时间、方法、单号、请求参数、响应体和耗时。

import logging logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s" ) logger = logging.getLogger("wms_connector") def log_call(method, order_no, req, resp, cost_ms): logger.info( "call method=%s order_no=%s req=%s resp=%s cost=%sms", method, order_no, req, resp[:500], cost_ms )

这个log_call函数看起来简单,但它在实际排查中的作用很大。对比响应体时,我只需要查日志里同一个order_no的请求,看看是否每次都返回同一个错误。如果耗时从200ms涨到5秒,那大概率不是业务代码问题,而是WMS侧变慢或者网段有问题。

日志里要注意脱敏:不要把完整的app_secret或私钥打印进去。我一般是打印请求参数时去掉sign字段,响应体截断前500字符。截断不是偷懒,是避免日志文件被超长报文撑爆。

6. 最后落地:用自测脚本核对库存一致性再切换生产

6.1 对比本地库存与WMS库存的核对脚本

代码写完并不代表对接完成。我最后一道关卡是自测脚本,用它对库存一致性做核对。这个脚本很简单,但能发现很多隐藏在状态流程里的问题。

def compare_stock(sku_list, warehouse_code=None): for sku in sku_list: local_qty = get_local_stock(sku) wms_qty = query_stock(sku, warehouse_code, app_key, secret, base_url) if local_qty != wms_qty: logger.error("sku=%s local=%s wms=%s", sku, local_qty, wms_qty) else: logger.info("sku=%s ok", sku)

使用方式很直接:上线前挑100个有代表性的SKU跑一遍抽查,再跑一次全量。如果两边对不上,先查push_log里是否有状态为processing或failed的记录。很多对不上的原因不是库存接口错了,而是入库单创建成功但confirm失败,库存一直没增加。

我在一次真实项目里就遇到这个问题,对账时发现本地库存比WMS多了12件,查了整整半天才找到一张被漏确认的入库单。从那以后,我每次上线前都会强制跑一遍这个比对脚本,并把差异单号发给仓库同事复核,再也不靠人工猜。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询