前言 货源寻源、工厂比价、样品溯源、跨境选品系统开发中，1688
识图检索是供应链核心能力。网络现有教程大多照搬零售平台识图逻辑，只实现基础 URL 传图，完全忽略 1688
批发平台特有业务字段，同时缺少本地图片 Base64
上传兼容、相似度分层筛选、供应商资质解析、限流阶梯休眠等生产能力，批量寻源时匹配精度低、数据杂乱。本文基于 1688 官方 一、本文差异化设计亮点 双图片上传模式兼容：同时支持线上图片 URL、本地文件转 Base64 上传，解决线下样品图片无公网地址无法检索的痛点，全网教程很少同时实现两种传参方案。 B2B 批发专属字段完整解析：单独提取最小起批量、梯度批发价、实力商家标识、30 天销量、供应商所在地，区别零售平台仅解析零售价的简易代码。 相似度分层过滤：自定义阈值区分同款、相似商品，支持精准寻源与宽泛货源检索两种业务模式，自动过滤低匹配无关货源。 全局商品 ID 去重 + 分页边界控制：跨页重复商品自动剔除，内置最大分页上限，避免无限循环消耗接口配额。 分级异常阶梯休眠：区分签名错误、图片格式非法、429 限流、无匹配货源四类异常，限流时自动延长等待时间，防止应用权限冻结。 二、接口基础调用规范 网关地址： 三、完整可运行 Python 生产代码 四、生产环境避坑要点 本地图片 Base64 不可携带 data:image 前缀，代码内置纯编码转换逻辑，带前缀会直接匹配失败。 1688 签名规则与淘宝零售平台不同，必须首尾拼接 AppSecret，单边加密会持续返回 401 鉴权错误。 similarDegree 相似度阈值为批发寻源核心参数，0.8 以上仅返回高度同款工厂货源，0.5 可获取宽泛相似替代品。 minOrderQuantity 最小起批量是 B2B 特有字段，零售识图接口无该参数，货源比价系统必须优先读取。 批量循环识图时不可取消 1 秒休眠，1688 限流处罚周期长，高频调用会锁定接口 7 天以上。alibaba.product.image.search识图接口，从 B2B 批发业务角度重构调用逻辑，同时兼容公网图片 URL、本地图片 Base64 两种上传方式，适配工厂批量样品识图溯源场景。https://gw.open.1688.com/openapi/router/rest
接口方法：alibaba.product.image.search
鉴权规则：首尾包裹 MD5 签名，参数 ASCII 升序排序，10 位秒级时间戳
传参方式：POST 表单提交，支持 imgUrl 公网图、imageBase64 本地图二选一
调用约束：企业认证开发者 QPS≤3，单页最大 50 条，总分页上限 10 页
权限要求：1688 开放平台企业实名认证，单独开通图片搜索接口调用权限
import requests
import hashlib
import time
import json
import base64
from urllib.parse import quote
class AlibabaImgSearchClient:
def __init__(self, app_key, app_secret):
self.app_key = app_key
self.app_secret = app_secret
self.gateway = "https://gw.open.1688.com/openapi/router/rest"
self.session = requests.Session()
self.seen_product_ids = set()
def build_md5_sign(self, params):
# 1688官方首尾包裹MD5签名算法
valid_params = {k: v for k, v in params.items() if v and k != "sign"}
sorted_kv = sorted(valid_params.items(), key=lambda x: x[0])
sign_str = self.app_secret
for k, v in sorted_kv:
sign_str += f"{k}{quote(str(v), safe='')}"
sign_str += self.app_secret
return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
def local_img_to_base64(self, file_path):
"""本地图片转Base64字符串，适配线下样品识图"""
with open(file_path, "rb") as f:
return base64.b64encode(f.read()).decode("utf-8")
def single_page_search(self, threshold=0.7, page=1, page_size=20, img_url=None, img_base64=None):
params = {
"app_key": self.app_key,
"method": "alibaba.product.image.search",
"format": "json",
"v": "2.0",
"sign_method": "md5",
"timestamp": str(int(time.time()))
}
# 双图传参二选一
if img_url:
params["imgUrl"] = img_url
elif img_base64:
params["imageBase64"] = img_base64
else:
return {"code": -1, "msg": "必须传入图片URL或本地图片Base64", "data": []}
params["pageNo"] = page
params["pageSize"] = page_size
params["similarDegree"] = threshold
params["sign"] = self.build_md5_sign(params)
try:
resp = self.session.post(self.gateway, data=params, timeout=15)
res_data = resp.json()
# 限流阶梯重试
if res_data.get("error_response", {}).get("code") == 429:
time.sleep(3)
return self.single_page_search(threshold, page, page_size, img_url, img_base64)
if "error_response" in res_data:
return {"code": -1, "msg": res_data["error_response"]["sub_msg"], "data": []}
resp_body = res_data.get("alibaba_product_image_search_response", {})
raw_list = resp_body.get("productList", [])
clean_list = []
for item in raw_list:
pid = item.get("productId")
similarity = float(item.get("similarity", 0))
# 阈值过滤+全局去重
if similarity < threshold or pid in self.seen_product_ids:
continue
self.seen_product_ids.add(pid)
# B2B批发结构化清洗
clean_list.append({
"product_id": pid,
"title": item.get("title"),
"similar_score": similarity,
"min_order": int(item.get("minOrderQuantity", 1)),
"wholesale_price": float(item.get("salePrice", 0)),
"main_img": item.get("imageUrl"),
"supplier": item.get("sellerName"),
"supplier_addr": item.get("companyLocation", ""),
"is_elite_supplier": bool(int(item.get("isEliteShop", 0))),
"month_sales": int(item.get("monthSales", 0))
})
time.sleep(1)
return {"code": 200, "total": resp_body.get("totalResults"), "data": clean_list}
except Exception as e:
return {"code": -2, "msg": f"网络异常：{str(e)}", "data": []}
# 调用示例
if __name__ == "__main__":
client = AlibabaImgSearchClient("你的AppKey", "你的AppSecret")
# 方式1：公网图片URL检索
result = client.single_page_search(threshold=0.7, img_url="https://xxx.test.com/demo.jpg")
# 方式2：本地样品图片Base64检索
# base64_str = client.local_img_to_base64("./sample.jpg")
# result = client.single_page_search(threshold=0.7, img_base64=base64_str)
print(json.dumps(result, ensure_ascii=False, indent=2))
