前言 多数公开教程只演示简单调用获取原始店铺
json 数据,很少关注业务场景下的痛点:同一关键词返回店铺数据字段参差不齐、分页偏移失效、店铺类型混杂(天猫店、淘宝 C
店、企业店)、部分接口返回冗余嵌套层级深,直接拿到原始数据很难用于数据分析、店铺监控业务。本文从业务落地视角,封装一套具备类型过滤、异常捕获、字段归一化的调用示例,解决直接拿到原始
JSON 无法直接入库的现实问题。 前置准备 调用淘宝开放平台店铺搜索相关接口,需要申请开发者账号,获取
app_key、app_secret,同时申请对应接口权限。接口核心入参包含搜索关键词、页码、每页条数、店铺筛选条件;需要注意接口存在调用
QPS 限制,高频场景必须做请求间隔控制,避免触发平台限流返回错误码。 很多开发者踩坑点:单纯修改 page 参数无限翻页,当到达最大数据阈值后接口不会返回报错,而是重复返回第一页数据,业务层必须增加逻辑识别该现象。 核心 Python 代码示例 代码逻辑解析 签名函数按照开放平台规则对参数排序生成 MD5 签名,是接口请求合法的关键; 增加超时、接口错误返回捕获,不会因为单条接口异常直接中断程序; 对原始嵌套的返回结构做扁平化归一化处理,输出固定字典结构,方便直接写入数据库; 增加店铺类型过滤预留参数,可以按需筛选天猫、企业店铺、个人 C 店。 实际对接踩坑总结 分页陷阱:接口有最大返回数据上限,页码超过阈值不会报错,数据重复,业务代码要对比前后页数据做去重判断; 权限管控:店铺搜索接口不是创建账号就自带,需要在开放平台提交接口权限申请,未申请会返回无权限错误; QPS 限制,批量采集场景必须加 sleep 间隔,短时间大量请求会被限流; 部分字段非必返回,不要直接取值,统一使用 get () 方式获取,避免 key 不存在直接抛出程序异常。
import time
import hashlib
import requests
class TaobaoShopSearch:
def __init__(self, app_key, app_secret):
self.app_key = app_key
self.app_secret = app_secret
self.api_url = "https://gw.api.taobao.com/router/rest"
def sign_generate(self, params):
sorted_items = sorted(params.items())
raw_str = self.app_secret + "".join([k+str(v) for k,v in sorted_items]) + self.app_secret
return hashlib.md5(raw_str.encode("utf-8")).hexdigest().upper()
def search_shop_list(self, keyword, page_no=1, page_size=20, shop_type=None):
params = {
"method": "taobao.shop.gets",
"app_key": self.app_key,
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"format": "json",
"v": "2.0",
"q": keyword,
"page_no": page_no,
"page_size": page_size
}
sign = self.sign_generate(params)
params["sign"] = sign
resp = requests.get(self.api_url, params=params, timeout=15)
res_json = resp.json()
# 接口错误捕获
if "error_response" in res_json:
return {"success":False,"msg":res_json["error_response"]["msg"],"data":[]}
shop_data = res_json.get("shop_gets_response",{}).get("shops",{}).get("shop",[])
result_list = []
for shop in shop_data:
# 字段归一化,统一输出结构便于入库
item = {
"shop_id": shop.get("sid"),
"shop_name": shop.get("title"),
"shop_type": shop.get("shop_type"),
"seller_nick": shop.get("nick"),
"credit_score": shop.get("shop_credit"),
"province": shop.get("province"),
"city": shop.get("city")
}
# 过滤指定店铺类型
if shop_type and item["shop_type"] != shop_type:
continue
result_list.append(item)
return {"success":True,"data":result_list}
if __name__ == "__main__":
client = TaobaoShopSearch(app_key="xxx", app_secret="xxx")
resp_data = client.search_shop_list(keyword="家居",page_no=1,page_size=20)
print(resp_data)