1688 的 item_search_shop 接口是阿里巴巴开放平台提供的重要接口,用于获取指定店铺的所有商品信息。该接口对于电商数据分析、竞品监控、供应链管理等场景具有重要价值,能够帮助开发者批量获取店铺商品的详细数据。
一、接口核心特性分析
1. 接口功能与定位
2. 接口认证机制
3. 核心参数与响应结构
请求参数
响应结构
二、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 开放平台的相关规定,确保数据使用的合法性和合规性。