阿里妈妈(淘宝联盟)的item_search接口(官方接口名为taobao.tbk.item.search)是通过关键词、类目、价格等条件批量获取推广商品列表的核心工具,广泛应用于淘宝客选品、推广平台搭建、电商营销分析等场景。该接口不仅能返回商品基础信息(标题、价格、销量),更聚焦推广属性(佣金比例、优惠券、推广链接),是构建 “选品 - 推广 - 变现” 闭环的关键。本文将系统讲解接口对接流程、参数配置、代码实现及最佳实践,帮助开发者从入门到精通。
一、接口基础认知(核心功能与场景)
二、对接前置准备(账号与环境)
三、接口调用流程(官方标准)
四、代码实现示例(Python)
import requests
import time
import hmac
import hashlib
import json
from urllib.parse import urlencode
from typing import List, Dict
class AlimamaSearchApi:
def __init__(self, app_key: str, app_secret: str, sandbox: bool = False):
self.app_key = app_key
self.app_secret = app_secret
self.sandbox = sandbox
self.base_url = "https://gw-api.tbsandbox.com/router/rest" if sandbox else "https://eco.taobao.com/router/rest"
self.format = "json"
self.version = "2.0"
self.method = "taobao.tbk.item.search" # 官方接口名
def generate_sign(self, params: Dict[str, str]) -> str:
"""生成HMAC-MD5签名"""
# 1. 按参数名ASCII升序排序
sorted_params = sorted(params.items(), key=lambda x: x[0])
# 2. 拼接为key=value字符串
sign_str = urlencode(sorted_params)
# 3. 拼接app_secret并加密
sign = hmac.new(
self.app_secret.encode(),
sign_str.encode(),
hashlib.md5
).hexdigest().lower()
return sign
def item_search(self,
q: str = "",
cat: str = "",
start_price: float = None,
end_price: float = None,
sort: str = "tk_total_sales",
page: int = 1,
page_size: int = 20,
platform: int = 1) -> Dict:
"""
搜索推广商品列表
:param q: 搜索关键词(与cat二选一)
:param cat: 类目ID
:param start_price: 起始价格(元)
:param end_price: 结束价格(元)
:param sort: 排序方式(默认销量降序)
:param page: 页码(1-50)
:param page_size: 每页条数(1-100)
:param platform: 平台类型(1-淘宝,2-天猫)
:return: 标准化搜索结果
"""
try:
# 1. 校验参数
if not q and not cat:
return {"success": False, "error_msg": "关键词(q)和类目(cat)至少需提供一个"}
if page < 1 or page > 50:
return {"success": False, "error_msg": "页码必须在1-50之间"}
if page_size < 1 or page_size > 100:
return {"success": False, "error_msg": "每页条数必须在1-100之间"}
# 2. 组装公共参数
public_params = {
"app_key": self.app_key,
"method": self.method,
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"format": self.format,
"v": self.version
}
# 3. 组装业务参数
biz_params = {
"sort": sort,
"page": str(page),
"page_size": str(page_size),
"platform": str(platform)
}
if q:
biz_params["q"] = q
if cat:
biz_params["cat"] = cat
if start_price is not None:
biz_params["start_price"] = str(start_price)
if end_price is not None:
biz_params["end_price"] = str(end_price)
# 4. 合并参数并生成签名
all_params = {**public_params,** biz_params}
all_params["sign"] = self.generate_sign(all_params)
# 5. 发送POST请求
response = requests.post(
url=self.base_url,
data=all_params,
timeout=15
)
response.raise_for_status()
result = response.json()
# 6. 处理错误响应
if "error_response" in result:
error = result["error_response"]
return {
"success": False,
"error_code": error["code"],
"error_msg": error["msg"],
"sub_msg": error.get("sub_msg", "")
}
# 7. 解析并标准化数据
search_resp = result.get("tbk_item_search_response", {})
items = search_resp.get("results", {}).get("n_tbk_item", [])
standardized_items = []
for item in items:
standardized_items.append({
"item_id": item.get("num_iid"),
"title": item.get("title"),
"main_image": item.get("pict_url"),
"url": item.get("item_url"),
"price": {
"original": float(item.get("reserve_price", 0)),
"final": float(item.get("zk_final_price", 0))
},
"commission": {
"rate": float(item.get("commission_rate", 0)) / 100, # 转为小数
"amount": round(float(item.get("zk_final_price", 0)) * (float(item.get("commission_rate", 0)) / 100), 2)
},
"sales": {
"volume_30d": int(item.get("volume", 0)),
"comment_count": int(item.get("comment_count", 0))
},
"coupon": {
"info": item.get("coupon_info", ""),
"url": item.get("coupon_click_url", "")
},
"promotion_url": item.get("click_url") # 推广链接
})
return {
"success": True,
"total": int(search_resp.get("total_results", 0)),
"page": page,
"page_size": page_size,
"total_pages": (int(search_resp.get("total_results", 0)) + page_size - 1) // page_size,
"items": standardized_items
}
except requests.exceptions.HTTPError as e:
return {"success": False, "error_msg": f"HTTP错误: {str(e)}"}
except Exception as e:
return {"success": False, "error_msg": f"搜索失败: {str(e)}"}
# 使用示例
if __name__ == "__main__":
# 替换为实际参数(沙箱环境可使用测试appkey)
APP_KEY = "your_app_key"
APP_SECRET = "your_app_secret"
# 初始化API客户端(sandbox=True为测试环境)
api = AlimamaSearchApi(APP_KEY, APP_SECRET, sandbox=False)
# 搜索“无线耳机”,价格50-200元,按销量降序,第1页,30条/页
result = api.item_search(
q="无线耳机",
start_price=50,
end_price=200,
sort="tk_total_sales",
page=1,
page_size=30,
platform=1 # 淘宝平台
)
if result["success"]:
print(f"搜索结果:共 {result['total']} 件商品,{result['total_pages']} 页")
print(f"当前第 {result['page']} 页,{len(result['items'])} 件商品\n")
# 打印前5件商品信息
for i, item in enumerate(result["items"][:5]):
print(f"商品 {i+1}:")
print(f"标题:{item['title'][:50]}...") # 截断长标题
print(f"价格:原价 {item['price']['original']} 元 → 券后 {item['price']['final']} 元")
print(f"佣金:{item['commission']['rate']*100}%(约 {item['commission']['amount']} 元/件)")
print(f"30天销量:{item['sales']['volume_30d']} 件")
if item['coupon']['info']:
print(f"优惠券:{item['coupon']['info']}(领取链接:{item['coupon']['url'][:30]}...)")
print(f"推广链接:{item['promotion_url'][:50]}...\n")
else:
print(f"搜索失败:{result['error_msg']}(错误码:{result.get('error_code')})")