微店搜索接口和主流电商存在明显差异,除常规商品基础信息外,自带分销佣金、拼团标记等特有业务字段。很多开发只做简单请求,上线后遇到分页穿透、token
过期无重试、佣金字段缺失、下架商品混入结果集等问题。本文不从基础参数讲解切入,聚焦生产环境的数据清洗、分页边界处理、自动令牌刷新,适配选品系统、ERP
商品同步业务场景。 微店开放平台搜索接口采用 OAuth2 鉴权,access_token
有效期仅 2
小时,批量任务很容易出现中途鉴权失效。另外接口存在分页黑洞:页码超过平台内部阈值,不会返回报错,只会持续返回空列表,若直接循环页码会造成死循环。同时价格单位为分,不做转换直接入库会出现价格放大
100 倍的 bug,分销佣金字段仅部分商品返回,直接读取会触发字典异常。 下面封装类集成 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)