前言 网络上大部分
1688 搜索接口示例仅实现单次请求调用，缺少业务场景适配。1688 作为 B2B
批发平台，搜索接口存在独有的特性：阶梯报价返回、起批量参数筛选、分页限制、开放平台签名校验严格、调用频率管控。本文围绕货源比价业务需求，构建带分页遍历、参数封装、数据标准化、异常熔断的调用方案，适合后台货源采集系统直接集成。 一、接入前置条件 注册 1688 开放平台应用，获取 目标接口：alibaba.offer.search（商品关键词搜索接口） 接口约束：单页最大 20 条数据；支持筛选起订量、产地；短时间高频请求容易触发限流。 二、Python 完整实现代码 三、方案差异化设计说明 增加最大页数限制 内置起批量筛选参数，贴合 1688 批发业务，这是普通零售平台搜索接口不存在的需求； 返回数据统一标准化，剔除大量无用原始字段，直接适配数据库存储结构； 可控请求间隔，规避开放平台限流拦截。 四、生产环境踩坑要点 签名易错点：时间戳必须毫秒级，参数名大小写不能修改，排序拼接不能漏任何公共参数； 权限限制：普通开发者应用无法获取真实最低批发价，需要企业资质认证； 关键词编码：中文关键词框架内部自动编码，不要手动二次 urlencode，避免参数错乱； 不要无限分页：接口返回商品达到平台搜索上限后不会继续返回数据，做好终止判断。appKey、appSecret；完成 OAuth 授权拿到access_token。
import time
import hashlib
import requests
class Ali1688SearchClient:
def __init__(self, app_key, app_secret, token):
self.app_key = app_key
self.app_secret = app_secret
self.access_token = token
self.gateway = "https://gw.open.1688.com/openapi/param2/import"
# 生成MD5签名，严格遵循平台升序拼接规则
def build_sign(self, param_dict):
sort_params = sorted(param_dict.items())
raw = "".join(f"{k}{v}" for k, v in sort_params) + self.app_secret
return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
def search_page(self, keyword, page_no=1, page_size=20, min_order=1):
params = {
"app_key": self.app_key,
"timestamp": str(int(time.time() * 1000)),
"format": "json",
"v": "1",
"method": "alibaba.offer.search",
"access_token": self.access_token,
"keywords": keyword,
"pageNo": page_no,
"pageSize": page_size,
"minOrderQuantity": min_order
}
params["sign"] = self.build_sign(params)
resp = requests.get(self.gateway, params=params, timeout=12)
return resp.json()
# 批量获取多页搜索结果，自动分页终止
def full_search(self, keyword, max_page=5):
result_list = []
current_page = 1
while current_page <= max_page:
try:
res_data = self.search_page(keyword, current_page)
if "error_response" in res_data:
print(f"接口异常：{res_data['error_response']['msg']}")
break
items = res_data.get("result", {}).get("resultList", [])
if not items:
break
result_list.extend(items)
current_page += 1
time.sleep(0.55)
except Exception as err:
print(f"网络异常：{str(err)}")
time.sleep(1)
# 标准化输出字段，方便入库
standardized = []
for item in result_list:
standardized.append({
"item_id": item.get("offerId"),
"title": item.get("subject"),
"price_info": item.get("priceRange"),
"main_image": item.get("mainImageUrl"),
"min_order": item.get("minOrderQuantity"),
"seller_id": item.get("sellerId")
})
return standardized
if __name__ == "__main__":
client = Ali1688SearchClient("appkey", "appsecret", "access_token")
goods = client.full_search("办公文件夹", max_page=3)
print(f"检索商品总数：{len(goods)}")max_page，防止无边界分页造成流量浪费，适配业务按需采集场景；
