天猫开放搜索接口常被用于竞品调研、品类行情统计。网上多数示例只完成基础调用,直接返回全部原始字段,业务系统接收大量冗余字段,内存占用高,同时忽略天猫特有规则:品牌筛选校验、部分商品为定向内部商品不对外展示、搜索结果存在少量已下架但仍出现在搜索列表的数据。 本文不做简单入门演示,重点围绕字段按需裁剪、结果集降噪、业务有效性校验进行封装,减少传输体积,过滤伪有效商品,适配后端统计类业务,给出可落地的 Python 实现。 实现代码 核心设计思路 按需字段裁剪:不直接透传接口全部返回字段,只提取业务需要字段,降低网络 IO 与内存开销,适合大批量循环拉取场景。 结果集降噪处理:内存集合做商品 ID 去重,过滤库存为 0 的残留搜索商品,这是天猫搜索接口很常见的脏数据问题。支持可选传入 brand_id 做品牌定向筛选,适配品牌维度数据分析。 完整异常分层:区分网络异常、平台返回错误码,不会因为单页失败直接导致整体任务崩溃。单页条数强制上限 30,符合平台接口约束。 对接踩坑总结 天猫搜索接口深度分页存在限制,页码过大时返回结果会出现错乱,业务层需要配置最大翻页阈值。调用需要对应接口权限,应用未开通权限会返回权限错误。批量循环调用时必须加入延时,避免触发平台风控限流。中文关键词不需要 url 手动编码,网关自动处理。
import requests
import time
import hashlib
class TmallKeywordSearch:
def __init__(self, app_key, app_secret):
self.app_key = app_key
self.app_secret = app_secret
self.gateway = "https://gw.api.tmall.com/router/rest"
self.seen_item_ids = set()
def gen_sign(self, param_dict):
sorted_items = sorted(param_dict.items())
raw = self.app_secret + "".join([f"{k}{v}" for k, v in sorted_items]) + self.app_secret
return hashlib.md5(raw.encode("utf‑8")).hexdigest().upper()
def search_goods(self, keyword, page_no=1, page_size=30, brand_id=None):
params = {
"method": "tmall.item.search",
"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": min(page_size,30)
}
if brand_id:
params["brand_id"] = brand_id
params["sign"] = self.gen_sign(params)
try:
resp = requests.get(self.gateway, params=params, timeout=12)
resp_json = resp.json()
except Exception:
return {"ok":False, "msg":"网络请求异常","data":[]}
if "error_response" in resp_json:
return {"ok":False, "msg":resp_json["error_response"]["msg"],"data":[]}
resp_body = resp_json.get("item_search_response",{})
raw_items = resp_body.get("items",{}).get("item",[])
output = []
for item in raw_items:
iid = item.get("num_iid")
if not iid or iid in self.seen_item_ids:
continue
# 过滤无对外售卖权限、库存为0商品
stock = int(item.get("stock_num",0))
if stock <= 0:
continue
self.seen_item_ids.add(iid)
# 按需选取字段,裁剪无用原始数据
output.append({
"item_id": iid,
"title": item.get("title"),
"price": float(item.get("price",0)),
"sales": int(item.get("sales_volume",0)),
"brand_name": item.get("brand_name",""),
"shop_name": item.get("nick","")
})
total = resp_body.get("total_results",0)
return {"ok":True,"total":total,"data":output}
if __name__ == "__main__":
cli = TmallKeywordSearch("your_app_key","your_app_secret")
ret = cli.search_goods("无线蓝牙耳机", page_no=1)
print("有效商品列表:", ret["data"])