创建应用：

申请 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

五、应用场景电商 APP：实现拍照搜商品功能，提升用户体验供应链管理：通过图片快速匹配供应商和货源竞品分析：抓取竞争对手的主打产品进行分析智能选品：根据热门商品图片推荐类似产品

通过以上步骤，开发者可以完整实现淘宝拍立淘 API 的接入与应用。建议在正式上线前，先在沙盒环境进行充分测试，并根据实际业务需求调整参数和处理逻辑。

标签：