×

唯品会 item_search_img 接口深度分析及 Python 实现

万邦科技Lex 万邦科技Lex 发表于2025-09-04 16:59:51 浏览250 评论0

抢沙发发表评论

       注册账号免费测试唯品会API接口

唯品会的 item_search_img 接口(又称 "唯品会拍立淘")是唯品会开放平台提供的图像搜索接口,支持通过图片(URL 或本地图片)搜索唯品会平台上的同款或相似商品。该接口基于图像识别技术,能够快速匹配视觉特征相似的商品,广泛应用于商品导购、同款比价、库存查找等场景。

一、接口核心特性分析

1. 接口功能与定位

  • 核心功能:通过图片特征匹配唯品会平台商品,返回相似商品列表及详细信息

  • 技术原理:基于深度学习的图像特征提取与比对,支持商品主体识别和特征匹配

  • 应用场景

    • 商品导购:用户上传图片快速找到同款商品

    • 比价工具:通过图片找到同款商品进行价格对比

    • 库存查询:商家通过样品图片查询平台库存情况

    • 市场分析:根据热销商品图片分析市场趋势

2. 认证机制

唯品会开放平台采用 appkey + access_token 的认证方式:


  • 开发者在唯品会开放平台注册应用,获取 appkey 和 appsecret

  • 使用 appkey 和 appsecret 获取 access_token(有有效期限制)

  • 每次接口调用时,在请求参数中携带 access_token 进行身份验证

3. 核心参数与响应结构

请求参数

参数名类型是否必填说明
image_urlString二选一图片 URL(公网可访问)
image_base64String二选一本地图片 Base64 编码(不含前缀)
pageInteger页码,默认 1
page_sizeInteger每页数量,默认 20,最大 50
sortString排序方式:similarity(相似度)、price_asc(价格升序)等
access_tokenString访问令牌

响应核心字段

  • 分页信息:总商品数、总页数、当前页码

  • 商品列表:每个商品包含

    • 基础信息:商品 ID、标题、主图、详情链接

    • 价格信息:原价、折扣价、折扣率

    • 相似度信息:与搜索图片的匹配度

    • 店铺信息:店铺名称、类型

    • 库存信息:是否有货、大致库存范围

二、Python 脚本实现

以下是调用唯品会 item_search_img 接口的完整实现,支持图片 URL 和本地图片两种搜索方式:

import requests
import time
import json
import base64
import logging
from typing import Dict, Optional, List
from requests.exceptions import RequestException
from pathlib import Path

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

class VipImageSearchAPI:
    def __init__(self, appkey: str, appsecret: str):
        """
        初始化唯品会图片搜索API客户端
        :param appkey: 唯品会开放平台appkey
        :param appsecret: 唯品会开放平台appsecret
        """
        self.appkey = appkey
        self.appsecret = appsecret
        self.base_url = "https://api.vip.com"
        self.access_token = None
        self.token_expires_at = 0  # token过期时间戳
        self.session = requests.Session()
        self.session.headers.update({
            "Content-Type": "application/json",
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
        })

    def _get_access_token(self) -> Optional[str]:
        """获取访问令牌"""
        # 检查token是否有效
        if self.access_token and self.token_expires_at > time.time() + 60:
            return self.access_token
            
        logging.info("获取新的access_token")
        url = f"{self.base_url}/oauth2/token"
        
        params = {
            "grant_type": "client_credentials",
            "client_id": self.appkey,
            "client_secret": self.appsecret
        }
        
        try:
            response = self.session.get(url, params=params, timeout=10)
            response.raise_for_status()
            result = response.json()
            
            if "access_token" in result:
                self.access_token = result["access_token"]
                self.token_expires_at = time.time() + result.get("expires_in", 3600)
                return self.access_token
            else:
                logging.error(f"获取access_token失败: {result.get('error_description', '未知错误')}")
                return None
                
        except RequestException as e:
            logging.error(f"获取access_token请求异常: {str(e)}")
            return None

    def _local_image_to_base64(self, image_path: str) -> Optional[str]:
        """将本地图片转换为Base64编码"""
        image_file = Path(image_path)
        
        # 检查文件是否存在
        if not image_file.exists() or not image_file.is_file():
            logging.error(f"图片文件不存在: {image_path}")
            return None
            
        # 检查文件格式
        valid_extensions = ['.jpg', '.jpeg', '.png', '.gif']
        if image_file.suffix.lower() not in valid_extensions:
            logging.error(f"不支持的图片格式: {image_file.suffix},支持格式: {valid_extensions}")
            return None
            
        try:
            # 读取并编码图片
            with open(image_path, 'rb') as f:
                image_data = f.read()
                # 检查图片大小,避免过大
                if len(image_data) > 10 * 1024 * 1024:  # 10MB
                    logging.error("图片大小超过10MB限制")
                    return None
                base64_str = base64.b64encode(image_data).decode('utf-8')
            return base64_str
        except Exception as e:
            logging.error(f"图片编码失败: {str(e)}")
            return None

    def search_by_image(self, 
                       image_url: Optional[str] = None, 
                       image_path: Optional[str] = None,
                       page: int = 1, 
                       page_size: int = 20,
                       sort: str = "similarity") -> Optional[Dict]:
        """
        按图片搜索唯品会商品
        :param image_url: 图片URL(二选一)
        :param image_path: 本地图片路径(二选一)
        :param page: 页码
        :param page_size: 每页数量
        :param sort: 排序方式
        :return: 搜索结果
        """
        # 验证图片参数
        if not image_url and not image_path:
            logging.error("必须提供image_url或image_path参数")
            return None
            
        # 获取有效的access_token
        if not self._get_access_token():
            return None
            
        url = f"{self.base_url}/item/search/img"
        
        # 构建请求参数
        params = {
            "page": page,
            "page_size": page_size,
            "sort": sort,
            "access_token": self.access_token
        }
        
        # 添加图片参数
        if image_url:
            params["image_url"] = image_url
        else:
            # 处理本地图片
            base64_image = self._local_image_to_base64(image_path)
            if not base64_image:
                return None
            params["image_base64"] = base64_image
            
        try:
            response = self.session.get(url, params=params, timeout=20)  # 图片搜索耗时较长
            response.raise_for_status()
            result = response.json()
            
            # 检查响应状态
            if result.get("code") == 0:
                # 格式化响应数据
                return self._format_response(result.get("data", {}))
            else:
                logging.error(f"图片搜索失败: {result.get('message', '未知错误')} (错误码: {result.get('code')})")
                return None
                
        except RequestException as e:
            logging.error(f"图片搜索请求异常: {str(e)}")
            return None
        except json.JSONDecodeError:
            logging.error(f"图片搜索响应解析失败: {response.text[:200]}...")
            return None

    def _format_response(self, response_data: Dict) -> Dict:
        """格式化响应数据"""
        # 分页信息
        pagination = {
            "total_items": int(response_data.get("total", 0)),
            "total_pages": (int(response_data.get("total", 0)) + int(response_data.get("page_size", 20)) - 1) // int(response_data.get("page_size", 20)),
            "current_page": int(response_data.get("page", 1)),
            "page_size": int(response_data.get("page_size", 20))
        }
        
        # 格式化商品列表
        products = []
        for item in response_data.get("items", []):
            products.append({
                "goods_id": item.get("goods_id"),
                "title": item.get("title"),
                "main_image": item.get("main_image"),
                "detail_url": item.get("detail_url"),
                "price": {
                    "original_price": float(item.get("original_price", 0)),
                    "vip_price": float(item.get("vip_price", 0)),
                    "discount": float(item.get("discount", 0))
                },
                "similarity": float(item.get("similarity", 0)),  # 相似度(0-100)
                "stock_status": item.get("stock_status"),  # 库存状态
                "shop_info": {
                    "shop_id": item.get("shop_id"),
                    "shop_name": item.get("shop_name"),
                    "shop_type": item.get("shop_type")
                },
                "brand_info": {
                    "brand_id": item.get("brand_id"),
                    "brand_name": item.get("brand_name")
                }
            })
            
        return {
            "pagination": pagination,
            "products": products,
            "search_id": response_data.get("search_id")  # 搜索ID,用于后续操作
        }

    def search_all_pages(self, image_url: Optional[str] = None, 
                        image_path: Optional[str] = None,
                        max_pages: int = 5) -> List[Dict]:
        """
        获取多页搜索结果
        :param image_url: 图片URL
        :param image_path: 本地图片路径
        :param max_pages: 最大页数限制
        :return: 所有商品列表
        """
        all_products = []
        page = 1
        
        while page <= max_pages:
            logging.info(f"获取第 {page} 页搜索结果")
            result = self.search_by_image(
                image_url=image_url,
                image_path=image_path,
                page=page,
                page_size=50  # 使用最大页大小减少请求次数
            )
            
            if not result or not result["products"]:
                break
                
            all_products.extend(result["products"])
            
            # 检查是否已到最后一页
            if page >= result["pagination"]["total_pages"]:
                break
                
            page += 1
            # 添加延迟,避免触发频率限制
            time.sleep(2)  # 图片搜索接口限制更严格,延迟更长一些
            
        return all_products


# 示例调用
if __name__ == "__main__":
    # 替换为实际的appkey和appsecret(从唯品会开放平台获取)
    APPKEY = "your_appkey"
    APPSECRET = "your_appsecret"
    
    # 初始化API客户端
    api = VipImageSearchAPI(APPKEY, APPSECRET)
    
    # 方式1:通过图片URL搜索
    # search_result = api.search_by_image(
    #     image_url="https://img.vip.com/xxx.jpg",  # 替换为实际图片URL
    #     page=1,
    #     page_size=20,
    #     sort="similarity"
    # )
    
    # 方式2:通过本地图片搜索
    search_result = api.search_by_image(
        image_path="./product_sample.jpg",  # 替换为本地图片路径
        page=1,
        page_size=20,
        sort="similarity"
    )
    
    # 方式3:获取多页结果
    # search_result = api.search_all_pages(
    #     image_path="./product_sample.jpg",
    #     max_pages=3
    # )
    
    if isinstance(search_result, dict) and "products" in search_result:
        print(f"共找到 {search_result['pagination']['total_items']} 件相似商品")
        print(f"当前第 {search_result['pagination']['current_page']}/{search_result['pagination']['total_pages']} 页\n")
        
        # 打印前5件商品信息
        for i, product in enumerate(search_result["products"][:5], 1):
            print(f"{i}. {product['title']} (相似度: {product['similarity']}%)")
            print(f"   品牌: {product['brand_info']['brand_name']}")
            print(f"   原价: {product['price']['original_price']}元 → 折扣价: {product['price']['vip_price']}元 ({product['price']['discount']}折)")
            print(f"   店铺: {product['shop_info']['shop_name']} ({product['shop_info']['shop_type']})")
            print(f"   库存状态: {product['stock_status']}")
            print(f"   链接: {product['detail_url']}")
            print("-" * 100)
    elif isinstance(search_result, list):
        # 处理多页结果
        print(f"共获取到 {len(search_result)} 件商品")   
 
 
三、接口调用关键技术与注意事项
1. 图片处理最佳实践
图片质量:清晰的商品主体图片(无水印、无遮挡)识别效果最佳
图片尺寸:建议图片尺寸在 500x500 到 1000x1000 像素之间
图片格式:优先使用 JPG 格式,识别成功率高于 PNG 和 GIF
Base64 处理:
必须移除 Base64 编码前缀(如data:image/jpeg;base64,)
图片大小控制在 10MB 以内,过大可能导致请求失败
URL 图片:确保图片 URL 为公网可访问,响应速度快的图片源识别效率更高
2. 常见错误及解决方案
错误码说明解决方案
400    请求参数错误    检查图片参数是否正确提供,格式是否符合要求    
401    未授权或 token 无效    重新获取 access_token    
403    权限不足    检查应用是否已申请图片搜索接口权限    
413    请求实体过大    图片大小超过限制,压缩图片后重试    
429    调用频率超限    降低调用频率,增加请求间隔    
500    服务器内部错误    稍后重试,或联系唯品会技术支持    
10001    图片处理失败    检查图片是否损坏,尝试更换图片    
3. 性能优化建议
请求频率控制:图片搜索接口 QPS 限制通常较低(5 次 / 秒左右),需严格控制调用频率
图片预处理:对本地图片进行压缩和裁剪,突出商品主体,提高识别准确率
结果缓存:相同图片的搜索结果可缓存 30-60 分钟,减少重复调用
异步处理:对于批量图片搜索,采用异步队列方式处理,避免超
相似度过滤:可设置相似度阈值(如 > 70%),过滤低相似度结果
四、应用场景与扩展
典型应用场景
智能导购系统:用户上传图片快速找到同款商品
跨平台比价工具:通过图片识别不同平台的同款商品进行价格对比
库存查询系统:商家通过样品图片查询平台库存情况
电商选品工具:根据市场热销商品图片寻找类似款式商品
扩展建议
结合商品详情接口获取更完整的产品信息
实现相似度排序和过滤,只展示高相似度商品
开发批量图片搜索功能,支持多张图片同时查询
添加图片编辑功能,允许用户裁剪、旋转图片以提高识别率
实现搜索历史记录功能,方便用户查看之前的搜索结果

通过合理使用唯品会图片搜索接口,开发者可以构建便捷的商品查找工具,提升用户体验,同时为电商运营提供市场分析数据支持。使用时需遵守唯品会开放平台的使用规范,确保数据使用的合法性。

群贤毕至

访客