×

1688 店铺列表搜索接口:B 端供应商检索的分页偏移与脏数据过滤方案

Ace Ace 发表于2026-08-21 17:29:26 浏览13 评论0

抢沙发发表评论

1688 店铺搜索接口多用于供应商挖掘、同行店铺盘点。网上大部分示例只演示基础签名与循环分页,忽略 B 端特有问题:搜索结果会混入关闭店铺、个体散户,并且服务端动态权重排序会造成分页偏移,同一店铺多次重复出现,直接循环翻页会造成统计结果失真。本文不做基础参数科普,聚焦业务层过滤、偏移去重、异常兼容,提供一套面向供应商筛选的生产级封装代码。

该接口核心痛点:total 总计数仅供参考,超过一定页码后统计值不准;部分店铺已经停业注销,但接口依然返回记录;翻页过程平台实时重排,造成店铺重复;中文关键词必须做好编码,否则签名直接报错。开发前需要在开放平台申请对应接口权限,获取 app_key、app_secret,服务器 IP 加入白名单。

python

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"])

代码做三处差异化处理:第一,以 memberId 作为唯一标识,集合全局去重,专门解决 B 端动态排序带来的分页偏移重复;第二,增加停业店铺过滤,剔除已经失效的供应商;第三,设置最大页码保护,避免 total 计数不准造成无限空跑。

对接调试要点:签名报错优先核对参数排序、毫秒时间戳;中文关键词不需要手动编码,底层请求库自动处理;不要去掉 sleep,平台会按 app_key 做限流。total 字段仅做参考,业务统计以实际返回并过滤后的数据集为准。

群贤毕至

访客