×

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

万邦科技Lex 万邦科技Lex 发表于2025-08-31 16:34:18 浏览235 评论0

抢沙发发表评论

注册账号免费测试1688API数据接口

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

一、接口核心特性分析

1. 接口功能与定位

  • 功能:获取指定 1688 店铺内的商品列表及详情信息

  • 应用场景

    • 竞品分析:监控竞争对手店铺的商品结构和价格策略

    • 供应链管理:批量获取供应商店铺的商品信息

    • 电商代运营:帮助商家管理多个店铺的商品数据

    • 数据分析:分析特定行业或类目下的商品趋势

2. 接口认证机制

1688 开放平台采用 appkey + 签名 的认证方式,具体流程如下:


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

  • 每次请求时,将所有参数按规则排序并生成签名

  • 签名与参数一同发送,1688 服务器验证签名有效性

3. 核心参数与响应结构

请求参数

参数名类型是否必填说明
methodString接口方法名,固定为 alibaba.item.search.shop
app_keyString应用的 appkey
timestampString时间戳,格式为 yyyy-MM-dd HH:mm:ss
formatString响应格式,默认 json
vString接口版本,固定为 2.0
signString签名
shop_idString店铺 ID
pageNumber页码,默认 1
page_sizeNumber每页条数,默认 40,最大 100
sortString排序方式,如 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 开放平台的相关规定,确保数据使用的合法性和合规性。

群贤毕至

访客