一、整体流程一览
| 步骤 | 动作 | 耗时 |
|---|---|---|
| ① 注册认证 | 淘宝开放平台 → 实名认证 | 1-3个工作日 |
| ② 创建应用 | 获取 App Key / App Secret | 即时 |
| ③ 申请权限 | 申请 taobao.items.search 等接口 | 1-2个工作日 |
| ④ 调接口 | 签名 → 请求 → 解析 → 存储 | 按需 |
二、接口选择
| 接口 | 适用场景 | 核心字段 | 频率限制 |
|---|---|---|---|
| taobao.items.search | 关键词搜商品、竞品监控 | num_iid, title, price, pic_url, volume | 免费500次/天，企业10万次/天 |
| taobao.tbk.item.get | 淘宝客推广、查优惠券/佣金 | zk_final_price(折后价), tk_rate(佣金), coupon_info | 约5000次/天 |
| taobao.item.search_shop | 按店铺拉全量商品 | 同上 + shop_id | 同上 |
本文主力演示 taobao.items.search（关键词搜索）。
三、签名机制（最核心，错了必报错）
淘宝 API 采用 MD5 签名，规则如下：
签名串 = AppSecret + 按ASCII升序拼接(key1value1key2value2...) + AppSecret
sign = MD5(签名串).toUpperCase()
签名步骤拆解：
- 收集所有参数（排除 sign 本身）
- 按参数名 ASCII 码升序排序
- 拼接为
key1value1key2value2...（无分隔符） - 首尾各追加 AppSecret
- MD5 加密 → 转大写
pythonimport hashlib
def generate_sign(params: dict, app_secret: str) -> str:
sorted_params = sorted(params.items(), key=lambda x: x[0])
param_str = app_secret + ''.join(f"{k}{v}" for k, v in sorted_params if k != 'sign') + app_secret
return hashlib.md5(param_str.encode('utf-8')).hexdigest().upper()
四、完整 Python 实战代码
pythonimport requests
import hashlib
import time
from datetime import datetime
class TaobaoAPI:
def __init__(self, app_key: str, app_secret: str):
self.app_key = app_key
self.app_secret = app_secret
self.base_url = "https://gw.api.taobao.com/router/rest"
def generate_sign(self, params: dict) -> str:
sorted_params = sorted(params.items(), key=lambda x: x[0])
param_str = self.app_secret + ''.join(
f"{k}{v}" for k, v in sorted_params if k != 'sign'
) + self.app_secret
return hashlib.md5(param_str.encode('utf-8')).hexdigest().upper()
def search_items(self, keyword: str, page_no: int = 1, page_size: int = 50,
fields: str = None) -> dict:
"""单页搜索"""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
params = {
"app_key": self.app_key,
"method": "taobao.items.search",
"timestamp": timestamp,
"format": "json",
"v": "2.0",
"sign_method": "md5",
"q": keyword,
"page_no": str(page_no),
"page_size": str(page_size),
}
if fields:
params["fields"] = fields # 如: num_iid,title,pict_url,price,volume
params["sign"] = self.generate_sign(params)
response = requests.get(self.base_url, params=params, timeout=10)
return response.json()
def batch_fetch(self, keyword: str, max_pages: int = 10, page_size: int = 50) -> list:
"""批量多页获取"""
all_items = []
for page in range(1, max_pages + 1):
print(f"🔍 第 {page} 页...")
result = self.search_items(keyword, page_no=page, page_size=page_size)
# 错误处理
if result.get("error_response"):
print(f"❌ {result['error_response']['sub_msg']}")
break
items = result.get("items_search_response", {}) \
.get("items", {}) \
.get("item", [])
if not items:
print(f"✅ 第 {page} 页无数据，停止")
break
all_items.extend(items)
time.sleep(1) # 防限流
return all_items
# ========== 调用示例 ==========
if __name__ == "__main__":
APP_KEY = "YOUR_APP_KEY"
APP_SECRET = "YOUR_APP_SECRET"
api = TaobaoAPI(APP_KEY, APP_SECRET)
data = api.batch_fetch(keyword="手机", max_pages=5, page_size=50)
print(f"共获取 {len(data)} 条商品")
for item in data[:3]:
print(f" {item.get('title')} | ¥{item.get('price')} | 销量{item.get('volume')}")
五、关键参数说明
| 参数 | 必填 | 说明 | 示例 |
|---|---|---|---|
q | ✅ | 搜索关键词 | 手机 |
page_no | ❌ | 页码，默认1 | 1 |
page_size | ❌ | 每页数量，默认20，最大100 | 50 |
fields | ❌ | 返回字段，减少传输量 | num_iid,title,price,pict_url,volume |
sort | ❌ | 排序：sale-desc(销量降) / price_asc(价格升) | sale-desc |
start_price / end_price | ❌ | 价格区间 | 0 / 1000 |
cat | ❌ | 类目ID | 50012025(女装) |
总页数计算： total_pages = ceil(total_results / page_size)，从首页响应的 total_results 字段获取。
六、响应数据结构
json{
"items_search_response": {
"total_results": "5428",
"items": {
"page": "1",
"real_total_results": "5428",
"page_size": 48,
"item": [
{
"num_iid": "123456789",
"title": "商品标题...",
"pict_url": "https://img.alicdn.com/...",
"price": "199.00",
"volume": "3000+",
"nick": "店铺名",
"seller_id": "898146183"
}
]
}
}
}
七、性能优化与避坑
| 问题 | 解决方案 |
|---|---|
| 限流 403 | time.sleep(1) 控制间隔，企业认证可到10万次/天 |
| 签名错误 | 检查参数是否按 ASCII 升序、是否排除了 sign 字段、时间戳是否在 ±5 分钟内 |
| 数据延迟 | 销量/价格有约3分钟延迟，需搭配消息订阅或定时全量同步 |
| 高频调用封禁 | 单应用每分钟不超过60次，建议用 Redis 缓存 + 异步框架(asyncio/Scrapy) |
| AppSecret 泄露 | 切勿硬编码在前端，服务端签名后再返回数据 |
八、数据存储建议
pythonimport pandas as pd
# 导出 CSV
df = pd.DataFrame(data)
df.to_csv("taobao_items.csv", index=False, encoding="utf-8-sig")
# 或存 MySQL
# INSERT INTO products (num_iid, title, price, volume, pic_url) VALUES (...)
# ON DUPLICATE KEY UPDATE price=VALUES(price), volume=VALUES(volume)
九、合规红线
- ✅ 走官方 API，合法合规
- ❌ 禁止爬取用户隐私（手机号、地址）
- ❌ 禁止直接展示价格做比价（需通过淘宝客链接跳转）
- 遵守《淘宝开放平台规则》及《个人信息保护法》
一句话总结： 注册拿密钥 → MD5签名 → 循环分页 → 控速防封 → 存库分析。整个链路跑通后，批量拉取万级商品数据只是几分钟的事。
