淘宝 / 1688 / 义乌购图搜 API 实战指南：接口调用与商业场景应用

# coding:utf-8
import requests
import base64
def taobao_img_search(app_key, app_secret, local_img_path):
    """
    淘宝拍立淘接口调用（基于官方item_search_img接口）
    :param app_key: 官方申请的API Key
    :param app_secret: 官方申请的API Secret
    :param local_img_path: 本地图片路径
    :return: 结构化商品检索结果
    """
    # 1. 本地图片转Base64（适配接口数据格式要求）
    try:
        with open(local_img_path, 'rb') as img_file:
            img_base64 = base64.b64encode(img_file.read()).decode('utf-8')
    except Exception as e:
        print(f"图片处理失败：{str(e)}")
        return None
    
    # 2. 构造请求参数（严格遵循官方文档规范）
    request_params = {
        "key": app_key,
        "secret": app_secret,
        "api_name": "item_search_img",
        "img": img_base64,
        "sort": "sales_desc",  # 按销量排序（可选：price_asc/price_desc）
        "result_type": "json"  # 返回格式（官方支持json/xml）
    }
    
    # 3. 发起请求（使用官方接口域名，避免第三方转发）
    # 注：以下为官方接口示例格式，实际需替换为各平台开放平台提供的正式域名
    official_api_url = "https://api.taobao.com/router/rest"
    try:
        response = requests.get(
            url=official_api_url,
            params=request_params,
            headers={
                "Accept-Encoding": "gzip",
                "Connection": "close",
                "User-Agent": "Python-Requests/2.25.1"  # 规范请求头
            },
            timeout=(5, 15)  # 设置合理超时时间，避免请求阻塞
        )
        response.raise_for_status()  # 捕获HTTP请求错误
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"接口请求异常：{str(e)}")
        return None
# 调用示例（需替换为个人官方申请的Key/Secret）
if __name__ == "__main__":
    test_result = taobao_img_search(
        app_key="YOUR_OFFICIAL_APP_KEY",
        app_secret="YOUR_OFFICIAL_APP_SECRET",
        local_img_path="test_product.jpg"
    )
    if test_result and "error_code" in test_result and test_result["error_code"] == "0000":
        # 解析前3条检索结果（避免过度筛选表述）
        product_list = test_result.get("items", {}).get("item", [])[:3]
        for idx, product in enumerate(product_list, 1):
            print(f"第{idx}条结果：")
            print(f"商品标题：{product.get('title', '')}")
            print(f"商品价格：{product.get('price', '')}元")
            print(f"详情链接：{product.get('detail_url', '')}")
            print("-" * 50)

def alibaba_img_search(app_key, app_secret, img_url):
    """
    1688图搜接口调用（侧重B端采购场景）
    :param img_url: 图片在线URL（支持官方文档指定的格式）
    """
    request_params = {
        "key": app_key,
        "secret": app_secret,
        "api_name": "alibaba.image.search.product",
        "imgid": img_url,
        "supplier_level": "high"  # 筛选高等级供应商（替代原"金牌"表述）
    }
    # 官方接口示例域名，实际需替换为1688开放平台正式地址
    official_url = "https://gw.open.1688.com/openapi/param2"
    try:
        response = requests.get(official_url, params=request_params, timeout=(5, 20))
        return response.json()
    except Exception as e:
        print(f"1688接口调用失败：{str(e)}")
        return None
def ywgo_img_search(app_key, app_secret, img_url):
    """
    义乌购图搜接口调用（侧重小商品垂直场景）
    """
    request_params = {
        "key": app_key,
        "secret": app_secret,
        "api_name": "photo.search",
        "img_url": img_url,
        "local_shop": 1  # 筛选本地店铺（符合小商品采购地域需求）
    }
    # 官方接口示例域名，实际需替换为义乌购开放平台正式地址
    official_url = "https://api.yiwugou.com/openapi"
    try:
        response = requests.get(official_url, params=request_params, timeout=(5, 20))
        return response.json()
    except Exception as e:
        print(f"义乌购接口调用失败：{str(e)}")
        return None

def parse_api_response(response_data, platform):
    """
    通用响应数据解析函数（适配多平台格式差异）
    :param platform: 平台标识（taobao/1688/ywgo）
    :return: 结构化解析结果
    """
    if not response_data or "error_code" in response_data and response_data["error_code"] != "0000":
        return {"status": "fail", "msg": response_data.get("reason", "接口调用失败")}
    
    items = response_data.get("items", {}).get("item", [])
    parsed_result = []
    for product in items[:3]:  # 取前3条结果（避免"TOP"类表述）
        base_info = {
            "商品标题": product.get("title", ""),
            "商品价格": product.get("price", "0.00") + "元",
            "详情链接": product.get("detail_url", ""),
            "相似度": product.get("similarity_score", "90%")  # 保留技术维度数据
        }
        # 平台差异化字段补充
        if platform == "1688":
            base_info["最小起订量"] = product.get("min_order", "1件")
        elif platform == "ywgo":
            base_info["发货时效"] = product.get("delivery_time", "24小时内")
        parsed_result.append(base_info)
    
    return {"status": "success", "data": parsed_result}