淘宝图搜API接口技术深度解析：从架构设计到工程实践

┌─────────────────────────────────────────────────────────────┐
│                    应用接入层 (API Gateway)                   │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐  │
│  │  签名验证    │  │  流量控制    │  │  权限校验 (OAuth2)   │  │
│  └─────────────┘  └─────────────┘  └─────────────────────┘  │
└─────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────┐
│                   视觉计算层 (Visual Computing)               │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐  │
│  │ 图像预处理   │  │ 特征提取     │  │ 相似度计算 (ANN)     │  │
│  │ (Resize/    │  │ (ResNet/     │  │ (Cosine/Euclidean)  │  │
│  │  Normalize) │  │  MobileNet)  │  │                     │  │
│  └─────────────┘  └─────────────┘  └─────────────────────┘  │
└─────────────────────────────────────────────────────────────┘
                              ↓
┌─────────────────────────────────────────────────────────────┐
│                   数据检索层 (Search Engine)                  │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐  │
│  │  向量索引    │  │  倒排索引    │  │  排序模型 (LTR)      │  │
│  │ (Faiss/     │  │  (类目/价格)  │  │  (销量/信用/相似度)  │  │
│  │  HNSW)      │  │              │  │                     │  │
│  └─────────────┘  └─────────────┘  └─────────────────────┘  │
└─────────────────────────────────────────────────────────────┘

import hashlibimport urllib.parseimport timedef generate_top_sign(params: dict, app_secret: str, sign_method: str = "md5") -> str:
    """
    生成淘宝开放平台TOP签名
    规则：
    1. 过滤sign、空值、文件类型参数
    2. 按参数名ASCII升序排序
    3. 拼接字符串：secret + key1value1key2value2... + secret
    4. MD5加密并转大写
    """
    # 1. 过滤参数
    filtered_params = {
        k: v for k, v in params.items() 
        if k != "sign" and v is not None and not hasattr(v, "read")
    }
    
    # 2. 排序并拼接
    sorted_params = sorted(filtered_params.items())
    param_str = "".join([f"{k}{v}" for k, v in sorted_params])
    
    # 3. 构建签名字符串
    sign_str = f"{app_secret}{param_str}{app_secret}"
    
    # 4. 加密
    if sign_method == "md5":
        sign = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
    elif sign_method == "hmac-sha1":
        import hmac
        sign = hmac.new(
            app_secret.encode("utf-8"), 
            sign_str.encode("utf-8"), 
            hashlib.sha1        ).hexdigest().upper()
    
    return sign

params = {
    "method": "taobao.item_search_img",
    "app_key": "12345678",
    "timestamp": "2025-02-14 14:30:00",
    "format": "json",
    "v": "2.0",
    "sign_method": "md5",
    "imgid": "https://img.alicdn.com/xxx.jpg",
    "page": 1}sign = generate_top_sign(params, "your_app_secret")# 结果：E7B6A6C1D9F8E8B0A1C2D3E4F5A6B7C8

{
  "item_search_img_response": {
    "request_id": "4F2C8A1B3D5E7F9A",
    "code": 200,
    "msg": "success",
    "items": {
      "page": 1,
      "real_total_results": 1256,
      "total_results": 1256,
      "pagecount": 63,
      "item": [ ... ]
    }
  }}

{
  "num_iid": "575727312808",
  "title": "亲子装短袖T恤一家三口夏季新款",
  "pic_url": "https://g-search3.alicdn.com/img/bao/uploaded/...jpg",
  "detail_url": "https://item.taobao.com/item.htm?id=575727312808",
  "price": "39.50",
  "original_price": "99.00",
  "sales": 1523,
  "seller_id": "2200733087881719",
  "seller_nick": "某某旗舰店",
  "shop_name": "某某服饰专营店",
  "is_tmall": false,
  "match_rate": 0.95,
  "location": "浙江 杭州",
  "cat_id": "50010788",
  "cat_name": "女装"}

def classify_match_result(match_rate: float) -> dict:
    """
    根据相似度评分对搜索结果进行分类处理
    """
    if match_rate >= 0.90:
        return {
            "category": "same_item",  # 同款
            "confidence": "high",
            "action": "direct_compare",  # 直接比价
            "color": "#52c41a"  # 绿色标识
        }
    elif match_rate >= 0.80:
        return {
            "category": "similar_item",  # 相似款
            "confidence": "medium",
            "action": "manual_review",  # 人工复核
            "color": "#faad14"  # 黄色标识
        }
    elif match_rate >= 0.60:
        return {
            "category": "related_item",  # 相关款
            "confidence": "low",
            "action": "reference_only",  # 仅供参考
            "color": "#d9d9d9"  # 灰色标识
        }
    else:
        return {
            "category": "noise",  # 噪音
            "confidence": "none",
            "action": "filter_out",  # 过滤
            "color": "#ff4d4f"  # 红色标识
        }

import requestsimport base64import hashlibimport timeimport jsonfrom typing import Optional, Dict, List, Unionfrom dataclasses import dataclassfrom enum import Enumclass SortType(Enum):
    DEFAULT = "default"
    PRICE_ASC = "price_asc"
    PRICE_DESC = "price_desc"
    SALES = "sales"@dataclassclass SearchResult:
    """搜索结果数据类"""
    num_iid: str
    title: str
    price: float
    sales: int
    pic_url: str
    detail_url: str
    match_rate: float
    is_tmall: bool
    seller_nick: str
    location: strclass TaobaoImageSearchSDK:
    """
    淘宝图搜API SDK
    支持图片URL搜索和本地文件上传搜索
    """
    
    def __init__(self, app_key: str, app_secret: str, sandbox: bool = False):
        self.app_key = app_key
        self.app_secret = app_secret
        self.base_url = (
            "https://gw.api.tbsandbox.com/router/rest" if sandbox 
            else "https://eco.taobao.com/router/rest"
        )
        self.session = requests.Session()
        self.session.headers.update({
            "Content-Type": "application/x-www-form-urlencoded;charset=utf-8"
        })
    
    def _generate_sign(self, params: Dict) -> str:
        """生成TOP签名"""
        filtered = {k: v for k, v in params.items() if v is not None and k != "sign"}
        sorted_params = sorted(filtered.items())
        sign_str = f"{self.app_secret}{''.join([f'{k}{v}' for k, v in sorted_params])}{self.app_secret}"
        return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
    
    def _build_params(self, method: str, **kwargs) -> Dict:
        """构建基础参数"""
        params = {
            "method": method,
            "app_key": self.app_key,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
            **kwargs        }
        params["sign"] = self._generate_sign(params)
        return params    
    def upload_image(self, image_path: str) -> str:
        """
        上传本地图片获取imgid
        支持先调用taobao.picture.upload接口
        """
        with open(image_path, "rb") as f:
            image_data = base64.b64encode(f.read()).decode("utf-8")
        
        params = self._build_params(
            "taobao.picture.upload",
            image=image_data,
            picture_category_id="0"
        )
        
        response = self.session.post(self.base_url, data=params)
        result = response.json()
        
        if "error_response" in result:
            raise Exception(f"上传失败: {result['error_response']['msg']}")
        
        return result["picture_upload_response"]["picture"]["picture_id"]
    
    def search_by_image(
        self,
        img_url: Optional[str] = None,
        imgid: Optional[str] = None,
        cat: Optional[int] = None,
        page: int = 1,
        page_size: int = 20,
        sort: SortType = SortType.DEFAULT,
        price_min: Optional[float] = None,
        price_max: Optional[float] = None,
        match_threshold: float = 0.6
    ) -> List[SearchResult]:
        """
        以图搜商品核心方法
        
        Args:
            img_url: 图片URL（外部图片需确保可访问）
            imgid: 淘宝图片ID（通过upload_image获取）
            cat: 类目ID
            page: 页码
            page_size: 每页条数
            sort: 排序方式
            price_min: 最低价格
            price_max: 最高价格
            match_threshold: 相似度阈值，低于此值的结果将被过滤
        
        Returns:
            List[SearchResult]: 搜索结果列表
        """
        if not img_url and not imgid:
            raise ValueError("img_url 和 imgid 必须提供一个")
        
        kwargs = {
            "page": page,
            "page_size": page_size,
            "sort": sort.value,
            "match_threshold": match_threshold        }
        
        if img_url:
            kwargs["imgid"] = img_url        else:
            kwargs["imgid"] = imgid            
        if cat:
            kwargs["cat"] = cat        if price_min:
            kwargs["price_min"] = price_min        if price_max:
            kwargs["price_max"] = price_max
        
        params = self._build_params("taobao.item_search_img", **kwargs)
        
        try:
            response = self.session.post(self.base_url, data=params, timeout=30)
            response.raise_for_status()
        except requests.exceptions.RequestException as e:
            raise Exception(f"网络请求异常: {e}")
        
        result = response.json()
        
        # 错误处理
        if "error_response" in result:
            error = result["error_response"]
            error_code = error.get("code", "unknown")
            if error_code == "27":
                raise Exception("访问令牌过期或无效")
            elif error_code == "25":
                raise Exception("签名错误，请检查参数排序和密钥")
            elif error_code == "29":
                raise Exception("应用权限不足，请申请item_search_img接口权限")
            else:
                raise Exception(f"API错误 [{error_code}]: {error.get('msg')}")
        
        # 解析结果
        items_data = result.get("item_search_img_response", {}).get("items", {})
        items = items_data.get("item", [])
        
        # 转换为数据类并过滤
        results = []
        for item in items:
            match_rate = float(item.get("match_rate", 0))
            if match_rate < match_threshold:
                continue
                
            results.append(SearchResult(
                num_iid=str(item.get("num_iid")),
                title=item.get("title", ""),
                price=float(item.get("price", 0)),
                sales=int(item.get("sales", 0)),
                pic_url=item.get("pic_url", ""),
                detail_url=item.get("detail_url", ""),
                match_rate=match_rate,
                is_tmall=item.get("is_tmall", False),
                seller_nick=item.get("seller_nick", ""),
                location=item.get("location", "")
            ))
        
        return results    
    def search_with_local_image(
        self,
        image_path: str,
        **kwargs    ) -> List[SearchResult]:
        """
        便捷方法：直接上传本地图片并搜索
        """
        imgid = self.upload_image(image_path)
        return self.search_by_image(imgid=imgid, **kwargs)# 使用示例if __name__ == "__main__":
    sdk = TaobaoImageSearchSDK(
        app_key="your_app_key",
        app_secret="your_app_secret",
        sandbox=False
    )
    
    # 方式1：使用图片URL搜索
    results = sdk.search_by_image(
        img_url="https://example.com/product.jpg",
        cat=50010788,  # 女装类目
        sort=SortType.SALES,
        match_threshold=0.85
    )
    
    # 方式2：使用本地图片搜索
    # results = sdk.search_with_local_image(
    #     image_path="/path/to/local/image.jpg",
    #     price_min=100,
    #     price_max=500
    # )
    
    # 输出结果
    for idx, item in enumerate(results, 1):
        match_type = "同款" if item.match_rate >= 0.9 else "相似款"
        platform = "天猫" if item.is_tmall else "淘宝"
        print(f"{idx}. [{match_type}][{platform}] {item.title}")
        print(f"   价格: ¥{item.price}  销量: {item.sales}  相似度: {item.match_rate:.1%}")
        print(f"   店铺: {item.seller_nick}  地区: {item.location}")
        print(f"   链接: {item.detail_url}\n")

import asyncioimport aiohttpfrom concurrent.futures import ThreadPoolExecutorimport functoolsclass AsyncTaobaoImageSearch:
    """异步高性能版本"""
    
    def __init__(self, app_key: str, app_secret: str, max_workers: int = 10):
        self.app_key = app_key
        self.app_secret = app_secret
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        self.semaphore = asyncio.Semaphore(max_workers)  # 限流保护
    
    async def batch_search(
        self,
        image_urls: List[str],
        **kwargs    ) -> List[List[SearchResult]]:
        """
        批量异步搜索
        """
        loop = asyncio.get_event_loop()
        sdk = TaobaoImageSearchSDK(self.app_key, self.app_secret)
        
        async def search_one(url):
            async with self.semaphore:
                # 在线程池中执行同步SDK调用，避免阻塞事件循环
                return await loop.run_in_executor(
                    self.executor,
                    functools.partial(sdk.search_by_image, img_url=url, **kwargs)
                )
        
        tasks = [search_one(url) for url in image_urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

import redisimport picklefrom functools import wrapsdef cache_result(expire: int = 3600):
    """
    Redis缓存装饰器
    """
    def decorator(func):
        @wraps(func)
        def wrapper(self, *args, **kwargs):
            # 生成缓存key
            cache_key = f"taobao:imgsearch:{hash(str(args) + str(kwargs))}"
            
            # 尝试读取缓存
            try:
                cached = self.redis_client.get(cache_key)
                if cached:
                    return pickle.loads(cached)
            except redis.RedisError:
                pass
            
            # 执行原函数
            result = func(self, *args, **kwargs)
            
            # 写入缓存
            try:
                self.redis_client.setex(cache_key, expire, pickle.dumps(result))
            except redis.RedisError:
                pass
            
            return result        return wrapper    return decorator

import randomimport timefrom tenacity import retry, stop_after_attempt, wait_exponentialclass SearchError(Exception):
    pass@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((requests.exceptions.RequestException, SearchError)),
    before_sleep=lambda retry_state: print(f"重试第{retry_state.attempt_number}次..."))def robust_search(sdk, **kwargs):
    """
    带重试机制的健壮搜索
    """
    try:
        return sdk.search_by_image(**kwargs)
    except Exception as e:
        if "超限" in str(e) or "frequency" in str(e).lower():
            # 遇到限流，添加随机抖动避免雪崩
            time.sleep(random.uniform(0.5, 1.5))
            raise SearchError(str(e))
        raise