1688平台API接口实战：Python实现店铺全量商品数据抓取

在电商数据驱动决策的时代，1688作为国内最大的B2B批发平台，其开放的API接口为商家提供了高效获取商品数据的通道。本文将以Python语言为例，详解如何通过官方接口实现店铺所有商品的自动化抓取。（综合参考）

一、接口核心能力与开发准备

接口功能概述
通过alibaba.trade.product.search4trade接口（或1688.item_search_shop），开发者可获取指定店铺的商品列表数据，涵盖商品ID、标题、主图、价格、库存、销量等核心字段，支持分页查询以覆盖全量商品。
开发前准备
• 账号注册：前往1688开放平台完成企业开发者实名认证，创建应用获取App Key和App Secret。
• 权限申请：在控制台申请“商品信息查询”类API权限，审核周期通常为1-3个工作日。
• 环境依赖：Python 3.6+、Requests库、Hashlib库（用于签名生成）。

二、接口调用全流程解析

1. 请求参数构建

关键参数包括： • method：接口方法名（如alibaba.trade.product.search4trade）
• seller_member_id：目标店铺ID（可通过商家后台或公开页面获取）
• page_no/page_size：分页参数（建议每页50-100条以平衡效率与稳定性）
• fields：指定返回字段（如product_id,title,price,pic_url）

params = {"method": "alibaba.trade.product.search4trade","app_key": "your_app_key","seller_member_id": "123456789","page_no": 1,"page_size": 50,"timestamp": int(time.time()),"v": "2.0","format": "json"
}

2. 签名生成机制

1688 API采用MD5加密签名验证请求合法性，核心步骤：

将除sign外的参数按字母升序排列
拼接键值对为字符串，末尾追加App Secret
对拼接结果进行MD5哈希并转为大写

def generate_sign(params, secret):params_sorted = sorted(params.items())sign_str = "&".join([f"{k}{v}" for k,v in params_sorted if k != "sign"])sign = hashlib.md5((sign_str + secret).encode('utf-8')).hexdigest().upper()return signparams['sign'] = generate_sign(params, APP_SECRET)

3. 分页数据抓取逻辑

由于单次请求最多返回100条数据，需通过循环翻页直至无新数据返回：

def fetch_all_products(max_retries=3):page_no = 1all_products = []while True:try:products = request_api(page_no)if not products: breakall_products.extend(products)page_no += 1except Exception as e:if max_retries <= 0: raisemax_retries -= 1time.sleep(5)return all_products

三、数据解析与存储优化

响应结构解析
成功响应为JSON格式，核心数据位于result.product_list.product数组中。需特别关注字段：
• product_id：商品唯一标识（用于详情接口二次调用）
• price_range：价格区间（批发场景常见）
• moq：最小起订量
存储建议
• 数据库选型：MySQL/MongoDB适用于中小规模数据；ClickHouse适合海量数据分析
• 增量同步：通过update_time字段实现增量更新，降低API调用压力

四、高频问题与避坑指南

QPS限制
免费版API通常限制5-10次/秒，企业版可申请提升至50+次/秒。建议添加time.sleep(0.2)避免触发限流。
数据一致性
• 定期校验数据总量（对比接口返回total_results与实际条目数）
• 设置重试机制应对网络抖动（示例代码包含3次重试）
安全规范
• 禁止硬编码密钥，建议使用环境变量或密钥管理系统
• HTTPS强制加密传输，避免中间人攻击

五、典型应用场景

竞品监控：周期性抓取竞对店铺SKU变化，分析价格策略调整
库存预警：实时同步库存数据，对接ERP系统自动补货
选品分析：统计热销商品类目分布，指导采购决策

完整代码示例

import requests
import hashlib
import timeAPP_KEY = "YOUR_APP_KEY"
APP_SECRET = "YOUR_APP_SECRET"
SHOP_ID = "TARGET_SHOP_ID"def generate_sign(params, secret):# 签名生成函数（同上）def get_products(page):url = "https://eco.1688.com/router/rest"params = {"method": "alibaba.trade.product.search4trade","app_key": APP_KEY,"seller_member_id": SHOP_ID,"page_no": page,"page_size": 50,"timestamp": int(time.time()),"v": "2.0","format": "json"}params["sign"] = generate_sign(params, APP_SECRET)try:resp = requests.get(url, params=params, timeout=10)if resp.status_code == 200:data = resp.json()if data.get("code") == "200":return data["result"]["product_list"]["product"]return Noneexcept Exception as e:print(f"Page {page} failed: {str(e)}")return None# 调用示例
all_products = []
current_page = 1
while True:products = get_products(current_page)if not products: breakall_products.extend(products)current_page += 1time.sleep(0.5)  # 控制请求频率print(f"Total products fetched: {len(all_products)}")