1688 item_search_shop 接口深度分析及 Python 实现

1688 的 item_search_shop 接口是阿里巴巴开放平台提供的重要接口，用于获取指定店铺的所有商品信息。该接口对于电商数据分析、竞品监控、供应链管理等场景具有重要价值，能够帮助开发者批量获取店铺商品的详细数据。
一、接口核心特性分析

1. 接口功能与定位
   功能：获取指定 1688 店铺内的商品列表及详情信息
   应用场景：
   竞品分析：监控竞争对手店铺的商品结构和价格策略
   供应链管理：批量获取供应商店铺的商品信息
   电商代运营：帮助商家管理多个店铺的商品数据
   数据分析：分析特定行业或类目下的商品趋势
2. 接口认证机制
   1688 开放平台采用 appkey + 签名 的认证方式，具体流程如下：
   开发者在 1688 开放平台注册应用，获取 appkey 和 appsecret
   每次请求时，将所有参数按规则排序并生成签名
   签名与参数一同发送，1688 服务器验证签名有效性

3. 核心参数与响应结构
   请求参数
   参数名 类型 是否必填 说明
   method String 是 接口方法名，固定为 alibaba.item.search.shop
   app_key String 是 应用的 appkey
   timestamp String 是 时间戳，格式为 yyyy-MM-dd HH:mm:ss
   format String 否 响应格式，默认 json
   v String 是 接口版本，固定为 2.0
   sign String 是 签名
   shop_id String 是 店铺 ID
   page Number 否 页码，默认 1
   page_size Number 否 每页条数，默认 40，最大 100
   sort String 否 排序方式，如 price_asc 按价格升序
   响应结构
   响应主要包含以下几个部分：
   分页信息：总页数、当前页、总商品数
   商品列表：每个商品包含基础信息、价格、库存、销量等
   店铺信息：店铺名称、信用等级、主营类目等
   二、Python 脚本实现
   下面是调用 item_search_shop 接口的完整 Python 实现，包含签名生成、接口调用、数据解析等功能：
   import requests
   import hashlib
   import time
   import json
   import logging
   from typing import Dict, Optional, List
   from requests.exceptions import RequestException

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

class AlibabaItemSearchShopAPI:
def init(self, appkey: str, appsecret: str):
"""
初始化1688 item_search_shop接口客户端
: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 get_shop_items(self, shop_id: str, page: int = 1, page_size: int = 40, 
                   sort: Optional[str] = None) -> Optional[Dict]:
    """
    获取店铺商品列表
    :param shop_id: 店铺ID
    :param page: 页码
    :param page_size: 每页数量
    :param sort: 排序方式
    :return: 商品列表数据
    """
    # 构造请求参数
    params = {
        "method": "alibaba.item.search.shop",
        "app_key": self.appkey,
        "timestamp": self._get_timestamp(),
        "format": "json",
        "v": "2.0",
        "shop_id": shop_id,
        "page": str(page),
        "page_size": str(page_size)
    }

    # 添加可选的排序参数
    if sort:
        params["sort"] = sort

    # 生成签名
    params["sign"] = self._generate_sign(params)

    try:
        # 发送请求
        response = self.session.get(
            self.base_url,
            params=params,
            timeout=15
        )
        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_shop_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))
    }

    # 提取店铺信息
    shop_info = {
        "shop_id": data.get("shop_id"),
        "shop_name": data.get("shop_name"),
        "shop_type": data.get("shop_type"),
        "main_category": data.get("main_category"),
        "credit_level": data.get("credit_level")
    }

    # 提取并格式化商品列表
    products = []
    for item in data.get("items", {}).get("item", []):
        product = {
            "product_id": item.get("product_id"),
            "title": item.get("title"),
            "image_url": item.get("image_url"),
            "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", "")
            },
            "trade_info": {
                "sales_count": int(item.get("sales_count", 0)),
                "trade_count": int(item.get("trade_count", 0)),
                "comment_count": int(item.get("comment_count", 0))
            },
            "stock": int(item.get("stock", 0)),
            "category": item.get("category"),
            "tags": item.get("tags", "").split(","),
            "location": item.get("location")
        }
        products.append(product)

    return {
        "pagination": pagination,
        "shop_info": shop_info,
        "products": products
    }

def get_all_shop_items(self, shop_id: str, max_pages: int = 10) -> List[Dict]:
    """
    获取店铺所有商品（分页获取）
    :param shop_id: 店铺ID
    :param max_pages: 最大页数限制，防止请求过多
    :return: 所有商品列表
    """
    all_products = []
    page = 1

    while True:
        logging.info(f"获取店铺 {shop_id} 第 {page} 页商品")
        result = self.get_shop_items(shop_id, page=page, page_size=100)

        if not result or not result["products"]:
            break

        all_products.extend(result["products"])

        # 检查是否已达最后一页或最大页数限制
        if page >= result["pagination"]["total_pages"] or page >= max_pages:
            break

        page += 1
        # 添加延迟，避免触发频率限制
        time.sleep(1)

    return all_products

示例调用
if name == "main":

# 替换为实际的appkey和appsecret（从1688开放平台获取）
APPKEY = "your_appkey"
APPSECRET = "your_appsecret"
# 替换为目标店铺ID
SHOP_ID = "12345678"

# 初始化API客户端
api = AlibabaItemSearchShopAPI(APPKEY, APPSECRET)

# 方式1：获取指定页的商品
# shop_items = api.get_shop_items(SHOP_ID, page=1, page_size=20, sort="price_asc")

# 方式2：获取店铺所有商品（分页获取）
shop_items = api.get_all_shop_items(SHOP_ID, max_pages=5)

if shop_items:
    # 如果是完整响应，处理方式略有不同
    if isinstance(shop_items, dict) and "products" in shop_items:
        print(f"店铺 {shop_items['shop_info']['shop_name']} 共有 {shop_items['pagination']['total_items']} 件商品")
        products = shop_items["products"]
    else:
        print(f"成功获取店铺 {SHOP_ID} 的 {len(shop_items)} 件商品")
        products = shop_items

    # 打印前5件商品信息
    print("\n前5件商品信息：")
    for i, product in enumerate(products[:5], 1):
        print(f"{i}. {product['title']}")
        print(f"   价格：{product['price']['min_price']}-{product['price']['max_price']}{product['price']['unit']}")
        print(f"   销量：{product['trade_info']['sales_count']} 件")
        print(f"   库存：{product['stock']}")
        print(f"   链接：{product['detail_url']}")
        print("-" * 80)

三、接口调用注意事项

1. 调用限制与规范
   QPS 限制：默认 QPS 限制为 10，超过会返回 429 错误
   数据缓存：相同店铺的商品信息建议设置缓存，避免频繁调用
   分页处理：单页最大只能获取 100 条数据，大量商品需要分页获取
   合规使用：获取的商品数据不得用于不正当竞争或商业侵权行为
2. 常见错误及解决方案
   错误码 说明 解决方案
   10001 签名错误 检查签名生成逻辑，确保参数排序和加密正确
   20001 店铺 ID 不存在 确认 shop_id 是否正确有效
   403 没有权限访问 检查应用是否已申请相关接口权限
   429 调用频率超限 降低调用频率，实现请求限流
   500 服务器内部错误 稍后重试，或联系 1688 开放平台技术支持
3. 数据解析要点
   价格信息可能返回区间值（min_price 和 max_price），对应不同规格
   销量数据为 30 天内累计销量，部分店铺可能不公开
   商品详情 URL 需要拼接域名才能直接访问
   库存数据可能不是实时数据，存在一定延迟
   四、应用场景与扩展建议
   典型应用场景
   店铺商品采集与分析系统
   竞品价格监控工具
   供应链管理系统
   电商数据分析平台
   扩展建议
   实现数据持久化存储，支持历史数据分析
   添加商品价格变化趋势追踪功能
   结合商品详情接口，获取更全面的商品信息
   开发可视化报表，展示店铺商品结构和销售情况
   实现异常监控，如商品价格异常波动提醒
   通过合理使用 item_search_shop 接口，开发者可以高效获取 1688 店铺的商品数据，为电商运营和数据分析提供有力支持。使用时需遵守 1688 开放平台的相关规定，确保数据使用的合法性和合规性。