在 B
端供应链数据分析场景中,1688 开放平台关键词搜索商品接口被大量用于货源监控、供应商筛选、价格行情统计。网络上大部分示例只演示简单调用获取原始
JSON,很少关注 B 端接口独有的问题:混合批发 /
一件代发商品、多规格冗余字段、接口返回部分空数据、访问频次限制。本文不从基础鉴权入门切入,重点讲解业务层数据清洗、异常报文兼容,提供可直接用于业务项目的封装代码。 1688
搜索接口和各大电商平台最大差异在于返回数据掺杂大量营销字段,直接原始输出会给后续统计带来干扰;同时接口存在部分商品字段返回
null,若不做容错处理,业务程序极易直接抛出异常中断。另外接口有 QPS 限制,短时间批量调用会触发返回错误码,需要做请求间隔控制。 生产示例代码 关键逻辑解读 offerId 集合去重:部分情况下接口分页会出现少量重复商品,利用 offerId 做集合去重,避免统计重复货源。 业务字段裁剪:舍弃营销标签、图片数组等大体积冗余数据,只保留供应链分析核心字段,减少内存占用。 空值兼容处理:对 title、供应商名称全部设置默认空字符串,防止 key 不存在直接导致程序崩溃。 主动休眠控频:内置请求间隔,规避开放平台 QPS 超限返回错误,降低任务失败概率。 实际对接踩坑小结 搜索结果同时包含批发、代发、定制类商品,业务需要时可以增加字段过滤筛选对应货源类型。 价格返回为区间字符串,不是数值,后续做价格统计需要二次解析处理。 token 存在有效期,批量长时间任务需要增加 token 过期捕获逻辑。
import time
import requests
class Ali1688SearchClient:
def __init__(self, app_key, app_secret, access_token):
self.app_key = app_key
self.app_secret = app_secret
self.access_token = access_token
self.api_url = "https://gw.open.1688.com/openapi/param2/1/com.alibaba.search/offerSearch"
def search_goods(self, keyword, page_size=20, max_fetch_page=2):
result_data = []
seen_offer_id = set()
for page in range(1, max_fetch_page + 1):
time.sleep(1.2)
params = {
"app_key": self.app_key,
"access_token": self.access_token,
"keywords": keyword,
"page": page,
"pageSize": page_size
}
resp = requests.get(self.api_url, params=params, timeout=12)
res_json = resp.json()
if res_json.get("error_response"):
break
offer_list = res_json.get("result", {}).get("offerList", [])
for offer in offer_list:
offer_id = offer.get("offerId")
if not offer_id or offer_id in seen_offer_id:
continue
seen_offer_id.add(offer_id)
item = {
"offerId": offer_id,
"title": offer.get("title", ""),
"price": offer.get("priceRange", ""),
"supplierName": offer.get("supplier", {}).get("companyName", ""),
"isOneBatch": offer.get("supportMixWholesale", False)
}
result_data.append(item)
return result_data
if __name__ == "__main__":
client = Ali1688SearchClient("your_app_key", "your_app_secret", "your_token")
output = client.search_goods("家用收纳盒", max_fetch_page=2)
print(output)