×

☁️《抖店API基础¥0.018/百次·增值¥0.05/百次:云内云外价差架构实战》(附Python源码)

万邦科技Lex 万邦科技Lex 发表于2026-08-06 14:52:35 浏览23 评论0

抢沙发发表评论

结论先拍:抖店开放平台(V1.3,2026.6.16发布)官方资费是 基础API云内0.018元/百次/AppKey、云外0.18元/百次/AppKey;增值API云内0.05元/百次/AppKey、云外0.5元/百次/AppKey,且增值API未经允许禁止云外调用。 价差不是“8折优惠”而是硬10倍,且增值层云外不是贵是不通。自研ERP的最优架构因此被倒推成一句:订单/解密/批量解密等敏感链路必须抖店云内,基础调用走云内+数据推送替轮询,增值调用单独隔离AppKey并强制云内。


一、官方资费表(V1.3锚定)

API类型
抖店云内
抖店云外
价差
云外增值约束
基础API(订单/售后/商品详情//product/addV2等)
¥0.018/百次/AppKey
¥0.18/百次/AppKey
×10
允许但×10
增值API(批量解密/敏感数据类)
¥0.05/百次/AppKey
¥0.5/百次/AppKey
×10
禁止调用(返回无权限)
结算特征:
  • 预充值按量,欠费硬断(非429限流);

  • 2026.7.1起 /product/addV2/product/addSchema 正式计入基础API收费(原来上新免费洼地被填平);

  • 数据推送(订单/售后推RDS)是免费产品,官方明确推荐“云内调用+数据推送”是费用最低方案。


二、价差在架构里意味着什么(不是算小账是定部署)

  • 云外跑基础API:单价×10,单店日1万次月费从¥54变¥540,20店铺货从¥1080变¥10800。

  • 云外碰增值API:不是¥0.5/百次扣费,是直接 no permission / 非云内调用 拦截,批量解密订单地址/手机在云外根本拿不到。

  • 官方入云截止期已过:2023.8.31起商家自研涉及订单解密的必须入抖店云,截止日后云外解密权限直接关闭。

所以“云内vs云外”对抖店不是成本选项,是敏感链路的准入条件;基础调用你可以理论上选云外,但×10账单会教做人。

三、推荐架构:云内收口 + 推送替轮询 + 增值隔离

商家浏览器
   │
抖店云内 ECS(自研ERP后端)
   ├─ DyBaseClient   基础API(订单/商品/库存)  令牌桶限速
   ├─ DyAdvClient    增值API(批量解密)        单独AppKey+强制云内+权限守卫
   ├─ PushConsumer   数据推送(订单/售后→RDS)  免费,替代轮询
   └─ Redis          幂等键 / 日调用计数 / 预估余额
   │
   └─ 增量补偿:每5~10min调一次 order/searchList 补漏
要点:
  1. 两个AppKey:基础一个、增值一个。增值Key不碰订单列表,只做解密,便于权限审计和额度隔离。

  2. 推送为主:开通抖店云数据推送,订单变更直接写RDS,ERP只读库;order/searchList 仅作5min补偿,轮询量砍80%+。

  3. 商品发布合并/product/addV2 本地校验+草稿缓存,失败重试计入同SKU,不盲目重提(7.1后这条开始烧钱)。

  4. 增值调用前守卫in_cloud=False 直接抛 PermissionError,不让代码走到网关才被发现。


四、Python:DyGuardedClient(基础/增值分层 + 云外熔断 + 预充值余额守卫)

# dy_guarded_client.py
"""
抖店开放平台 GuardedClient
- 基础/增值分层计价(0.018 / 0.05 云内;0.18 / 0.5 云外)
- 增值API云外直接熔断(官方禁止)
- 预充值模式:本地反推预估余额,低于N天预估熔断非核心
- 捕获费用类错误特征
"""
import time, hashlib, requests, json
from typing import Dict, Optional

GW = "https://openapi-fxg.jinritemai.com/gateway"
IN_BASE = 0.018 / 100
OUT_BASE = 0.18 / 100
IN_ADV = 0.05 / 100
OUT_ADV = 0.5 / 100

class DyGuardedClient:
    def __init__(self, app_key, app_secret,
                 in_cloud=True,
                 cached_balance=None,
                 est_daily_calls=10_000,
                 warn_days=3,
                 recharge_in_flight=False):
        self.ak = app_key
        self.ask = app_secret
        self.in_cloud = in_cloud
        self.official_balance = cached_balance
        self.local_spent = 0.0
        self.est_daily = est_daily_calls
        self.warn_days = warn_days
        self.recharge_in_flight = recharge_in_flight

    @property
    def est_balance(self) -> Optional[float]:
        if self.official_balance is None:
            return None
        return self.official_balance - self.local_spent

    def sync_official_balance(self, fresh: float):
        self.official_balance = fresh
        self.local_spent = 0.0
     # 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
    def _sign(self, params: Dict) -> str:
        # 抖店:按key升序,拼接 param=val 后 md5(secret+拼接+secret)
        f = 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 f)
        return hashlib.md5(f"{self.ask}{qs}{self.ask}".encode()).hexdigest().lower()

    def _check_tier(self, is_value: bool):
        if is_value and not self.in_cloud:
            raise PermissionError("❌ 抖店增值API禁止云外调用,必须抖店云内(返回no permission)")

    def _before_call(self, is_value: bool, is_core: bool):
        unit = (IN_ADV if is_value else IN_BASE) if self.in_cloud else (OUT_ADV if is_value else OUT_BASE)
        bal = self.est_balance
        if bal is None:
            return
        day_cost = self.est_daily * unit
        if bal <= 0:
            if self.recharge_in_flight and not is_core:
                raise RuntimeError("⏸ 预估余额≤0,充值在途,非核心熔断")
            raise RuntimeError("🚨 预充值余额≤0,抖店已硬切断,请充值待入账")
        if bal < day_cost * self.warn_days and not is_core:
            raise RuntimeError(f"⏸ 余额¥{bal:.2f}<{self.warn_days}天预估¥{day_cost*self.warn_days:.2f},非核心熔断")

    @staticmethod
    def _is_fee_err(d: Dict) -> bool:
        if "error_response" not in d:
            return False
        er = d["error_response"]
        code = str(er.get("code", ""))
        blob = json.dumps(er, ensure_ascii=False).lower()
        return code in ("1003", "50001") or "balance" in blob or "fee" in blob or "insufficient" in blob

    def safe_call(self, method, biz, token=None, *, is_core=True,
                  is_value=False, max_retry=3):
        self._check_tier(is_value)
        self._before_call(is_value, is_core)
        unit = (IN_ADV if is_value else IN_BASE) if self.in_cloud else (OUT_ADV if is_value else OUT_BASE)

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

        for att in range(max_retry):
            try:
                r = requests.post(GW, data=params, timeout=15)
                d = r.json()
                if self._is_fee_err(d):
                    self.local_spent += self.est_daily * unit
                    if self.recharge_in_flight and att < max_retry - 1:
                        time.sleep(20); continue
                    raise RuntimeError(f"🚨 疑似欠费断调: {d['error_response']}")
                if "error_response" in d:
                    raise Exception(f"DY_ERR[{d['error_response'].get('code')}]: {d['error_response'].get('message')}")
                self.local_spent += unit
                return d
            except requests.RequestException:
                if att < max_retry - 1:
                    time.sleep(2 ** att); continue
                raise

    # 基础业务
    def list_orders(self, token, start, end, page=1):
        return self.safe_call("order.searchList", {
            "start_time": start, "end_time": end,
            "page": page, "size": 50
        }, token, is_core=True, is_value=False)

    def add_product(self, token, payload):
        # 7.1起计费
        return self.safe_call("product.addV2", payload, token, is_core=True, is_value=False)

    # 增值业务(批量解密)
    def batch_decrypt(self, token, order_ids):
        return self.safe_call("order.batchDecrypt", {"order_ids": order_ids},
                              token, is_core=True, is_value=True)


if __name__ == "__main__":
    # 云内基础Client
    base = DyGuardedClient("AK", "AS", in_cloud=True, cached_balance=12.0, est_daily_calls=10_000)
    try:
        base.list_orders("TOKEN", "2026-08-01 00:00:00", "2026-08-01 00:05:00")
    except RuntimeError as e:
        print(e)

    # 云外硬调增值 → 直接PermissionError
    adv_bad = DyGuardedClient("AK", "AS", in_cloud=False)
    try:
        adv_bad.batch_decrypt("TOKEN", ["123"])
    except PermissionError as e:
        print(e)

    # 余额预警演示
    low = DyGuardedClient("AK", "AS", in_cloud=True, cached_balance=0.25, est_daily_calls=10_000)
    try:
        low.safe_call("product.addV2", {"a": 1}, is_core=False, is_value=False)
    except RuntimeError as e:
        print(e)

五、成本对照(单店日1万次,云内 vs 云外)

场景
月API费
其中增值(若日500次解密)
云内基础+推送替轮询(实际轮询降到日2千)
≈¥10.8
增值¥0.75
云内全轮询(日1万)
¥54
增值¥0.75
云外全轮询
¥540
增值云外禁止/若强行基础¥540
推送替轮询这一招,在抖店官方文档里被点名“费用最低方案”,不是玄学: 100笔订单5状态变更场景下,云内+推送比云内纯轮询少调大量 orderDetail,比云外轮询省一个数量级。

六、落地顺序(自研ERP)

  1. 企业实名自研应用 → 标签“商家后台系统/电商ERP” → 拿AppKey;

  2. ECS买抖店云内(前后端+RDS都在内,解密落库也内网);

  3. 数据推送订单/售后到RDS,写PushConsumer;

  4. 建两个AppKey:base(订单/商品/库存)、adv(batchDecrypt),adv只云内跑;

  5. 把上面 DyGuardedClientin_cloud 写死 True,单元测试里用 False 跑断言确保没人误改;

  6. 每天凌晨拉账单校准 sync_official_balance,余额<3天预估企微红警。


一句话定性:抖店0.018和0.05不是“便宜档”,是云内准入价;云外0.18/0.5是劝退价,增值云外更是权限墙。自研ERP把“抖店云内+推送为主+增值隔离AppKey+余额守卫”焊进Client,单店月API费可以压到一杯咖啡钱,架构错把ECS放在阿里云公网跑解密,不是贵10倍是跑不通。
要不要我把 DyGuardedClient 扩成 Redis中心化预估余额(多进程共享)+ 数据推送Consumer模板(订单/售后写PG)+ 商品发布调用合并器,直接拼进你前面那套九家中台调度器?


群贤毕至

访客