1688
店铺搜索接口多用于供应商挖掘、同行店铺盘点。网上大部分示例只演示基础签名与循环分页,忽略 B
端特有问题:搜索结果会混入关闭店铺、个体散户,并且服务端动态权重排序会造成分页偏移,同一店铺多次重复出现,直接循环翻页会造成统计结果失真。本文不做基础参数科普,聚焦业务层过滤、偏移去重、异常兼容,提供一套面向供应商筛选的生产级封装代码。 该接口核心痛点:total 总计数仅供参考,超过一定页码后统计值不准;部分店铺已经停业注销,但接口依然返回记录;翻页过程平台实时重排,造成店铺重复;中文关键词必须做好编码,否则签名直接报错。开发前需要在开放平台申请对应接口权限,获取 app_key、app_secret,服务器 IP 加入白名单。 python 代码做三处差异化处理:第一,以 memberId 作为唯一标识,集合全局去重,专门解决 B 端动态排序带来的分页偏移重复;第二,增加停业店铺过滤,剔除已经失效的供应商;第三,设置最大页码保护,避免 total 计数不准造成无限空跑。 对接调试要点:签名报错优先核对参数排序、毫秒时间戳;中文关键词不需要手动编码,底层请求库自动处理;不要去掉 sleep,平台会按 app_key 做限流。total 字段仅做参考,业务统计以实际返回并过滤后的数据集为准。
import requests
import time
import hashlib
class Ali1688ShopSearch:
def __init__(self, app_key, app_secret):
self.app_key = app_key
self.app_secret = app_secret
self.api_url = "https://gw.open.1688.com/openapi/http"
self.shop_member_set = set() # 店铺memberId全局去重,解决分页偏移
self.session = requests.Session()
def make_sign(self, params):
sorted_param = sorted(params.items(), key=lambda x:x[0])
raw_str = self.app_secret + "".join(f"{k}{v}" for k,v in sorted_param) + self.app_secret
return hashlib.md5(raw_str.encode("utf-8")).hexdigest().upper()
def query_single_page(self, keyword, page_no, page_size=20):
timestamp = str(int(time.time()*1000))
params = {
"app_key": self.app_key,
"method": "alibaba.seller.search",
"timestamp": timestamp,
"format": "json",
"v": "2.0",
"q": keyword,
"pageNo": page_no,
"pageSize": min(page_size,20)
}
params["sign"] = self.make_sign(params)
try:
resp = self.session.post(self.api_url, data=params, timeout=12)
return resp.json()
except Exception:
return {}
def search_shop_list(self, keyword, max_page=30):
shop_result = []
page = 1
while page <= max_page:
time.sleep(1.0) # QPS限流适配
resp_json = self.query_single_page(keyword, page)
resp_body = resp_json.get("alibaba_seller_search_response", {})
shop_list = resp_body.get("sellerList", [])
if not shop_list:
break
for shop in shop_list:
member_id = shop.get("memberId")
if not member_id or member_id in self.shop_member_set:
continue
# 过滤停业、无效店铺
if shop.get("isClosed"):
continue
self.shop_member_set.add(member_id)
shop_result.append({
"member_id": member_id,
"shop_name": shop.get("shopName",""),
"company_name": shop.get("companyName",""),
"is_factory": shop.get("isFactory",False),
"credit_level": shop.get("creditLevel","")
})
page += 1
return {"valid_shop_count": len(shop_result), "shop_data": shop_result}
if __name__ == "__main__":
client = Ali1688ShopSearch("your_app_key","your_app_secret")
res = client.search_shop_list("五金配件")
print("筛选后有效供应商数量:", res["valid_shop_count"])