前言 市面上多数教程仅简单调用单页商品接口，忽略
1688B2B
平台特有限制：店铺商品分页上限、批发规格字段缺失、账号权限分级、高频调用风控拦截。本文从业务落地角度，封装一套具备分页循环、异常重试、字段统一清洗的工具类，适配采购比价、货源爬虫后台等商用场景，区别于基础单请求示例。 一、前置接入准备 入驻 1688 开放平台，创建应用获取 核心接口：alibaba.offer.list.get（店铺商品列表查询）； 关键限制：单页最多 20 条，单次请求间隔最低 500ms，非企业应用无法读取批发底价。 二、完整 Python 封装代码 三、核心逻辑拆解（差异化要点） 自动分页循环：无需手动循环页码，接口无数据自动终止，避免死循环； 风控限流机制：每页请求强制休眠 600ms，解决平台高频封禁问题； 数据归一化处理：原始返回字段繁杂，统一提取业务核心字段，直接入库； 异常捕获机制：网络超时、接口报错自动捕获，中断时打印错误日志便于排查。 四、生产环境避坑总结 权限区分：个人开发者接口仅能获取商品标价，企业认证应用才可读取阶梯批发价； 分页阈值：pageSize 最大 20，传入更大数值会被平台强制截断； 签名规范：参数必须按字母升序拼接，时间戳为毫秒级，大小写 MD5 不匹配会鉴权失败； 数据存储：全店商品量超千件时，建议分批次落库，一次性存储易造成内存溢出。appKey、appSecret，完成店铺授权；
import time
import hashlib
import requests
class Ali1688ShopSpider:
def __init__(self, app_key, app_secret, access_token):
self.app_key = app_key
self.app_secret = app_secret
self.token = access_token
self.api_url = "https://gw.open.1688.com/openapi/param2/import"
# 生成平台标准MD5签名
def get_sign(self, params):
sorted_str = "".join([k+str(params[k]) for k in sorted(params)]) + self.app_secret
return hashlib.md5(sorted_str.encode()).hexdigest().upper()
# 单页商品请求
def get_page_goods(self, shop_id, page_num=1, page_size=20):
params = {
"app_key": self.app_key,
"timestamp": str(int(time.time()*1000)),
"format": "json",
"v": "1",
"method": "alibaba.offer.list.get",
"access_token": self.token,
"sellerId": shop_id,
"pageNo": page_num,
"pageSize": page_size
}
params["sign"] = self.get_sign(params)
resp = requests.get(self.api_url, params=params, timeout=10)
return resp.json()
# 全店商品循环拉取，自带限流重试
def get_all_shop_goods(self, shop_id):
all_goods = []
page = 1
while True:
try:
res = self.get_page_goods(shop_id, page)
if res.get("error_response"):
print("接口报错：", res["error_response"]["msg"])
break
data = res["result"]["resultList"]
if not data:
break
all_goods.extend(data)
page += 1
time.sleep(0.6) # 风控延时
except Exception as e:
print("请求异常重试", e)
time.sleep(2)
# 统一字段清洗归一化
clean_data = [{
"item_id": item["offerId"],
"title": item["subject"],
"price_range": item["priceRange"],
"pic_url": item["mainImageUrl"],
"stock": item["stockNumber"]
} for item in all_goods]
return clean_data
# 调用示例
if __name__ == "__main__":
client = Ali1688ShopSpider("你的appKey", "你的appSecret", "授权token")
goods_list = client.get_all_shop_goods("目标店铺sellerId")
print(f"共获取商品{len(goods_list)}件")
