《1688 API：落地方案、接口边界与业务踩坑 —— 从采购到分销的全链路对接》（附Python源码）
1688 不是“另一个闲鱼”。它是B2B 供货底座：你这边是买家/分销商，1688 商家是供应商。接口边界必须按角色切：选品（读）→ 采购下单（写）→ 支付（资金）→ 履约回传（物流/退款）→ 分销回流（下游平台）。最坑的不是签名，而是：库存不是强一致、密文地址要白名单、分销价靠 flow、物流回传会静默失败、Token 会过期、QPS 不是日额度。
一、1688 对接的角色模型（先别写代码）
角色 | 在 1688 的身份 | 持有凭证 | 能干什么 |
|---|---|---|---|
下游卖家 | 1688 买家 / 分销商 | 买家 access_token | 搜品、下单、查单、处理退款 |
供应商 | 1688 商家 | 商家 session | 接单、发货、退换货、回传物流 |
ISV/ERP | 开放平台应用 | AppKey/AppSecret | 代买家/代分销商调 API |
跨境卖家 | 寻源通/跨境分销买家 | 跨境标签应用 | 拿 HS Code、申报价、账期支付 |
⚠️ 和闲鱼不同：闲鱼 ISV 是“代卖家管自己店”；1688 ISV 是“代买家去买货 / 代分销商管供应链”。所以 1688 的trade.order.create是用买家 token 调的，不是商家 token。
二、全链路接口地图（采购 → 分销）
选品层（读） alibaba.product.search / product.get / product.batch.get alibaba.product.category.get alibaba.cpsMedia.productInfo（分销价/佣金） │ ▼ 采购层（写） alibaba.createOrder.preview 价格/运费/优惠预校验 alibaba.trade.fastCreateOrder 轻量下单（代发首选） alibaba.trade.order.create 完整下单 alibaba.trade.fenxiaoOrder.create 分销/代发回流单（带下游渠道+下游单号） │ ▼ 支付层（资金） alibaba.trade.payWay.query alibaba.trade.pay / protocolPay.preparePay（免密，需签约） alibaba.crossBorderPay.url.get（跨境宝） │ ▼ 履约层（供应商动作 + 回传） alibaba.trade.order.get 查状态/运单号 alibaba.trade.getLogisticsTraceInfo alibaba.logistics.repush_tracking 物流重推（单号丢了用） │ ▼ 售后层 alibaba.trade.refund.get 分销退货退款创建 / 售后订单回传（分销方案专用）
三、六大业务踩坑（比技术坑更贵）
1️⃣ 库存不是强一致：“显示有货”≠“能下单”
product.get的库存是快照，不是占用后库存1688 一件代发是订单触发式扣减：下游付钱 → 才向供应商占库存
正确做法：下单前调
createOrder.preview，下单失败按SUB_ORDER_NOT_ENOUGH类错误回退到备供
2️⃣ 分销价靠 flow 和 retailPrice
老严选：
isPftOffer=true+ttpft新严选：
isJxhyOffer=true+boutiquefenxiao/boutiquepifa用错 flow → 价格不对 / 下单失败“无 retailPrice”
3️⃣ 密文地址不是“传进去就行”
抖音/淘宝/拼多多下游订单：姓名手机脱敏
1688 侧要用
encryptOutOrderInfo之类字段传密文，不是自己解密再传明文没白名单：接口直接权限拒绝
传明文：占用解密额度，甚至违规
4️⃣ 物流回传会“静默成功”
供应商点了发货，分销商后台“已发货但无单号”
原因：物流渠道没做电子面单授权 / 网络抖动 / 回传超时
解法：定时扫
status=shipped and tracking_no is null→ 调重推接口，重推前先查渠道授权
5️⃣ Token 有效期短，必须刷
1688 买家 access_token 通常几小时到一天级，不是永久
生产里要有：刷新器 + 多卖家 token 表 + 过期重授权二维码
把 token 写死在配置里 = 一周后半夜报警
6️⃣ QPS ≠ 日调用量
商品搜索默认 ~10/s，订单类 ~20/s
500 个 offer × 30s 轮询 = 16.7/s，单 Key 必限流
正解：事件订阅（价格/库存变更）+ 下单前预览 + 本地缓存 + 多 AppKey 分域
四、生产向源码：1688 采购→分销骨架
# commerce_mesh/suppliers/ali1688/client.py
import time
import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
class Ali1688Env(str, Enum):
SANDBOX = "https://gw-api.passport.alibaba.com/sandbox"
PROD = "https://gw.open.1688.com/openapi"
class OrderStatus(str, Enum):
WAIT_PAY = "wait_pay"
PAID = "paid"
SHIPPED = "shipped"
SIGNED = "signed"
REFUNDING = "refunding"
CLOSED = "closed"
@dataclass
class BuyerToken:
seller_erp_id: str
access_token: str
refresh_token: str
expires_at: float
account: str = ""
@dataclass
class PurchaseReq:
offer_id: str
sku_id: str
qty: int
receiver: dict # 明文/密文由 downstream 决定
outer_order_no: str # 下游平台订单号
channel: str = "shopee" # shopee/tiktok/taobao/douyin
flow: str = "" # boutiquefenxiao / ttpft / ""
encrypt_out_order_info: Optional[str] = None
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class TokenStore:
"""聚石塔/内网：token 不出内网"""
def __init__(self):
self._t: dict[str, BuyerToken] = {}
def save(self, t: BuyerToken):
self._t[t.seller_erp_id] = t
def usable(self, erp_id: str) -> Optional[str]:
t = self._t.get(erp_id)
if not t:
return None
if t.expires_at - 300 < time.time():
return None
return t.access_token
class Ali1688PurchaseClient:
def __init__(self, token_store: TokenStore, qps_limiter):
self.tokens = token_store
self.limiter = qps_limiter
self.preview_cache: dict[str, float] = {}
# ---------- 1. 下单前预览（防超卖/防错价） ----------
def preview(self, erp_id: str, req: PurchaseReq) -> dict:
token = self.tokens.usable(erp_id)
if not token:
return {"ok": False, "code": "TOKEN_EXPIRED"}
self.limiter.acquire("createOrder.preview") # 令牌桶
# 伪调用：alibaba.createOrder.preview
preview = {
"offer_id": req.offer_id,
"sku_id": req.sku_id,
"unit_price": 39.9,
"freight": 4.0,
"retail_price": 39.9 if req.flow else None,
"available": True,
}
if req.flow and preview["retail_price"] is None:
return {"ok": False, "code": "NO_RETAIL_PRICE", "msg": "该SKU未报名严选"}
return {"ok": True, "preview": preview}
# ---------- 2. 创建采购单 ----------
def create_purchase_order(self, erp_id: str, req: PurchaseReq) -> dict:
pre = self.preview(erp_id, req)
if not pre["ok"]:
return pre
token = self.tokens.usable(erp_id)
self.limiter.acquire("trade.order.create")
# 分销/代发单：带下游单号 + 密文
params = {
"offerId": req.offer_id,
"skuId": req.sku_id,
"quantity": req.qty,
"outerOrderNo": req.outer_order_no,
"downstreamChannel": req.channel,
}
if req.flow:
params["flow"] = req.flow
if req.encrypt_out_order_info:
params["encryptOutOrderInfo"] = req.encrypt_out_order_info
else:
params["receiver"] = req.receiver # 明文仅限非密文下游
# resp = top_execute(method="alibaba.trade.fenxiaoOrder.create", ...)
resp = {
"success": True,
"purchase_order_id": f"1688-{uuid.uuid4().hex[:12]}",
"payable_amount": 39.9 * req.qty + 4.0,
}
return {"ok": True, **resp}
# ---------- 3. 物流回传补偿 ----------
def repair_missing_tracking(self, erp_id: str, shipped_orders: list[dict]) -> list[dict]:
fixed = []
for o in shipped_orders:
if o.get("status") == "shipped" and not o.get("tracking_no"):
# 1. 查渠道授权
# 2. 调 alibaba.logistics.repush_tracking
fixed.append({
"purchase_order_id": o["purchase_order_id"],
"action": "repush_tracking",
"note": "渠道未授权时重推无效，需先电子面单授权",
})
return fixed五、和前面系列的拼接关系
《聚石塔入塔》：1688 买家端可以公网调，但下游 PII / 密文 / 分销商关系建议内网存
《闲鱼 order.ship》：闲鱼是“卖家用 token 发自己单”；1688 是“买家用 token 买别人货”
《两套接口边界》：1688 商品采集 ≠ 1688 代发下单；选品池 → 采购池必须过一道“供应商白名单 + 分销协议”
《六大坑》：时区/税价/超卖/编码/映射/消息丢失，在 1688 里变成：库存快照、retailPrice、密文面单、物流静默、SKU spec 映射、回调丢失
《中台调度》：
Ali1688Adapter只负责“采购侧统一模型”，不要让它知道 TikTok/Shopee 的店铺逻辑
六、落地检查表（上线前）
[ ] 企业实名 + 应用权限（商品/交易/物流/退款）都开了
[ ] 跨境场景额外开“寻源通 / 跨境分销”标签
[ ] Token 有刷新，不下发到前端
[ ] 下单前必调
createOrder.preview[ ] 分销单用对
flow和retailPrice[ ] 下游密文订单走密文字段，不解密
[ ] 物流回传有“已发无单”补偿任务
[ ] 库存变更走消息订阅，不 30s 全量轮询
[ ] QPS 令牌桶 + 多 AppKey 分域
[ ] 供应商维度建“备供表”，主供售罄自动切
七、一句话收口
1688 对接的成熟度，不看你会不会调trade.order.create，而看你能不能处理：库存快照、严选 flow、密文面单、物流静默、token 过期、供应商跑路。选品是数学，采购是状态机，分销是合规，履约是补偿——四件事用一套 ERP 模型撑起来，才算“对接完了”。
