分享淘宝拍立淘 API 全实操指南

2025-09-22 16:05:09 6阅读

创建应用：

申请 API 权限：

3. 获取密钥二、API 调用流程1. 认证与授权

拍立淘 API 支持两种认证方式：

python

# 方式一：使用AppKey+AppSecret直接调用（无用户授权）
import hashlib
import time
import requests
app_key = "你的AppKey"
app_secret = "你的AppSecret"
def generate_sign(params):
    """生成API签名"""
    # 1. 参数排序
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    # 2. 拼接字符串
    sign_str = app_secret + ''.join([f"{k}{v}" for k, v in sorted_params]) + app_secret
    # 3. MD5加密
    return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
# 方式二：OAuth2.0授权（需用户授权，获取access_token）
# 1. 引导用户访问授权页面
auth_url = f"https://oauth.taobao.com/authorize?response_type=code&client_id={app_key}&redirect_uri=你的回调URL&state=STATE"
# 2. 用户授权后，通过code换取access_token
def get_access_token(code):
    token_url = "https://oauth.taobao.com/token"
    data = {
        "grant_type": "authorization_code",
        "client_id": app_key,
        "client_secret": app_secret,
        "code": code,
        "redirect_uri": "你的回调URL"
    }
    response = requests.post(token_url, data=data)
    return response.json()["access_token"]

2. 图片搜索 API 调用

支持图片 URL和图片文件上传两种方式：

# 方式一：通过图片URL搜索
def search_by_image_url(image_url):
    url = "https://eco.taobao.com/router/rest"
    timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    
    # 请求参数
    params = {
        "method": "taobao.image.search",
        "app_key": app_key,
        "timestamp": timestamp,
        "format": "json",
        "v": "2.0",
        "sign_method": "md5",
        "image_url": image_url,  # 图片URL
        "q": "服装",  # 搜索关键词（可选）
        "cat": "50008898",  # 类目ID（可选）
    }
    
    # 生成签名
    params["sign"] = generate_sign(params)
    
    # 发送请求
    response = requests.post(url, data=params)
    return response.json()
# 方式二：通过图片文件上传搜索
def search_by_image_file(file_path):
    url = "https://eco.taobao.com/router/rest"
    timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    
    # 请求参数
    params = {
        "method": "taobao.image.search",
        "app_key": app_key,
        "timestamp": timestamp,
        "format": "json",
        "v": "2.0",
        "sign_method": "md5",
        "q": "服装",
        "cat": "50008898",
    }
    
    # 生成签名
    params["sign"] = generate_sign(params)
    
    # 读取图片文件
    files = {"image": open(file_path, "rb")}
    
    # 发送请求（包含文件上传）
    response = requests.post(url, data=params, files=files)
    return response.json()

3. 响应结果解析

成功的响应包含商品列表、匹配度等信息：

# 解析搜索结果
def parse_search_result(result):
    if "error_response" in result:
        error = result["error_response"]
        print(f"API错误: {error['code']} - {error['msg']}")
        return []
    
    items = []
    if "image_search_response" in result:
        response = result["image_search_response"]
        if "item_list" in response and response["item_list"]:
            for item in response["item_list"]["item"]:
                items.append({
                    "item_id": item.get("item_id"),
                    "title": item.get("title"),
                    "price": item.get("price"),
                    "pic_url": item.get("pic_url"),
                    "match_rate": item.get("match_rate"),  # 匹配度（百分比）
                    "seller_nick": item.get("seller_nick")
                })
    
    return items
# 使用示例
result = search_by_image_url("https://example.com/test.jpg")
items = parse_search_result(result)
for item in items:
    print(f"商品: {item['title']}, 价格: {item['price']}, 匹配度: {item['match_rate']}%")

三、常见问题与解决方案1. 签名错误2. 权限不足3. 图片处理限制4. 频率限制四、最佳实践1. 图片预处理

from PIL import Image
import io
def optimize_image(file_path, max_size=5*1024*1024):
    """优化图片以符合API要求"""
    img = Image.open(file_path)
    
    # 调整尺寸
    width, height = img.size
    if width < 300 or height < 300:
        img = img.resize((max(300, width), max(300, height)), Image.LANCZOS)
    
    # 压缩质量
    img_byte_arr = io.BytesIO()
    img.save(img_byte_arr, format='JPEG', quality=85)
    img_byte_arr = img_byte_arr.getvalue()
    
    # 如果仍超过大小限制，进一步压缩
    while len(img_byte_arr) > max_size:
        img = Image.open(io.BytesIO(img_byte_arr))
        img_byte_arr = io.BytesIO()
        quality = max(50, int(quality * 0.9))  # 每次降低10%质量
        img.save(img_byte_arr, format='JPEG', quality=quality)
        img_byte_arr = img_byte_arr.getvalue()
    
    return img_byte_arr

2. 异步批量处理

import asyncio
import aiohttp
async def search_image_async(session, image_url):
    """异步调用拍立淘API"""
    timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    params = {
        "method": "taobao.image.search",
        "app_key": app_key,
        "timestamp": timestamp,
        "format": "json",
        "v": "2.0",
        "sign_method": "md5",
        "image_url": image_url,
    }
    params["sign"] = generate_sign(params)
    
    async with session.post("https://eco.taobao.com/router/rest", data=params) as response:
        return await response.json()
async def batch_search(image_urls):
    """批量异步搜索"""
    async with aiohttp.ClientSession() as session:
        tasks = []
        for url in image_urls:
            tasks.append(search_image_async(session, url))
        
        # 控制并发数，避免触发限流
        semaphore = asyncio.Semaphore(10)  # 限制为10个并发
        async def sem_task(task):
            async with semaphore:
                return await task
        
        results = await asyncio.gather(*[sem_task(task) for task in tasks])
        return results

3. 缓存策略

import redis
# 初始化Redis缓存
redis_client = redis.Redis(host='localhost', port=6379, db=0)
def cached_image_search(image_url):
    """带缓存的图片搜索"""
    # 生成缓存键
    cache_key = f"image_search:{hashlib.md5(image_url.encode()).hexdigest()}"
    
    # 检查缓存
    cached_result = redis_client.get(cache_key)
    if cached_result:
        return json.loads(cached_result)
    
    # 无缓存，调用API
    result = search_by_image_url(image_url)
    
    # 缓存结果（有效期1小时）
    redis_client.setex(cache_key, 3600, json.dumps(result))
    
    return result