1688 的 item_search_img 接口（又称 "1688 拍立淘"）是阿里巴巴开放平台提供的图像搜索接口，支持通过图片（URL 或本地图片）搜索 1688 平台上的同款或相似商品。该接口基于阿里巴巴的图像识别技术，能够快速匹配商品库中视觉特征相似的商品，广泛应用于供应商查找、同款比价、供应链管理等场景。
一、接口核心特性分析
1. 接口功能与定位
2. 认证机制
3. 核心参数与响应结构
请求参数
响应核心字段
二、Python 脚本实现
import requests
import hashlib
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 AlibabaImageSearchAPI:
def __init__(self, appkey: str, appsecret: str):
"""
初始化1688图片搜索接口客户端
:param appkey: 1688开放平台appkey
:param appsecret: 1688开放平台appsecret
"""
self.appkey = appkey
self.appsecret = appsecret
self.base_url = "https://gw.open.1688.com/openapi/http/1/system.oauth2"
self.session = requests.Session()
def _generate_sign(self, params: Dict) -> str:
"""生成1688 API签名"""
# 1. 按参数名ASCII升序排序
sorted_params = sorted(params.items(), key=lambda x: x[0])
# 2. 拼接为key=value&key=value格式
sign_str = "&".join([f"{k}={v}" for k, v in sorted_params])
# 3. 拼接appsecret并MD5加密
sign_str += self.appsecret
return hashlib.md5(sign_str.encode()).hexdigest().upper()
def _get_timestamp(self) -> str:
"""生成时间戳（yyyy-MM-dd HH:mm:ss）"""
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
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()
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 = 40,
sort: Optional[str] = None) -> Optional[Dict]:
"""
按图片搜索1688商品
: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
# 构建基础参数
params = {
"method": "alibaba.item.search.img",
"app_key": self.appkey,
"timestamp": self._get_timestamp(),
"format": "json",
"v": "2.0",
"page": str(page),
"page_size": str(page_size)
}
# 添加排序参数
if sort:
params["sort"] = sort
# 添加图片参数
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_image
# 生成签名
params["sign"] = self._generate_sign(params)
try:
# 发送请求
response = self.session.get(
self.base_url,
params=params,
timeout=20 # 图片搜索耗时较长，设置较长超时
)
response.raise_for_status()
result = response.json()
# 处理错误响应
if "error_response" in result:
error = result["error_response"]
logging.error(f"接口错误: {error.get('msg', '未知错误')} (错误码: {error.get('code', '未知')})")
return None
# 格式化响应数据
return self._format_response(result)
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: Dict) -> Dict:
"""格式化响应数据"""
result = response.get("alibaba_item_search_img_response", {})
data = result.get("result", {})
# 提取分页信息
pagination = {
"total_items": int(data.get("total_results", 0)),
"total_pages": (int(data.get("total_results", 0)) + int(data.get("page_size", 40)) - 1) // int(data.get("page_size", 40)),
"current_page": int(data.get("page", 1)),
"page_size": int(data.get("page_size", 40))
}
# 格式化商品列表
products = []
for item in data.get("items", {}).get("item", []):
product = {
"product_id": item.get("product_id"),
"title": item.get("title"),
"main_image": item.get("main_image"),
"detail_url": item.get("detail_url"),
"price": {
"min_price": float(item.get("min_price", 0)),
"max_price": float(item.get("max_price", 0)),
"unit": item.get("unit", "")
},
"wholesale_price": item.get("wholesale_price", []), # 批发价格区间
"moq": int(item.get("moq", 0)), # 最小起订量
"transaction": {
"sales_count": int(item.get("sales_count", 0)), # 30天销量
"buyer_count": int(item.get("buyer_count", 0)), # 买家数
"repeat_rate": float(item.get("repeat_rate", 0)) # 重复采购率
},
"supplier": {
"supplier_id": item.get("supplier_id"),
"supplier_name": item.get("supplier_name"),
"location": item.get("location"),
"credit_level": item.get("credit_level"),
"year": int(item.get("year", 0)) # 经营年限
},
"similarity": float(item.get("similarity", 0)) # 相似度（0-100）
}
products.append(product)
return {
"pagination": pagination,
"products": products,
"search_id": 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=100 # 使用最大页大小减少请求次数
)
if not result or not result["products"]:
break
all_products.extend(result["products"])
# 检查是否已到最后一页
if page >= result["pagination"]["total_pages"]:
break
page += 1
# 添加延迟，避免触发频率限制
time.sleep(1.5)
return all_products
# 示例调用
if __name__ == "__main__":
# 替换为实际的appkey和appsecret（从1688开放平台获取）
APPKEY = "your_appkey"
APPSECRET = "your_appsecret"
# 初始化API客户端
api = AlibabaImageSearchAPI(APPKEY, APPSECRET)
# 方式1：通过图片URL搜索
# search_result = api.search_by_image(
# image_url="https://img.alicdn.com/imgextra/i3/xxx.jpg", # 替换为实际图片URL
# page=1,
# page_size=20,
# sort="sales"
# )
# 方式2：通过本地图片搜索
search_result = api.search_by_image(
image_path="./product_sample.jpg", # 替换为本地图片路径
page=1,
page_size=20,
sort="sales"
)
# 方式3：获取多页结果
# search_result = api.search_all_pages(
# image_path="./product_sample.jpg",
# max_pages=3
# )
if search_result:
# 处理单页结果
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['price']['min_price']}-{product['price']['max_price']}{product['price']['unit']}")
print(f" 起订量: {product['moq']}{product['price']['unit']}")
print(f" 30天销量: {product['transaction']['sales_count']} 件")
print(f" 供应商: {product['supplier']['supplier_name']} ({product['supplier']['location']})")
print(f" 链接: {product['detail_url']}")
print("-" * 100)
else:
# 处理多页结果
print(f"共获取到 {len(search_result)} 件商品")