🏭 1688供应链中台必备：商品·订单·库存API全链路集成方案（附Python源码）

一、全链路时序（中台视角）

┌─────────────┐  ① 商品同步        ┌──────────────────┐
│  1688开放平台│◀──────────────────│  中台商品服务     │
│  offer.search│  item.get         │  (SKU映射+MOQ)   │
└─────────────┘                    └──────┬───────────┘
                                        │ 运营选品确认
                                        ▼
                              ② 创建采购单 alibaba.trade.create
                                        │
┌─────────────┐  ③ 轮询订单+物流    ┌──────────────────┐
│  1688开放平台│── trade.buyer.list─▶│  中台采购服务     │
│  logistics   │── logistics.ship   │  (状态机:待发→在途→│
└─────────────┘                     │   已签收→入库)    │
                                    └──────────────────┘

⚠️ 库存说明：1688 普通接口返回的是可售库存快照（非实时锁定），严格防超卖建议：① 以 sku.amount_on_sale做预警 ② 扣减后二次校验返回库存；实时锁定库存属增值接口（需资源包）。

二、Python：1688 供应链全链路 Client

# ali1688_supply_chain.py
import hashlib
import time
import requests
import urllib.parse
from typing import Dict, List, Optional
from datetime import datetime, timedelta

# ───────────────────────────────────────────────────────
# 令牌桶 — 保守 QPS=8（免费上限≈10），避免触发流控
# ───────────────────────────────────────────────────────
class _TokenBucket:
    def __init__(self, rate=8, cap=None):
        self.rate, self.cap = rate, cap or int(rate)
        self.tok = float(self.cap)
        self.ts = time.monotonic()

    def wait(self):
        now = time.monotonic()
        self.tok = min(self.cap, self.tok + (now - self.ts) * self.rate)
        self.ts = now
        if self.tok >= 1:
            self.tok -= 1
            return
        time.sleep((1 - self.tok) / self.rate + 0.01)
        self.tok = 0


class Ali1688SupplyChainClient:
    """
    1688 供应链中台 Client
    覆盖: 商品搜索/详情 · 采购订单创建/查询 · 物流跟踪
    """
    GW = "https://gw.open.1688.com/openapi"

    def __init__(self, app_key: str, app_secret: str, access_token: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.token = access_token
        self.bucket = _TokenBucket(rate=8)

    # ───────────────────────────────────────────────────
    # 签名 MD5（标准1688规则）
    # ───────────────────────────────────────────────────
    def _sign(self, params: Dict) -> str:
        filt = sorted((k, v) for k, v in params.items()
                      if v is not None and str(v).strip() != '' and k != 'sign')
        qs = ''.join(f"{k}{v}" for k, v in filt)
        return hashlib.md5(
            f"{self.app_secret}{qs}{self.app_secret}".encode()
        ).hexdigest().upper()

    def _call(self, url: str, method: str, biz: Dict) -> Dict:
        self.bucket.wait()
        p = {
            "method": method,
            "app_key": self.app_key,
            "session": self.token,
            "timestamp": str(int(time.time() * 1000)),
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
        }
        p["param2" if "param2" in url or method == "alibaba.offer.search"
                    else "param"] = urllib.parse.quote_plus(
            str(biz).replace("'", '"')
        )
        p["sign"] = self._sign(p)

        for att in range(3):
            r = requests.get(url, params=p, timeout=15)
            r.raise_for_status()
            d = r.json()
            if "error_response" in d:
                ec = str(d["error_response"].get("code", ""))
                if "FLOW_CONTROL" in ec or ec == "429":
                    if att < 2:
                        time.sleep(2 ** att)
                        continue
                raise Exception(f"1688 FlowLimit: {d['error_response'].get('msg')}")
            return d.get(list(d.keys() - {'error_response'})[0], {})
        raise Exception("1688 call failed after retries")

    # ========== ① 商品域 ==========
    def search_offers(self, kw: str, pg=1, sz=40,
                       price_min=None, price_max=None) -> Dict:
        biz = {"keywords": kw, "pageNo": pg, "pageSize": min(sz, 50),
               "sortType": "booked"}
        if price_min: biz["beginPrice"] = str(int(price_min * 100))
        if price_max: biz["endPrice"] = str(int(price_max * 100))
        return self._call(
            f"{self.GW}/param2/2/alibaba.offer.search/2.0",
            "alibaba.offer.search", biz
        )

    def get_item(self, offer_id: str, fields: str = None) -> Dict:
        biz = {"item_id": offer_id}
        if fields: biz["fields"] = fields
        res = self._call(f"{self.GW}/http/2/1", "alibaba.item.get", biz)
        return res.get("alibaba_item_get_response", {}).get("item", {})

    # ========== ② 订单域 ==========
    def list_orders(self, status="waitsellersend",
                     hours_back=48, page=1, sz=50) -> List[Dict]:
        biz = {
            "orderStatus": status,
            "gmtCreateStart": (datetime.now() - timedelta(hours=hours_back))
                               .strftime("%Y-%m-%d %H:%M:%S"),
            "gmtCreateEnd": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "pageNo": page, "pageSize": sz
        }
        res = self._call(f"{self.GW}/http/2/1", "alibaba.trade.buyer.list", biz)
        return res.get("alibaba_trade_buyer_list_response", {}
                      ).get("tradeModelList", []) or []

    def get_order_detail(self, order_id: str) -> Dict:
        biz = {"orderId": str(order_id)}
        res = self._call(f"{self.GW}/http/2/1", "alibaba.trade.get", biz)
        return res.get("result", {})

    # ========== ③ 物流/库存快照 ==========
    def get_logistics(self, order_id: str) -> List[Dict]:
        biz = {"orderId": str(order_id)}
        res = self._call(f"{self.GW}/http/2/1", "alibaba.logistics.trade.ship", biz)
        return res.get("logisticsOrders", []) or []

    def get_stock_snapshot(self, offer_id: str) -> Dict:
        """可售库存快照（取自商品详情 amount_on_sale）"""
        item = self.get_item(offer_id, fields="item_id,sku_list")
        skus = item.get("sku_list", []) or []
        return {
            "offer_id": offer_id,
            "skus": [{
                "sku_id": s.get("sku_id"),
                "spec": s.get("spec_attributes"),
                "price": s.get("price"),
                "stock": s.get("amount_on_sale", 0)
            } for s in skus]
        }


# =============================================================
# 使用示例 — 全链路演示
# =============================================================
if __name__ == "__main__":
    cli = Ali1688SupplyChainClient(
        app_key="YOUR_APP_KEY",
        app_secret="YOUR_APP_SECRET",
        access_token="YOUR_ACCESS_TOKEN"
    )

    try:
        # ① 搜索 + 取详情
        srch = cli.search_offers("不锈钢保温杯 定制", price_min=15, price_max=60)
        offers = srch.get("offers", [])
        print(f"✅ 搜到 {srch.get('totalResult')} 条，取首条")

        if offers:
            offer_id = str(offers[0]["offerId"])
            item = cli.get_item(
                offer_id,
                fields="item_id,title,price,sku_list,pics,min_order_quantity"
            )
            print(f"   商品: {item.get('title')}  MOQ:{item.get('min_order_quantity')}")

            # ③ 库存快照
            stock = cli.get_stock_snapshot(offer_id)
            print(f"   SKU库存快照: {len(stock['skus'])} 个规格")

        # ② 查待发货采购单（中台定时任务入口）
        orders = cli.list_orders(status="waitsellersend", hours_back=72)
        print(f"\n📋 待发货采购单: {len(orders)} 笔")
        for o in orders[:3]:
            oid = o.get("id")
            lg = cli.get_logistics(str(oid))
            print(f"   单 {oid} → 物流单: {[x.get('billNo') for x in lg]}")

    except Exception as e:
        print("❌", e)

三、中台字段映射要点（面试/设计必问）

四、生产级注意事项

1. 订单创建接口：示例中未展示 alibaba.trade.create（需传 offerId + skuId + quantity + consignee），中台应在运营确认采购→生成内部PO→调此接口→回填1688 orderId

2. 库存防超卖：展示层用 amount_on_sale做预警；扣减后若1688返回库存不足，中台标记采购异常走人工复核

3. 物流轮询：已付款待发货订单每30分钟查一次 list_orders(waitsellersend)→ get_logistics，签收到后调ERP入库接口

4. 幂等：以1688 orderId作为中台采购单幂等键，重复回调不重复建单

5. Access Token刷新：token通常1年过期，中台需实现OAuth2 refresh_token定时刷新

五、一句话总结（方案汇报版）

1688供应链中台 = 商品API建SKU映射 → 采购订单API创建并回填1688单号 → 定时轮询订单/物流API回写发货/签收状态 → 库存快照做采购预警，全链路用官方API免费完成，关键是SKU对照表 + 状态机 + QPS限速。