简介这套代码包围绕旺店通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必填签名密钥妥善保存不要提交到Gitbase_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(keylambda 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, dataparams, timeout10)这里的核心设计是biz_params先通过urlencode转成一个字符串再放进外层参数字典。这个做法在不同WMS项目里不一定通用部分版本要求把业务参数摊平到最外层也就是每个业务字段都直接参与签名。我拆包时发现代码里保留了SIGN_MODE开关nested表示业务参数整体参与签名flatten表示摊平你可以按实际文档切换到对应模式。timeout10是我在对接过程里刻意加的。有些生产场景里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)。推送前先尝试插入一条statusprocessing的记录如果插入成功说明这个单号之前没处理过可以继续调用WMS如果插入时唯一键冲突说明之前已经推过那就直接查询order_id不要在WMS重复建单。我多次强调这个表要放在业务库不要放在WMS接口模块的本地文件里。因为业务系统可能有多实例部署文件锁在跨实例时是失效的而数据库唯一键是全局生效的。另外不要把成功记录删掉成功记录是后续对账和排查的凭证。5.2 补偿与重试失败单要能重新跑只做幂等还不够。很多订单因为网络超时、网关限流、参数写错第一次推送失败后就停在那里。如果不做补偿这些单子会一直卡在草稿状态仓库的人就会说“你推的单怎么没下来”。def retry_failed_orders(): # 只重试失败且未超过上限的记录 rows db.query( SELECT * FROM push_log WHERE statusfailed 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( levellogging.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_codeNone): 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件查了整整半天才找到一张被漏确认的入库单。从那以后我每次上线前都会强制跑一遍这个比对脚本并把差异单号发给仓库同事复核再也不靠人工猜。希望帮到你。本文还有配套的精品资源点击获取