1688订单对接ERP，本质上就是实现采购单自动创建、状态回传、发货同步与入库确认的完整闭环。如果你正在开发这套流程，下面这份Python实现的方案可以直接复用，代码中还标注了1688开放平台几个容易踩坑的地方，帮你节省调试时间。

一、 1688订单对接的四大核心场景与接口选择

核心场景其实只有四个，把逻辑理清后，剩下的就是代码层面的落地工作：

拉取新订单，通过 alibaba.trade.buyer.list 接口进行定时轮询，建议频率控制在5到10分钟一次，获取数据后在ERP端创建采购单。

订单明细，使用 alibaba.trade.get，根据orderId查询，补全SKU、金额等信息。

发货状态同步，借助 alibaba.logistics.trade.ship，轮询或等待推送均可，获取物流单号后更新到ERP。

确认收货，调用 alibaba.trade.confirmReceive，在ERP完成入库操作后触发，关闭在途状态。

⚠️ 特别注意：应用必须先申请采购单查询、物流查询、确认收货这几个权限。建议使用自用型应用，如果非第三方ISV，可以省掉复杂的OAuth授权流程，直接用session key即可。

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

下面这个客户端类，将上述四个场景全部封装完毕，直接替换app_key和access_token即可运行。

ali1688_order_sync.py

import hashlib
import time
import requests
from datetime import datetime, timedelta
from typing import Dict, List, Optional

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)

使用示例非常直接：拉取待发货订单、查询明细、查看物流，一条龙走完：

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"? 订单 {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。定时任务负责触发，逻辑上分四步走：先查询待发货订单，然后写入ERP采购单表，接着把物流状态同步到出库单，最后入库完成后再调用确认收货，将整个流程闭环。

字段映射方面，建议提前梳理清楚：1688的 id 对应ERP的 outer_order_no（外部单号），productItems[].specId 对应 sku_code，但这需要预先维护一张映射表。其他字段如单价、数量、物流单号基本直接对应，没有复杂逻辑。

四、 关键避坑清单与实战经验

Access Token过期：session（access_token）有效期通常为一年，到期后需前往1688后台重新授权，切勿硬编码死token。

时间窗口限制：gmtCreateStart/End 的跨度不能超过90天。建议每次只拉取近2小时的数据，再补跑前一天的数据，这样不容易漏单。

部分发货订单：1688支持分批发货，partconsigned 状态需要单独处理，按 subOrderId 进行匹配，不要整体处理。

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

接口限流：默认QPS大约为10，建议每条请求之间加 sleep(0.2)，或用队列控制并发，防止被封。

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

这个问题本质上考察系统设计能力。两个核心思路：

增量轮询 + 兜底全量。常规情况下每5分钟按 gmtModified 增量拉取一次。但仅靠增量不够保险，每天凌晨再跑一次近48小时的全量数据做比对，基本能覆盖所有遗漏。

幂等写入。ERP侧以 order_id + sub_order_id 做主键，即使接口重推数据，也不会重复建单。

异常重试。API超时或触发限流时，采用指数退避策略重试3次。如果连续失败，将记录丢入死信表，留待人工复核。