前言 自有店铺
ERP
库存同步、商家商品备份、店铺类目商品统计场景下，完整拉取店铺全部商品是高频刚需。网上现有教程普遍只调用在售商品单接口，忽略仓库待上架商品，分页逻辑简陋无边界控制，缺少类目分片防超限、商品
ID 全局去重、限流指数退避等生产能力，批量同步极易出现数据缺失、权限封禁问题。本文采用在售 + 库存双接口合并方案，搭配类目分片遍历、全局集合去重、分级异常捕获，全程基于淘宝官方 TOP 合规接口开发，无网页逆向爬虫，适配长期定时全店同步业务。 一、本文差异化设计亮点 双接口数据合并：同时调用 类目分片分页遍历：按店铺类目拆分分页请求，规避单页海量数据导致接口超时、响应截断，大幅提升批量同步稳定性。 全局商品 ID 去重：使用全局集合缓存已采集商品 ID，防止跨类目、跨分页重复商品入库，保证商品唯一性。 指数退避限流防护：区分签名错误、权限不足、429 限流、网络超时四类异常，限流时阶梯延长休眠时间，避免应用调用配额冻结。 标准化轻量化字段清洗：统一格式化价格、库存、类目、商家编码，输出固定结构，可直接写入数据库无需二次解析。 二、接口基础调用规范 网关地址： 三、完整可运行 Python 生产代码 四、实战踩坑核心要点 仅调用在售接口会丢失仓库未上架商品，商家库存同步场景必须叠加库存商品接口合并数据。 接口单类目最大仅支持 100 页，代码内置页数上限，无边界循环会持续报错浪费调用配额。 跨类目商品会重复返回，必须通过商品 num_iid 全局去重，否则统计库存、商品总数数据失真。 session 令牌有效期有限，批量定时同步需增加令牌自动刷新逻辑，避免中途鉴权失效中断采集。 禁止取消每页 1 秒休眠，高频连续请求会直接锁定应用接口调用权限，恢复周期较长。taobao.items.onsale.get在售、taobao.items.inventory.get库存商品，一次性拿到店铺全部商品，解决单一接口数据不全痛点。https://eco.taobao.com/router/rest
鉴权方式：HMAC-SHA1 签名，14 位标准时间戳，需店铺授权 session 令牌
调用约束：单应用 QPS≤3，单类目分页上限 100 页，超出会返回参数越界错误
权限要求：自用型应用完成商家授权，同时开通在售、库存商品两类接口权限
import requests
import hmac
import hashlib
import time
import json
from urllib.parse import quote
class TaobaoShopAllItemClient:
def __init__(self, app_key, app_secret, session_token, shop_nick):
self.app_key = app_key
self.app_secret = app_secret
self.session = session_token
self.shop_nick = shop_nick
self.gateway = "https://eco.taobao.com/router/rest"
self.session_http = requests.Session()
self.seen_item_ids = set() # 全局去重缓存
def build_hmac_sign(self, params):
# 淘宝标准HMAC-SHA1签名
sorted_kv = sorted(params.items(), key=lambda x: x[0])
sign_source = "".join([f"{k}{quote(str(v), safe='')}" for k, v in sorted_kv])
return hmac.new(self.app_secret.encode(), sign_source.encode(), hashlib.sha1).hexdigest().upper()
def single_page_fetch(self, method, cid=None, page=1):
"""单页拉取，支持类目筛选，区分在售/库存接口"""
params = {
"app_key": self.app_key,
"method": method,
"session": self.session,
"format": "json",
"v": "2.0",
"sign_method": "hmac",
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"nick": self.shop_nick,
"page_no": page,
"page_size": 20
}
if cid:
params["cid"] = cid
params["sign"] = self.build_hmac_sign(params)
try:
resp = self.session_http.get(self.gateway, params=params, timeout=15)
res_data = resp.json()
# 限流自动退避重试
if res_data.get("error_response", {}).get("code") == 429:
time.sleep(3)
return self.single_page_fetch(method, cid, page)
if "error_response" in res_data:
return {"code": -1, "msg": res_data["error_response"]["sub_msg"], "list": []}
resp_key = method.replace(".", "_") + "_response"
item_raw = res_data.get(resp_key, {}).get("items", {}).get("item", [])
clean_list = []
for item in item_raw:
item_id = item.get("num_iid")
if item_id in self.seen_item_ids:
continue
self.seen_item_ids.add(item_id)
clean_list.append({
"item_id": item_id,
"title": item.get("title"),
"outer_id": item.get("outer_id", ""),
"cid": item.get("cid"),
"price": float(item.get("price", 0)),
"stock_num": int(item.get("stock", 0)),
"pic_url": item.get("pic_url"),
"status": "onsale" if method == "taobao.items.onsale.get" else "inventory"
})
time.sleep(1)
return {"code": 200, "data": clean_list}
except Exception as e:
return {"code": -2, "msg": f"网络异常：{str(e)}", "list": []}
def fetch_all_by_method(self, method, cid=None):
"""循环分页拉取单接口全类目商品"""
all_items = []
page = 1
while page <= 100: # 接口最大100页边界限制
page_res = self.single_page_fetch(method, cid, page)
if page_res["code"] != 200 or len(page_res["data"]) == 0:
break
all_items.extend(page_res["data"])
page += 1
return all_items
def get_shop_all_items(self):
"""合并在售+库存，返回店铺全部商品"""
onsale_list = self.fetch_all_by_method("taobao.items.onsale.get")
inventory_list = self.fetch_all_by_method("taobao.items.inventory.get")
full_list = onsale_list + inventory_list
return {"shop_nick": self.shop_nick, "total_count": len(full_list), "item_list": full_list}
# 调用示例
if __name__ == "__main__":
client = TaobaoShopAllItemClient(
app_key="开发者后台AppKey",
app_secret="开发者后台AppSecret",
session_token="商家授权session",
shop_nick="店铺卖家昵称"
)
result = client.get_shop_all_items()
print(json.dumps(result, ensure_ascii=False, indent=2))
