万邦api博客

《亚马逊 SP-API：落地方案、接口边界与业务踩坑 —— “年费 $1400 + 超量 $0.40/千次” 的真相与 2026 反转》（附 Python 源码）

Sat, 17 Oct 2026 10:05:53 +0800

《亚马逊 SP-API：落地方案、接口边界与业务踩坑 —— “年费 $1400 + 超量$ 0.40/千次” 的真相与 2026 反转》（附 Python 源码）

0. 先纠正一个时间线（很重要）

你标题里的 “年费 $1400 + GET 超量$ 0.40/千次” 是亚马逊 2025-11-03 公布、原定 2026-01-31 / 2026-04-30 生效 的收费方案，但：

2026-01：延期
2026-03：无限期推迟
2026-05-12：Amazon 正式邮件宣布 “will not move forward with the SP-API usage and annual fees at this time”

所以 2026-10-17 今天写这篇文章，正确姿态是：

把 $1400 /$ 0.40 当“成本模型教材”和“未来可能再来一次的防御设计”，但不要把生产架构建立在这个账单一定存在上。
Amazon 原话留了 at this time，以后换种形态收费完全可能。

原方案回顾（仅作历史/压测口径）：

项目	数值
年费（第三方开发者）	$1400 / 年
Basic 档包含 GET	250 万次 / 月
超量 GET	$0.40 / 1000 次
Pro / Plus	$1000 /$ 10000 月费
PUT/POST/PATCH	不按量计费，但受限流
自营私域 App（只给自己店铺用）	不收开发者费

一、SP-API 和前面几家根本不是一类东西

平台	模型	计费
1688 / 淘宝 / 京东 / 拼多多 / 抖店	国内开放平台，按调用/云内云外	预充值 / 塔内塔外 / 云外×10
亚马逊 SP-API	LWA OAuth + Role/App 授权 + REST + 通知/报告分离	原拟收费，现免费，但限流极狠

SP-API 核心组成：

Auth：Login with Amazon（LWA）→ access_token（约 1h）+ refresh_token（长期）
Selling Partner API：orders / catalog / listings / inventory / feeds / reports / notifications / finances / shipments
EventBridge / SQS 通知：ORDER_CHANGE、REPORT_PROCESSING_FINISHED 等
Reports API：建报告 → 轮询状态 → 下载文档（大批量正道）
Dynamic Usage Plan（DUP）：你看到的文档限流是最低档，Amazon 会根据历史自动调

二、订单接口边界（最容易被国内团队写错）

接口	角色	限流（默认）
`getOrders`	按 CreatedAfter / LastUpdatedAfter 拉订单列表	0.0167 rps = 1/min，burst 20
`getOrder`	单订单头信息	0.5 rps / burst 30
`getOrderItems`	订单行项目（价格/数量/ASIN/买家配送）	0.5 rps / burst 30
Orders Report	批量订单 CSV	建报告 1/min，下载另算
ORDER_CHANGE 通知	订单变更事件	不耗 getOrders 额度

⚠️ 国内团队最常犯的错：

每 5 秒调一次 getOrders 查新单 → 10 秒内把 burst 打光 → 后面全 429

Amazon 官方结论：

日单 < 2 万：getOrders + getOrderItems 可行
大单量：ORDER_CHANGE 通知 + Orders Report

还有一条隐藏规则：

LastUpdatedAfter 查到的时间是 “当前时间 - 2 分钟”，新单有 2 分钟延迟。

三、生产架构：通知优先，Reports 批量，REST 兜底

EventBridge / SQS
  └─ ORDER_CHANGE (orderId, changeType)
        └─ 入队 (amazonOrderId)

Worker:
  getOrder(orderId)         # 订单头
  getOrderItems(orderId)    # 订单行
  → 写 order 快照

每 30~60 min：
  Orders Report (GET_FLAT_FILE_ORDER_REPORT)
    → 批量对账，补漏

财务/库存：
  GET_FBA_MYI_UNSUPPRESSED_INVENTORY_DATA
  GET_LEDGER_SUMMARY
  → Reports API，不逐 SKU 调 getInventorySummaries

规则（和淘宝 DSS / 京东 JMQ / 抖店推送同构）：

通知做触发器，Report 做批量真相，getOrder 做单条补全，轮询 getOrders 只做兜底。*

四、Python：LWA 令牌管理（refresh 不互踢）

# spapi/auth.py
import time
import requests


class LwaTokenProvider:
    def __init__(self, client_id, client_secret, refresh_token,
                 redis=None, ttl_margin=300):
        self.client_id = client_id
        self.client_secret = client_secret
        self.refresh_token = refresh_token
        self.redis = redis
        self.margin = ttl_margin
        self._cache = {}

    def _redis_key(self):
        return f"spapi:lwa:access:{self.client_id}"

    def get_access_token(self) -> str:
        if self.redis:
            tok = self.redis.get(self._redis_key())
            if tok:
                return tok

        resp = requests.post("https://api.amazon.com/auth/o2/token", data={
            "grant_type": "refresh_token",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "refresh_token": self.refresh_token,
        }, timeout=5)
        resp.raise_for_status()
        data = resp.json()
        token = data["access_token"]
        expires_in = int(data.get("expires_in", 3600))

        if self.redis:
            self.redis.set(self._redis_key(), token, ex=expires_in - self.margin)
        return token

多实例用 Redis 锁防并发刷新：

def refresh_with_lock(provider, r):
    lock = r.lock("spapi:lwa:refresh:lock", timeout=10)
    if lock.acquire(blocking=False):
        try:
            r.delete(provider._redis_key())
            provider.get_access_token()
        finally:
            lock.release()

五、Python：限流感知的 SP-API 客户端

# spapi/client.py
import time
import requests


class SpApiThrottleError(Exception):
    def __init__(self, retry_after):
        super().__init__("throttled")
        self.retry_after = retry_after


class SpApiClient:
    def __init__(self, token_provider, region_endpoint,
                 rate_per_sec=0.0167, burst=20):
        self.tp = token_provider
        self.base = region_endpoint   # NA/EU/FE 不同
        self.rate = rate_per_sec
        self.burst = burst
        self._tokens = burst
        self._last = time.monotonic()

    def _acquire(self):
        while True:
            now = time.monotonic()
            self._tokens = min(
                self.burst,
                self._tokens + (now - self._last) * self.rate
            )
            self._last = now
            if self._tokens >= 1:
                self._tokens -= 1
                return
            time.sleep(0.2)

    def _request(self, method, path, params=None, retries=5):
        for i in range(retries):
            self._acquire()
            token = self.tp.get_access_token()
            headers = {
                "x-amz-access-token": token,
                "Accept": "application/json",
            }
            url = f"{self.base}{path}"
            resp = requests.request(method, url, headers=headers,
                                    params=params, timeout=10)
            if resp.status_code == 429:
                ra = int(resp.headers.get("Retry-After", 60))
                time.sleep(ra + (i * 2))
                continue
            if resp.status_code >= 500:
                time.sleep(2 ** i)
                continue
            resp.raise_for_status()
            return resp.json()
        raise RuntimeError("sp-api retry exhausted")

    def get_orders(self, last_updated_after, next_token=None):
        p = {"MarketplaceIds": "...", "LastUpdatedAfter": last_updated_after}
        if next_token:
            p["NextToken"] = next_token
        return self._request("GET", "/orders/v0/orders", p)

    def get_order_items(self, amazon_order_id):
        return self._request("GET", f"/orders/v0/orders/{amazon_order_id}/orderItems")

六、订单同步：通知 + Report + getOrders 兜底

# spapi/order_sync.py
from datetime import datetime, timedelta


class SpOrderSync:
    def __init__(self, client, repo, dedup):
        self.client = client
        self.repo = repo
        self.dedup = dedup

    def on_order_change(self, amazon_order_id: str):
        # 通知只给 id，不信任通知体里的状态
        self.dedup.mark(f"order_change:{amazon_order_id}")
        self.fill_order(amazon_order_id)

    def fill_order(self, amazon_order_id):
        head = self.client._request("GET", f"/orders/v0/orders/{amazon_order_id}")
        items = self.client.get_order_items(amazon_order_id)
        snap = self.normalize(head, items)
        self.apply(snap)
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    def poll_compensation(self, marketplace_ids, now: datetime):
        # 结束时间留 2 分钟缓冲（Amazon 订单有 2 分钟延迟）
        end = now - timedelta(minutes=2)
        start = end - timedelta(minutes=30)
        next_token = None
        while True:
            resp = self.client.get_orders(
                last_updated_after=start.isoformat(),
                next_token=next_token,
            )
            for o in resp["payload"]["Orders"]:
                self.on_order_change(o["AmazonOrderId"])
            next_token = resp["payload"].get("NextToken")
            if not next_token:
                break
            # getOrders 默认 1/min，别在循环里硬跑：
            time.sleep(60)

七、Reports 模式：大单量别用 getOrders

# spapi/reports.py
class SpReportClient:
    def __init__(self, client):
        self.client = client

    def pull_orders_report(self, marketplace_ids, start, end):
        create = self.client._request("POST", "/reports/2021-06-30/reports", {
            "reportType": "GET_FLAT_FILE_ORDER_REPORT",
            "marketplaceIds": marketplace_ids,
            "dataStartTime": start.isoformat(),
            "dataEndTime": end.isoformat(),
        })
        report_id = create["reportId"]

        # 轮询状态：别狂轮，指数退避
        for _ in range(30):
            status = self.client._request(
                "GET", f"/reports/2021-06-30/reports/{report_id}"
            )["payload"]
            if status["processingStatus"] == "DONE":
                return self.download(status["reportDocumentId"])
            time.sleep(min(60, 5 * (2 ** _)))
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    def download(self, doc_id):
        doc = self.client._request(
            "GET", f"/reports/2021-06-30/documents/{doc_id}"
        )["payload"]
        # doc["url"] 是带签名的临时地址
        r = requests.get(doc["url"])
        return r.text  # TSV/CSV

降本效果：

1 万订单：getOrderItems 可能 1 万次 GET
Orders Report：建 1 次 + 状态轮询 ~5 次 + 下载 1 次 = 7 次调用

八、限流 / 错误分类（Amazon 特有）

现象	类型	处理
429 `QuotaExceeded`	限流	读 `Retry-After`，指数退避
403 Access Denied	授权/Roles 错	不重试，查 IAM/Data Rights
404	orderId 错 / 区域错	不重试
500/503	平台抖动	退避重试
`x-amzn-RateLimit-Limit`	当前档位	监控用
DUP 提额	历史好	别依赖，仍要做退避

Amazon 限流算法 = token bucket，但每个 operation 独立桶，且按 account × app 双维度限流。

九、“如果真收 $0.40/千次” 的防御代码

# spapi/cost_meter.py
class SpApiGetMeter:
    def __init__(self, redis, app_id, budget_usd_per_month=50.0):
        self.r = redis
        self.key = f"spapi:get:cost:{app_id}:{month_now()}"
        self.price_per_call = 0.40 / 1000

    def count(self, n=1):
        total = self.r.incrby(self.key, n)
        cost = total * self.price_per_call
        if cost > 50.0:
            raise BudgetExceeded(f"SP-API GET 预算将超: ${cost:.2f}")

    def estimated_cost(self):
        return int(self.r.get(self.key) or 0) * self.price_per_call

调用侧：

meter.count(1)
resp = client.get_orders(...)

规则：

GET 才计费，POST/PUT 不按量
Reports 把 N 次 GET 压成几次 → 天然省钱
通知替代轮询 → 把“心跳轮询”成本降到 0
多卖家共享 App → 调用量聚合，更容易撞档 → 私域 App / 多 App 分租户要算清

十、和前几篇收口

1688：secret 头尾 MD5，QPS 卡脖
淘宝：聚石塔内外 10 倍，DSS 推送免费
京东：云鼎内稳，POP 时间窗
拼多多/抖店：预充值，云外 ×10，增值禁外
亚马逊 SP-API：不限国内云内云外概念，限流极狠 + 通知/报告分离 + 原拟 $1400+$ 0.40 但 2026-05 取消

一句话：

国内平台是“钱换调用”，亚马逊是“限流换架构”。
你用国内 ERP 的轮询本能写 SP-API，会先被 429 打服，再被（可能的）账单打醒。
SP-API 的正解永远是：EventBridge 叫你、Reports 扛批量、getOrder 只补刀、NextToken 别硬翻、2 分钟延迟写进时间窗。*

《抖音订单API：/trade/order/detail 落地方案 —— 消息推送 + 补偿查询的架构设计》（附 Python 源码）

Sat, 17 Oct 2026 09:15:55 +0800

《抖音订单API：/trade/order/detail 落地方案 —— 消息推送 + 补偿查询的架构设计》（附 Python 源码）

一、先拍结论

抖店订单同步的成熟架构只有一句话：
“消息/数据推送做触发器，/order/searchList 做时间窗兜底，/order/orderDetail 拉真相，本地状态机做仲裁。”
你写的 /trade/order/detail 在抖店开放平台里更接近“抖音开放平台担保支付/交易订单”语义；抖店（抖店开放平台）真实订单详情是 /order/orderDetail，列表是 /order/searchList。

官方自己都明说：

订单详情接口存在部分数据返回延时，订单生成后延时 10s 再查；极端情况延迟 20 分钟，要用补偿机制再查一次
数据推送（抖店云 RDS）按 B 端事件触发，不是底层表每行变更都推
只靠消息推送 = 高峰丢包/重试失败/服务器慢响应 → 漏单
正确姿势：消息推送 + 接口轮询补偿

二、抖店订单接口边界（先纠名）

接口	角色	说明
消息推送 `doudian_trade_TradeCreate/Paid/...`	触发器	只给 order_id / 事件类型
数据推送（抖店云 RDS）	准实时落库	推密文字段到绑定库，订单/售后/商品可配
`/order/searchList`	补偿/初始化	按创建时间或更新时间拉列表
`/order/orderDetail`	真相源	子单、优惠、收货密文、履约、售后态
`/afterSale/List` + `/afterSale/Detail`	售后真相	退款/退货/换货独立轴

规则：
消息 ≠ 订单数据，列表 ≠ 订单详情，数据推送 ≠ 业务已处理。

三、整体架构

┌──────────────────────────────┐
│ 抖店平台                      │
│  - 消息推送（TradeCreate等）  │
│  - 数据推送（RDS 密文）       │
└────────────┬─────────────────┘
             │
             ▼
┌──────────────────────────────┐
│ Push Ingest（2~5s 返回 success）│
│  - msg_id 去重                │
│  - 只入队 (shop_id, order_id) │
└────────────┬─────────────────┘
             ▼
     ┌─────────────────┐
     │  Order Detail    │
     │  Worker          │
     │ /order/orderDetail│
     │ +10s 延迟 + 20min │
     │   补偿            │
     └────────┬──────────┘
              ▼
     ┌─────────────────┐
     │ 状态机 + 幂等写   │
     │ version 仲裁     │
     └────────┬──────────┘
              ▼
     ┌─────────────────┐
     │ 补偿轮询 Cron     │
     │ /order/searchList│
     │ 更新时间窗 滚动   │
     └─────────────────┘

双通道：

推模型：快、省调用、但会丢/重/乱序
拉模型：慢、烧调用、但能兜底

四、为什么不能直接“消息来了就发货”

抖店订单详情有延迟：

新单 10s 内可能拿不到完整数据
达人信息要等联盟结算，最长延迟 1 小时
数据推送的 version 不再是时间戳，不能拿来做时间排序
翻页时订单状态变化 → 正向翻页跳单 → 必须倒序/时间窗重叠

所以消息只干一件事：

order_id 进队，标记 need_detail=true

五、Python：抖店订单 Client

# douyin/order_client.py
import time
import hashlib
import hmac
import requests


class DouyinOrderClient:
    GATEWAY = "https://openapi-fxg.jinritemai.com/"

    def __init__(self, app_key, app_secret, access_token, sign_method="hmac-sha256"):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        self.sign_method = sign_method

    def _sign(self, params: dict) -> str:
        qs = "&".join(f"{k}={params[k]}" for k in sorted(params))
        if self.sign_method == "hmac-sha256":
            return hmac.new(self.app_secret.encode(), qs.encode(), hashlib.sha256).hexdigest()
        return hashlib.md5((self.app_secret + qs + self.app_secret).encode()).hexdigest().upper()
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    def _common(self, method: str):
        return {
            "method": method,
            "app_key": self.app_key,
            "access_token": self.access_token,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "v": "2",
            "sign_method": self.sign_method,
            "format": "json",
        }

    def _post(self, method: str, biz: dict):
        params = self._common(method)
        params["param_json"] = str(biz).replace("'", '"')
        params["sign"] = self._sign(params)
        resp = requests.post(self.GATEWAY, data=params, timeout=5)
        return resp.json()

    def search_list(self, start_time, end_time, order_by=1, page=1, page_size=100, order_status=None):
        biz = {
            "start_time": start_time,
            "end_time": end_time,
            "order_by": order_by,       # 1=按更新时间倒序更稳
            "page": page,
            "page_size": page_size,
            "is_desc": 1,
        }
        if order_status is not None:
            biz["order_status"] = order_status
        return self._post("order.searchList", biz)

    def order_detail(self, order_id: str):
        return self._post("order.orderDetail", {"order_id": order_id})

六、Detail Worker：10s 延迟 + 20min 补偿

# douyin/order_detail_worker.py
import time


class OrderDetailWorker:
    def __init__(self, client, repo, dedup, bucket, queue):
        self.client = client
        self.repo = repo
        self.dedup = dedup
        self.bucket = bucket
        self.queue = queue

    def handle(self, shop_id: str, order_id: str, delay=10):
        # 第一次：延 10s 再查
        time.sleep(delay)
        snap = self._fetch(order_id)
        if not self._data_complete(snap):
            # 极端延迟：20 分钟后补偿
            self.queue.enqueue_delay(
                {"shop_id": shop_id, "order_id": order_id, "phase": "compensate"},
                delay=20 * 60,
            )
            return

        self._apply(shop_id, snap)

    def _fetch(self, order_id: str):
        self.bucket.acquire()
        resp = self.client.order_detail(order_id)
        return (resp.get("data") or {}).get("order_info") or resp
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    def _data_complete(self, snap) -> bool:
        if not snap:
            return False
        # 刚支付时可能缺收件/子单/优惠拆分
        if snap.get("pay_status") in (1, 2) and not snap.get("sku_order_list"):
            return False
        return True

    def _apply(self, shop_id, snap):
        event_key = f"detail:{snap.get('order_id')}:{snap.get('update_time')}"
        if not self.dedup.mark(shop_id, snap.get("order_id"), event_key):
            return
        self.repo.upsert_order(shop_id, snap)
        self.repo.emit_business_events(snap)

七、补偿轮询：时间窗前移 + 倒序 + 重叠

# douyin/order_compensator.py
from datetime import datetime, timedelta


class OrderCompensator:
    def __init__(self, client, repo, bucket):
        self.client = client
        self.repo = repo
        self.bucket = bucket

    def run(self, shop_id, now: datetime):
        # 结束时间前移 2~5 分钟，避开平台入库延迟
        end = now - timedelta(minutes=3)
        start = end - timedelta(minutes=30)

        page = 1
        min_update_seen = end

        while True:
            self.bucket.acquire()
            resp = self.client.search_list(
                start_time=int(start.timestamp()),
                end_time=int(end.timestamp()),
                order_by=1,     # 更新时间倒序
                page=page,
            )
            orders = self._extract(resp)
            if not orders:
                break

            for o in orders:
                self.repo.mark_need_detail(shop_id, o["order_id"], o.get("update_time"))
                u = int(o.get("update_time", 0))
                if u < min_update_seen.timestamp():
                    min_update_seen = datetime.fromtimestamp(u)

            if len(orders) < 100 or page >= 50:
                break
            page += 1

        # 下一轮从最小 update_time 往前，并留 1 分钟重叠防跳单
        return min_update_seen - timedelta(minutes=1)

规则（抖店专用）：

结束时间 前移 2~5 分钟
按更新时间倒序，从后往前翻
下一轮起点 = 本批最小 update_time - 重叠窗口
不深翻页，不靠 page=9999

八、幂等键与状态机

def order_version(snap) -> int:
    # 抖店 version 不是时间戳了，用 update_time 做业务版本
    return int(snap.get("update_time") or 0)


def apply_order(repo, shop_id, snap):
    cur = repo.get_order(shop_id, snap["order_id"])
    new_ver = order_version(snap)

    if cur is None:
        repo.insert_order(shop_id, snap)
        fire_events(None, snap)
        return

    if new_ver < cur.version:
        return  # 旧消息晚到
    if new_ver == cur.version and snap.get("order_status") == cur.order_status:
        return  # 完全幂等

    check_order_transition(cur.order_status, snap["order_status"])
    rows = repo.update_order_if_newer(shop_id, snap, cur.version)
    if rows == 1:
        fire_events(cur, snap)

SQL 核心：

UPDATE douyin_order
SET order_status=#, pay_amount=#, post_addr=#,
    version=#, update_time=#
WHERE shop_id=# AND order_id=#
  AND version < #{new_version};

九、消息推送接入（只做触发器）

# douyin/push_webhook.py
def on_push(body: dict):
    # 2~5 秒内必须返回 success
    msg_id = body["msg_id"]
    if dedup.msg_seen(msg_id):
        return "success"

    tag = body["tag"]                 # TradeCreate / TradePaid ...
    order_id = body["data"]["order_id"]
    shop_id = body["shop_id"]

    queue.publish({"shop_id": shop_id, "order_id": order_id, "trigger": tag})
    dedup.mark_msg(msg_id)
    return "success"

消息类型（节选）：

doudian_trade_TradeCreate：下单
doudian_trade_TradePaid：支付成功
doudian_trade_TradePartlySellerShip / TradeSellerShip：发货
doudian_trade_TradeSuccess：完成
doudian_trade_TradeCanceled：取消
doudian_trade_AuthorInfo：达人信息补推

消息只告诉你要“去看哪个 order_id”，别把消息体当订单真相。

十、数据推送（抖店云 RDS）怎么摆

抖店云数据推送：

推到绑定 RDS，不是推到你的业务库
订单/售后/商品可配
历史 60 天、保留 60 天可配
version 不再当时间戳

生产摆法：

抖店云 RDS（密文）
   └─ CDC / 定时读
        └─ 业务库：order_raw（密文）
             └─ 解密服务（云内）
                  └─ order_biz（明文只在云内）

合规底线：

明文收货信息/手机号解密服务必须抖店云内；云外只持有 order_id / 状态 / 金额 / 商品，不持有明文。

十一、售后轴独立（和拼多多同思路）

def sync_aftersale(shop_id, start, end):
    while True:
        resp = client._post("afterSale.List", {
            "start_time": start, "end_time": end, "page": page, "page_size": 100
        })
        for a in extract(resp):
            detail = client._post("afterSale.Detail", {
                "order_id": a["order_id"], "after_sale_id": a["after_sale_id"]
            })
            apply_aftersale(shop_id, detail)

订单表只存：

after_sale_status = 最新一条售后单的投影

真实仲裁看 douyin_after_sale 表。

十二、调用配额与降本

消息推送：不按 API 次数烧钱
数据推送：走抖店云 RDS，不是 /order/orderDetail 轮询
/order/searchList：补偿用，别当实时主源
/order/orderDetail：按 order_id 触发，不批量盲拉
新单：消息 → 10s 后 detail
老单/漏单：30min 窗补偿
大促：searchList 窗缩到 5~10min，detail 桶单独限流

十三、落地检查表

[ ] 消息 webhook 2~5s 返回 success，不写业务
[ ] msg_id 去重，order_id 进队
[ ] detail 延 10s 查，不完整就排 20min 补偿
[ ] searchList 结束时间前移 3min，倒序，重叠 1min
[ ] order_id 唯一键 + event_key 去重 + version(update_time) 仲裁
[ ] 订单轴 / 售后轴 / 达人信息轴分开
[ ] 达人信息用 doudian_trade_AuthorInfo 覆盖，别信创建时刻
[ ] 明文解密服务在抖店云内，云外只留密文/状态/金额
[ ] 数据推送 RDS 是“准实时源”，不是“业务已处理”
[ ] 大促补偿窗缩短、detail QPS 单独桶、余额守卫在 Client 层

十四、一句话收口

淘宝用 DSS 推、京东用 JMQ 推、拼多多靠增量轮询、抖店是“消息推送 + 抖店云数据推送 + searchList 补偿 + orderDetail 拉真相”四件套。
抖店订单最阴的三件事是：
新单详情延迟 10s~20min、version 不再是时间戳、达人信息要等联盟结算补推。
所以别把消息当数据、别把列表当详情、别把创建时刻的达人信息当最终口径。
消息负责“叫你起床”，searchList 负责“查谁没醒”，orderDetail 负责“看清楚这张脸”。

《抖音商品API：/product/detail 与 /product/list 的字段边界与图片合规踩坑》（附 Python 源码）

Fri, 16 Oct 2026 14:19:29 +0800

《抖音商品API：/product/detail 与 /product/list 的字段边界与图片合规踩坑》（附 Python 源码）

一、先拍结论

抖店商品域最容易出事的地方不是“调不通”，而是：
把 /product/list 当详情用、把 pic 当 SKU 图用、把白底图当首图用、把平台返回图链当永久 CDN 用、把类目属性当可选填。
/product/list：列表视图，适合兜底/增量/比状态，字段是“缩略版”
/product/detail：商品真相源，返回规格、SKU、图片、类目、属性、审核态
pic：商品轮播主图
spec_pics：SKU 规格图，要用 spec_detail_id 关联，不是用 spec_id 硬取
图片合规：首图实物无字、白底图单独传、3:4 或 1:1、无二维码/无极限词/无第三方水印
类目属性：product_format_new 等必填属性不填 → 发品直接被审核打回

二、两个接口的边界（别再混）

维度	`/product/list` / `/product/listV2`	`/product/detail`
定位	商品列表/分页/状态扫描	单品真相
入参	`page/page_size/status/update_time`	`product_id` 或 `out_product_id`
返回	基础字段：标题、状态、价格、主图、类目	完整：spec、sku、spec_pics、desc、类目属性
成本	列表类调用，店群别乱翻页	详情类调用，按商品触发
用法	兜底漏单、状态机推进	写商品主数据、SKU 映射、图片合规校验
坑	以为列表里有完整 SKU 图	以为 detail 不收费/可以无限拉

生产铁律：

list  = 找哪些商品变了
detail = 把变了的那个商品看清楚
sku/list = 拿 SKU 级 spec_detail_id
spec_pics = 按 spec_detail_id 关联 SKU 图

官方明确说过：product/detail 返回 spec_pics，但 不能用 /sku/list 的 spec_id 去取 spec_pics，要用 spec_detail_id1 和 spec_pics.spec_detail_id 关联；图片链接往往还要拼固定域名。

三、核心字段边界

1. 商品级

{
  "product_id": "3558192687276554544",
  "out_product_id": "dy001",        // 新推荐；out_product_id 是老字段，逐渐废弃
  "name": "纯棉短袖T恤 男 夏季",
  "pic": ["https://p3-.../a.jpg"],  // 商品轮播主图，不是 SKU 图
  "status": 1,                       // 上下架
  "check_status": 2,                 // 审核状态
  "cid": 12345,                     // 类目
  "product_format_new": {...},       // 类目必填属性
  "description": "<p>详情页HTML</p>"
}

2. SKU 级

{
  "sku_id": "123",
  "out_sku_id": 456,
  "outer_sku_id": "A001",            // 新推荐 string 版
  "spec_detail_id1": "1001",
  "spec_detail_id2": "1002",
  "spec_detail_name1": "颜色",
  "spec_detail_name2": "尺码",
  "price": 3900,                     // 分
  "stock_num": 10
}

3. 规格图级

{
  "spec_pics": [
    {"spec_detail_id": "1001", "url": "https://p3-.../red.png"},
    {"spec_detail_id": "1002", "url": "https://p3-.../blue.png"}
  ]
}

正确关联：

sku_spec_detail_id = sku["spec_detail_id1"]
sku_img = next(p["url"] for p in spec_pics if p["spec_detail_id"] == sku_spec_detail_id)

错误关联：

sku_img = spec_pics[sku["spec_id"]]   # 错！spec_id ≠ spec_detail_id

四、图片合规：抖店不是“能传上去就合规”

1. 主图规则（官方口径）

第一张主图：商品主体正面实物图
所有主图：除品牌 logo 外，不得有文字/水印
辅图：侧面/背面/平铺/细节，不能全一个角度
主图商品数量/颜色/规格要和 SKU 一致
不得有二维码、微信号、第三方平台水印、非平台链接

2. 比例与尺寸

1:1：800×800，建议 1200×1200
3:4：750×1000，信息流更吃这个
白底图：纯白、无阴影、无 logo、主体占 70%+
详情页：宽 750，总高别超 10000
主图视频：9:16，1080×1920，MP4

3. 白底图误区

白底图不是“首图必须白底”，而是“商城流量/活动报名的加分素材”。

服装鞋包类：首图用实拍/纯色底图，白底图单独传；

全 AI 生成主体、抠图假白底、牛皮癣大字：会被打压/下架/降权。

五、类目与必填属性：发品被打回的重灾区

/product/detail 里你能看到类目和属性，但发品/编辑时：

cid 选错 → 类目错放（严重违规，下架+扣分）
product_format_new 必填属性没填 → 审核不通过
必填属性要通过 /product/getCatePropertyV2 拿，不是拍脑袋
编辑商品时：spec_prices 是大 JSON，不传的字段不更新，传就要传完整结构

类目属性归一：

pdd_cat_path = "男装 > T恤 > 短袖T恤"
tb_cid3     = 50012029
jd_cid3     = 998
douyin_cid  = 12345

# 跨平台别用 cat_id 直连，用“类目路径 + 属性词”做映射

六、Python：detail / list 拉取 + 字段归一

# douyin/product_client.py
import time
import hashlib
import requests
import hmac


class DouyinProductClient:
    GATEWAY = "https://openapi-fxg.jinritemai.com/"

    def __init__(self, app_key, app_secret, access_token, sign_method="hmac-sha256"):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        self.sign_method = sign_method

    def _sign(self, params: dict) -> str:
        qs = "&".join(f"{k}={params[k]}" for k in sorted(params))
        if self.sign_method == "hmac-sha256":
            return hmac.new(
                self.app_secret.encode(), qs.encode(), hashlib.sha256
            ).hexdigest()
        return hashlib.md5(
            (self.app_secret + qs + self.app_secret).encode()
        ).hexdigest().upper()

    def _common(self, method: str):
        return {
            "method": method,
            "app_key": self.app_key,
            "access_token": self.access_token,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
            "v": "2",
            "sign_method": self.sign_method,
            "format": "json",
        }

    def list_products(self, status=None, page=1, page_size=100, start=None, end=None):
        biz = {"page": page, "page_size": page_size, "order_by": 1}
        if status is not None:
            biz["status"] = status
        if start:
            biz["start_time"] = start
        if end:
            biz["end_time"] = end
        params = self._common("product.list")
        params["param_json"] = str(biz).replace("'", '"')
        params["sign"] = self._sign(params)
        return requests.post(self.GATEWAY, data=params, timeout=5).json()

    def get_detail(self, product_id=None, out_product_id=None, show_draft=False):
        biz = {"show_draft": show_draft}
        if product_id:
            biz["product_id"] = product_id
        elif out_product_id:
            biz["out_product_id"] = out_product_id
        else:
            raise ValueError("product_id/out_product_id 至少传一个")
        params = self._common("product.detail")
        params["param_json"] = str(biz).replace("'", '"')
        params["sign"] = self._sign(params)
        return requests.post(self.GATEWAY, data=params, timeout=5).json()

抖店 param_json 是字符串化 JSON，不是散参数；timestamp 是 yyyy-MM-dd HH:mm:ss（GMT+8）。

七、Python：SKU 图按 spec_detail_id 关联

# douyin/spec_pic_resolver.py
from dataclasses import dataclass
from typing import Dict, List, Optional


@dataclass
class ResolvedSku:
    sku_id: str
    outer_sku_id: Optional[str]
    attrs: Dict[str, str]
    price_yuan: float
    stock: int
    sku_img: Optional[str]

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def resolve_skus(detail: dict) -> List[ResolvedSku]:
    d = (detail.get("data") or {})
    spec_pics = {p["spec_detail_id"]: p["url"] for p in d.get("spec_pics", [])}

    out = []
    for sku in d.get("skus", d.get("spec_prices", [])):
        attrs = {}
        for i in (1, 2, 3):
            name = sku.get(f"spec_detail_name{i}")
            val = sku.get(f"spec_detail_value{i}") or sku.get(f"spec{i}_name")
            if name:
                attrs[name] = val

        # 一级规格图最常见；多级可拼 [sd1, sd2]
        sd = sku.get("spec_detail_id1")
        sku_img = spec_pics.get(sd) if sd else None

        price_fen = sku.get("price") or sku.get("sku_price") or 0
        stock = sku.get("stock_num") or sku.get("available_stock") or 0

        out.append(ResolvedSku(
            sku_id=str(sku["sku_id"]),
            outer_sku_id=sku.get("outer_sku_id") or str(sku.get("out_sku_id", "")),
            attrs=attrs,
            price_yuan=price_fen / 100.0,
            stock=int(stock),
            sku_img=sku_img,
        ))
    return out

八、Python：图片合规预检（上架前拦废图）

# douyin/image_compliance.py
import re
from urllib.parse import urlparse


FORBIDDEN_TEXT_HINTS = ["微信", "加我", "扫码", "薇信", "私聊", "淘宝", "京东", "拼多多"]
EXT_OK = {".jpg", ".jpeg", ".png"}
LIMIT_MB = 3


class ImageComplianceError(Exception):
    pass


def check_main_images(pics: list, first_must_be_real=True):
    if not pics or len(pics) < 3:
        raise ImageComplianceError("主图至少 3 张")
    if len(pics) > 9:
        raise ImageComplianceError("主图最多 9 张")

    for url in pics:
        p = urlparse(url)
        ext = "." + p.path.rsplit(".", 1)[-1].lower()
        if ext not in EXT_OK:
            raise ImageComplianceError(f"格式不支持: {url}")
        if any(h in url.lower() for h in ("qrcode", "wechat", "vip")):
            raise ImageComplianceError(f"疑似二维码/站外水印: {url}")

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def check_title(title: str):
    if not title:
        raise ImageComplianceError("标题为空")
    if len(title) > 30:
        raise ImageComplianceError("标题超过 30 个汉字")
    bad = ["第一", "最高级", "全网最低", "百分百", "根治", "包治", "最便宜"]
    hit = [w for w in bad if w in title]
    if hit:
        raise ImageComplianceError(f"标题含极限/违规词: {hit}")


def check_attr_completeness(product_format_new: dict, required_keys: list):
    missing = [k for k in required_keys if not product_format_new.get(k)]
    if missing:
        raise ImageComplianceError(f"类目必填属性缺失: {missing}")

调用顺序：

check_title(name)
check_main_images(pic)
check_attr_completeness(product_format_new, required_keys)
# 过了再进 addV2 / editV2

九、list → detail → sku 的生产流水线

/product/listV2
   └─ 拿到 product_id + status + update_time
        └─ 状态/时间变化 → 进待同步队列

Worker:
   /product/detail(product_id)
      ├─ 归一商品主数据（标题/类目/主图/详情）
      ├─ resolve_skus()  → SKU 表
      ├─ spec_pics 按 spec_detail_id 关联
      ├─ check_title / check_main_images / check_attr_completeness
      └─ 写 erp_spu / erp_sku / platform_product_snapshot

编辑回写：
   /product/editV2
      - spec_prices 要传完整结构
      - 不传的字段不会被更新

十、编辑商品的两个暗雷

1. `spec_prices` 是整体结构

编辑时你只改价格：

{"spec_prices": [{"sku_id": 1, "price": 3990}]}

❌ 会丢库存/规格名/编码。

✅ 必须从 detail 里读完整 spec_prices，改其中字段再回写。

2. 审核中和在售商品

编辑在售商品：审核通过后才会覆盖线上数据
审核中商品：别并发编辑，别用 list 里的缩略数据回写 detail 级字段

十一、跨平台 SKU 归一（接前几篇）

{
  "platform": "douyin",
  "spu_id": product_id,
  "sku_id": sku_id,
  "outer_sku_id": outer_sku_id,
  "attrs": {"颜色": "红", "尺码": "M"},
  "sku_fingerprint": sha1(json.dumps(sorted(attrs))),
  "price": 39.9,
  "stock": 10,
  "images": {
     "product_pics": [...],     # 商品主图
     "sku_spec_pic": "https://..."  # 规格图，按 spec_detail_id 关联
  }
}

规则：

pic ≠ spec_pics
spec_id ≠ spec_detail_id
out_sku_id(int) ≠ outer_sku_id(string)
跨平台别用平台 cat_id 直连，用类目路径+属性词

十二、落地检查表

[ ] list 只做“哪些商品变了”，不写主数据
[ ] detail 才写 SPU/SKU/图片/属性
[ ] SKU 图用 spec_detail_id 关联 spec_pics
[ ] 主图 ≥3 张、≤9 张，首图实物无字
[ ] 白底图单独传，不强制当首图
[ ] 3:4 / 1:1 都支持，服装鞋包优先 3:4
[ ] 标题 ≤30 汉字，无极限词/站外词
[ ] 详情页无二维码/微信号/第三方水印
[ ] product_format_new 必填属性用 getCatePropertyV2 拿
[ ] editV2 的 spec_prices 传完整结构
[ ] 图片 URL 拼域名后再存，别直接信相对路径
[ ] 列表调用和详情调用分桶计费（2026.7.1 起都值钱）

十三、一句话收口

淘宝商品难在 properties_name，京东难在 skuList 嵌套，拼多多难在 specs + sku_list.specs 自己笛卡尔，
抖店商品难在：list 和 detail 的字段边界、spec_id 和 spec_detail_id 的图关联、首图和白底图的合规语义、类目必填属性的审核拦截。
谁把 /product/list 的 pic 当成 SKU 图、把白底图塞成首图、把 spec_id 拿去查 spec_pics、把必填属性当可选填——
谁就会看到：商品能发上去，但流量起不来、审核被打回、SKU 图串色、跨平台主数据全歪。

《抖店开放平台API：落地方案、接口边界与业务踩坑 —— 2026.7.1 商品发布接口收费》（附 Python 源码）

Fri, 16 Oct 2026 13:47:43 +0800

《抖店开放平台API：落地方案、接口边界与业务踩坑 —— 2026.7.1 商品发布接口收费》（附 Python 源码）

一、2026.7.1 新规先拍死（官方口径）

抖店开放平台 2026-06-16 发《收费范围调整公告》+《技术服务费规则 V1.3》，2026-07-01 起生效：

类型	接口	抖店云内	抖店云外（公网/ECS/本地）	计费
基础 API	`/product/addV2`、`/product/addSchema` 等	¥0.018 / 百次 / AppKey	¥0.18 / 百次 / AppKey（×10）	按实际调用量
基础 API	订单/售后/商品详情等	¥0.018 / 百次	¥0.18 / 百次	按实际调用量
增值 API	批量解密等	¥0.05 / 百次	禁止云外调用	按实际调用量

新增收费类目：

电商软件服务商 → 商品供应链
电商软件服务商 → 选品铺货-微应用

结算模型：

预充值。每月 5 号扣上月费用；余额不足 = 硬断调用（不是 429 限流）；已充金额不退、不提现、不转账。

这点和拼多多一样毒：

淘宝/京东是“用后结”，抖店/拼多多是“先充钱，欠费断气”。

二、抖店 API 的“视角”和别家不同

平台	商品发布	订单真相	推送
淘宝	`item.add` 聚石塔内	`trade.fullinfo.get`	DSS 推送
京东	`jd.goods.create` / read 模型	`jd.order.detail.get`	JMQ
拼多多	`pdd.goods.add`	`pdd.order.information.get`	增量轮询为主
抖店	`/product/addV2` / `/product/addSchema`	`/order/orderDetail`	数据推送（订单/售后→RDS）免费

抖店最舒服的一点：

订单/售后数据推送是免费产品，官方明着建议你“云内调用 + 数据推送”代替轮询。

抖店最坑的一点：

商品发布以前是铺货 ERP 的“免费洼地”，2026.7.1 起填平了。
店群/选品铺货/微应用 ISV 的“日发几万 SKU”模型直接变成本中心。

三、商品发布接口边界（别把 addV2 当 detail 用）

接口	作用	收费？
`/product/publishPreCheck`	发品前置校验（类目/资质/门槛）	官方定位为前置校验，别当 addV2 发
`/product/addSchema`	Schema 草稿/结构化发品	收费（基础 API）
`/product/addV2`	直接发布商品	收费（基础 API）
`/product/editV2`	编辑商品	是否收费看当期收费列表
`/product/listQuery`	商品列表	看收费列表
`/product/detail`	商品详情	看收费列表
`/spu/query`	SPU 是否存在	批量化摊薄
数据推送（订单/售后）	变更写 RDS	免费

生产铁律：

免费前置校验 → 拦掉必败请求
批量查 SPU / 类目 / 属性 → 摊薄
addSchema 建草稿 → 同源商品复用
addV2 正式发 → 只发一次
listQuery 兜底 → 别逐单 detail

四、成本测算（店群一眼看懂）

# douyin/cost_calc.py
class DouyinErpCost:
    RATES = {
        "cloud_inner": 0.018 / 100,   # 元/次
        "cloud_outer": 0.18 / 100,    # 元/次
    }

    def monthly(self, calls_per_day: int, shops: int, env: str, days=30):
        if env not in self.RATES:
            raise ValueError(env)
        unit = self.RATES[env]
        total_calls = calls_per_day * shops * days
        return total_calls * unit


if __name__ == "__main__":
    c = DouyinErpCost()
    # 单店日 1 万次发布类调用
    print("云内 1店:", c.monthly(10_000, 1, "cloud_inner"))   # 54.0
    print("云外 1店:", c.monthly(10_000, 1, "cloud_outer"))   # 540.0
    print("云内 20店:", c.monthly(10_000, 20, "cloud_inner")) # 1080.0
    print("云外 20店:", c.monthly(10_000, 20, "cloud_outer")) # 10800.0

店群场景：

100 店 × 日 3000 发品调用
云内：≈ ¥1620 / 月
云外：≈ ¥16200 / 月

云外不是“贵一点”，是把商品发布 ERP 的利润模型打穿。

五、抖店签名 Client（method + timestamp 秒级）

# douyin/client.py
import hashlib
import time
import requests


class DouyinOpenClient:
    GATEWAY = "https://openapi-fxg.jinritemai.com/"

    def __init__(self, app_key, app_secret, access_token=None, timeout=5):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        self.timeout = timeout
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    def _sign(self, params: dict) -> str:
        flat = {k: v for k, v in params.items() if v is not None and k != "sign"}
        qs = "".join(f"{k}{v}" for k, v in sorted(flat.items()))
        raw = self.app_secret + qs + self.app_secret
        return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()

    def execute(self, method: str, biz: dict):
        params = {
            "app_key": self.app_key,
            "method": method,
            "timestamp": str(int(time.time())),
            "v": "2.0",
            "sign_method": "md5",
            "format": "json",
        }
        if self.access_token:
            params["access_token"] = self.access_token
        params.update(biz)
        params["sign"] = self._sign(params)

        resp = requests.post(self.GATEWAY, data=params, timeout=self.timeout)
        return resp.json()

⚠️ 抖店：

时间戳秒级
app_secret 头尾包
公共参数 + 业务参数一起排
和拼多多很像，但 gateway / method 命名体系不同

六、预充值余额守卫（和拼多多同思路，但按月扣）

# douyin/balance_guard.py
class DouyinBalanceGuard:
    def __init__(self, redis, client,
                 warn_yuan=200.0, stop_yuan=50.0, alarm=None):
        self.r = redis
        self.client = client
        self.warn = warn_yuan
        self.stop = stop_yuan
        self.alarm = alarm
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    def cached(self):
        v = self.r.get("douyin:balance")
        return float(v) if v else 999999.0

    def refresh(self):
        # 调开放平台账户/费用查询接口（伪代码）
        resp = self.client.execute("open.account.balance.query", {})
        bal = float(resp["data"]["balance"])
        self.r.set("douyin:balance", bal, ex=120)
        return bal

    def pre_check(self, is_paid_api=True):
        if not is_paid_api:
            return
        bal = self.cached()
        if bal <= self.stop:
            self.alarm and self.alarm(f"抖店余额熔断: {bal}")
            raise CircuitOpen("douyin balance too low")
        if bal <= self.warn:
            self.alarm and self.alarm(f"抖店余额预警: {bal}")

规则：

订单推送（免费）：照常消费
addV2 / addSchema / 收费详情：先过 guard
欠费不是限流，是发品链路整段熄火

七、发品降本：把“N 次散弹”合并成“1 次收费”

# douyin/product_publisher.py
class ProductPublisher:
    def __init__(self, client, guard, bucket):
        self.client = client
        self.guard = guard
        self.bucket = bucket
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    def publish_batch(self, products: list):
        # 1) 免费前置校验
        pre = self.client.execute("/product/publishPreCheck", {
            "products": products
        })
        valid = [p for p in products if pre_ok(pre, p)]
        if not valid:
            return []

        # 2) 批量查 SPU 是否存在（摊薄）
        existing = self.batch_spu_query([p["outer_spu_id"] for p in valid])

        # 3) 同源商品先建 schema 草稿
        drafts = self.build_schema_drafts(valid)

        # 4) 只对新 SPU 调 addV2
        for p in drafts:
            if p["spu_id"] in existing:
                continue
            self.guard.pre_check(is_paid_api=True)
            self.bucket.acquire()
            self.client.execute("/product/addV2", p)

    def batch_spu_query(self, outer_ids: list):
        out = {}
        for i in range(0, len(outer_ids), 50):
            chunk = outer_ids[i:i+50]
            resp = self.client.execute("/spu/query", {"outer_spu_ids": chunk})
            out.update(parse_spu(resp))
        return out

    def build_schema_drafts(self, products):
        # 同源（同图/同详情/不同规格）合成一个 schema
        groups = group_by_template(products)
        drafts = []
        for g in groups:
            self.guard.pre_check(is_paid_api=True)
            self.bucket.acquire()
            d = self.client.execute("/product/addSchema", g)
            drafts.append(d)
        return drafts

降本公式（经验）：

散弹式：preCheck(0?) + spu查×N + addV2×N + detail×N
合并式：preCheck(1) + spu查(N/50) + addSchema(组数) + addV2(新SPU数) + listQuery(1/20)

铺货型 ERP 通常能砍掉 60%~75% 收费调用。

八、订单侧：推送为主，轮询兜底

抖店数据推送（免费）
  └─ 订单/售后变更 → RDS
        └─ ERP 读库 + 状态机

每 5~10 min：
  /order/searchList（收费基础API）
     └─ 补漏单/晚到事件

详情：
  /order/orderDetail
     └─ 真相源，别拿列表写业务

和淘宝 DSS、京东 JMQ 同构：

推送/消息 = 触发器，列表 = 兜底，详情 = 法官。

九、云内部署决策（抖店比拼多多更硬）

场景	建议
自研单店 ERP	云外能跑，但发品/订单量大就亏
铺货/店群/选品微应用	必须抖店云内
订单地址/手机解密	云外直接无权限（和淘宝增值/拼多多增值同逻辑）
大促	云内 + 推送 + 限流桶 + 余额守卫

架构：

抖店云内 ECS
 ├─ DyBaseClient   基础API（订单/商品/库存）  令牌桶
 ├─ DyAdvClient    增值API（解密）            单独AppKey + 强制云内
 ├─ PushConsumer   数据推送 → RDS             免费
 └─ Redis          幂等键 / 日调用计数 / 余额缓存

十、错误分类（抖店特有）

现象	处理
`sign invalid`	secret/排序/时间戳
`access_token expired`	刷新 1 次
`no permission / 非云内调用`	增值 API 别云外调
系统错误/余额类	熔断 + 充值，不重试业务
限流	退避
网络超时	退避，但别当“发布失败”直接重发（会双发）
发品审核挂起	不是接口错，进人工/异步审核流

发品最阴的 bug：

超时 → 你重试 addV2 → 同商品发两次 → 重复 SPU / 类目违规 / 被平台判铺货异常。

解决：

idempotency_key = sha1(shop_id + outer_spu_id + spec_fingerprint)
if dedup.exists(idempotency_key):
    return existing_product

十一、和前几篇收口

1688：基础免费，QPS 卡脖，签名 secret 头尾
淘宝 TOP：聚石塔内外 10 倍，DSS 推送免费
京东 JOS：云鼎内稳，POP 时间窗滚动
拼多多：预充值，云外 ×10，增值禁外
抖店：预充值 + 云外 ×10 + 商品发布 2026.7.1 起收费 + 数据推送免费

一句话：

抖店 2026.7.1 之前，铺货 ERP 的潜规则是“商品发布不要钱，拼命发”；
2026.7.1 之后，规则变成“发一次就计费，云外发一次顶云内十次，欠费不是限流是断气”。
能活下来的抖店 ERP 不是“能发品”，而是：
前置校验拦废单、SPU 批量查、schema 并草稿、addV2 只发新、推送替轮询、余额守卫在 Client 层。

《拼多多订单API：pdd.order.list.get 落地方案 —— 退款/售后状态机的幂等设计》（附 Python 源码）

Thu, 15 Oct 2026 13:36:51 +0800

《拼多多订单API：pdd.order.list.get 落地方案 —— 退款/售后状态机的幂等设计》（附 Python 源码）

先拍结论：

pdd.order.list.get 不是实时增量主接口，它是“按成交时间捞历史/初始化”的接口（时间窗 ≤24h，历史 90 天）。
实时同步要用 pdd.order.number.list.increment.get，真相用 pdd.order.information.get，售后用 pdd.refund.list.increment.get + pdd.refund.information.get。
拼多多订单最坑的是：订单发货态（order_status）和售后态（after_sales_status / refund_status）是两条独立轴，不能只存一个“状态”字段。

正确模型：

order_status : 1待发货 / 2已发货待签收 / 3已签收
refund_status: 1无售后或关闭 / 2售后处理中 / 3退款中 / 4退款成功
after_sales_status: 0/2/3/4/5/6/7/8/9/10/11/...（细粒度售后单状态）
risk_control_status: 0正常 / 1审核中

生产里“订单状态机”必须是 (order_status, refund_status, after_sales_status, risk_control_status, version) 五维收敛，不是 status 一个字段来回覆盖。

一、pdd.order.list.get 的定位（别用错）

接口	时间维度	用途	生产角色
`pdd.order.list.get`	成交时间，窗 ≤24h，历史 90 天	历史初始化	新店接入/补数据
`pdd.order.number.list.increment.get`	更新时间，窗 ≤30min	实时增量	主轮询
`pdd.order.information.get`	order_sn	订单真相	拉子单/优惠/地址/售后态
`pdd.refund.list.increment.get`	售后更新时间，窗 ≤30min	售后增量	售后单触发器
`pdd.refund.information.get`	order_sn + after_sales_id	售后真相	退款/退货/换货处理

官方口径很明确：

新用户用 order.list.get 同步历史，实时用 order.number.list.increment.get，消息/增量只当触发器，information 才是真相。

二、拼多多订单/售后状态轴（必须分开建）

1. 订单发货轴 `order_status`

1 待发货
2 已发货待签收
3 已签收

（仅已成交订单：confirm_status=1）

2. 售后粗轴 `refund_status`（列表接口用）

1 无售后或售后关闭
2 售后处理中
3 退款中
4 退款成功
5 全部

3. 售后细轴 `after_sales_status`（详情/售后接口用）

0  无售后
2  买家申请退款，待商家处理
3  退货退款，待商家处理
4  商家同意退款，退款中
5  平台同意退款，退款中
6  驳回退款，待买家处理
7  已同意退货退款，待用户发货
8  平台处理中
9  平台拒绝退款，退款关闭
10 退款成功
11 买家撤销
12 买家逾期未处理，退款失败
13 买家逾期，超过有效期
14 换货补寄待商家处理
15 换货补寄待用户处理
16 换货补寄成功
17 换货补寄失败
18 换货补寄待用户确认完成
21 待商家同意维修
22 待用户确认发货
24 维修关闭
25 维修成功
27 待用户确认收货
31 已同意拒收退款，待用户拒收
32 补寄待商家发货
33 同意召回后退款，待商家召回

4. 风控轴

risk_control_status = 1  → 审核订单，不能发货

发货铁律（官方）：
order_status=1 AND refund_status=1 AND risk_control_status=0 才能发。

三、幂等键设计（三层叠加）

L1 唯一键：   (shop_id, order_sn)                → 防重复插入
L2 去重表：   (shop_id, order_sn, event_key)     → 防重复消费
L3 状态机：   UPDATE ... WHERE order_sn=? AND version < new_version AND status 合法

event_key 可以是：

incr:{updated_at}（增量轮询）
msg:{msg_id}（消息推送）
refund:{after_sales_id}:{after_sales_status}:{updated_at}（售后事件）

四、统一订单快照模型

# pdd/order_model.py
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class PddOrderSnapshot:
    shop_id: str
    order_sn: str

    order_status: int = 0           # 1/2/3
    refund_status: int = 1          # 1/2/3/4
    after_sales_status: int = 0     # 细粒度
    risk_control_status: int = 0    # 0/1

    confirm_status: int = 0         # 0未成交 1已成交 2取消
    pay_amount: float = 0.0
    updated_at: int = 0             # 平台更新时间（秒）
    version: int = 0                # = updated_at，单调推进

    after_sales_id: Optional[str] = None
    refund_amount: float = 0.0
    raw: dict = field(default_factory=dict)


def snapshot_from_list(o: dict, shop_id: str) -> PddOrderSnapshot:
    updated = int(o.get("updated_at") or o.get("last_updated_time") or 0)
    return PddOrderSnapshot(
        shop_id=shop_id,
        order_sn=o["order_sn"],
        order_status=int(o.get("order_status", 0)),
        refund_status=int(o.get("refund_status", 1)),
        after_sales_status=int(o.get("after_sales_status", 0)),
        risk_control_status=int(o.get("risk_control_status", 0)),
        confirm_status=int(o.get("confirm_status", 0)),
        pay_amount=float(o.get("pay_amount", 0) or 0),
        updated_at=updated,
        version=updated,
        raw=o,
    )

五、状态机：订单轴 × 售后轴分别收敛

# pdd/order_state_machine.py
from enum import IntEnum


class OrderShipStatus(IntEnum):
    UNKNOWN = 0
    WAIT_SHIP = 1
    SHIPPED = 2
    SIGNED = 3


class AfterSalesStatus(IntEnum):
    NONE = 0
    BUYER_APPLY_REFUND = 2
    RETURN_REFUND_PENDING = 3
    SELLER_AGREED_REFUNDING = 4
    PLATFORM_AGREED_REFUNDING = 5
    REJECTED_PENDING_BUYER = 6
    AGREED_RETURN_WAIT_BUYER_SHIP = 7
    PLATFORM_PROCESSING = 8
    REFUND_CLOSED = 9
    REFUND_SUCCESS = 10
    BUYER_CANCELED = 11
    EXCHANGE_PENDING_SELLER = 14
    EXCHANGE_PENDING_BUYER = 15
    EXCHANGE_SUCCESS = 16
    REPAIR_PENDING_SELLER = 21
    REPAIR_SUCCESS = 25


# 订单发货态：只允许前进，不允许被旧事件覆盖
ORDER_ALLOWED = {
    OrderShipStatus.WAIT_SHIP: {OrderShipStatus.SHIPPED},
    OrderShipStatus.SHIPPED: {OrderShipStatus.SIGNED},
    OrderShipStatus.SIGNED: set(),
}

# 售后态：可以“升级/关闭/成功”，但不能从 退款成功 退回 待处理
AFTERSALES_BLOCKED_FROM = {
    AfterSalesStatus.REFUND_SUCCESS: {
        AfterSalesStatus.BUYER_APPLY_REFUND,
        AfterSalesStatus.RETURN_REFUND_PENDING,
        AfterSalesStatus.SELLER_AGREED_REFUNDING,
    },
    AfterSalesStatus.REFUND_CLOSED: {
        AfterSalesStatus.SELLER_AGREED_REFUNDING,
        AfterSalesStatus.PLATFORM_AGREED_REFUNDING,
    },
}


class IllegalTransition(Exception):
    pass


def check_order_transition(cur: int, new: int):
    cur_s = OrderShipStatus(cur)
    new_s = OrderShipStatus(new)
    if new_s == cur_s:
        return
    if new_s not in ORDER_ALLOWED.get(cur_s, set()):
        raise IllegalTransition(f"order {cur} -> {new} not allowed")
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex

def check_aftersales_transition(cur: int, new: int):
    new_s = AfterSalesStatus(new)
    if new_s == AfterSalesStatus(cur):
        return
    blocked = AFTERSALES_BLOCKED_FROM.get(AfterSalesStatus(cur), set())
    if new_s in blocked:
        raise IllegalTransition(f"after_sales {cur} -> {new} rollback blocked")

六、幂等落库（SQL 层兜底）

CREATE TABLE pdd_order (
    shop_id        VARCHAR(32)  NOT NULL,
    order_sn       VARCHAR(64)  NOT NULL,
    order_status   SMALLINT     NOT NULL DEFAULT 0,
    refund_status  SMALLINT     NOT NULL DEFAULT 1,
    after_sales_status SMALLINT NOT NULL DEFAULT 0,
    risk_control_status SMALLINT NOT NULL DEFAULT 0,
    confirm_status SMALLINT     NOT NULL DEFAULT 0,
    pay_amount     DECIMAL(12,2) NOT NULL DEFAULT 0,
    version        BIGINT       NOT NULL DEFAULT 0,
    updated_at     BIGINT       NOT NULL DEFAULT 0,
    PRIMARY KEY (shop_id, order_sn)
);

CREATE TABLE pdd_order_event_dedup (
    shop_id    VARCHAR(32) NOT NULL,
    order_sn   VARCHAR(64) NOT NULL,
    event_key  VARCHAR(128) NOT NULL,
    created_at BIGINT NOT NULL,
    PRIMARY KEY (shop_id, order_sn, event_key)
);

更新语句（防乱序核心）：

UPDATE pdd_order
SET
    order_status = #{new_order_status},
    refund_status = #{new_refund_status},
    after_sales_status = #{new_after_sales},
    risk_control_status = #{new_risk},
    pay_amount = #{pay_amount},
    version = #{new_version},
    updated_at = #{new_updated_at}
WHERE shop_id = #{shop_id}
  AND order_sn = #{order_sn}
  AND version < #{new_version};

version < new_version 一行搞定：
旧消息晚到 → 0 行
同版本重复 → 0 行
新状态推进 → 1 行

七、Python：增量同步 + 幂等消费

# pdd/order_sync.py
import time


class PddOrderSync:
    def __init__(self, client, repo, dedup, bucket):
        self.client = client
        self.repo = repo
        self.dedup = dedup
        self.bucket = bucket

    def poll_increment(self, start_sec: int, end_sec: int):
        page = 1
        while True:
            self.bucket.acquire()
            resp = self.client.execute("pdd.order.number.list.increment.get", {
                "start_updated_at": start_sec,
                "end_updated_at": end_sec,
                "page": page,
                "page_size": 100,
            })
            orders = self._extract(resp)
            if not orders:
                break

            for o in orders:
                self.ingest(o)

            if len(orders) < 100:
                break
            page += 1

    def ingest(self, o: dict):
        shop_id = o["shop_id"]
        order_sn = o["order_sn"]
        updated = int(o.get("updated_at") or 0)
        event_key = f"incr:{updated}"

        # L2 去重
        if not self.dedup.mark(shop_id, order_sn, event_key):
            return

        # 列表字段不全 → 拉详情
        detail = self.fetch_detail(order_sn)
        snap = snapshot_from_list(detail if detail else o, shop_id)

        self.apply_snapshot(snap)

    def fetch_detail(self, order_sn: str):
        try:
            resp = self.client.execute("pdd.order.information.get", {"order_sn": order_sn})
            return (resp.get("order_info_get_response") or {}).get("order_info")
        except RetriableError:
            return None   # 下次增量/兜底再补
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    def apply_snapshot(self, snap: PddOrderSnapshot):
        cur = self.repo.get(shap=snap)
        if cur is None:
            self.repo.insert(snap)
            self.emit_order_events(None, snap)
            return

        if snap.version < cur.version:
            return  # 旧事件
        if snap.version == cur.version and snap.order_status == cur.order_status \
           and snap.after_sales_status == cur.after_sales_status:
            return  # 完全幂等

        # 状态机校验
        try:
            check_order_transition(cur.order_status, snap.order_status)
            check_aftersales_transition(cur.after_sales_status, snap.after_sales_status)
        except IllegalTransition as e:
            self.repo.append_suspect(cur, snap, str(e))
            return

        rows = self.repo.update_if_newer(snap)
        if rows == 1:
            self.emit_order_events(cur, snap)

八、售后单独立建模（别塞进订单表覆盖）

@dataclass
class PddAfterSale:
    shop_id: str
    order_sn: str
    after_sales_id: str
    after_sales_status: int
    after_sales_type: int        # 2仅退款 3退货退款 4换货 5补寄 6维修
    refund_amount: float
    version: int
    raw: dict

售后增量：

def sync_refund(shop_id, start_sec, end_sec):
    while True:
        resp = client.execute("pdd.refund.list.increment.get", {
            "start_updated_at": start_sec,
            "end_updated_at": end_sec,
            "after_sales_status": 0,   # 0=全部
            "page": page,
            "page_size": 100,
        })
        for r in extract_refunds(resp):
            if not dedup.mark(shop_id, r["order_sn"], f"rf:{r['after_sales_id']}:{r['after_sales_status']}:{r['updated_at']}"):
                continue
            detail = client.execute("pdd.refund.information.get", {
                "order_sn": r["order_sn"],
                "after_sales_id": r["after_sales_id"],
            })
            apply_aftersale(detail)

订单上的 after_sales_status 只是“最新一条售后单的投影”，

真实仲裁看 pdd_after_sale 表。

九、业务动作守卫（最值钱的部分）

def can_ship(snap: PddOrderSnapshot) -> (bool, str):
    if snap.confirm_status != 1:
        return False, "not_confirmed"
    if snap.order_status != 1:
        return False, "not_wait_ship"
    if snap.risk_control_status == 1:
        return False, "risk_review"
    if snap.refund_status in (2, 3):
        return False, "refund_in_process"
    if snap.after_sales_status in (
        2, 3, 4, 5, 7, 8, 14, 15, 21, 31, 33
    ):
        return False, "after_sale_open"
    return True, "ok"

调用发货前：

ok, reason = can_ship(snap)
if not ok:
    freeze_and_alert(snap.order_sn, reason)
else:
    do_ship(snap)

这就是前面说的：“货发出去了，用户申请退款”的坑。
拼多多买家仅退款很早就能发起，ERP 不冻结 = 资损。

十、消息推送 + 轮询双通道（和淘宝/京东同构）

消息推送
  └─ 快返回 success（2~5s 内）
       └─ 入 MQ
            └─ pdd.order.information.get 拉真相

pdd.order.number.list.increment.get（30min 窗，每 60s）
  └─ 兜底漏推送/超时

pdd.refund.list.increment.get（30min 窗）
  └─ 售后单独立推进

三者都写：
  pdd_order(order_sn 主键)
  pdd_after_sale(after_sales_id 主键)
  pdd_order_event_dedup(event_key 主键)

十一、时间窗与限流

order.list.get：成交时间窗 ≤ 24h，历史 90 天，别做实时主源
number.list.increment.get：更新时间窗 ≤ 30min，实时主源
refund.list.increment.get：更新时间窗 ≤ 30min
列表 QPS / 详情 QPS / 售后 QPS 分桶
预充值余额守卫（上篇）必须套在这里：欠费不是 429，是订单同步直接断气

十二、落地检查表

[ ] order.list.get 只用于初始化/补历史，不用于实时
[ ] 实时用 order.number.list.increment.get
[ ] 订单轴 / 售后轴 / 风控轴三轴分开存
[ ] (shop_id, order_sn) 唯一键 + (order_sn, event_key) 去重表 + version<new_version 状态机
[ ] 售后单独立表，订单表只存“最新售后投影”
[ ] 发货前 order_status=1 & refund_status=1 & risk_control_status=0 & 无未结售后
[ ] 消息 2~5s 返回 success，业务异步跑
[ ] 旧事件晚到 → 0 行更新；同版本重发 → 0 行更新
[ ] 退款成功/售后关闭不允许回退到“待处理”
[ ] 余额守卫在 Client 层，不在业务层事后发现

十三、一句话收口

淘宝订单难在“消息+增量+详情”三角，
京东难在“时间窗+鼎内+POP/FBP 分域”，
拼多多订单难在：一个订单号下面，发货态、售后态、风控态、成交态是四根平行轴。
你用 status 一个字段存拼多多订单，迟早会把“已发货 + 退款成功 + 换货中”存成“已发货”。
正确做法：
列表做触发器，详情做真相，售后单独立建模，version 做时间仲裁，状态机做乱序防火墙。

《拼多多商品API：pdd.goods.detail.get 与 pdd.goods.cats.get 的类目与规格踩坑》（附 Python 源码）

Thu, 15 Oct 2026 13:19:13 +0800

《拼多多商品API：pdd.goods.detail.get 与 pdd.goods.cats.get 的类目与规格踩坑》（附 Python 源码）

先拍结论：

拼多多商品模型里最坑的不是“拿不到数据”，而是类目有两套、规格是组合爆炸、cat_id 不能直接当发布类目、spec 不是 SKU。
pdd.goods.cats.get：标准类目树（全平台类目字典），不是“当前店铺能发什么”
pdd.goods.authorization.cats：店铺可发布类目（受经营范围限制）
pdd.goods.detail.get：商品视图，cat_id / cat_ids 是归属类目，sku_list / goods_specs 才是可售单元
拼多多的“规格”是 维度(spec_key) × 值(spec_value) → SKU 组合，直接平铺就会下错“红色 / M”

一、两个接口的职责边界

接口	视角	主要用途	常见误用
`pdd.goods.cats.get`	平台标准类目树	类目 ID ↔ 名称、层级、父子关系	拿它当“本店可发类目”
`pdd.goods.authorization.cats`	商家可发布类目	发布商品时选类目	忘了循环下钻到叶子
`pdd.goods.detail.get`	单个商品详情	标题/价格/图/类目/规格/SKU	以为 `specs` 就是可售 SKU

官方商品发布说明里写得很清楚：

标准类目用 pdd.goods.cats.get
商家可发类目用 pdd.goods.authorization.cats
要发商品必须循环下钻到最子集（叶子）类目，拼多多商品必须有三级或四级分类

二、类目踩坑 1：`cat_id` ≠ 可发布类目

pdd.goods.detail.get 返回：

{
  "cat_id": 12345,
  "cat_ids": [100, 200, 12345]
}

含义：

cat_id：商品当前归属类目（可能是三级/四级）
cat_ids：一级 → 二级 → 三级 → 四级路径

但：

店铺能不能发这个类目，要看 pdd.goods.authorization.cats。
不在店铺经营范围 → 用 cat_id 调 pdd.goods.add 直接报：
“当前商品分类不在店铺经营范围内”

✅ 正确动作：

authorization.cats(parent_cat_id=0) 拿一级可发类目
逐级下钻到叶子
再用叶子 cat_id 发商品
goods.cats.get 只做“类目 ID → 名称 / 路径展示 / 跨平台映射”

三、类目踩坑 2：类目会到四级，不是淘宝三级就完事

pdd.goods.cats.get 返回结构：

{
  "goods_cats_get_response": {
    "goods_cats_list": [
      {
        "cat_id": 239,
        "cat_name": "男装",
        "level": 1,
        "parent_cat_id": 0
      },
      {
        "cat_id": 242,
        "cat_name": "T恤",
        "level": 3,
        "parent_cat_id": 241
      }
    ]
  }
}

规则：

level=1 一级，level=2 二级，level=3 三级，level=4 四级
有些类目三级是叶子，有些要四级
发布必须用叶子；用父级 cat_id 发商品 → 审核挂起/类目校验失败

四、规格踩坑 3：specs 不是 SKU，是“维度值清单”

拼多多常见返回（简化）：

{
  "goods_specs": [
    {"spec_key": "颜色", "spec_value": "红色", "spec_id": 1},
    {"spec_key": "颜色", "spec_value": "蓝色", "spec_id": 2},
    {"spec_key": "尺码", "spec_value": "M", "spec_id": 3},
    {"spec_key": "尺码", "spec_value": "L", "spec_id": 4}
  ],
  "sku_list": [
    {"sku_id": "111", "specs": [1, 3], "price": 3900, "quantity": 10},
    {"sku_id": "112", "specs": [1, 4], "specs_info": [{"key":"颜色","value":"红色"},{"key":"尺码","value":"L"}], "price": 4100, "quantity": 8},
    {"sku_id": "113", "specs": [2, 3], "price": 3900, "quantity": 5}
  ]
}

❌ 错误模型：

options = ["红色", "蓝色", "M", "L"]   # 用户选规格时下成 红色 + M 两个标签

✅ 正确模型：

goods_specs：规格维度值池
sku_list[].specs：该 SKU 引用的 spec_id 组合
真正下单/库存/价格单位是 sku_id

五、Python：类目树加载器

# pdd/category_tree.py
from typing import Optional


class PddCategoryTree:
    def __init__(self, client):
        self.client = client
        self.cache = {}   # cat_id -> node
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    def load_children(self, parent_cat_id: int = 0):
        resp = self.client.execute("pdd.goods.cats.get", {
            "parent_cat_id": parent_cat_id,
        })
        nodes = (
            resp.get("goods_cats_get_response", {})
                .get("goods_cats_list", [])
        )
        for n in nodes:
            self.cache[n["cat_id"]] = n
        return nodes

    def load_full_path(self, cat_ids: list[int]):
        """cat_ids=[100,200,12345] -> 名称路径"""
        return [self.cat_name(cid) for cid in cat_ids]

    def cat_name(self, cat_id: int) -> Optional[str]:
        if cat_id in self.cache:
            return self.cache[cat_id]["cat_name"]
        # 没缓存就查一次
        children = self.load_children(0)
        return self.cache.get(cat_id, {}).get("cat_name")

    def is_leaf(self, cat_id: int) -> bool:
        children = self.load_children(cat_id)
        return len(children) == 0

店铺可发类目（发布用）：

def shop_publishable_leaf_cats(client):
    leaves = []

    def dfs(parent):
        nodes = client.execute("pdd.goods.authorization.cats", {
            "parent_cat_id": parent
        }).get("goods_auth_cats_get_response", {}).get("goods_cats_list", [])
        if not nodes:
            leaves.append(parent)
            return
        for n in nodes:
            dfs(n["cat_id"])

    dfs(0)
    return leaves

六、Python：规格 → SKU 组合解析

# pdd/spec_resolver.py
from dataclasses import dataclass
from typing import List, Dict


@dataclass
class ResolvedSku:
    sku_id: str
    spec_combo: Dict[str, str]   # {"颜色": "红色", "尺码": "M"}
    price_yuan: float
    stock: int


def build_spec_index(goods_specs: list) -> Dict[int, dict]:
    """
    spec_id -> {key, value}
    """
    idx = {}
    for s in goods_specs or []:
        idx[s["spec_id"]] = {
            "key": s.get("spec_key"),
            "value": s.get("spec_value"),
        }
    return idx
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex

def resolve_skus(goods_detail: dict) -> List[ResolvedSku]:
    specs = goods_detail.get("goods_specs", [])
    spec_idx = build_spec_index(specs)

    out = []
    for sku in goods_detail.get("sku_list", []):
        combo = {}
        for sid in sku.get("specs", []):
            meta = spec_idx.get(sid)
            if meta:
                combo[meta["key"]] = meta["value"]

        # 拼多多价格单位：分
        price_fen = sku.get("price") or sku.get("sku_price") or 0
        stock = sku.get("quantity") or sku.get("sku_quantity") or 0

        out.append(ResolvedSku(
            sku_id=str(sku["sku_id"]),
            spec_combo=combo,
            price_yuan=price_fen / 100.0,
            stock=int(stock),
        ))
    return out

消费侧：

skus = resolve_skus(detail)
for s in skus:
    print(s.sku_id, s.spec_combo, s.price_yuan, s.stock)
# 111 {'颜色': '红色', '尺码': 'M'} 39.0 10

⚠️ 有的接口返回 sku_list[].specs_info=[{"key":"颜色","value":"红色"}]，

有的只给 specs=[spec_id,...]。

生产里两个都兼容，别只信一种。

七、跨平台规格归一（接前几篇）

平台	规格模型	可售主键
1688	`skuInfo` / `skuArray`，属性对	`offerId + skuId`
淘宝	`skus.sku.properties_name` = `pid:vid:名:值`	`num_iid + sku_id`
京东	`skuList[].specText`	`skuId + skuList.skuId`
拼多多	`goods_specs` 维度值 + `sku_list.specs` 引用	`goods_id + sku_id`

统一模型：

{
  "platform": "pdd",
  "spu_id": goods_id,
  "sku_id": sku_id,
  "attrs": {"颜色": "红色", "尺码": "M"},
  "price": 39.00,
  "currency": "CNY",
  "stock": 10,
  "stock_trusted": True,
}

规则：

跨平台别用“颜色/尺码”硬字段
用 attrs: dict + sku_fingerprint = sha1(json(sorted(attrs)))

八、类目映射踩坑：用 cat_id 跨平台直连会翻车

淘宝 cid3=50012029 ≠ 京东 cid3=123 ≠ 拼多多 cat_id=12345

三者没有全局主键。

正确做法：

各平台拉类目树
建 platform_cat_id ↔ 标准类目词（女装/T恤/3C）
用“类目路径文本 + 属性词”做向量/规则匹配
低置信度 → 人工审

def pdd_cat_path(client, cat_ids):
    tree = PddCategoryTree(client)
    return " > ".join(tree.cat_name(c) for c in cat_ids)
# "男装 > T恤 > 短袖T恤"

九、发布商品时的类目/规格顺序（很重要）

1. authorization.cats 下钻到叶子 cat_id
2. goods.cat.rule.get 拿类目必填属性
3. goods.spec.get / goods.spec.id.get 建规格维度
4. 组装 sku_list：spec 组合 → 价格 → 库存
5. goods.img.upload 传图
6. goods.add / goods.edit.goods.commit
7. goods.commit.status.get 查审核

最常见失败：

用标准类目 cats.get 的 id 直接发 → 不在经营范围
用父级类目发 → 类目校验失败
只传 specs 不传 sku_list → 没可售单元
把“颜色:红”和“尺码:M”当两个 SKU → 下单串规格

十、字段缺失清单（拼多多特有）

cat_ids 可能空 → 用 cat_id 反查，但别信它完整
goods_desc 可能带 HTML / 被审核截断
min_group_price / min_normal_price 单位是分
sku_list 可能空（仅 SPU 无 SKU 的商品）
多多进宝接口 pdd.ddk.goods.detail 返回结构和商家接口不同，别混 parser
goods_sign 是加密商品标识，goods_id 在某些场景会被引导用 goods_sign 替代

十一、和前几篇收口

1688：offerId+skuId，批发视角，库存快照不可信
淘宝：item.get 公开 / item.seller.get 私有，SKU 是 properties_name
京东：jingdong.item.read.get 是 SPU+skuList+priceInfo+stockInfo
拼多多：goods_specs 是维度值池，sku_list.specs 才是组合，类目有“标准类目 / 可发类目”两套

一句话：

淘宝把 SKU 藏在 properties_name 里，京东把 SKU 放在 skuList.specText 里，
拼多多把 SKU 拆成 spec 维度再让你自己笛卡尔——
谁直接把 颜色:红, 尺码:M 当两个下拉项，谁就会把“红色M”卖成“红色 + M”。

下一篇可以写：

《拼多多 × 淘宝 × 京东 × 1688 SKU 指纹统一：spec_dict → sha1 → erp_sku_id 的跨平台商品主数据模型》

要我直接写吗？

《拼多多开放平台API：落地方案、接口边界与业务踩坑 —— 预充值模式的成本陷阱》（附 Python 源码）

Wed, 14 Oct 2026 10:27:08 +0800

《拼多多开放平台API：落地方案、接口边界与业务踩坑 —— 预充值模式的成本陷阱》（附 Python 源码）

一、先拍结论

拼多多开放平台最坑的不是“接口难调”，而是计费模型：
预充值 + 云外 ×10 + 欠费硬切断 + 控制台余额 T-1 滞后。
单价 0.01 元/百次/client_id 看起来像白送，
但生产里真正烧钱/漏单的三件事是：
公网 ECS（阿里云/腾讯云/本地机房）调 = 云外 ×10
地址/手机号解密等增值能力 = 云外直接禁调，不是加钱
余额 ≤ 0 = 不是 429 限流，是系统级错误把订单同步打断

官方收费规则写得很直白：基础 API 云内 0.01、云外 0.10；增值 API 云内 0.03、云外 0.30，且增值 API 未经允许禁止云外调用；按日统计并从预充值余额扣除。

二、拼多多 API 的“视角”和别家不同

平台	主视角	订单真相接口
1688	买家采购	`alibaba.trade.*`
淘宝	卖家经营	`taobao.trade.fullinfo.get`
京东	POP 商家	`jd.order.detail.get`
拼多多	商家/服务商	`pdd.order.information.get` / `pdd.order.list.get` / `pdd.order.increment.get`

拼多多核心订单族：

pdd.order.list.get：按状态/时间拉列表
pdd.order.increment.get：增量变更
pdd.order.information.get：订单详情（子单/优惠/收货/售后）
pdd.logistics.tracking.get：物流轨迹
pdd.erp.* / pdd.goods.* / pdd.refund.*：商品、售后、ERP 类扩展

规则：
list/increment 是“捕捞网”，information 是“法官”，tracking 是“物流视角”。
拼多多没有“消息推送替代轮询”的淘宝 DSS 级体验，小 ISV 经常是“增量 + 定时兜底”。

三、预充值模式的四个隐形陷阱

1️⃣ 先充钱，不是月底结账

淘宝/京东：超免额后月底结。

拼多多：调用前账户必须有余额，今天跑的调用明天才扣，但“现在能不能调”看“现在余额”。

2️⃣ 云外 ×10 倍

云内（拼多多云）：基础 0.01 / 百次 / client_id
云外（阿里云/腾讯云/本地）：基础 0.10 / 百次 / client_id
增值：云内 0.03，云外 0.30 且禁止

单店日 1 万次：

云内：30 天 × 100 百次 × 0.01 = ¥30/月
云外：= ¥300/月
10 店 ISV 云外：¥3000/月，一年 3.6 万纯部署费

3️⃣ 欠费是“断气”，不是限流

余额 0 / 负数：

不返回 429
返回系统级错误（社区常见如 50001 类费用/权限特征，具体以当期文档为准）
订单同步直接停，不会优雅降级

4️⃣ 控制台余额是 T-1

按日结算，界面看到的是“昨天扣完后的残值”。

你以为还有 8 毛，其实这一小时已经欠了 3 块。

四、签名：拼多多是 MD5(client_secret + kv + client_secret)

官方规则：

所有参数（公共 + 业务，不含 sign）
key 按 ASCII 升序
拼成 keyvaluekeyvalue
头尾拼 client_secret
MD5 → 大写

# pdd/client.py
import hashlib
import time
import requests


class PddClient:
    GATEWAY = "https://gw-api.pinduoduo.com/api/router"
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    def __init__(self, client_id, client_secret, access_token=None, timeout=5):
        self.client_id = client_id
        self.client_secret = client_secret
        self.access_token = access_token
        self.timeout = timeout

    def _sign(self, params: dict) -> str:
        flat = {k: v for k, v in params.items() if v is not None and k != "sign"}
        qs = "".join(f"{k}{v}" for k, v in sorted(flat.items()))
        raw = f"{self.client_secret}{qs}{self.client_secret}"
        return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()

    def execute(self, type_: str, biz: dict):
        params = {
            "type": type_,
            "client_id": self.client_id,
            "timestamp": str(int(time.time())),   # 秒级，误差 <10min
            "data_type": "JSON",
            "sign_method": "md5",
        }
        if self.access_token:
            params["access_token"] = self.access_token
        params.update(biz)
        params["sign"] = self._sign(params)

        resp = requests.post(self.GATEWAY, data=params, timeout=self.timeout)
        return resp.json()

⚠️ 网上有文章写“HMAC-SHA256 / key=value&...&client_secret=xxx”，那是错配/第三方网关写法。

开放平台官方是 MD5 + 无分隔拼接 + secret 头尾。

五、余额守卫：把“欠费断气”变成“熔断+告警”

# pdd/balance_guard.py
import time

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class PddBalanceGuard:
    def __init__(self, redis, client,
                 warn_yuan=50.0, stop_yuan=10.0,
                 alarm=None):
        self.r = redis
        self.client = client
        self.warn = warn_yuan
        self.stop = stop_yuan
        self.alarm = alarm
        self.key = "pdd:balance:cached"

    def cached_balance(self) -> float:
        v = self.r.get(self.key)
        return float(v) if v else 999999.0

    def refresh_from_platform(self):
        # 实际调用拼多多费用/账户查询接口；这里用伪接口
        resp = self.client.execute("pdd.finance.account.balance.get", {})
        bal = float(resp["xxx"]["balance"])
        self.r.set(self.key, bal, ex=60)
        return bal

    def pre_check(self):
        bal = self.cached_balance()
        if bal <= self.stop:
            self.alarm and self.alarm(f"PDD 余额熔断: {bal} 元，停止非核心调用")
            raise CircuitOpen("pdd balance too low")
        if bal <= self.warn:
            self.alarm and self.alarm(f"PDD 余额预警: {bal} 元，准备充值")

    def on_api_billing_error(self):
        # 命中费用类系统错误：立刻刷新余额并降级
        try:
            self.refresh_from_platform()
        except Exception:
            self.r.set(self.key, "0", ex=30)
        raise CircuitOpen("pdd billing/permission error")

调用侧：

guard.pre_check()
try:
    resp = client.execute("pdd.order.increment.get", {...})
except BillingError:
    guard.on_api_billing_error()

六、订单同步：增量 + 兜底，不深翻页

# pdd/order_scanner.py
from datetime import datetime, timedelta
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex

class PddOrderScanner:
    def __init__(self, client, guard, bucket, slide_back_sec=60):
        self.client = client
        self.guard = guard
        self.bucket = bucket
        self.slide_back = slide_back_sec

    def scan(self, start: datetime, end: datetime, order_status: int):
        while start < end:
            window_end = min(start + timedelta(hours=1), end)
            self._scan_window(start, window_end, order_status)
            start = window_end - timedelta(seconds=self.slide_back)

    def _scan_window(self, start, end, order_status, page=1):
        while True:
            self.guard.pre_check()
            self.bucket.acquire()

            resp = self.client.execute("pdd.order.increment.get", {
                "start_updated_at": int(start.timestamp()),
                "end_updated_at": int(end.timestamp()),
                "order_status": order_status,
                "page": page,
                "page_size": 100,
            })

            if is_billing_error(resp):
                self.guard.on_api_billing_error()
                return

            orders = resp.get("order_increment_get_response", {}).get("order_list", [])
            if not orders:
                break

            for o in orders:
                yield o

            if len(orders) < 100 or page >= 50:
                break
            page += 1

详情补全：

def fill_detail(client, guard, order_sn):
    guard.pre_check()
    resp = client.execute("pdd.order.information.get", {
        "order_sn": order_sn,
        "is_hit_rule": 1,   # 按权限要返的字段
    })
    return resp.get("order_info_get_response", {}).get("order_info")

七、云内/云外部署决策

def pdd_deploy_recommend(shops: int, need_decrypt: bool) -> str:
    if need_decrypt:
        return "必须拼多多云内；云外增值/解密直接禁调"
    if shops >= 5:
        return "拼多多云内；云外 10 店 ISV 一年多花几万"
    return "小自研可云外，但订单/地址解密迟早要回云内"

经验：

商品/订单列表：云外能跑，但 ×10
收货人姓名/手机/地址解密：敏感数据强制云内
增值 API：云外不是贵，是不允许
大促：云外网络抖动 + 余额临界 = 双杀

八、成本守卫（本地计数器兜底）

# pdd/cost_meter.py
from datetime import date


class CostMeter:
    def __init__(self, redis, client_id, price_per_100, budget_per_day):
        self.r = redis
        self.client_id = client_id
        self.price = price_per_100
        self.budget = budget_per_day

    def incr(self, calls=1):
        key = f"pdd:cost:{self.client_id}:{date.today()}"
        total = self.r.incrby(key, calls)
        self.r.expire(key, 60 * 60 * 48)
        if total / 100 * self.price > self.budget:
            raise BudgetExceeded(f"今日预算将超：{total} calls")

    def estimated_cost(self):
        key = f"pdd:cost:{self.client_id}:{date.today()}"
        total = int(self.r.get(key) or 0)
        return total / 100 * self.price

规则：

云端部署 ≠ 忘了记账
本地 meter 用来“提前刹车”，平台账单用来“事后对账”
预算超 80% → 降频；超 100% → 停非核心同步

九、错误分类（拼多多特有）

现象	类型	处理
`sign invalid`	签名错	改拼接/时间戳/secret
`timestamp expired`	时钟偏差	校准机器时间
无接口权限	权限没申请	后台点申请，不重试
系统错误/费用类（如 50001 特征）	欠费/余额	熔断 + 充值 + 不重试业务
限流/频繁访问	可重试	指数退避
网络超时	可重试	退避，但别当“没单”
解密返回空	云外调增值	必须云内，不靠加钱解决

十、和前几篇收口

1688：基础免费但 QPS 卡脖，跨境/分销另立域
淘宝 TOP：聚石塔内外 10 倍，增值禁外
京东 JOS：云鼎内稳，POP 订单时间窗滚动
拼多多：预充值硬断，云外 10 倍，增值禁外
共同真理：
消息/增量做触发器，详情接口做真相，本地守卫做安全带，云内合规做底线。

一句话：

拼多多 API 的单价是糖，预充值模型是刀。
你以为在省 0.01 元/百次，
其实在赌“余额别在双 11 凌晨归零、云外别被禁调解密、控制台别拿 T-1 余额骗你”。
把 BalanceGuard 写进 Client，比把 sign 写对更重要。

《京东订单API jingdong.pop.order.search 落地方案：鼎内调用与QPS配额管理》（附 Python 源码）

Wed, 14 Oct 2026 09:20:52 +0800

《京东订单API jingdong.pop.order.search 落地方案：鼎内调用与QPS配额管理》（附 Python 源码）

一、先拍结论

jingdong.pop.order.search 是 POP 商家订单检索接口，不是订单真相源（真相是 jd.order.detail.get / 订单消息）。
它的生产定位是：“按状态+时间窗捞订单列表”，再拿 orderId 去拉详情。
官方已经明说：
创建时间支持查近 2 年，按更新时间只支持近 3 个月
开始/结束时间不得相差超过 1 个月
“建议通过时间滚动的方式拉取订单，页数要尽量少”
云鼎内调用是官方标注的调用方式之一

所以京东订单同步的成熟写法不是“翻页翻到天荒地老”，而是：

时间窗滚动（30天/窗）
  → pop.order.search 捞 orderId
  → jd.order.detail.get 拉真相
  → (order_id, order_state, operate_time) 做幂等
  → JMQ/订单消息做触发器，search 做兜底

二、接口边界（别和这些搞混）

接口	作用	备注
`jingdong.pop.order.search`	POP 订单列表检索	创建时间近2年 / 更新时间近3月 / 时间窗≤1月
`jingdong.pop.order.fbp.search`	FBP 订单检索	京东仓配履约订单
`jd.order.detail.get`	单订单详情	子单/优惠/发票/收货
JMQ 订单消息	变更推送	触发器，不是真相
`jingdong.pop.order.print.data.get`	发货/打单数据	打单用
`jingdong.etms.waybill.*`	电子面单/轨迹	履约用

POP 和 VOP（企业采购）权限完全隔离，别用同一个 AppKey 想通吃。

三、时间窗与分页：官方给的“反模式”

官方文档原话要点：

start_date / end_date 差值 不得超过 1 个月
WAIT_SELLER_STOCK_OUT 时起止可空，默认前一个月
其他状态 start_date 必传
page_size 最大 100
“建议您之前频繁的翻页，页数较大，请尽快调整为时间滚动方式”

❌ 错误写法

page=1..5000，固定最近30天，order_state=ALL

问题：

大商家一窗几万单 → 翻页深 → 超时/限流
订单在窗口内被修改 → 翻页过程中顺序漂移 → 漏单
状态 ALL + 深翻页 = 京东最喜欢限你流

✅ 正确写法

按 order_state 分桶
每个状态：时间窗 30 天，sortType=0 升序，page 不超 50
窗滚动：end = now - 30s，start = end - 30d
每次同步记录 (state, last_operate_time)
下次从 last_operate_time 继续

四、鼎内调用是什么

京东有“云鼎/鼎内”概念：应用/服务在京东云鼎环境内调用，某些订单/数据/合规链路更稳，官方接口文档也会标“云鼎内调用”。

生产含义：

鼎内：延迟低、订单/隐私字段合规链路更完整、大促抖动小
鼎外（公网 IDC）：能调，但隐私字段/大促稳定性/审计不如鼎内
订单类接口必须卖家 OAuth access_token，不管鼎内鼎外

经验口径（非官方保证）：
京东商家 JOS 订单类接口，鼎内通常比鼎外更适合做“持续同步 + 大促兜底”；
但“鼎内”不是免限流金牌，QPS 配额照样按应用/权限包来。

五、QPS 与配额：社区经验口径

官方不把“所有应用的 QPS”写死成一刀切，社区实测大致是：

应用类型	订单类经验 QPS	说明
个人应用	1~2/s	订单类基本别上生产
企业实名（基础）	5~10/s	`pop.order.search` 常见档
买资源包	20 / 50 / 100/s	需平台审批/购包
鼎内企业应用	视权限包	不是无限，别睡大觉

费用口径（社区经验）：

商家 JOS 有日免费额度
pop.order.search 超量按量计费
鼎内/云内单价低于公网（类似淘宝聚石塔内外价差逻辑）
联盟 API（jd.union.open.*）是 CPS 选品体系，和商家订单同步是两回事

六、Python：京东 JOS 基础 Client（秒级时间！）

# jd/jos_client.py
import hashlib
import time
import requests


class JdJosClient:
    GATEWAY = "https://api.jd.com/routerjson"

    def __init__(self, app_key, app_secret, access_token=None, timeout=5):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        self.timeout = timeout

    def _sign(self, params: dict) -> str:
        flat = {k: v for k, v in params.items() if v is not None and k != "sign"}
        qs = "".join(f"{k}{v}" for k, v in sorted(flat.items()))
        raw = f"{self.app_secret}{qs}{self.app_secret}"
        return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()

    def execute(self, method: str, biz: dict):
        params = {
            "app_key": self.app_key,
            "method": method,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),  # 京东用秒级/字符串
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
        }
        if self.access_token:
            params["access_token"] = self.access_token
        params.update(biz)
        params["sign"] = self._sign(params)

        resp = requests.post(self.GATEWAY, data=params, timeout=self.timeout)
        return resp.json()

京东坑：把 1688 的毫秒时间戳带过来 → timestamp invalid。
京东要 yyyy-MM-dd HH:mm:ss，允许误差 10 分钟。

七、Python：时间滚动订单检索器

# jd/pop_order_scanner.py
from datetime import datetime, timedelta
from ratelimit.token_bucket import TokenBucket


class PopOrderScanner:
    MAX_WINDOW = timedelta(days=30)
    SLIDE_BACK = timedelta(seconds=30)   # 防修改时间漂移漏单

    def __init__(self, client: JdJosClient, bucket: TokenBucket):
        self.client = client
        self.bucket = bucket

    def scan_state(self, order_state: str, start: datetime, end: datetime):
        while start < end:
            window_end = min(start + self.MAX_WINDOW, end)
            yield from self._scan_window(order_state, start, window_end)
            # 下一窗往前滚，并回退 30s 防丢
            start = window_end - self.SLIDE_BACK
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    def _scan_window(self, order_state, start, end, page_size=100):
        page = 1
        while True:
            self.bucket.acquire()
            resp = self.client.execute("jingdong.pop.order.search", {
                "start_date": start.strftime("%Y-%m-%d %H:%M:%S"),
                "end_date": end.strftime("%Y-%m-%d %H:%M:%S"),
                "order_state": order_state,
                "dateType": 0,                 # 0=按创建时间；1=按修改时间
                "sortType": 0,                 # 0=升序，滚动拉取稳
                "page": str(page),
                "page_size": str(page_size),
                "optional_fields": (
                    "orderId,orderState,operateTime,orderStartTime,"
                    "itemInfoList,payType,orderSellerPrice,consigneeInfo"
                ),
            })

            node = resp.get("jingdong_pop_order_search_responce") or {}
            result = node.get("searchorderinfo_result") or {}
            api_result = result.get("apiResult") or {}

            if api_result.get("success") == "false":
                raise JdApiError(api_result.get("chineseErrCode"), api_result.get("numberCode"))

            orders = result.get("orderInfoList") or []
            if not orders:
                break

            for o in orders:
                yield o

            # 官方建议：别深翻页
            if len(orders) < page_size or page >= 50:
                break
            page += 1

状态分桶示例：

STATES = [
    "NOT_PAY",
    "WAIT_SELLER_STOCK_OUT",
    "WAIT_GOODS_RECEIVE_CONFIRM",
    "FINISHED_L",
    "TRADE_CANCELED",
    "LOCKED",
]

八、QPS 配额管理器（单实例 + Redis 分布式）

单实例令牌桶

# ratelimit/token_bucket.py
import time

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class TokenBucket:
    def __init__(self, rate: float, capacity: float = None, jitter: float = 0.05):
        self.rate = rate
        self.capacity = capacity or rate
        self.tokens = self.capacity
        self.last = time.monotonic()
        self.jitter = jitter

    def acquire(self, n: int = 1):
        while True:
            now = time.monotonic()
            self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.rate)
            self.last = now
            if self.tokens >= n:
                self.tokens -= n
                return
            sleep = (n - self.tokens) / self.rate + self.jitter
            time.sleep(sleep)

多实例分布式限流

# ratelimit/redis_sliding.py
import time


class RedisQpsLimiter:
    def __init__(self, redis, key: str, qps: int):
        self.r = redis
        self.key = key
        self.qps = qps

    def allow(self) -> bool:
        now = time.time()
        self.r.zremrangebyscore(self.key, 0, now - 1)
        self.r.zadd(self.key, {str(now): now})
        self.r.expire(self.key, 2)
        return self.r.zcard(self.key) <= self.qps

调用侧：

if not redis_limiter.allow():
    time.sleep(0.1)
    continue
bucket.acquire()

规则：

企业 10 QPS → 客户端设 8
鼎内企业 20 QPS → 客户端设 16
多店铺：按 shop_id 分桶，别全店共享一个令牌桶
列表 8 QPS + 详情 5 QPS 分开桶，别互相挤死

九、限流/错误分类（京东特有）

class JdApiError(Exception):
    def __init__(self, msg, code=None):
        super().__init__(msg)
        self.code = code


def classify(resp):
    node = resp.get("jingdong_pop_order_search_responce", {})
    api = (node.get("searchorderinfo_result") or {}).get("apiResult") or {}
    if api.get("success") == "false":
        msg = api.get("chineseErrCode", "")
        code = api.get("numberCode", "")
        # 限额/限流
        if "accesscontrol" in msg or "limited" in msg or str(code) == "16":
            raise RetriableError(f"jd limit: {msg}")
        # 参数/权限
        if "permission" in msg or "validate" in msg:
            raise FatalError(f"jd param/perm: {msg}")
        raise RetriableError(f"jd api err: {msg}")

    # HTTP 层
    # code=16 / 429 / ISP_FLOW_CONTROL_LIMIT → 退避

重试策略：

code=16 / 429 / flow control → 指数退避（10s, 30s, 60s）
token 过期 → 刷 access_token 再试 1 次
参数错 / 无权限 / 店铺未授权 → 不重试，告警
网络超时 → 重试，但别把“超时”当成“没数据”

十、幂等与状态机

pop.order.search 只给列表，别直接写业务：

def upsert_from_search(order):
    order_id = str(order["orderId"])
    state = order["orderState"]
    op_time = parse_time(order.get("operateTime") or order.get("orderStartTime"))

    cur = order_repo.get(order_id)
    version = int(op_time.timestamp() * 1000)

    if cur and version < cur.version:
        return  # 旧窗数据晚到

    # 只存“列表级快照”，详情再去 jd.order.detail.get 拉
    order_repo.upsert({
        "order_id": order_id,
        "order_state": state,
        "operate_time": op_time,
        "version": version,
        "need_detail": True,
    })

详情拉取：

def fill_detail(order_id):
    resp = client.execute("jd.order.detail.get", {"order_id": order_id})
    trade = extract_trade(resp)
    order_repo.update_detail(trade)

十一、消息 + 搜索双通道（生产必做）

JMQ 订单消息
   └─> 入队 order_change(order_id)

定时器（每 60s）
   └─> pop.order.search 按状态滚动
         └─> 补消息漏掉的 order_id

两者都只干一件事：
   mark need_detail = True

Worker：
   jd.order.detail.get → 写 order_snapshot → 跑状态机

铁律：

消息：快返回、msgId 去重、异步处理
搜索：兜底、时间滚动、别深翻页
详情：真相源
(order_id, order_state, operate_time)：幂等键

十二、大促/店铺多时的配置

JD_POP_CONFIG = {
    "bucket_qps": 8,                 # 企业 10 留余量
    "detail_qps": 5,
    "window_days": 30,
    "slide_back_sec": 30,
    "max_page": 50,
    "states": STATES,
    "per_shop_bucket": True,
    "use_jmq": True,
    "search_fallback_interval": 60,  # 秒
    "dingnei_preferred": True,       # 鼎内网关/部署优先
}

鼎内部署时：

订单同步服务放鼎内容器
Redis/DB 也走鼎内内网
公网只暴露“内部调度 API”，不暴露 JOS 调用层
大促前压测的是 detail + 发货 + 面单，不是 search 翻页

十三、落地检查表

[ ] 时间戳用 yyyy-MM-dd HH:mm:ss，别用毫秒
[ ] pop.order.search 只捞列表，不写业务
[ ] 时间窗 ≤ 30 天，状态分桶，回退 30s
[ ] page 不超 50，不深翻页
[ ] 列表 QPS / 详情 QPS / 面单 QPS 分桶
[ ] 多实例用 Redis 滑动窗口，不用单机 time.sleep 假装限流
[ ] 限流退避、token 过期刷一次、权限错告警
[ ] JMQ 做触发器，search 做兜底，detail 做真相
[ ] (order_id, order_state, operate_time) 幂等
[ ] POP / FBP / VOP 应用分开，不串 AppKey
[ ] 鼎内优先，但鼎内也不假设“无限 QPS”

十四、一句话收口

jingdong.pop.order.search 不是“订单同步主接口”，是“订单 ID 捕捞网”。
生产里把它当主角 → 深翻页、限流、漏单、大促崩；
把它当配角 → 时间滚动捞 ID、消息做闹钟、detail 做法官、QPS 桶做安全带。
鼎内调用解决的是合规和稳定性，不解决“你不会限流”的问题。

《京东商品API：从 item.jd.get 迁移到 jingdong.item.read.get 的字段映射与踩坑》（附 Python 源码）

Tue, 13 Oct 2026 09:36:34 +0800

《京东商品API：从 item.jd.get 迁移到 jingdong.item.read.get 的字段映射与踩坑》（附 Python 源码）

先拍结论：

京东这波迁移不是重构业务，而是“宙斯 JOS → 京东商家开放平台”的命名/入口收敛。
网关还是 https://api.jd.com/routerjson，签名还是 secret + kv(ASCII) + secret，AccessToken 逻辑不变；
真正要改的是三件事：
method 从 item.jd.get / jd.item.get / jingdong.ware.get 换成 jingdong.item.read.get
返回根节点从 item_get_response / jingdong_ware_get_response 换成 jingdong_item_read_get_response
旧接口“单 SKU 平铺”，新接口“SPU + skuList + priceInfo + stockInfo”嵌套化
老代码直接 item["title"] 的，迁移后必炸。

一、老 vs 新：先对齐接口身份

维度	老 `item.jd.get` / `jd.item.get`	新 `jingdong.item.read.get`
平台	宙斯 JOS（已融合到 open.jd.com）	京东商家开放平台
定位	商品详情只读（旧命名）	商品基础信息只读（标准化）
网关	`api.jd.com/routerjson`	同上
签名	MD5(secret+kv+secret)	同上（一行不改）
入参	`sku_id` / `item_id` + `fields`	`sku_id` / `sku_ids` + `fields`
批量	不支持或各版本不一	支持 `sku_ids`，最多 20 个
返回根	`item_get_response`	`jingdong_item_read_get_response`
结构	平铺多	`goodsInfoList / priceInfo / imageInfo / skuList / stockInfo`
新应用申请	维护期，不鼓励	官方推荐

经验口径：
网关/签名/鉴权零改动，解析层重写，批量能力顺手用上。

二、旧接口“脑内模型”（为什么会写错）

老 item.jd.get 很多人写成：

{
  "item_get_response": {
    "item": {
      "sku_id": 100012345678,
      "title": "XX 笔记本",
      "price": 4999.00,
      "stock": 120,
      "image": "https://img10.360buyimg.com/..."
    }
  }
}

于是老代码：

item = resp["item_get_response"]["item"]
title = item["title"]
price = float(item["price"])
stock = int(item["stock"])

三、新接口真实结构（简化但接近生产）

{
  "jingdong_item_read_get_response": {
    "code": "0",
    "message": "success",
    "result": {
      "goodsInfoList": [
        {
          "skuId": "100012345678",
          "title": "XX 笔记本",
          "shortTitle": "轻薄本",
          "saleState": 1,
          "priceInfo": {
            "marketPrice": "5999.00",
            "jdPrice": "4999.00",
            "promotionPrice": "4799.00",
            "currency": "CNY"
          },
          "imageInfo": {
            "mainImg": "https://img10.360buyimg.com/n1/...",
            "imgList": ["https://img10...", "https://img10..."]
          },
          "brand": { "brandId": 123, "brandName": "Lenovo" },
          "category": {
            "cid1": 1, "cid1Name": "电脑办公",
            "cid2": 12, "cid2Name": "笔记本",
            "cid3": 123, "cid3Name": "轻薄本"
          },
          "stockInfo": {
            "stockNum": 120,
            "stockStatus": "available"
          },
          "salesInfo": {
            "totalSold": 3200,
            "commentCount": 1450,
            "goodRate": "97.2"
          },
          "skuList": [
            {
              "skuId": "10001234567801",
              "specText": "16G+512G",
              "price": "4999.00",
              "stock": 60
            },
            {
              "skuId": "10001234567802",
              "specText": "32G+1T",
              "price": "5799.00",
              "stock": 60
            }
          ]
        }
      ]
    }
  }
}

⚠️ 不同权限包返回的字段名会有差异：

有的返回 jdPrice，有的返回 salePrice
有的返回 stockNum，有的返回 stock
有的返回 goodsInfoList，联盟接口返回 data
下架商品 saleState=0，价格/库存可能直接为空

所以迁移第一原则：别信字段名，先打原始 JSON 建映射表。

四、字段映射表（ERP 落地用）

老 item.jd.get	新 jingdong.item.read.get	说明
`item.sku_id`	`goodsInfoList[].skuId`	最小销售单元
`item.item_id` / `ware_id`	SPU 需另查 / 用 sku 反推	京东强 SKU 模型
`item.title`	`goodsInfoList[].title`
`item.sub_title`	`shortTitle`	不一定有
`item.price`	`priceInfo.jdPrice`	当前京东价
`item.market_price`	`priceInfo.marketPrice`	划线价
`item.promotion_price`	`priceInfo.promotionPrice`	活动价
`item.stock`	`stockInfo.stockNum` / `skuList[].stock`	区域库存才准
`item.img_url`	`imageInfo.mainImg`	360buy 图链有防盗链
`item.images`	`imageInfo.imgList`
`item.brand`	`brand.brandName`
`item.cid`	`category.cid3`	三级类目
`item.sku_attrs`	`skuList[].specText`	“颜色:黑;尺码:XL”
`item.desc`	新接口常不返回	详情 HTML 要别的权限/接口
`item.seller_nick`	`shopInfo.shopName`	商家接口才有
`item.outer_id`	商家后台字段，新只读接口未必返	自营运配编码要店铺商品接口

跨平台统一模型里：
京东主键用 jd_<skuId>，不是 item_id；
1688 是 offerId+skuId，淘宝是 num_iid+sku_id，京东是 skuId + skuList[].skuId。

五、Python：兼容老新的客户端

# jd/client.py
import hashlib
import time
import requests

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class JDClient:
    GATEWAY = "https://api.jd.com/routerjson"

    def __init__(self, app_key, app_secret, access_token=None, timeout=5):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        self.timeout = timeout

    def _sign(self, params: dict) -> str:
        flat = {k: v for k, v in params.items() if v is not None and k != "sign"}
        qs = "".join(f"{k}{v}" for k, v in sorted(flat.items()))
        raw = f"{self.app_secret}{qs}{self.app_secret}"
        return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()

    def execute(self, method: str, biz: dict):
        params = {
            "app_key": self.app_key,
            "method": method,
            "timestamp": str(int(time.time() * 1000)),
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
        }
        if self.access_token:
            params["access_token"] = self.access_token
        params.update(biz)
        params["sign"] = self._sign(params)

        resp = requests.post(self.GATEWAY, data=params, timeout=self.timeout)
        return resp.json()

六、Python：旧→新适配层（重点）

# jd/item_adapter.py
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class JDItem:
    sku_id: str
    spu_hint: Optional[str] = None
    title: Optional[str] = None
    short_title: Optional[str] = None
    market_price: Optional[float] = None
    sale_price: Optional[float] = None
    promo_price: Optional[float] = None
    main_img: Optional[str] = None
    img_list: list[str] = field(default_factory=list)
    brand_name: Optional[str] = None
    cid3: Optional[str] = None
    cid3_name: Optional[str] = None
    total_stock: Optional[int] = None
    on_sale: bool = True
    skus: list[dict] = field(default_factory=list)
    raw: dict = field(default_factory=dict)


def _f(x):
    try:
        return float(x)
    except (TypeError, ValueError):
        return None


def _i(x):
    try:
        return int(x)
    except (TypeError, ValueError):
        return None


def parse_legacy_item(resp: dict) -> JDItem:
    """老 item.jd.get"""
    item = (resp.get("item_get_response") or {}).get("item") or {}
    return JDItem(
        sku_id=str(item.get("sku_id") or item.get("skuId")),
        title=item.get("title"),
        market_price=_f(item.get("market_price")),
        sale_price=_f(item.get("price")),
        promo_price=_f(item.get("promotion_price")),
        main_img=item.get("img_url") or item.get("image"),
        total_stock=_i(item.get("stock")),
        skus=[{
            "sku_id": str(item.get("sku_id")),
            "spec": "",
            "price": _f(item.get("price")),
            "stock": _i(item.get("stock")),
        }],
        raw=item,
    )

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def parse_new_item(resp: dict) -> JDItem:
    """新 jingdong.item.read.get"""
    node = resp.get("jingdong_item_read_get_response") or {}
    if str(node.get("code")) not in ("0", "", None) and "result" not in node:
        raise RuntimeError(node.get("message") or "jd item read error")

    goods_list = (
        (node.get("result") or {}).get("goodsInfoList") or []
    )
    if not goods_list:
        raise RuntimeError("empty goodsInfoList")

    g = goods_list[0]
    price_info = g.get("priceInfo") or {}
    stock_info = g.get("stockInfo") or {}
    image_info = g.get("imageInfo") or {}
    category = g.get("category") or {}

    return JDItem(
        sku_id=str(g.get("skuId")),
        title=g.get("title"),
        short_title=g.get("shortTitle"),
        market_price=_f(price_info.get("marketPrice")),
        sale_price=_f(price_info.get("jdPrice") or price_info.get("salePrice")),
        promo_price=_f(price_info.get("promotionPrice")),
        main_img=image_info.get("mainImg"),
        img_list=image_info.get("imgList") or [],
        brand_name=(g.get("brand") or {}).get("brandName"),
        cid3=str(category.get("cid3") or ""),
        cid3_name=category.get("cid3Name"),
        total_stock=_i(stock_info.get("stockNum") or stock_info.get("stock")),
        on_sale=g.get("saleState", 1) == 1,
        skus=[{
            "sku_id": str(s.get("skuId")),
            "spec": s.get("specText") or s.get("spec") or "",
            "price": _f(s.get("price")),
            "stock": _i(s.get("stock")),
        } for s in (g.get("skuList") or [])],
        raw=g,
    )

调用侧：

client = JDClient(APP_KEY, APP_SECRET, ACCESS_TOKEN)

# 老
legacy = parse_legacy_item(client.execute("item.jd.get", {"sku_id": sku}))
# 新
new = parse_new_item(client.execute("jingdong.item.read.get", {
    "sku_id": sku,
    "fields": "skuId,title,shortTitle,priceInfo,imageInfo,brand,category,stockInfo,skuList,salesInfo"
}))

七、批量迁移红利：别再单 SKU 狂打

def batch_get(client, sku_ids: list[str]):
    if len(sku_ids) > 20:
        raise ValueError("jingdong.item.read.get 单次最多 20 个")
    resp = client.execute("jingdong.item.read.get", {
        "sku_ids": ",".join(sku_ids),
        "fields": "skuId,title,priceInfo,stockInfo,imageInfo,skuList"
    })
    node = resp.get("jingdong_item_read_get_response") or {}
    return (node.get("result") or {}).get("goodsInfoList") or []

限流口径（社区经验）：

基础 2 QPS
企业 5~50 QPS
批量 20 个 ≠ 20 次免费，仍按平台计费/限流模型走

BUCKET = TokenBucket(rate=8)   # 企业 10 的话留余量

def safe_batch(skus):
    for chunk in chunks(skus, 20):
        BUCKET.acquire()
        yield batch_get(client, chunk)

八、迁移期双跑校验（很重要）

def migrate_check(sku):
    old = safe_parse(parse_legacy_item(client.execute("item.jd.get", {"sku_id": sku})))
    new = safe_parse(parse_new_item(client.execute("jingdong.item.read.get", {"sku_id": sku})))

    diffs = []
    if old.title != new.title:
        diffs.append(("title", old.title, new.title))
    if abs((old.sale_price or 0) - (new.sale_price or 0)) > 0.01:
        diffs.append(("price", old.sale_price, new.sale_price))
    if (old.total_stock or 0) != (new.total_stock or 0):
        diffs.append(("stock", old.total_stock, new.total_stock))
    return diffs

规则：

标题/价格不一致 → 先查 area 区域价、查促销时间窗
库存不一致 → 京东库存是区域可售库存，不传 area 默认北京
老接口有、新接口没有（desc / outer_id / cost_price）→ 不是丢了，是权限域不同

九、区域库存坑（京东特有）

client.execute("jingdong.item.read.get", {
    "sku_id": sku,
    "area": "19_1607_3157_0",   # 省_市_区_镇
    "fields": "skuId,stockInfo,priceInfo"
})

不传 area：

价格用默认区域
库存用平台默认仓配视角
跨境/家电/冷链会明显失真

淘宝/1688 很少强调“区域库存”，京东 POP/自营必须传 area。

十、和前几篇收口

1688：offerId+skuId 采购视角，amountOnSale=-1/999999 不可信
淘宝：num_iid 卖家视角，item.get 公开 / item.seller.get 私有
京东：skuId 是王，jingdong.item.read.get 是商家/POP 只读视图；联盟 jd.union.open.goods.detail.query 是 CPS 视角，别混用
迁移本质：京东不是“接口变了”，是“返回从平铺商品变成 SPU/SKU 价格库存对象树”

一句话：

老京东商品接口教你怎么“取一个商品”；
新 jingdong.item.read.get 教你怎么“理解一个京东 SPU：哪些 SKU、各区多少钱、哪区有货”。
迁移别只改 method，要把“商品=一条记录”的脑模型，升级成“商品=SPU 聚合 + SKU 集合 + 区域库存”。

《京东JOS API：落地方案、接口边界与业务踩坑 —— 联盟免费 vs 商家按量的双轨制》（附python源码）

Tue, 13 Oct 2026 09:16:31 +0800

下面按「双轨制定位 → 接口边界 → 落地架构 → 常见踩坑 → Python 双 Client 源码」给你一套可直接抄的工程化方案。

结论先拍：
京东联盟 API（jd.union.open.）= 选品/CPS/比价，调用本身接近免费，赚钱靠成交抽佣*
京东商家 JOS（jingdong. / jd.）= 店铺订单/真实库存/发货/电子面单，有日免额，超量按量计费**
两套 Key、两套授权、两套配额，绝不能混用。

一、双轨制全景对照

维度	京东联盟 API	京东商家 JOS
前缀	`jd.union.open.*`	`jingdong.` / `jd.pop.order.` / `jd.order.*`
目的	选品、券价、佣金、CPS 推广	ERP、订单同步、真实库存、发货回写
资质	联盟实名（可不绑店铺）	企业认证 + 卖家 OAuth AccessToken
库存	有货 / 无货 / 预售	可售库存、锁定库存、区域仓存
价格	促销价、券后价、佣金率	商家后台基准价、协议价
费用	调用免费，成交抽佣 5%~20%	日免额内免费，超量云内 ¥0.01~0.05/百次，云外 ×3~×10
电子面单	无	按单计费（几分/单）
典型 QPS	5~30/s（可申）	个人 1~2/s，企业 5~10/s
适合	导购站、比价、铺货、达人系统	商家自建 ERP / WMS / 打单系统

⚠️ 最贵误区：

用联盟 API 拿“自己店铺真实库存”→ 拿不到；

用商家 JOS 批量爬竞品比价 → 3000 次/天或企业免额瞬间烧完，云外再放大 10 倍账单。

二、接口边界（什么时候用哪套）

✅ 用「联盟 API」

商品搜索：jd.union.open.goods.query
商品详情：jd.union.open.goods.detail.query
推广链接：jd.union.open.goods.promotiongoodsinfo.query
券后价 / 佣金率 / 类目 CPS
跨店选品、内容电商、返利机器人

✅ 用「商家 JOS」

商品主数据：jingdong.ware.read.get
SKU：jingdong.ware.sku.read
真实库存：jingdong.stock.get / ware.stock.get
店铺订单：jingdong.pop.order.search / jd.order.detail.get
发货回写 / 物流轨迹：jingdong.etms.trace.get
电子面单：jingdong.etms.waybill.get

❌ 禁止用法

联盟 Key 查自己店铺订单 → 无权
商家 Key 爬全站竞品 → 超量 + 合规风险
把联盟调用量“无限轮询”思维套到商家 Key → 月烧几千块

三、落地架构（生产可用）

┌──────────────┐
 定时任务/Webhook│              │
 ───────────────▶│  API Router  │
                 └──────┬───────┘
                        │
        ┌───────────────┴───────────────┐
        ▼                                ▼
 UnionClient                      JosMerchantClient
（联盟：选品/比价/推广）          （商家：订单/库存/发货）
        │                                │
        ▼                                ▼
 Quota: 只记调用量                  Quota: 日额度 + 超量预警
 Billing: CPS 佣金                   Billing: 按量 + 电子面单

工程原则：

Key 不混用：配置层就隔离
商家接口走令牌桶：QPS 保守（企业 8/s，个人 1/s）
订单/库存优先消息推送，别轮询
商家调用打 meter：日调用量、接口级、店铺级
联盟接口可高频，但别用来“补商家数据”

四、业务踩坑清单（很值钱）

时间戳用秒不是毫秒
JOS 要 int(time.time())，用毫秒 → Invalid Timestamp
签名必须 ASCII 升序 + 空值剔除 + 紧凑 JSON
数组/JSON 字段别用 json.dumps(obj) 默认空格
公开商品接口 ≠ 自己店铺数据
jingdong.ware.read.get 不传 token 也能查公开字段
但库存 / 商家编码 / 区域仓 → 必须卖家 AccessToken
订单类接口要“企业应用 + 接口权限申请 + token”
个人应用直接 403 / no permission
日免额耗尽返回类似
accesscontrol.limited-by-app-access-count
不是代码错，是没钱了
电子面单“取号即计费”
取消未揽收通常不扣，但取号动作本身计费
沙箱只验签名
沙箱返 mock 商品/空订单是正常的，别调试到怀疑人生
云外调用贵到离谱
商家系统尽量放京东云/聚石塔体系内

五、Python 双 Client 源码（可直接用）

# jd_dual_api_client.py
"""
京东双通道 API Client
1. UnionClient       -> 京东联盟 API (jd.union.open.*)
2. JosMerchantClient -> 京东商家 JOS  (jingdong.* / jd.*)
网关统一: https://api.jd.com/routerjson
"""

import hashlib
import json
import time
import requests
from typing import Any, Dict


class BaseJDClient:
    GATEWAY = "https://api.jd.com/routerjson"

    def __init__(self, app_key: str, app_secret: str):
        self.app_key = app_key
        self.app_secret = app_secret

    def _build_sign(self, params: Dict[str, Any]) -> str:
        """
        MD5 签名：
        - 剔除 sign
        - 空字符串可保留（京东一般要求非空，但实现上建议剔除 None）
        - key 按 ASCII 升序
        - secret + kv + secret
        """
        filtered = {
            k: v for k, v in params.items()
            if k != "sign" and v is not None and v != ""
        }
        raw = self.app_secret
        for k in sorted(filtered.keys()):
            v = filtered[k]
            if isinstance(v, (dict, list)):
                v = json.dumps(v, ensure_ascii=False, separators=(",", ":"))
            raw += f"{k}{v}"
        raw += self.app_secret
        return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()

    def _request(self, method: str, extra: Dict[str, Any], token: str = None) -> dict:
        params = {
            "app_key": self.app_key,
            "method": method,
            "timestamp": str(int(time.time())),
            "format": "json",
            "v": "1.0",
            "sign_method": "md5",
        }
        if token:
            params["access_token"] = token

        # 业务参数
        for k, v in extra.items():
            if isinstance(v, (dict, list)):
                v = json.dumps(v, ensure_ascii=False, separators=(",", ":"))
            params[k] = v

        params["sign"] = self._build_sign(params)

        resp = requests.post(self.GATEWAY, data=params, timeout=10)
        resp.raise_for_status()
        return resp.json()


class UnionClient(BaseJDClient):
    """
    京东联盟 API
    场景：选品 / 比价 / 券后价 / 佣金 / CPS
    """

    def goods_query(self, keyword: str, page_index: int = 1, page_size: int = 20):
        biz = {
            "goodsReqDTO": {
                "keyword": keyword,
                "pageIndex": page_index,
                "pageSize": page_size,
            }
        }
        return self._request(
            method="jd.union.open.goods.query",
            extra={"360buy_param_json": biz},
        )

    def goods_detail(self, sku_ids: list):
        biz = {
            "skuIds": sku_ids
        }
        return self._request(
            method="jd.union.open.goods.detail.query",
            extra={"360buy_param_json": biz},
        )

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class JosMerchantClient(BaseJDClient):
    """
    京东商家 JOS
    场景：商品主数据 / 真实库存 / 店铺订单 / 发货
    """

    def ware_read(self, ware_id: int, access_token: str):
        biz = {
            "wareId": ware_id
        }
        return self._request(
            method="jingdong.ware.read.get",
            extra={"360buy_param_json": biz},
            token=access_token,
        )

    def stock_get(self, sku_id: int, access_token: str):
        biz = {
            "skuId": sku_id
        }
        return self._request(
            method="jingdong.stock.get",
            extra={"360buy_param_json": biz},
            token=access_token,
        )

    def pop_order_search(self, start_date: str, end_date: str, page: int, access_token: str):
        biz = {
            "orderReq": {
                "startDate": start_date,
                "endDate": end_date,
                "page": page,
                "pageSize": 100,
            }
        }
        return self._request(
            method="jingdong.pop.order.search",
            extra={"360buy_param_json": biz},
            token=access_token,
        )


if __name__ == "__main__":
    union = UnionClient("UNION_APP_KEY", "UNION_APP_SECRET")
    print(union.goods_query("机械键盘"))

    merchant = JosMerchantClient("JOS_APP_KEY", "JOS_APP_SECRET")
    print(merchant.ware_read(1000123, "SELLER_ACCESS_TOKEN"))

六、配额护栏（生产必加）

商家侧一定要加：

from collections import defaultdict
from datetime import date

daily_counter = defaultdict(int)

def count_and_guard(app_name: str, cost: int, daily_limit: int):
    today = str(date.today())
    key = f"{app_name}:{today}"
    daily_counter[key] += cost
    if daily_counter[key] > daily_limit:
        raise RuntimeError(f"{app_name} 日调用量即将超免额：{daily_counter[key]}/{daily_limit}")

规则建议：

联盟：只统计，不扣费
商家：日调用 > 80% 免额 → 告警
订单列表：能用 Webhook/消息就不翻页
库存：本地缓存 30~60s，别每秒查

七、一句话选型

**“看别人商品”用联盟，“管自己店铺”用 JOS；
联盟烧的是佣金，JOS 烧的是调用次数和面单。”**

如果你愿意，我可以把这套再扩成：

✅ 联盟推广链生成 + 佣金回执校验
✅ 商家订单增量同步（cursor / 消息 / 防超卖）
✅ 京东 + 淘宝 + 拼多多三平台 API 统一抽象层

你更想先要哪一个？

万邦api博客

《亚马逊 SP-API：落地方案、接口边界与业务踩坑 —— “年费 $1400 + 超量 $0.40/千次” 的真相与 2026 反转》（附 Python 源码）

《亚马逊 SP-API：落地方案、接口边界与业务踩坑 —— “年费 1400+超量0.40/千次” 的真相与 2026 反转》（附 Python 源码）

0. 先纠正一个时间线（很重要）

一、SP-API 和前面几家根本不是一类东西

二、订单接口边界（最容易被国内团队写错）

三、生产架构：通知优先，Reports 批量，REST 兜底

四、Python：LWA 令牌管理（refresh 不互踢）

五、Python：限流感知的 SP-API 客户端

六、订单同步：通知 + Report + getOrders 兜底

七、Reports 模式：大单量别用 getOrders

八、限流 / 错误分类（Amazon 特有）

九、“如果真收 $0.40/千次” 的防御代码

十、和前几篇收口

《抖音订单API：/trade/order/detail 落地方案 —— 消息推送 + 补偿查询的架构设计》（附 Python 源码）

《抖音订单API：/trade/order/detail 落地方案 —— 消息推送 + 补偿查询的架构设计》（附 Python 源码）

一、先拍结论

二、抖店订单接口边界（先纠名）

三、整体架构

四、为什么不能直接“消息来了就发货”

五、Python：抖店订单 Client

六、Detail Worker：10s 延迟 + 20min 补偿

七、补偿轮询：时间窗前移 + 倒序 + 重叠

八、幂等键与状态机

九、消息推送接入（只做触发器）

十、数据推送（抖店云 RDS）怎么摆

十一、售后轴独立（和拼多多同思路）

十二、调用配额与降本

十三、落地检查表

十四、一句话收口

《抖音商品API：/product/detail 与 /product/list 的字段边界与图片合规踩坑》（附 Python 源码）

《抖音商品API：/product/detail 与 /product/list 的字段边界与图片合规踩坑》（附 Python 源码）

一、先拍结论

二、两个接口的边界（别再混）

三、核心字段边界

1. 商品级

2. SKU 级

3. 规格图级

四、图片合规：抖店不是“能传上去就合规”

1. 主图规则（官方口径）

2. 比例与尺寸

3. 白底图误区

五、类目与必填属性：发品被打回的重灾区

六、Python：detail / list 拉取 + 字段归一

七、Python：SKU 图按 spec_detail_id 关联

八、Python：图片合规预检（上架前拦废图）

九、list → detail → sku 的生产流水线

十、编辑商品的两个暗雷

1. spec_prices 是整体结构

2. 审核中和在售商品

十一、跨平台 SKU 归一（接前几篇）

十二、落地检查表

十三、一句话收口

《抖店开放平台API：落地方案、接口边界与业务踩坑 —— 2026.7.1 商品发布接口收费》（附 Python 源码）

《抖店开放平台API：落地方案、接口边界与业务踩坑 —— 2026.7.1 商品发布接口收费》（附 Python 源码）

一、2026.7.1 新规先拍死（官方口径）

二、抖店 API 的“视角”和别家不同

三、商品发布接口边界（别把 addV2 当 detail 用）

四、成本测算（店群一眼看懂）

五、抖店签名 Client（method + timestamp 秒级）

六、预充值余额守卫（和拼多多同思路，但按月扣）

七、发品降本：把“N 次散弹”合并成“1 次收费”

八、订单侧：推送为主，轮询兜底

九、云内部署决策（抖店比拼多多更硬）

十、错误分类（抖店特有）

十一、和前几篇收口

《拼多多订单API：pdd.order.list.get 落地方案 —— 退款/售后状态机的幂等设计》（附 Python 源码）

《拼多多订单API：pdd.order.list.get 落地方案 —— 退款/售后状态机的幂等设计》（附 Python 源码）

一、pdd.order.list.get 的定位（别用错）

二、拼多多订单/售后状态轴（必须分开建）

1. 订单发货轴 order_status

2. 售后粗轴 refund_status（列表接口用）

3. 售后细轴 after_sales_status（详情/售后接口用）

4. 风控轴

三、幂等键设计（三层叠加）

四、统一订单快照模型

五、状态机：订单轴 × 售后轴 分别收敛

六、幂等落库（SQL 层兜底）

七、Python：增量同步 + 幂等消费

八、售后单独立建模（别塞进订单表覆盖）

《亚马逊 SP-API：落地方案、接口边界与业务踩坑 —— “年费 $1400 + 超量$ 0.40/千次” 的真相与 2026 反转》（附 Python 源码）

1. `spec_prices` 是整体结构

1. 订单发货轴 `order_status`

2. 售后粗轴 `refund_status`（列表接口用）

3. 售后细轴 `after_sales_status`（详情/售后接口用）

五、状态机：订单轴 × 售后轴分别收敛

十、消息推送 + 轮询双通道（和淘宝/京东同构）

二、类目踩坑 1：`cat_id` ≠ 可发布类目

十一、消息 + 搜索双通道（生产必做）