1688订单对接ERP的核心目标是实现采购单自动创建→状态回传→发货同步→入库确认的全链路自动化。下面给出可直接落地的Python实现，并标注1688开放平台的关键避坑点。

一、 1688订单对接的四种核心场景

⚠️ 前置条件：应用需申请采购单查询、物流查询、确认收货权限，且是自用型应用（非第三方ISV可免OAuth复杂授权，用session key即可）。

二、 Python封装：1688订单同步客户端

# ali1688_order_sync.py
import hashlib
import time
import requests
from datetime import datetime, timedelta
from typing import Dict, List, Optional
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex 注册链接
class Ali1688OrderClient:
    """1688 采购订单同步客户端"""

    def __init__(self, app_key: str, app_secret: str, access_token: str):
        self.app_key = app_key
        self.app_secret = app_secret
        # access_token 通过1688 OAuth2或后台获取(session key)
        self.access_token = access_token
        self.gateway = "https://gw.open.1688.com/openapi/http/2/1"

    # ──────────────────────────────────────────────
    # 签名：1688 HMAC-MD5 标准方式
    # ──────────────────────────────────────────────
    def _sign(self, params: Dict) -> str:
        sorted_items = sorted((k, v) for k, v in params.items() if v is not None)
        qs = ''.join(f"{k}{v}" for k, v in sorted_items)
        raw = f"{self.app_secret}{qs}{self.app_secret}"
        return hashlib.md5(raw.encode('utf-8')).hexdigest().upper()

    def _call(self, method: str, biz_params: Dict) -> Dict:
        """通用调用封装"""
        params = {
            "method": method,
            "app_key": self.app_key,
            "session": self.access_token,
            "timestamp": str(int(time.time() * 1000)),
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
        }
        # 1688把业务参数整体做URL编码放进 'param2'
        import urllib.parse
        params["param2"] = urllib.parse.quote_plus(str(biz_params).replace("'", '"')))
        params["sign"] = self._sign(params)

        resp = requests.get(self.gateway, params=params, timeout=15)
        resp.raise_for_status()
        result = resp.json()

        if "error_response" in result:
            err = result["error_response"]
            raise Exception(f"1688 API Error [{err.get('code')}]: {err.get('msg')}")
        return result

    # ──────────────────────────────────────────────
    # 1. 拉取买家订单列表（新订单/待发货）
    # ──────────────────────────────────────────────
    def list_orders(self, status: str = "waitbuyerpay",
                     start_hours: int = 2, page_no: int = 1,
                     page_size: int = 50) -> List[Dict]:
        """
        status 可选值:
          waitbuyerpay 待付款
          waitsellersend 待卖家发货(已付款) ← 通常用这个
          partconsigned   部分发货
          finshed        已完成
        """
        biz = {
            "pageNo": page_no,
            "pageSize": page_size,
            "gmtCreateStart": (datetime.now() - timedelta(hours=start_hours)).strftime("%Y-%m-%d %H:%M:%S"),
            "gmtCreateEnd": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "orderStatus": status,
            "isHis": "false"   # false=查近90天
        }
        res = self._call("alibaba.trade.buyer.list", biz)
        data = res.get("alibaba_trade_buyer_list_response", {})
        orders = data.get("tradeModelList", []) or []
        total = data.get("totalRecord", 0)
        return orders, total

    # ──────────────────────────────────────────────
    # 2. 获取订单明细（含SKU、单价、数量）
    # ──────────────────────────────────────────────
    def get_order_detail(self, order_id: str) -> Dict:
        biz = {"orderId": order_id}
        res = self._call("alibaba.trade.get", biz)
        return res.get("alibaba_trade_get_response", {}).get("result", {})

    # ──────────────────────────────────────────────
    # 3. 查询发货物流（卖家已发货时同步到ERP）
    # ──────────────────────────────────────────────
    def get_logistics(self, order_id: str) -> List[Dict]:
        biz = {"orderId": order_id}
        res = self._call("alibaba.logistics.trade.ship", biz)
        return res.get("result", {}).get("logisticsOrders", []) or []

    # ──────────────────────────────────────────────
    # 4. 确认收货（ERP入库后调用）
    # ──────────────────────────────────────────────
    def confirm_receive(self, order_id: str, sub_order_ids: Optional[List[str]] = None) -> bool:
        biz = {"orderId": order_id}
        if sub_order_ids:
            biz["subOrderIds"] = ",".join(sub_order_ids)
        res = self._call("alibaba.trade.confirmReceive", biz)
        return res.get("alibaba_trade_confirmreceive_response", {}).get("result", False)


# ============================================================
# 使用示例：同步待发货订单 → 写ERP本地库 → 查询物流
# ============================================================
if __name__ == "__main__":
    client = Ali1688OrderClient(
        app_key="YOUR_APP_KEY",
        app_secret="YOUR_APP_SECRET",
        access_token="YOUR_ACCESS_TOKEN"   # 1688后台获取session key
    )

    try:
        # ① 拉取待发货订单
        orders, total = client.list_orders(
            status="waitsellersend",   # 已付款待发货
            start_hours=48,            # 查近48h新建订单
            page_no=1
        )
        print(f"✅ 发现待处理订单 {len(orders)} / 总计 {total}")

        for order in orders:
            order_id = str(order.get("id"))
            print(f"\n📦 订单 {order_id} | 金额:{order.get('totalAmount')} | 状态:{order.get('status')}")

            # ② 获取明细（SKU匹配ERP商品编码）
            detail = client.get_order_detail(order_id)
            products = detail.get("productItems", []) or []
            for p in products:
                print(f"   └ SKU:{p.get('specId')} 商品:{p.get('name')} "
                      f"单价:{p.get('price')} × {p.get('quantity')}")

            # ③ 查物流（若卖家已发货）
            logistics = client.get_logistics(order_id)
            if logistics:
                for lg in logistics:
                    print(f"   🚚 物流单号:{lg.get('logisticsCode')} {lg.get('billNo')}")

            # ④ ERP入库后可调用确认收货
            # client.confirm_receive(order_id)

    except Exception as e:
        print(f"❌ 同步失败: {e}")

三、 ERP侧处理逻辑建议

┌──────────────┐    定时任务     ┌────────────────────┐
│ 1688开放平台 │ ──────────────▶ │ 同步服务(Python)   │
│ (订单/物流)  │ ◀────────────── │ ①查待发货订单      │
└──────────────┘  回写确认收货   │ ②写ERP采购单表     │
                                │ ③物流状态→出库单    │
                                │ ④入库后调confirmReceive│
                                └────────────────────┘

四、 关键避坑清单

1. Access Token过期：session(access_token)有效期通常1年，过期需在1688后台重新授权刷新，不要硬编码。

2. 时间窗口限制：gmtCreateStart/End跨度≤90天，建议每次拉近2小时+兜底补跑前一天。

3. 部分发货订单：1688支持分批发货，partconsigned状态需单独处理，按subOrderId匹配。

4. SKU映射：1688的specId是平台规格ID，ERP内需建1688_spec_id ↔ 内部sku_code映射表，否则无法自动匹配。

5. 接口限流：默认QPS≈10，建议加sleep(0.2)或使用队列控制并发。

五、 面试官追问：如何保证不丢单？

• 增量轮询+兜底全量：常规每5分钟按gmtModified增量拉取，每日凌晨跑一次48h全量比对。

• 幂等写入：ERP以order_id + sub_order_id为主键，重复推送不重复建单。

• 异常重试：API超时/限流用指数退避重试3次，失败记入死信表人工复核。

如需我补充OAuth2获取AccessToken完整流程或ERP入库SQL模板，直接说即可 👍