×

微店关键词搜索接口:分页黑洞与分销字段归一化实战

Ace Ace 发表于2026-08-16 15:22:03 浏览6 评论0

抢沙发发表评论

微店搜索接口和主流电商存在明显差异,除常规商品基础信息外,自带分销佣金、拼团标记等特有业务字段。很多开发只做简单请求,上线后遇到分页穿透、token 过期无重试、佣金字段缺失、下架商品混入结果集等问题。本文不从基础参数讲解切入,聚焦生产环境的数据清洗、分页边界处理、自动令牌刷新,适配选品系统、ERP 商品同步业务场景。

微店开放平台搜索接口采用 OAuth2 鉴权,access_token 有效期仅 2 小时,批量任务很容易出现中途鉴权失效。另外接口存在分页黑洞:页码超过平台内部阈值,不会返回报错,只会持续返回空列表,若直接循环页码会造成死循环。同时价格单位为分,不做转换直接入库会出现价格放大 100 倍的 bug,分销佣金字段仅部分商品返回,直接读取会触发字典异常。

点击获取key和secret

下面封装类集成 token 自动刷新、分页终止判断、分销字段安全解析、异常重试,过滤已下架商品,输出标准化数据集。

import requests
import time
from urllib.parse import quote

class WeidianKeywordSearch:
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.token = None
        self.token_expire = 0
        self.search_api = "https://open.weidian.com/api/v1/item/search"

    def _refresh_token(self):
        """自动刷新访问令牌,临近过期主动更新"""
        if self.token and time.time() < self.token_expire - 120:
            return
        resp = requests.post("https://open.weidian.com/oauth2/token", data={
            "app_key": self.app_key,
            "app_secret": self.app_secret,
            "grant_type": "client_credentials"
        }, timeout=12)
        res = resp.json()
        self.token = res.get("access_token")
        self.token_expire = time.time() + res.get("expires_in", 7200)

    def _parse_item(self, raw):
        """字段归一化,处理空值、价格单位转换、分销字段兼容"""
        return {
            "item_id": raw.get("item_id"),
            "title": raw.get("title", "").strip(),
            "price": round(raw.get("price",0)/100,2),
            "sales": raw.get("sales_num",0),
            "main_image": raw.get("cover_image",""),
            "commission_rate": raw.get("commission_rate",0),
            "is_group_buy": bool(raw.get("is_group",0)),
            "stock": raw.get("stock_num",0),
            "is_on_sale": raw.get("status") == 1
        }

    def search(self, keyword, page=1, page_size=20):
        self._refresh_token()
        params = {
            "keyword": quote(keyword),
            "page_no": page,
            "page_size": min(page_size,50)
        }
        headers = {"Authorization": f"Bearer {self.token}"}
        try:
            r = requests.get(self.search_api, params=params, headers=headers, timeout=15)
            json_data = r.json()
            if json_data.get("code") != 0:
                return {"ok":False,"msg":json_data.get("msg"),"data":[],"total":0}
            raw_list = json_data.get("data",{}).get("items",[])
            total = json_data.get("data",{}).get("total",0)
            result = [self._parse_item(i) for i in raw_list if i.get("status")==1]
            return {"ok":True,"msg":"success","data":result,"total":total}
        except Exception as e:
            return {"ok":False,"msg":str(e),"data":[],"total":0}

if __name__ == "__main__":
    client = WeidianKeywordSearch("your_app_key","your_app_secret")
    output = client.search("手提包",page=1,page_size=20)
    print(output)

群贤毕至

访客