万邦api博客

《拼多多开放平台API预充值踩坑：0.01元/百次背后的到账延迟与欠费停调》（附python源码）

Sat, 18 Jul 2026 09:48:26 +0800

拼多多开放平台（POP Open Platform）采用预充值扣费模式，基础API云内调用标准为 0.01元/百次，云外为 0.1元/百次，增值API分别为 0.03元/百次 和 0.3元/百次。其核心坑点在于：充值到账非实时（财务审核/银行通道延迟）、扣费按日结算滞后、余额不足会直接切断API调用（返回错误而非降级），且不会在调用失败时给出明确的“余额不足”提示，极易导致ERP漏单。

一、三大核心踩坑点拆解

到账延迟（隐形断层）：充值后通常需要财务审核或走银行通道，并非像云服务器那样秒级到账。如果在余额临界点时充值，短时间内仍未到账，调用会继续失败。
欠费停调（硬切断）：拼多多不像某些平台有缓冲期，当预充值账户余额为 0 或负数时，接口会直接返回调用失败（通常为系统级错误码，如 50001 权限/费用相关错误，具体以当期文档为准），不会触发限流（429）而是直接业务中断，订单同步瞬间停止。
扣费滞后感知：官方按日统计扣除，控制台看到的余额可能不是实时扣减后的数值，容易误判还有余量而去睡大觉，醒来发现半夜断了好几个小时。

二、防御性编码：余额监控 + 预充值预警 + 调用熔断（Python源码）

建议在ERP调度任务中嵌入一个 GuardedPddClient，在每次批量调用前校验本地缓存的余额水位，低于阈值直接告警并暂停非核心任务。

# pdd_api_guard.py
"""
拼多多API预充值防御客户端
- 模拟获取账户余额（实际需调用拼多多费用查询接口或读取邮件/站内信）
- 本地水位阈值判断
- 欠费熔断 + 企微/钉钉告警
依赖: requests
"""
import time
import requests
import hashlib
import json
from datetime import datetime

# ================== 配置区 ==================
APP_KEY = "YOUR_PDD_APP_KEY"
APP_SECRET = "YOUR_PDD_APP_SECRET"
# 拼多多网关
GW = "https://gw-api.pinduoduo.com/api/router"
# 预警阈值：当预估或查询余额 < 预计1天消耗量时告警
WARNING_AMOUNT = 10.0  # 单位：元（例如预留10元≈100万次基础调用）
CHECK_INTERVAL = 1800  # 每30分钟校验一次余额
# ============================================
封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class PddAccountGuard:
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.last_check = 0
        self.cached_balance = None

    def _sign(self, params: dict) -> str:
        """拼多多MD5签名: secret + sorted_kv + secret"""
        filtered = sorted(
            (k, v) for k, v in params.items()
            if v is not None and str(v).strip() != "" and k != "sign"
        )
        qs = "".join(f"{k}{v}" for k, v in filtered)
        raw = f"{self.app_secret}{qs}{self.app_secret}"
        return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()

    def _call_raw(self, method: str, biz: dict) -> dict:
        """底层裸调用（不带守卫）"""
        params = {
            "client_id": self.app_key,
            "method": method,
            "timestamp": str(int(time.time())),
            "data_type": "JSON",
            "v": "V1.0",
            **biz
        }
        params["sign"] = self._sign(params)
        r = requests.post(GW, data=params, timeout=15)
        r.raise_for_status()
        return r.json()

    def fetch_balance_mock(self) -> float:
        """
        【重要】
        拼多多开放平台需查阅是否有公开的“查询账户余额”接口。
        若无直接接口，通常做法：
        1. 爬取/解析站内信的扣费通知
        2. 根据本地调用计数器 * 单价反向估算（误差大，仅作预警参考）
        3. 人工充值后在配置中心手动更新 balance
        这里用 mock 模拟一个余额值
        """
        # TODO: 替换为真实获取逻辑
        mock_balance = 8.50  # 模拟余额只剩8.5元
        self.cached_balance = mock_balance
        return mock_balance

    def check_balance_safe(self, est_daily_cost: float):
        """
        检查余额是否安全
        est_daily_cost: 预估单日消耗成本（调用次数/100 * 0.01）
        """
        now = time.time()
        if now - self.last_check < CHECK_INTERVAL and self.cached_balance is not None:
            bal = self.cached_balance
        else:
            bal = self.fetch_balance_mock()
            self.last_check = now

        if bal < WARNING_AMOUNT:
            self._alert(f"🚨 拼多多API余额预警：剩余 ¥{bal:.2f}，低于安全阈值 ¥{WARNING_AMOUNT}，预估日耗 ¥{est_daily_cost:.2f}，请立即充值！")
            # 可选：触发非核心任务熔断
            return False
        print(f"✅ 余额安全：剩余 ¥{bal:.2f}，日耗预估 ¥{est_daily_cost:.2f}")
        return True

    def _alert(self, msg: str):
        """简易告警（实际接入企微/钉钉Webhook）"""
        ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        print(f"[ALERT {ts}] {msg}")
        # 示例：requests.post(WECOM_WEBHOOK, json={"msgtype":"text","text":{"content":msg}})

    def safe_call(self, method: str, biz: dict, est_daily_cost: float = 0):
        """
        带余额守卫的调用
        """
        if not self.check_balance_safe(est_daily_cost):
            raise RuntimeError("API调用已熔断：账户余额不足，请充值后再试")

        try:
            resp = self._call_raw(method, biz)
            # 拼多多错误码判断（示例，以官方文档为准）
            if "error_response" in resp:
                err = resp["error_response"]
                code = err.get("code")
                msg = err.get("message", "")
                # 如果是费用/权限类错误，立即告警
                if code in (50001, 40001):  # 示意：可能涉及权限/费用/密钥错误
                    self._alert(f"❌ API返回疑似欠费/权限错误 [{code}]: {msg}")
                raise Exception(f"PDD[{code}]: {msg}")
            return resp
        except requests.exceptions.RequestException as e:
            # 网络层错误也要排查是否因欠费被网关拒绝
            self._alert(f"网络层异常，需排查是否因欠费停调：{e}")
            raise


# ===================== 使用示例 =====================
if __name__ == "__main__":
    client = PddAccountGuard(APP_KEY, APP_SECRET)

    # 假设每天调用 50万次基础API，云内成本 = 5000 * 0.01 = 50元/天
    EST_COST_PER_DAY = 50.0

    try:
        # 在定时任务（如APScheduler）拉取订单前先过守卫
        result = client.safe_call(
            "pdd.order.list.get",
            {
                "start_confirm_at": 1620000000,
                "end_confirm_at": 1620086400,
                "page": 1,
                "page_size": 100
            },
            est_daily_cost=EST_COST_PER_DAY
        )
        print("调用成功:", result.get("order_list_get_response", {}).get("total_count"))
    except RuntimeError as re:
        print(f"业务中断: {re}")
    except Exception as e:
        print(f"接口错误: {e}")

三、生产级避坑清单

双账户隔离：不要用主商家账号直接创建应用扣费，尽量用服务商/子账号隔离API预充值账户，防止误扣店铺保证金关联资金。
充值节奏：不要等告警再充，设置“余额 < 3天预估消耗”自动触发充值流程（甚至写脚本调用企业对公转账API，若开放），拼多多到账有延迟，必须留缓冲。
区分错误码：在统一异常处理里，把拼多多的 系统级错误（如签名错、权限错、费用相关错误） 与 业务级错误（如参数非法） 分开日志存储，欠费停调通常伴随特定系统级code，需第一时间钉钉轰炸。
本地计数器兜底：即使拿不到实时余额，也在程序里维护一个 (调用次数 // 100) * 0.01 的本地累减计数器，结合最后一次查到的官方余额做“双重校验”，防止控制台数据滞后导致击穿。

一句话总结：拼多多API 0.01元/百次 虽便宜，但预充值+硬切断机制是ERP杀手，必须在代码层做余额守卫、提前充值、错误码甄别，否则一次半夜欠费能让全店订单同步瘫痪数小时。

如果你愿意，我可以帮你把这套 余额守卫逻辑 嵌进你现有的 APScheduler 订单增量同步脚本 里，并补一份 拼多多常见系统级错误码与欠费特征对照表 吗？

《1688开放平台基础API真免费？QPS 10的限制怎么破（附GuardedClient源码）》

Sat, 18 Jul 2026 09:02:34 +0800

1688开放平台的基础API（商品搜索alibaba.offer.search、详情alibaba.item.get、订单查询等）确实是免费的，但默认受QPS限制（通常搜索类≈10/s，订单类≈20/s），超限会返回 ISP_FLOW_CONTROL_LIMIT 或 HTTP 429。

“破”QPS 10 的本质解法有三层：

客户端自律：令牌桶限速在 8~9 QPS（留余量防触发风控）。
架构级扩容：申请多个同企业实名下的AppKey做轮询分担（理论QPS = 单Key上限 × N）。
官方提额：在开放平台购买资源包将单Key提至50/100/200 QPS，或升级企业应用。

以下是集成了令牌桶限速 + 多Key轮询 + 限流退避的 GuardedClient 源码：

# guarded_1688_client.py
"""
1688开放平台 GuardedClient
- 多AppKey轮询突破单Key QPS限制
- 令牌桶单Key限速 (防触发ISP_FLOW_CONTROL_LIMIT)
- 捕获限流异常自动CoolDown + 指数退避重试
依赖: pip install requests
"""
import hashlib
import json
import time
import requests
from typing import Dict, List
from threading import Lock

# ------------------ 配置区 ------------------
APP_KEYS_SECRETS = [
    # 同一企业实名下创建的多个自用型应用
    {"key": "YOUR_APP_KEY_1", "secret": "YOUR_APP_SECRET_1"},
    {"key": "YOUR_APP_KEY_2", "secret": "YOUR_APP_SECRET_2"},
    # {"key": "...", "secret": "..."},
]
GW = "https://gw.api.1688.com/openapi"
MAX_QPS_PER_KEY = 8          # 单Key保守限速（官方≈10，留余量）
TOKEN_BUCKET_CAP = MAX_QPS_PER_KEY
# --------------------------------------------

封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ===== 单Key令牌桶 =====
class TokenBucket:
    def __init__(self, rate: float):
        self.rate = rate
        self.tokens = rate
        self.ts = time.monotonic()
        self.lock = Lock()

    def wait(self):
        with self.lock:
            now = time.monotonic()
            self.token = min(self.rate, self.tokens + (now - self.ts) * self.rate)
            self.ts = now
            if self.tokens < 1:
                time.sleep((1 - self.tokens) / self.rate + 0.005)
                self.tokens = 0
            else:
                self.tokens -= 1


# ===== 多Key轮询Guard =====
class Guarded1688Client:
    def __init__(self, apps: List[Dict], gw=GW):
        self.apps = apps
        self.gw = gw
        self.idx = 0
        self.idx_lock = Lock()
        # 为每个Key初始化独立令牌桶
        self.buckets = [TokenBucket(MAX_QPS_PER_KEY) for _ in apps]
        self.err_streak = [0] * len(apps)  # 连续限流计数

    def _next_app(self) -> (int, Dict, TokenBucket):
        with self.idx_lock:
            # 简单轮询，可改为按err_streak跳过冷却Key
            i = self.idx
            self.idx = (self.idx + 1) % len(self.apps)
            return i, self.apps[i], self.buckets[i]

    def _sign_1688(self, params: Dict, secret: str) -> str:
        """
        1688签名: md5(secret + sort(KV) + secret) -> 大写
        注意: 1688部分网关用 HmacMD5，这里以通用MD5为例，实际请对齐文档
        """
        filtered = sorted(
            (k, v) for k, v in params.items()
            if v is not None and str(v).strip() != "" and k != "sign"
        )
        qs = "".join(f"{k}{v}" for k, v in filtered)
        raw = f"{secret}{qs}{secret}"
        return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()

    def safe_call(self, method: str, biz: Dict, token: str = None, max_retry=5):
        """
        method: 如 alibaba.item.get
        biz: 业务参数
        """
        for att in range(max_retry):
            i, app, bucket = self._next_app()
            bucket.wait()  # 单Key QPS守护

            # 构造1688 param2风格或router风格参数（示意，按你网关调整）
            api_params = {
                "method": method,
                "app_key": app["key"],
                "timestamp": str(int(time.time() * 1000)),  # 1688用毫秒
                "format": "json",
                "v": "2.0",
                "sign_method": "md5",
                **biz
            }
            if token:
                api_params["access_token"] = token
            api_params["sign"] = self._sign_1688(api_params, app["secret"])

            try:
                # 1688 param2 路径示例: /openapi/param2/1/{method}/{app_key}
                url = f"{self.gw}/param2/1/{method}/{app['key']}"
                resp = requests.post(url, data=api_params, timeout=15)
                resp.raise_for_status()
                d = resp.json()

                # 1688成功通常在顶层，错误在 error_response 或 code!=200
                if "error_response" in d:
                    err = d["error_response"]
                    code = str(err.get("code", ""))
                    msg = err.get("msg", "")
                    # 限流判定
                    if "FLOW_CONTROL" in msg or code == "15" or resp.status_code == 429:
                        self.err_streak[i] += 1
                        back = min(2 ** self.err_streak[i], 8)
                        print(f"⚠️ Key[{i}] 限流，冷却 {back}s 后重试...")
                        time.sleep(back)
                        continue
                    raise Exception(f"1688[{code}]: {msg}")

                # 成功重置错误计数
                self.err_streak[i] = 0
                return d

            except requests.exceptions.HTTPError as e:
                if "429" in str(e):
                    self.err_streak[i] += 1
                    time.sleep(2 ** att)
                    continue
                raise

        raise RuntimeError("1688 API 多次重试仍限流，建议降低QPS或增加AppKey数量")


# ===================== 使用示例 =====================
if __name__ == "__main__":
    client = Guarded1688Client(APP_KEYS_SECRETS)

    # 查商品详情（免费接口，受QPS限制）
    try:
        result = client.safe_call(
            "alibaba.item.get",
            {"item_id": "612345678901", "fields": "title,price,sku_list,amount_on_sale"},
            token=None  # 商品公开字段可不传token
        )
        print("✅ 1688商品数据:", result.get("alibaba_item_get_response", {}).get("result", {}).get("title"))
    except Exception as e:
        print("❌", e)

关键设计说明

多Key轮询：通过 Guarded1688Client 在多个AppKey间分发请求，理论上N个Key可将总QPS提升至 N × 10。
令牌桶限速：每个Key独立令牌桶，强制单Key ≤ 8~9 QPS，避免触碰官方阈值被拉黑。
限流自愈：捕获 FLOW_CONTROL / 429 后对该Key做指数退避冷却，并自动切换下一个Key重试。
合规提醒：多Key必须是同一企业主体下合法创建的不同应用，用于分散合法业务流量，非绕过封禁。

如果日调用量极大（>50万次）或对实时性要求高，首选买官方资源包提单Key QPS至50/100，成本远低于维护多应用复杂度；多Key方案更适合初创期零成本扛中等并发的场景。

如果你愿意，我可以帮你把这套 GuardedClient 改成 Redis中心化令牌桶（支持分布式多进程调度），或者补一份 1688订单增量同步（alibaba.trade.buyer.list）的完整断点续跑脚本 吗？

🔄《京东JOS宙斯迁移实录：jingdong.ware.get → jingdong.ware.read.get 平滑切（2026版）》（附Python源码）

Fri, 17 Jul 2026 09:41:16 +0800

🔄《京东JOS宙斯迁移实录：`jingdong.ware.get` → `jingdong.ware.read.get` 平滑切（2026版）》（附Python源码）

京东宙斯（JOS）在2024~2026年间持续推进接口标准化，老接口jingdong.ware.get（及别名360buy.ware.get）已进入维护期，新应用不再支持申请，官方推荐统一迁移至jingdong.ware.read.get（只读商品读取接口，权限更清晰、支持SKU快照字段）。

迁移只需改method + 微调fields，网关/签名/AccessToken完全不变，下面给你一份可直接执行的平滑切换实录。

一、老 vs 新接口对照（一目了然）

维度	老接口（即将下线）	新接口（推荐）
Method	`jingdong.ware.get` / `360buy.ware.get`	✅ `jingdong.ware.read.get`
网关	`https://api.jd.com/routerjson`	完全相同
签名算法	MD5(AppSecret+KV_ASCII+AppSecret) 秒级timestamp	完全相同（一行不改）
必传参数	`wareId`, `fields`	完全相同（`wareId`, `fields`）
返回根节点	`jingdong_ware_get_response.ware`	`jingdong_ware_read_get_response.ware`
SKU库存可见	需传seller token	相同（传seller token可看`stockNum` in `skus_json`）
权限申请	老入口（已关闭新申请）	应用→接口权限→申请`jingdong.ware.read.get`（即时生效）

✅ 迁移量：代码中只改 method 字符串 + fields 补全 skus_json,stock_num，其余（Client/签名/网关/token）一行不动。

二、Python：老接口 vs 新接口调用对比 + 兼容封装

# jd_ware_migration_2026.py
"""
京东JOS宙斯迁移实录
old: jingdong.ware.get  →  new: jingdong.ware.read.get
同一个 JdClient 复用，仅 method + fields 不同
支持 沙箱/生产 切换
"""
import hashlib
import json
import requests
import time
from typing import Dict, Optional


class JdWareClient:
    """京东JOS商品接口客户端（兼容新旧两套）"""
    GW_PROD = "https://api.jd.com/routerjson"
    GW_SANDBOX = "https://api.sandbox.jd.com/routerjson"

    def __init__(self, app_key: str, app_secret: str, sandbox: bool = False):
        self.ak = app_key
        self.ask = app_secret
        self.gw = self.GW_SANDBOX if sandbox else self.GW_PROD

    # ───── JOS MD5签名（秒级timestamp，紧凑JSON）─────
    def _sign(self, params: Dict) -> str:
        filt = sorted(
            (k, v) for k, v in params.items()
            if k != "sign" and v is not None and str(v).strip() != ""
        )
        qs = "".join(f"{k}{v}" for k, v in filt)
        raw = f"{self.ask}{qs}{self.ask}"
        return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()

    def _call(self, method: str, biz: Dict, token: str = None) -> Dict:
        p = {
            "app_key": self.ak,
            "method": method,
            "timestamp": str(int(time.time())),   # ← 秒级！不是毫秒
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
            "360buy_param_json": json.dumps(biz, ensure_ascii=False,
                                             separators=(',', ':'))
        }
        if token:
            p["access_token"] = token
        p["sign"] = self._sign(p)

        r = requests.post(self.gw, data=p, timeout=15)
        r.raise_for_status()
        d = r.json()

        # 提取响应（兼容新旧method对应的resp_key）
        resp_key = method.replace(".", "_") + "_response"
        if resp_key not in d:
            for k in d:
                if k.endswith("_response"):
                    resp_key = k
                    break
        data = d.get(resp_key, d)

        if "error_response" in str(data):
            err = d.get(resp_key, {}).get("error_response") or d.get("error_response")
            if err:
                code = str(err.get("code", ""))
                zh = err.get("zh_desc") or err.get("en_desc")
                sub = err.get("sub_code", "")
                if "no permission" in zh:
                    raise PermissionError(
                        f"❌ 无权限：确认已申请 {method} 接口，订单类需传卖家AccessToken\n原始: {zh}"
                    )
                raise Exception(f"JOS[{code}]: {zh} sub={sub}")
            raise Exception(f"JOS error: {d}")
        return data

    # ─── 老接口（不推荐，仅作对比/过渡期兼容）───
    def get_ware_old(self, ware_id: str,
                     fields: str = None, token: str = None) -> Dict:
        fields = fields or (
            "ware_id,title,price,jd_price,stock_num,skus_json,image_path"
        )
        return self._call(
            "jingdong.ware.get",          # ← 老 method
            {"wareId": ware_id, "fields": fields},
            token
        ).get("ware", {})

    # ─── ✅ 新接口（推荐迁移目标）───
    def get_ware_new(self, ware_id: str,
                     fields: str = None, token: str = None) -> Dict:
        fields = fields or (
            "ware_id,title,brand_name,cid,category_name,"
            "price,jd_price,market_price,stock_num,"
            "skus_json,image_path,desc,outer_id,approve_status"
        )
        return self._call(
            "jingdong.ware.read.get",     # ← 新 method
            {"wareId": ware_id, "fields": fields},
            token
        ).get("ware", {})


# =========================================================
# 使用示例：平滑切换
# =========================================================
if __name__ == "__main__":
    client = JdWareClient(
        app_key="YOUR_JD_APP_KEY",
        app_secret="YOUR_JD_APP_SECRET",
        sandbox=False          # 沙箱测试 = True
    )
    WARE_ID = "100012345678"   # ← 替换真实 wareId
    SELLER_TOKEN = None        # 查自己店铺SKU库存需传卖家AccessToken

    try:
        # ★ 迁移后只用这一行 ★
        ware = client.get_ware_new(WARE_ID, token=SELLER_TOKEN)

        print("✅ [新接口] 商品标题:", ware.get("title"))
        print("   京东价:", ware.get("jd_price"))
        print("   SPU库存:", ware.get("stock_num"))

        # skus_json 解析（查自己店铺时才有真实库存）
        import json as _j
        skus_raw = ware.get("skus_json") or ""
        if skus_raw:
            skus = _j.loads(skus_raw)
            print(f"   SKU数: {len(skus)}")
            for s in skus[:3]:
                print(f"     skuId:{s.get('skuId')} 价:{s.get('jdPrice')} "
                      f"库:{s.get('stockNum')} 商家码:{s.get('outerId','')}")

        print("\n→ 迁移完成！仅需改 method='jingdong.ware.read.get' + fields补skus_json")

    except PermissionError as pe:
        print(pe)
    except Exception as e:
        print("❌", e)
        print("→ 确认: AppKey/Secret正确、接口已申请、wareId有效")

三、迁移 Checklist（照着做，5分钟切完）

应用后台申请新权限
→ 接口权限 → 勾选 jingdong.ware.read.get → 提交（通常即时生效）

代码改动（极小）

- method = "jingdong.ware.get"
+ method = "jingdong.ware.read.get"
- fields = "ware_id,title,price"
+ fields = "ware_id,title,price,jd_price,stock_num,skus_json,image_path"

wareId 参数名不变（驼峰，同老接口）

回归测试（5分钟）

沙箱调通签名 ✓
生产比对新旧返回字段一致性（标题/价格/SKU数应一致）
自家店铺查 SKU 库存需传 access_token（同旧逻辑）

下线老调用
确认所有环境切新 method 后，删除 get_ware_old() 分支

四、避坑清单

坑	现象	解决
新接口返回 403	未申请 `jingdong.ware.read.get`	应用→接口权限→申请
`skus_json` 空	公开查询别人商品 / 未传 token	查自己店铺商品须传卖家 AccessToken
老代码 `wareId` 写成 `ware_id`	部分SDK自动映射，推荐用 `wareId`(首字母大写)	与文档保持一致
返回结构 key 变	`jingdong_ware_get_response` → `jingdong_ware_read_get_response`	解析用 `.get("ware")` 不依赖外层 key
timestamp毫秒	`Invalid Timestamp`	JOS用秒级 `int(time.time())`
沙箱返回 mock 数据	正常，沙箱只验签不返真数据	切生产验证

五、面试/方案一句话

京东JOS宙斯迁移：老接口 jingdong.ware.get → 新接口 jingdong.ware.read.get，只改 method 字符串 + fields 补 skus_json,stock_num，网关/签名/AccessToken 完全不变；应用需申请新接口权限，返回值结构外层key变但内层 ware 节点字段一致，解析取 .get("ware") 即兼容，5分钟平滑切换。

需要我补 京东订单增量同步(jingdong.pop.order.search) 或库存快照定时校正脚本 的完整迁移示例吗？

🏯《淘宝TOP API聚石塔内 vs 聚石塔外价差10倍：ERP迁云到底值不值？》（附Python源码）

Fri, 17 Jul 2026 08:51:25 +0800

结论先行：淘宝TOP基础API收费差距确实是10倍（聚石塔内¥0.02/百次 vs 聚石塔外¥0.20/百次），增值API差距更是达10倍（内¥0.06 vs 外¥0.60且外禁止调用）。

对于日调用量超免费额度（企业通常数万~百万/天）的ERP，迁云不仅值，而且是必选项；但对于极小规模自用（日调用<1万次），外网费用一年也就几十块，迁云的ECS成本反而更高。

一、价差账本：10倍差在哪？

维度	聚石塔内（ECS/RDS）	聚石塔外（自建/IDC/公网）
基础API单价	¥0.02 / 百次	¥0.20 / 百次
增值API单价	¥0.06 / 百次	¥0.60 / 百次（且禁止调用）
网络链路	阿里内网（低延迟、免公网带宽费）	公网出口（高延迟、有带宽成本）
隐性成本	云主机租赁费（约¥100~数千/月）	服务器折旧、运维人力、安全风控（御城河）
稳定性	内网互通，抗大促限流更强	易触发跨网限流（isv.rate-limit-exceed）

算一笔账：

假设你的ERP日调用订单接口 50万次（约8333次/小时，中等卖家）：

塔内：5000百次 × ¥0.02 = ¥100 / 天
塔外：5000百次 × ¥0.20 = ¥1000 / 天
月差：(1000-100)×30 = ¥27,000。这还没算塔外可能被限流导致的漏单损失。

二、ERP迁云值不值？决策矩阵

✅ 强烈建议迁（ROI极高）

ISV服务商：服务多家店铺，调用量百万级/天，API费省下的钱远超ECS租金。
中大型卖家（日单>3000）：超免费额度后按量计费，内网调用省10倍，且大促弹性扩容不怕崩。
需敏感数据：买家手机号、地址解密等增值API强制只能在塔内调用，塔外根本拿不到。

❌ 暂时不必迁（没必要）

微型卖家/个人自用：日调用<1万次，还在免费额度内（¥0），迁云反而多花ECS钱。
纯商品爬取（公开字段）：taobao.item.get 等免费接口未超量时不涉及计费，无需为了省不存在的钱迁云。

三、附：Python 探测当前调用环境成本预警脚本

这个脚本模拟统计你的日调用量，并自动计算塔内/塔外月费差额，当差额覆盖ECS成本时给出迁云预警。

# top_just_cost_guard.py
"""
淘宝TOP API 聚石塔内 vs 塔外 成本测算器
逻辑：
1. 记录日调用量（实际应从Redis/DB读，这里模拟输入）
2. 扣除免费额度（假设企业版10万次/天免费，仅示意）
3. 计算超额部分的塔内/塔外费用
4. 对比ECS最低成本（假设小型ECS ¥150/月），给出迁云建议
"""
import sys
封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ===== 配置（按2026淘宝官方费率）=====
FREE_DAILY_QUOTA = 100_000        # 企业应用假设免费额度 10万次/天（示意）
BASE_IN_PRICE = 0.02              # 塔内 元/百次
BASE_OUT_PRICE = 0.20             # 塔外 元/百次
VALUE_IN_PRICE = 0.06             # 增值塔内
VALUE_OUT_PRICE = 0.60            # 增值塔外（禁）
ECS_MONTH_COST = 150              # 聚石塔最小ECS+RDS估算成本/月
DAYS_PER_MONTH = 30

def calc_monthly_cost(daily_calls: int, api_ratio=1.0, value_ratio=0.0):
    """
    daily_calls: 日均总调用次数
    api_ratio: 其中基础收费API占比 (0~1)
    value_ratio: 其中增值API占比 (0~1)
    """
    # 1. 计算超额次数（免费额度按天扣，超了才计费）
    over_daily = max(0, daily_calls - FREE_DAILY_QUOTA)
    over_monthly = over_daily * DAYS_PER_MONTH
    over_hundred = over_monthly / 100.0

    # 2. 塔内费用
    cost_in = over_hundred * (BASE_IN_PRICE * api_ratio + VALUE_IN_PRICE * value_ratio)
    # 3. 塔外费用（增值禁止，这里仅测算理论价差）
    cost_out = over_hundred * (BASE_OUT_PRICE * api_ratio + VALUE_OUT_PRICE * value_ratio)

    return cost_in, cost_out, over_monthly

def advise(daily_calls):
    cost_in, cost_out, over_m = calc_monthly_cost(daily_calls, api_ratio=0.8, value_ratio=0.2)
    
    print(f"📊 日均调用: {daily_calls:,} 次 | 月超额: {over_m:,.0f} 次")
    print(f"💰 聚石塔内月费(超量): ¥{cost_in:.2f}")
    print(f"💸 聚石塔外月费(超量): ¥{cost_out:._2f} (价差 ¥{cost_out-cost_in:.2f})")
    
    diff = cost_out - cost_in
    
    if daily_calls <= FREE_DAILY_QUOTA:
        print("✅ 当前在免费额度内，塔外零费，暂无需迁云")
    elif diff > ECS_MONTH_COST:
        print(f"🚨 **建议迁云**：API省下的 ¥{diff:.0f}/月 > ECS成本 ¥{ECS_MONTH_COST}，净赚 ¥{diff-ECS_MONTH_COST:.0f}/月")
        print("   + 额外收益：增值API可用、内网稳、御城河安全、弹性扩容")
    else:
        print(f"⚠️ 当前差额 ¥{diff:.0f}/月 < ECS成本 ¥{ECS_MONTH_COST}，微小规模可暂缓，但需承担限流/安全风险")

if __name__ == "__main__":
    # 模拟几个量级
    for d_calls in [5_000, 50_000, 100_000, 500_000, 1_000_000]:
        print("-" * 50)
        advise(d_calls)

运行输出逻辑示例（日50万次）：

月超额约1200万次
塔内超量费 ≈ ¥2400/月
塔外超量费 ≈ ¥24000/月
差额¥21600 > ECS¥150 → 强烈建议迁云

四、迁云不止是省钱，还有3个隐形红利

数据安全（御城河）：塔内有完整的主机安全、访问控制，防止员工离职拖库或黑客爬取买家隐私（违规会被罚巨额）。
数据推送（RDS）：塔内可以用订单数据同步服务（基于RDS Binlog/推送），比硬轮询API稳定10倍，还能规避429限流。
大促弹性：双11流量翻10倍，塔内ECS可分钟级扩容，塔外自建机房容易直接被打挂。

一句话总结：

只要你的ERP日调用超过免费额度（约10万+）或需要增值/解密数据，聚石塔内比塔外省10倍API费+送安全与稳定，迁云必值；只有微型自用且在免费额度内才建议苟在塔外。

要不要我帮你把这个成本测算脚本改成读取你真实Redis日调用计数，并自动对接淘宝error_response里的计费码做实时预警呢？

《别再被"免费API"误导：五家开放平台真实计费口径拆解（附决策清单）》（附python源码）

Thu, 16 Jul 2026 09:22:58 +0800

市面上常说“淘宝/京东/1688 API 免费”，但真实计费口径里藏着三层隐形门槛：免费额度上限、超额按量计费、增值/高级接口必须买包。一旦日调用量稍大或碰了敏感数据，账单就会悄悄产生。下面把五大平台（淘宝TOP / 京东JOS / 1688 / 拼多多 / 抖店）的真实计费口径拆开，并给你一份决策清单 + 用量防超脚本。

一、五大平台真实计费口径拆解（2026版）

1. 淘宝 TOP 开放平台

免费层：商品查询、基础订单、物流轨迹、淘宝客有日免费额度（企业≈5万~100万次/天，个人≈5000~1万），QPS 免费≈5~20/s。
基础收费层：taobao.trades.sold.get等订单类超免费额度后计费，聚石塔内≈¥0.02/百次，聚石塔外≈¥0.20/百次。
增值层：数据罗盘、CRM高级(RFM)、竞品分析需购增值包，聚石塔内≈¥0.06/百次，且禁止公网调用。
误区：“订单接口完全免费” → 错，超日额度立刻按量扣费；“个人号能查订单” → 错，需企业+卖家 OAuth。

2. 京东 JOS（宙斯）

免费层：商品查询、联盟选品、基础物流有免费额度（企业≈30万~100万次/天，个人≈500~2000），QPS≈5~10/s（可购包提至50/100）。
基础收费层：高频订单同步、电子面单取号超量按基础计费≈¥0.02~0.10/百次（聚石塔内远低于公网）。
增值层：会员RFM、数据洞察、无界电子面单按面单/包计费。
误区：“JOS 全免费” → 错，日调用超几十万或提 QPS 必涉及资源包；“联盟接口能看库存” → 错，联盟只返回有货/无货状态。

3. 1688 开放平台

免费层：商品搜索、详情、订单、物流完全免费，受 QPS 限制（搜索≈10/s，订单≈20/s），无明确硬顶但异常高频触发风控。
资源包层：提升 QPS、高级实时库存、跨境洞察、数据推送需买年费资源包（几百~几千/年）。
按量层：超资源包后≈¥0.001~0.01/次，极少触发。
误区：“1688 没有任何收费” → 错，高级实时库存 / 提频才收费，普通同步零成本；“日限很低” → 错，企业自用型几十万次/天实测零扣费。

四、拼多多开放平台

无完全免费：基础接口有小额免费额度，超额按调用量计费（基础≈¥0.01~0.10/百次），且涉及敏感/加密数据必须在拼多多云内调用（云主机年费约¥2460起）。
部分场景还有订单金额技术服务费（如0.6%）。
误区：“拼多多 API 和个人号一样免费” → 错，自研对接基本都要预充值或云费用。

五、抖店（抖音）开放平台

免费层：自研店铺应用有基础免费额度。
收费层：

基础 API：云内≈¥0.018/百次，云外≈¥0.18/百次；
增值 API：云内≈¥0.05/百次，云外≈¥0.5/百次（增值禁止云外调用）。

ISV 上架还可能抽成 10~30%。
误区：“抖店接口随便调不花钱” → 错，超额度或走云外成本陡增 10 倍。

二、别被“免费API”误导的 5 个核心真相

“免费”通常指“基础接口 + 日额度内 + 云内/聚石塔内”，离开这三个条件就可能计费。
订单 / 库存 / 用户隐私类接口，淘宝京东1688 都放在基础收费或增值层，不是永久免费。
公网调用比云内贵 10 倍（淘宝聚石塔外 ¥0.20 vs ¥0.02；抖店云外 ¥0.18 vs ¥0.018）。
QPS 超限只返回限流码（code=7 / 16），不扣费但业务中断，提频就要买包。
淘宝客 / 京东联盟本身接口免费，但成交后抽佣金，不是“零成本”。

三、决策清单：你的场景用哪家、会不会花钱？

你的场景	推荐平台	是否会花钱	关键动作
中小卖家商品同步+订单回写	淘宝TOP / 京东JOS / 1688	大概率零费（日<10万次）	用增量 modified、fields 过滤、聚石塔/云内调用
B2B批发找货、分销管理	1688	零费（不碰高级库存）	企业实名+自用型应用，别手滑买资源包
选品比价、CPS推广	淘宝客 / 京东联盟	接口零费（成交抽佣）	备案推广位，不调订单私有接口
高频大卖（>50万次/天）	淘宝/京东/抖店	超量计费（几分钱/百次）	提前算日调用量+买 QPS 包+云内部署
敏感数据（手机号/加密地址）	拼多多 / 抖店增值	必花钱（云费用+增值费）	评估是否必须拿明文，能用脱敏就用脱敏

四、附：通用 API 用量探测 + 防超阈值脚本（Python）

不管哪家，都建议在代码里埋一个日计数器 + 限流保护，快触免费额度时主动停任务，避免莫名其妙被扣费或限流。

# api_cost_guard.py
"""
五家电商通用：日调用计数 + 令牌桶限速 + 超额度抛异常
用法：每次真实请求前先 guard.check_and_wait()
"""
import time
from datetime import datetime
封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class ApiCostGuard:
    def __init__(self, app_key: str, daily_limit: int = 100_000, qps: float = 4.0):
        self.app_key = app_key
        self.daily_limit = daily_limit
        self.count = 0
        self.qps = qps
        self.tokens = qps
        self.ts = time.monotonic()

    def _bucket_wait(self):
        now = time.monotonic()
        self.tokens = min(self.qps, self.tokens + (now - self.ts) * self.qps)
        self.ts = now
        if self.tokens < 1:
            time.sleep((1 - self.tokens) / self.qps + 0.005)
            self.tokens = 0
        else:
            self.tokens -= 1

    def check_and_wait(self):
        if self.count >= self.daily_limit:
            raise RuntimeError(
                f"🚨 今日[{self.app_key}]已达{daily_limit}次（免费额度耗尽）\n"
                f"→ 再调将触发平台按量计费\n"
                f"→ 建议：断点续跑 / 增量同步 / 申请提额"
            )
        self._bucket_wait()
        self.count += 1

    def reset_if_new_day(self):
        # 简单示例：实际可落文件/Redis按日期重置
        if not hasattr(self, "_day"):
            self._day = datetime.now().date()
        if datetime.now().date() != self._day:
            self.count = 0
            self._day = datetime.now().date()
封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex

# ===== 使用示例 =====
if __name__ == "__main__":
    guard = ApiCostGuard("TAOBAO_APP_KEY", daily_limit=100_000, qps=4)

    for i in range(3):
        guard.reset_if_new_day()
        guard.check_and_wait()
        print(f"✅ 模拟调用成功 今日:{guard.count}/{guard.daily_limit}")
        # 这里放真实 platform_client.call(...)

怎么用：把 check_and_wait()塞进你封装好的 TopClient.call()/ JdClient.call()里，超限直接抛异常停任务，比后台突然发现被扣费舒服得多。

五、一句话总结

所谓“免费API”= 基础接口 + 在日额度内 + 云内/聚石塔内调用 + 不碰增值；
淘宝TOP/京东JOS/1688 中小规模基本零费，拼多多/抖店自带按量或云费，超额度后各家都是每百次几分钱但公网贵10倍，真要规模化记得增量同步 + 令牌桶 + 云内部署 + 买QPS包，别被“免费”两个字一路冲到账单上。

需要我帮你把这套 ApiCostGuard直接嵌进你现有的 淘宝TOP / 京东JOS / 1688 Client 里，按各平台返回码（code=7/ limited-by-app-access-count）做自动退避与告警吗？

要不要我帮你把这套 ApiCostGuard直接嵌进你现有的 淘宝TOP / 京东JOS / 1688 Client 里，按各平台返回码（code=7/ limited-by-app-access-count）做自动退避与告警呢？

《2026五大电商API收费全景对照：淘宝TOP/京东JOS/1688/拼多多/抖店，哪家用得起？》（附python源码）

Thu, 16 Jul 2026 09:06:34 +0800

💰 2026五大电商API收费全景对照：淘宝TOP / 京东JOS / 1688 / 拼多多 / 抖店，哪家用得起？

做电商ERP、选品系统或自动铺货，API成本往往是隐形的大头。2026年各大平台基本都走了「基础免费额度 + 超额按量计费 + 增值/资源包收费」的路线。下面给你一张能直接拍板的对照表，再附一套通用的用量探测与限流保护Python源码。

一、五大平台API收费全景对照表（2026版）

维度	淘宝 TOP	京东 JOS	1688	拼多多	抖店（抖音）
免费额度	企业≈1万~100万次/天个人≈500~5000次/天	企业≈30万~100万次/天个人≈500~2000次/天	基础接口¥0 默认QPS≈10~20/s 日量较宽松	企业≈10万次/天个人≈2000次/天有QPS限制	自研应用有免费额度 ISV按量阶梯云内更便宜
默认QPS	企业5~50/s 个人≤2/s	企业5~10/s（可购包提至50/100）个人1~2/s	搜索≈10/s 订单≈20/s	基础接口宽松商品≈5/s 订单≈2/s	基础接口有限额云内调用更稳定
超量计费 (聚石塔/云内)	基础API：¥0.02/百次增值API：¥0.06/百次	超免费额度按量计费 ≈¥0.02~0.10/百次	超资源包≈¥0.001~0.01/次	基础API：¥0.01/百次增值API：¥0.03/百次	基础API：¥0.018/百次增值API：¥0.05/百次
超量计费 (公网/云外)	基础API：¥0.20/百增值API：¥0.60/百	不建议公网超量成本陡增	无直接按量需买包提频	基础API：¥0.10/百增值API：¥0.30/百	基础API：¥0.18/百增值API：¥0.50/百
增值/高级收费	数据罗盘、CRM高级、竞品分析需购增值包	会员RFM、数据洞察电子面单取号按单算	实时库存高级、跨境洞察需购资源包（¥千~万/年）	敏感数据/加密信息需购云主机+权限	数据推送高级、定制分析 ISV上架抽成10~30%
淘宝客/联盟	`taobao.tbk.*`免费成交后扣佣金	`jd.union.`免费备案* 成交抽佣	无对应	无	无
认证要求	企业最优（订单需OAuth）	企业实名（订单需卖家Token）	企业实名（自研免费）	企业/个人均可	企业/个体户（需资质）
哪家用得起？	中小免费够用高频超量也极便宜聚石塔内部署	中小免费够用日调用<30万次零费买QPS包才花钱	B2B最划算商品/订单全免费只有提频/高级库存才买包	预充值模式钱花在调用量+云主机适合已有拼多多店	自研店铺零费 ISV做SaaS才分成云内调用省成本

✅ 结论先给：
预算极紧 / 中小卖家：淘宝、京东、1688 的基础接口免费额度完全够用（日几千~几万次同步零成本），拼多多需预充值但单价极低，抖店自研也免费。
高频大卖 / SaaS服务商：淘宝聚石塔内、京东购QPS包、1688买资源包、拼多多/抖店走云内调用，成本都在每百次几分钱级别，远底线性硬件成本。
千万别踩的坑：增值接口（数据洞察/CRM/实时高级库存）不免费，没需求别申请；公网调用比云内/聚石塔内贵10倍。

二、通用API用量探测 + 限流保护源码（Python）

不管哪个平台，签名→调用→捕获限流→退避重试→用量计数的逻辑是通用的。下面以通用REST风格+模拟限流码为例，你可把call_api替换成对应平台的Client。

# ecom_api_cost_guard.py
"""
五大电商API通用防护壳：
1. 本地日调用计数（防不知不觉超免费额度）
2. 令牌桶限速（QPS ≤ 免费上限×0.8）
3. 捕获 429 / code=16(FLOW_CONTROL) 指数退避
4. 区分 免费层 / 计费层 报错提示
依赖: pip install requests
"""
import hashlib
import time
import requests
from datetime import datetime

封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class ApiCostGuard:
    def __init__(self, app_key, daily_limit=100_000, qps=4):
        self.app_key = app_key
        self.daily_limit = daily_limit
        self.counter_key = f"api_cnt_{app_key}_{datetime.now().date()}"
        self.today_count = 0

        # 简易令牌桶
        self.rate = qps
        self.tokens = qps
        self.ts = time.monotonic()

    def _bucket_wait(self):
        now = time.monotonic()
        self.tokens = min(self.rate, self.tokens + (now - self.ts) * self.rate)
        self.ts = now
        if self.tokens < 1:
            time.sleep((1 - self.tokens) / self.rate + 0.01)
            self.tokens = 0
        else:
            self.tokens -= 1

    def _check_daily(self):
        if self.today_count >= self.daily_limit:
            raise RuntimeError(
                f"🚨 今日调用已达 {self.daily_limit} 次（免费额度耗尽）\n"
                f"→ 超量将按平台计费（淘宝¥0.02~0.20/百次 京东≈¥0.02~0.10/百次）\n"
                f"→ 建议：断点续跑 / 增量同步 / 申请提额"
            )

    def safe_call(self, platform_name, method, biz_params, sign_func, token=None, max_retry=3):
        """
        platform_name: "淘宝TOP"/"京东JOS"/"1688"/"拼多多"/"抖店"
        sign_func: 对应平台的签名函数 (params, secret)->sign
        """
        for att in range(max_retry):
            self._check_daily()
            self._bucket_wait()

            # ===== 这里替换为真实平台Client =====
            try:
                # 模拟构造参数并签名（示意）
                params = {"method": method, "app_key": self.app_key, **biz_params}
                if token: params["access_token"] = token
                params["sign"] = sign_func(params, "APP_SECRET")

                # resp = requests.post(gateway, data=params, timeout=15)
                # d = resp.json()
                # 下面用伪响应模拟 429 / 正常 分支
                fake_status = 200  # 改成 429 测限流
                if fake_status == 429:
                    raise requests.HTTPError("429 Client Error")

                self.today_count += 1
                print(f"✅ [{platform_name}] {method} 调用成功 今日:{self.today_count}/{self.daily_limit}")

                # return d
                return {"mock": "ok"}

            except Exception as e:
                err_str = str(e)
                if "429" in err_str or "FLOW_CONTROL" in err_str or "code=16" in err_str:
                    if att < max_retry - 1:
                        back = 2 ** att
                        print(f"⚠️  {platform_name} QPS限流，{back}s后第{att+1}次重试...")
                        time.sleep(back)
                        continue
                    raise RuntimeError(
                        f"❌ {platform_name} 持续限流，建议：\n"
                        f"  1) 降低QPS（当前设{self.rate}）\n"
                        f"  2) 买资源包/提额（京东/1688/淘宝聚石塔）\n"
                        f"  3) 增量同步减少无效调用"
                    )
                raise

        raise RuntimeError("重试耗尽")

封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# -------- 示例：一个通用MD5签名（淘宝/京东风格简化） --------
def generic_md5_sign(params: dict, secret: str) -> str:
    from typing import Tuple
    filtered = sorted((k, v) for k, v in params.items()
                      if k != "sign" and v is not None and str(v).strip() != "")
    qs = "".join(f"{k}{v}" for k, v in filtered)
    raw = f"{secret}{qs}{secret}"
    return hashlib.md5(raw.encode()).hexdigest().upper()


if __name__ == "__main__":
    guard = ApiCostGuard(
        app_key="YOUR_APP_KEY",
        daily_limit=100_000,   # ← 按平台免费额度填（如淘宝企业10万）
        qps=4                 # ← 免费上限5时设4，防超限
    )

    try:
        for i in range(3):
            guard.safe_call(
                platform_name="淘宝TOP",
                method="taobao.item.get",
                biz_params={"num_iid": "654321098765", "fields": "title,price"},
                sign_func=generic_md5_sign
            )
    except RuntimeError as e:
        print(e)

三、选型与省钱建议（落地用）

只用基础商品/订单同步
→ 淘宝（企业）、京东（企业）、1688 全免费额度内搞定，别手滑申请增值接口。
日调用超50万次
→ 淘宝/京东放聚石塔/京东云内调用（价差10倍）；1688买对应资源包提QPS；拼多多/抖店尽量走各自云内通道。
做SaaS卖给别人（ISV）
→ 抖店/淘宝要算平台抽成；1688/京东按调用量+资源包；拼多多预充值最稳。
防超支第一线
→ 代码里加日计数器 + 断点续跑 + 令牌桶，超限直接抛异常停任务，比被扣几百块舒服。

一句话总结：

2026年做电商API对接，中小玩家淘宝/京东/1688基础接口基本“白嫖”，拼多多几分钱一百次，抖店自研也免；真烧钱的是乱开增值包、公网高频硬冲、不做限流保护。上面的Python防护壳直接套进你现有Client里，就能安心对着账单睡觉。

要不要我帮你把某一具体平台（比如淘宝TOP或京东JOS）的签名+令牌桶+断点续跑写成完整可落地的同步模块？

《多店铺京东订单统一拉取：JOS API + 分布式调度 + 幂等设计实战》（附python源码）

Wed, 15 Jul 2026 09:45:58 +0800

多店铺京东订单统一拉取的核心在于：每个店铺独立 AccessToken + 增量时间窗拉取 jingdong.pop.order.search+ 以 (tid, shop_id)为唯一键做数据库幂等 + 分布式锁/Redis NX 防多节点并发重复处理 + 断点续跑防漏单。下面给你一套可直接落地的 Python 实战源码。

一、整体架构

┌────────────────────┐
│ 分布式调度器         │  APScheduler / Celery Beat
│ 遍历 shop_auth 表    │
└───────┬────────────┘
        │ 每店铺独立 token
        ▼
┌────────────────────┐
│ JdOrderSyncPerShop │
│ 1. 获取分布式锁     │  Redis SET shop_id:sync  NX EX
│ 2. 增量拉取订单     │  start_modified ~ end_modified
│ 3. 逐单查明细       │  jd.order.detail.get
│ 4. 幂等入库         │  INSERT ON CONFLICT(tid,shop_id) DO NOTHING
│ 5. 更新断点时间     │  last_sync_time
│ 6. 释放锁           │
└────────────────────┘

二、核心源码（Python · 含签名 / 分布式锁 / 幂等 / 断点）

# jd_multi_shop_order_sync.py
"""
多店铺京东订单统一拉取实战
- JOS MD5签名（秒级timestamp）
- Redis分布式锁（防止多节点同时跑同一店铺）
- 增量同步 modified 时间窗
- 幂等入库 (tid, shop_id) UNIQUE
- 断点续跑（记录 last_sync_time）
依赖:
  pip install requests redis apscheduler
"""
import hashlib
import json
import time
import datetime
import requests
import redis
from typing import Dict, List

# ================== 配置 ==================
APP_KEY = "YOUR_JD_APP_KEY"
APP_SECRET = "YOUR_JD_APP_SECRET"
REDIS_HOST = "localhost"
REDIS_PORT = 6379
REDIS_DB = 0
SYNC_INTERVAL_MIN = 30          # 增量窗口分钟数
LOCK_TTL_SEC = 300               # 分布式锁过期时间
# ==========================================

rds = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)


# ---------- JOS Client ----------
class JdJosClient:
    GW = "https://api.jd.com/routerjson"

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

    def _sign(self, params: Dict) -> str:
        filt = sorted(
            (k, v) for k, v in params.items()
            if k != "sign" and v is not None and str(v).strip() != ""
        )
        qs = "".join(f"{k}{v}" for k, v in filt)
        return hashlib.md5(f"{self.ask}{qs}{self.ask}".encode()).hexdigest().upper()

    def call(self, method, biz, token):
        p = {
            "app_key": self.ak,
            "method": method,
            "timestamp": str(int(time.time())),   # 秒级！
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
            "360buy_param_json": json.dumps(biz, ensure_ascii=False,
                                             separators=(',', ':')),
            "access_token": token
        }
        p["sign"] = self._sign(p)
        r = requests.post(self.GW, data=p, timeout=15)
        r.raise_for_status()
        d = r.json()

        resp_key = method.replace(".", "_") + "_response"
        if resp_key not in d:
            for k in d:
                if k.endswith("_response"):
                    resp_key = k
                    break
        data = d.get(resp_key, d)
        if "error_response" in str(data):
            err = d.get(resp_key, {}).get("error_response") or d.get("error_response")
            if err:
                raise Exception(f"JOS[{err.get('code')}]: {err.get('zh_desc')}")
        return data

    def list_orders(self, token, start, end, page=1, page_size=100):
        return self.call("jingdong.pop.order.search", {
            "start_modified": start,
            "end_modified": end,
            "order_state": "WAIT_SELLER_STOCK_OUT,WAIT_BUYER_CONFIRM_GOODS,FINISHED",
            "page": page,
            "page_size": min(page_size, 100)
        }, token)

    def order_detail(self, token, order_id):
        return self.call("jd.order.detail.get", {
            "orderId": str(order_id),
            "fields": "orderId,status,payment,buyer_nick,receiver_name,"
                      "receiver_mobile,orders.skuId,orders.outerSkuId,"
                      "orders.num,orders.price,created,modified"
        }, token)


# ---------- 分布式锁 ----------
def acquire_lock(shop_id: str) -> str:
    """Redis SET NX EX，返回 identifier 或 None"""
    identifier = str(time.time())
    lock_key = f"jd_sync_lock:{shop_id}"
    ok = rds.set(lock_key, identifier, nx=True, ex=LOCK_TTL_SEC)
    return identifier if ok else None


def release_lock(shop_id: str, identifier: str):
    lock_key = f"jd_sync_lock:{shop_id}"
    # Lua等效：仅当值匹配才删，防误释放
    with rds.pipeline() as pipe:
        while True:
            try:
                pipe.watch(lock_key)
                if pipe.get(lock_key) == identifier.encode():
                    pipe.multi()
                    pipe.delete(lock_key)
                    pipe.execute()
                break
            except redis.WatchError:
                continue


# ---------- 断点 & 幂等（SQLite示例，生产换 MySQL/PG） ----------
def init_db(db="jd_erp.db"):
    conn = __import__("sqlite3").connect(db)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS erp_order(
            tid TEXT,
            shop_id TEXT,
            status TEXT,
            payment REAL,
            buyer_nick TEXT,
            created TEXT,
            PRIMARY KEY(tid, shop_id)
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS shop_sync_point(
            shop_id TEXT PRIMARY KEY,
            last_sync_time TEXT
        )
    """)
    conn.commit()
    return conn


def load_last_time(conn, shop_id):
    cur = conn.execute("SELECT last_sync_time FROM shop_sync_point WHERE shop_id=?",
                       (shop_id,))
    row = cur.fetchone()
    return row[0] if row else None


def save_last_time(conn, shop_id, t):
    conn.execute("""
        INSERT INTO shop_sync_point(shop_id,last_sync_time)
        VALUES(?,?) ON CONFLICT(shop_id) DO UPDATE SET last_sync_time=excluded.last_sync_time
    """, (shop_id, t))
    conn.commit()


def upsert_order(conn, shop_id, order):
    conn.execute("""
        INSERT INTO erp_order(tid,shop_id,status,payment,buyer_nick,created)
        VALUES(?,?,?,?,?,?)
        ON CONFLICT(tid,shop_id) DO NOTHING
    """, (
        str(order.get("orderId")),
        shop_id,
        order.get("status"),
        float(order.get("payment") or 0),
        order.get("buyer_nick", ""),
        order.get("created", "")
    ))
    conn.commit()


# ---------- 单店铺同步 ----------
def sync_one_shop(client: JdJosClient, conn, shop: Dict):
    shop_id = shop["shop_id"]
    token = shop["access_token"]

    lock_id = acquire_lock(shop_id)
    if not lock_id:
        print(f"⚠️  店铺[{shop_id}] 正在其他节点同步，跳过")
        return

    try:
        now = datetime.datetime.now()
        last = load_last_time(conn, shop_id)
        if last:
            start = last
        else:
            start = (now - datetime.timedelta(minutes=SYNC_INTERVAL_MIN * 2)
                    ).strftime("%Y-%m-%d %H:%M:%S")
        end = now.strftime("%Y-%m-%d %H:%M:%S")

        print(f"▶ 店铺[{shop_id}] 增量 {start} ~ {end}")

        page = 1
        while True:
            rsp = client.list_orders(token, start, end, page=page)
            order_list = (rsp.get("popOrderSearch", {})
                          .get("orderSearch", {})
                          .get("orderInfoList", [])) or []
            total = rsp.get("popOrderSearch", {}).get("orderSearch", {}
                                                      ).get("orderTotal", 0)

            for o in order_list:
                tid = str(o.get("orderId"))
                # 明细（按需，列表已有基础字段可先upsert）
                detail = client.order_detail(token, tid)
                order_data = detail.get("orderInfo", o)
                upsert_order(conn, shop_id, order_data)

            if len(order_list) < 100 or page * 100 >= total:
                break
            page += 1
            time.sleep(0.15)   # QPS保护

        # 成功更新断点（用当前end，下次从此继续）
        save_last_time(conn, shop_id, end)
        print(f"✅ 店铺[{shop_id}] 同步完成，断点更新至 {end}")

    except Exception as e:
        print(f"❌ 店铺[{shop_id}] 同步异常: {e}")
    finally:
        release_lock(shop_id, lock_id)


# ---------- 多店铺入口 ----------
def sync_all_shops(shop_list: List[Dict]):
    client = JdJosClient(APP_KEY, APP_SECRET)
    conn = init_db()
    for shop in shop_list:
        try:
            sync_one_shop(client, conn, shop)
        except Exception as e:
            print(f"❌ 顶层异常 shop={shop.get('shop_id')}: {e}")


# ===================== 示例 shop_list =====================
if __name__ == "__main__":
    SHOP_LIST = [
        {"shop_id": "SHOP001", "access_token": "SELLER_TOKEN_A"},
        {"shop_id": "SHOP002", "access_token": "SELLER_TOKEN_B"},
    ]
    sync_all_shops(SHOP_LIST)

    # ---- 常驻调度示例（取消注释）----
    # from apscheduler.schedulers.blocking import BlockingScheduler
    # sched = BlockingScheduler()
    # sched.add_job(lambda: sync_all_shops(SHOP_LIST),
    #               'interval', minutes=SYNC_INTERVAL_MIN, id='jd_multi_shop_sync')
    # sched.start()

三、关键设计点说明

多店铺隔离
每个店铺一条 shop_auth记录，存 shop_id / access_token / refresh_token，同步时按 shop_id独立跑、独立锁、独立断点时间。
分布式锁（Redis SET NX EX）
防止多节点 / 多进程同时跑同一店铺导致重复调用API、重复入库。
幂等设计
本地订单表 PRIMARY KEY (tid, shop_id)，处理前数据库层 ON CONFLICT DO NOTHING，即使重试/重复推送也不重复建单。
增量 + 断点续跑
记录 last_sync_time，每次用 start_modified=last_sync_time, end_modified=now拉取，服务重启从中继续，避免漏单与超日额度。
QPS保护
分页间 time.sleep(0.15)+ 令牌桶（前文京东限流篇已给），遇 code=16 / ISP_FLOW_CONTROL指数退避重试。
对账兜底
每日凌晨全量比对一次最近 N 天订单，修正极端情况下漏掉的状态变更（回调丢失/网络抖动）。

四、避坑清单

坑	现象	解决
用毫秒时间戳	`Invalid Timestamp`	JOS用秒级 `int(time.time())`
订单403	传买家token / 个人应用	企业认证 + 卖家OAuth AccessToken
多节点重复入库	无幂等/无锁	`(tid,shop_id)`唯一索引 + Redis锁
重启后重复拉全量	无断点记录	持久化 `last_sync_time`
超日额度	`limited-by-app-access-count`	增量同步 + 断点，避免全量翻页

一句话总结：多店铺京东订单同步 = 每店铺独立 Seller Token → 增量 modified 时间窗 → Redis NX 锁防并发 → (tid,shop_id)幂等入库 → 记录 last_sync_time 断点续跑 → 日对账兜底，签名/QPS 同单店铺，仅调度与隔离维度不同。

《京东联盟API爬坑：不用店铺授权也能拿商品详情，但库存只有“有货/无货”》（附Python源码）

Wed, 15 Jul 2026 09:31:28 +0800

这是京东开放体系里最容易混淆、也最容易踩坑的点之一：

京东联盟API（jd.union.*）≠ 京东开放平台JOS（商家ERP接口），两者在数据深度与授权要求上完全不同。

一、核心结论先给

京东联盟API（jd.union.open.goods.detail.query）
✅ 不需要店铺授权，只要京东联盟实名应用 + AppKey/Secret 就能调
❌ 库存只返回状态（有货 / 无货 / 预售），不返回具体可售库存数量
✅ 适合：选品、比价、佣金监控、CPS推广、价格追踪
京东开放平台JOS（jingdong.ware.read.get/ jingdong.stock.get）
✅ 需要企业认证 + 店铺卖家OAuth授权
✅ 返回真实SKU库存数量、区域库存、锁定库存
✅ 适合：ERP商品同步、防超卖、WMS出入库、店铺自营管理

如果你做电商ERP、库存预占、发货回写，联盟接口库存不够用，必须走JOS商家接口。

二、京东联盟接口“坑点”拆解

坑序号	现象	真相
①	以为不用授权就能拿全量数据	联盟无店铺权限，看不到真实库存、成本价、商家编码
②	`skuList[].stock`要么没有，要么是1/0	联盟返回的是 stockState（1=有货 / 0=无货），不是 `stockNum`
③	价格字段和京东前台不一致	联盟返回促销价/券后价/佣金价，不是商家后台“基准库存价”
④	调订单/物流接口报403	联盟只有商品/推广/订单概览，没有 `jd.order.detail.get`这种商家接口
⑤	以为AppKey通用	联盟应用 ≠ JOS应用，网关不同，权限体系隔离

三、Python：京东联盟商品详情调用（含库存状态解析）

# jd_union_detail_demo.py
"""
京东联盟API 商品详情查询
jd.union.open.goods.detail.query
- 无需店铺授权
- 库存仅返回 有货(1) / 无货(0) / 预售 状态
依赖: requests
"""
import hashlib
import json
import time
import requests
from typing import Dict, List


class JdUnionClient:
    GW = "https://api.jd.com/routerjson"

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

    # ───── JOS/联盟通用 MD5签名（秒级timestamp）─────
    def _sign(self, params: Dict) -> str:
        filt = sorted(
            (k, v) for k, v in params.items()
            if k != "sign" and v is not None and str(v).strip() != ""
        )
        qs = "".join(f"{k}{v}" for k, v in filt)
        raw = f"{self.ask}{qs}{self.ask}"
        return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()

    def get_goods_detail(self, sku_ids: List[int], fields: str = None) -> Dict:
        """
        sku_ids: 最多20个SKU ID
        fields 建议: skuId,productName,price,promotionPrice,couponInfo,
                      stockState,commissionInfo,salesCount,shopName
        """
        fields = fields or (
            "skuId,productName,price,promotionPrice,couponInfo,"
            "stockState,commissionInfo,salesCount,shopName,logisticsType"
        )
        biz = {"skuIds": sku_ids, "fields": fields}

        params = {
            "app_key": self.ak,
            "method": "jd.union.open.goods.detail.query",
            "timestamp": str(int(time.time())),   # ← 秒级！
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
            "360buy_param_json": json.dumps(biz, ensure_ascii=False,
                                             separators=(',', ':'))
        }
        params["sign"] = self._sign(params)

        r = requests.post(self.GW, data=params, timeout=15)
        r.raise_for_status()
        d = r.json()

        resp_key = "jd_union_open_goods_detail_query_response"
        data = d.get(resp_key, {})
        if "error_response" in str(data):
            err = data.get("error_response")
            raise Exception(f"Union API Err: {err.get('code')} {err.get('zh_desc')}")
        return data

    # ───── 解析库存状态 ─────
    def parse_stock(self, item: Dict) -> str:
        """
        联盟接口库存只有状态，不是数量
        stockState: 1=有货, 0=无货, 可能含预售标识
        """
        state = item.get("stockState")
        if state == 1:
            return "有货"
        elif state == 0:
            = "无货"
        else:
            return f"未知状态({state})"

        # 有些返回在 skuList[].stock 用 1/0 表示
        skus = item.get("skuList", [])
        if skus:
            print("   SKU库存示意:", [(s.get("skuId"), s.get("stock", "N/A"))
                                       for s in skus[:3]])


# =========================================================
# 使用示例
# =========================================================
if __name__ == "__main__":
    client = JdUnionClient("YOUR_JD_APP_KEY", "YOUR_JD_APP_SECRET")
    SKU_IDS = [100012345678, 100087654321]   # ← 替换真实SKU ID

    try:
        result = client.get_goods_detail(SKU_IDS)
        data_list = result.get("data", [])
        if isinstance(data_list, dict):
            data_list = data_list.get("data") or []

        for item in (data_list or []):
            title = item.get("productName")
            price = item.get("promotionPrice") or item.get("price")
            stock_flag = client.parse_stock(item)
            commission = item.get("commissionInfo", {}).get("commission", "N/A")

            print(f"✅ SKU:{item.get('skuId')}")
            print(f"   标题:{title}")
            print(f"   促销价:¥{price}")
            print(f"   库存状态:{stock_flag}  ← 仅状态，无具体库存数")
            print(f"   预估佣金:{commission}")
            client.parse_stock(item)  # 打印sku示意

    except Exception as e:
        print("❌", e)
        print("→ 确认: 京东联盟应用已实名、jd.union.open.goods.detail.query 已申请")

四、如果你必须要真实库存数量，怎么办？

走 京东JOS商家接口（需企业+店铺授权）：

接口	能力
`jingdong.ware.sku.read`/ `jingdong.ware.stock.get`	返回 `stockNum`（可售库存）、`lockStock`（锁定）、`hsStockNum`（在途）
`jingdong.stock.get`	多仓库存、区域库存
需 `access_token`	卖家OAuth2授权（不是联盟token）

这部分代码在之前《京东商品详情API（jingdong.item.read.get）》《京东库存API（jd.stock.get）实时同步方案》中已经给过完整Python封装。

五、选型一句话总结（面试/方案）

京东联盟API（jd.union.open.goods.detail.query）不用店铺授权就能拿商品标题/价格/券/佣金/销量，但库存只有“有货/无货”状态；要做ERP库存同步、防超卖、发货回写，必须用京东开放平台JOS商家接口（jingdong.ware.read.get/ jd.stock.get）并走企业认证+卖家OAuth授权获取真实stockNum。

需要我帮你把“联盟选品监控 + JOS商家库存校正”写成一套双通道同步Demo（Python）吗？

《京东JOS vs 淘宝TOP vs 1688：接口免费额度、QPS、收费全方位对比》（附python源码）

Tue, 14 Jul 2026 10:04:33 +0800

这是一份基于当前（2026年）三大主流电商平台开放平台政策的核心维度对比。由于各平台费率与额度会随版本迭代调整，以下数据综合了官方文档与近期开发者实测，重点突出差异与选型痛点。

京东JOS vs 淘宝TOP vs 1688 全方位对比

对比维度	京东JOS (宙斯)	淘宝TOP (开放平台)	1688 (阿里巴巴中国站)
适用场景	自营/POP订单履约、实时库存、价格同步	淘宝/天猫商品、交易、营销、评价全链路	B2B批发找货、分销管理、代发库存
免费额度	个人：500次/天，20万次/年企业：1000次/天，40万次/年	基础接口免费（有日调用上限），部分高阶接口需订购应用收费套餐	基础查询免费（有日限），批量/深度接口需购买套餐包
默认QPS	2 QPS（免费版）可付费扩容至50/100/500+	依应用等级而定，普通应用通常 5~20 QPS 高级应用可达数百	依应用等级，通常 5~10 QPS 起步，高阶可扩容
收费模式	预付费资源包 + QPS扩容（例：0<QPS≤50 约1000元/QPS/月）	应用套餐包（按月/年）+ 超额流量费部分接口按成功率计费	应用版本套餐（基础/高级/旗舰）+ 接口调用包
认证要求	个人/企业实名（企业额度更高）	企业认证为主（部分基础接口个人可测）	企业认证（强B端属性，需营业执照）
核心优势	响应快（约200ms），库存/价格实时性强	生态最成熟，接口超2000个，文档与SDK极全	B2B供应链数据权威，支持拿货价、混批条件查询
主要坑点	免费额度偏低，商用成本高；订单类需OAuth授权	风控严，频繁调用易触发限流或验签失败；部分字段脱敏	接口返回结构偏重B端逻辑（最小起订量、加工方式等）

注：淘宝TOP与1688的具体收费金额随应用标签（如ISV、自研、商家后台）差异极大，自研型应用通常比第三方ISV享有更多免费权益。

附：通用Python封装思路（三平台兼容签名请求）

为了横向对比，你可以写一个通用的OpenClient，通过切换sign_method和gateway来测试三家的限额与QPS表现：

import hashlib
import time
import requests
from typing import Dict, Literal
封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class EcomClient:
    def __init__(self, platform: Literal["jd", "tb", "al1688"], app_key: str, app_secret: str, gateway: str):
        self.platform = platform
        self.app_key = app_key
        self.app_secret = app_secret
        self.gateway = gateway

    def _sign(self, params: Dict) -> str:
        """京东MD5(secret+kv+secret) vs 淘宝/1688 MD5(secret+kv_sorted+secret) 细微差异处理"""
        filtered = sorted(
            (k, v) for k, v in params.items() if k != "sign" and v is not None and str(v).strip() != ""
        )
        kv = "".join(f"{k}{v}" for k, v in filtered)
        
        if self.platform == "jd":
            raw = self.app_secret + kv + self.app_secret
        else:
            # 淘宝/TOP 与 1688 通常也是 secret前缀后缀，但部分旧版1688用不同编码，此处演示通用版
            raw = self.app_secret + kv + self.app_secret
        
        return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()

    def execute(self, method: str, biz_params: Dict, token: str = "") -> dict:
        params = {
            "app_key": self.app_key,
            "method": method,
            "timestamp": self._get_timestamp(),
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
            **biz_params,
        }
        if token:
            params["access_token"] = token
        params["sign"] = self._sign(params)

        try:
            resp = requests.post(self.gateway, data=params, timeout=10)
            return resp.json()
        except Exception as e:
            return {"error": str(e)}

    def _get_timestamp(self):
        """京东用秒(10位)，淘宝/1688用毫秒(13位) —— 这是最容易踩的坑！"""
        if self.platform == "jd":
            return str(int(time.time()))
        return str(int(time.time() * 1000))


# ===== 测试调用示例 =====
# jd_client = EcomClient("jd", "KEY", "SEC", "https://api.jd.com/routerjson")
# tb_client = EcomClient("tb", "KEY", "SEC", "https://gw.api.taobao.com/router/rest")
# ali_client = EcomClient("al1688", "KEY", "SEC", "https://gw.open.1688.com/openapi/param2/1/com.alibaba.product/alibaba.cps.op.search")

# print(jd_client.execute("jingdong.ware.sku.read", {"sku_id": "12345"}))

⚠️ 源码关键点提醒（对比实测必备）

时间戳差异：京东JOS用10位秒级，淘宝TOP与1688用13位毫秒级，混用会直接报“Invalid Signature”。
QPS压测：免费账号建议用time.sleep(1/QPS)控制频率，京东免费2 QPS，若瞬间并发超了会返回limit frequency而非扣费。
额度耗尽：淘宝TOP在日额度用完后会返回request count limit，京东JOS则会阻断调用并提示升级资源包。

如果你愿意，我可以帮你写一个三平台并行的小脚本，用同一个SKU ID分别调京东/淘宝/1688的商品详情接口，直观打印各自的响应耗时、剩余额度报错、返回字段丰富度，方便你做技术选型压测。

《仿电商ERP：Spring Boot整合京东JOS实现商品/订单/库存三件套自动同步》（附python源码）

Tue, 14 Jul 2026 09:48:14 +0800

虽然标题是《仿电商ERP：Spring Boot整合京东JOS...》，但你明确要求了附Python源码，下面我给你一套“架构思路（Spring Boot视角）+ 可直接跑的Python实现”，聚焦商品 / 订单 / 库存三件套自动同步，方便你对照理解或先用Python做原型验证。

一、仿ERP同步整体架构（对照Spring Boot思路）

在Java Spring Boot里通常是分层设计，Python这边可以一一对应：

配置层：AppKey / AppSecret / AccessToken统一管理
→ 对应 @ConfigurationProperties
平台适配层：京东JOS签名、网关、鉴权封装
→ 对应 JdClient / JosTemplate
业务同步层：商品、订单、库存增量拉取 + 本地落库
→ 对应 XxxSyncService
调度层：定时任务（APScheduler / Spring @Scheduled）
→ 对应 ScheduledExecutorService
幂等与重试：订单号去重、异常重试、分页游标
→ 对应数据库唯一索引 + @Retryable

二、京东JOS通用API Client（Python，含MD5签名）

京东JOS特点：

网关：https://api.jd.com/routerjson
签名：MD5(AppSecret + 排序KV + AppSecret) → 大写
timestamp是秒级（10位），别用毫秒

# jd_jos_client.py
import hashlib
import time
import requests
from typing import Dict

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

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

    def _sign(self, params: Dict) -> str:
        # 剔除 sign、空值
        filtered = sorted(
            (k, v) for k, v in params.items()
            if k != "sign" and v is not None and str(v).strip() != ""
        )
        kv = "".join(f"{k}{v}" for k, v in filtered)
        raw = self.app_secret + kv + self.app_secret
        return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()

    def execute(self, method: str, biz_params: Dict) -> dict:
        params = {
            "app_key": self.app_key,
            "method": method,
            "timestamp": str(int(time.time())),
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
            **biz_params,
        }
        if self.access_token:
            params["access_token"] = self.access_token

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

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

三、三件套同步示例（商品 / 订单 / 库存）

1️⃣ 商品同步（SKU读取）

对应JOS接口：jingdong.ware.sku.read

def sync_products(client: JdJosClient, sku_ids: str):
    """
    sku_ids: 逗号分隔的skuId字符串
    """
    result = client.execute("jingdong.ware.sku.read", {
        "sku_id": sku_ids,
        "field": "sku_id,outer_id,stock_num,price,ware_name"
    })
    data = result.get("jingdong_ware_sku_read_responce", {})
    skus = data.get("skus", [])
    for sku in skus:
        print({
            "sku_id": sku.get("sku_id"),
            "name": sku.get("ware_name"),
            "price": sku.get("price"),
            "stock": sku.get("stock_num"),
        })

2️⃣ 订单同步（增量分页拉取）

对应接口：jd.order.search/ jingdong.pop.order.search

def sync_orders(client: JdJosClient, start_time: str, end_time: str, page: int = 1):
    """
    start_time / end_time: yyyy-MM-dd HH:mm:ss
    """
    result = client.execute("jingdong.pop.order.search", {
        "start_date": start_time,
        "end_date": end_time,
        "order_state": "WAIT_SELLER_STOCK_OUT,WAIT_BUYER_CONFIRM_GOODS,FINISHED",
        "page": page,
        "page_size": 100,
    })
    orders = result.get("jingdong_pop_order_search_responce", {}).get("order_list", [])
    for o in orders:
        print({
            "order_id": o.get("order_id"),
            "state": o.get("order_state"),
            "amount": o.get("order_total_price"),
        })
    return len(orders)

ERP实践：
用 modified时间做增量
本地表用 order_id做唯一索引防重
分页直到返回条数为0

3️⃣ 库存同步（读取 + 回写思路）

读取：复用 jingdong.ware.sku.read里的 stock_num
回写（ERP→京东）：用库存更新接口（需对应权限），一般按 sku_id + outer_id映射本地仓库库存

伪代码思路：

def sync_stock_to_jd(client, sku_id, new_stock):
    # 实际接口以京东文档为准，示例仅为结构
    client.execute("jingdong.ware.stock.update", {
        "sku_id": sku_id,
        "quantity": new_stock,
    })

四、定时任务（仿 `@Scheduled`）

from apscheduler.schedulers.blocking import BlockingScheduler

def job():
    client = JdJosClient("APP_KEY", "APP_SECRET", "ACCESS_TOKEN")
    sync_products(client, "123456,789012")
    sync_orders(client, "2026-07-13 00:00:00", "2026-07-14 23:59:59")

if __name__ == "__main__":
    scheduler = BlockingScheduler()
    scheduler.add_job(job, "interval", minutes=30, id="jd_sync_job")
    scheduler.start()

五、常见坑（京东JOS特有）

时间戳用秒：京东是10位秒级，淘宝才是毫秒，混用必报签名错误
参数名ASCII排序：必须严格排序再拼串
订单接口要access_token：商品公开读可能不用，但订单/库存一定需要OAuth授权
分页+增量：不要全量扫历史订单，用 modified时间窗口 + 页游标

如果你愿意，我可以帮你把这套 Python 同步脚本改成标准 Spring Boot 分层结构（Config / Client / Service / Scheduled）的Java示例代码，方便你直接往ERP项目里搬。

万邦api博客

《拼多多开放平台API预充值踩坑：0.01元/百次背后的到账延迟与欠费停调》（附python源码）

一、三大核心踩坑点拆解

二、防御性编码：余额监控 + 预充值预警 + 调用熔断（Python源码）

三、生产级避坑清单

《1688开放平台基础API真免费？QPS 10的限制怎么破（附GuardedClient源码）》

关键设计说明

🔄《京东JOS宙斯迁移实录：jingdong.ware.get → jingdong.ware.read.get 平滑切（2026版）》（附Python源码）

🔄《京东JOS宙斯迁移实录：jingdong.ware.get → jingdong.ware.read.get 平滑切（2026版）》（附Python源码）

一、老 vs 新接口对照（一目了然）

二、Python：老接口 vs 新接口调用对比 + 兼容封装

三、迁移 Checklist（照着做，5分钟切完）

四、避坑清单

五、面试/方案一句话

🏯《淘宝TOP API聚石塔内 vs 聚石塔外价差10倍：ERP迁云到底值不值？》（附Python源码）

结论先行：淘宝TOP基础API收费差距确实是10倍（聚石塔内¥0.02/百次 vs 聚石塔外¥0.20/百次），增值API差距更是达10倍（内¥0.06 vs 外¥0.60且外禁止调用）。

一、价差账本：10倍差在哪？

二、ERP迁云值不值？决策矩阵

✅ 强烈建议迁（ROI极高）

❌ 暂时不必迁（没必要）

三、附：Python 探测当前调用环境成本预警脚本

四、迁云不止是省钱，还有3个隐形红利

《别再被"免费API"误导：五家开放平台真实计费口径拆解（附决策清单）》（附python源码）

一、五大平台真实计费口径拆解（2026版）

1. 淘宝 TOP 开放平台

2. 京东 JOS（宙斯）

3. 1688 开放平台

四、拼多多开放平台

五、抖店（抖音）开放平台

二、别被“免费API”误导的 5 个核心真相

三、决策清单：你的场景用哪家、会不会花钱？

四、附：通用 API 用量探测 + 防超阈值脚本（Python）

五、一句话总结

《2026五大电商API收费全景对照：淘宝TOP/京东JOS/1688/拼多多/抖店，哪家用得起？》（附python源码）

💰 2026五大电商API收费全景对照：淘宝TOP / 京东JOS / 1688 / 拼多多 / 抖店，哪家用得起？

一、五大平台API收费全景对照表（2026版）

二、通用API用量探测 + 限流保护源码（Python）

三、选型与省钱建议（落地用）

《多店铺京东订单统一拉取：JOS API + 分布式调度 + 幂等设计实战》（附python源码）

一、整体架构

二、核心源码（Python · 含签名 / 分布式锁 / 幂等 / 断点）

三、关键设计点说明

四、避坑清单

《京东联盟API爬坑：不用店铺授权也能拿商品详情，但库存只有“有货/无货”》（附Python源码）

这是京东开放体系里最容易混淆、也最容易踩坑的点之一：

一、核心结论先给

二、京东联盟接口“坑点”拆解

三、Python：京东联盟商品详情调用（含库存状态解析）

四、如果你必须要真实库存数量，怎么办？

五、选型一句话总结（面试/方案）

《京东JOS vs 淘宝TOP vs 1688：接口免费额度、QPS、收费全方位对比》（附python源码）

京东JOS vs 淘宝TOP vs 1688 全方位对比

附：通用Python封装思路（三平台兼容签名请求）

⚠️ 源码关键点提醒（对比实测必备）

《仿电商ERP：Spring Boot整合京东JOS实现商品/订单/库存三件套自动同步》（附python源码）

一、仿ERP同步整体架构（对照Spring Boot思路）

二、京东JOS通用API Client（Python，含MD5签名）

三、三件套同步示例（商品 / 订单 / 库存）

1️⃣ 商品同步（SKU读取）

2️⃣ 订单同步（增量分页拉取）

3️⃣ 库存同步（读取 + 回写思路）

四、定时任务（仿 @Scheduled）

五、常见坑（京东JOS特有）

🔄《京东JOS宙斯迁移实录：`jingdong.ware.get` → `jingdong.ware.read.get` 平滑切（2026版）》（附Python源码）

四、定时任务（仿 `@Scheduled`）