唯品会以特卖闪购为核心,其开放平台关键词搜索接口和普通电商逻辑差异很大。网上多数示例只演示基础调用,忽略特卖商品实时上下架带来的分页漂移、结果重复、混入过期专场商品等生产问题,直接上线会出现数据重复、统计失真。本文不从基础参数讲解切入,聚焦业务侧的去重、异常过滤、签名容错,给出一套可用于选品分析的封装实现。 该接口最大的坑点并非签名本身,而是业务层面:特卖商品会动态上下架,翻页过程中排序发生变动,同一商品会出现在多个页码;同时接口会返回已经结束特卖的商品,如果不做过滤,业务统计会引入大量无效数据。另外时间戳必须使用
13 位毫秒,时间偏差过大会直接鉴权失败,很多示例没有做校验处理。 前置条件:完成开放平台企业认证,申请搜索接口权限,获取 app_key、app_secret,服务器 IP 加入白名单。 python 代码中核心做了三处业务优化:第一,使用 set 集合保存商品 ID,专门解决特卖场景分页漂移带来的数据重复;第二,增加 对接调试常见问题:签名报错优先检查参数排序、毫秒时间戳、密钥前后有无空格;返回结果为空,排查 IP 白名单、接口权限是否开通;批量采集时不可去掉 sleep,平台会按 app_key+IP 双重限流。
import requests
import time
import hashlib
class VipKeywordSearch:
def __init__(self, app_key, app_secret):
self.app_key = app_key
self.app_secret = app_secret
self.api_url = "https://openapi.vip.com/router/rest"
self.item_set = set() # 用于去重,解决分页漂移
self.session = requests.Session()
def build_sign(self, params):
# 唯品会签名:按键名ASCII升序排序
sorted_items = sorted(params.items(), key=lambda x: x[0])
raw = self.app_secret + "".join(f"{k}{v}" for k, v in sorted_items) + self.app_secret
return hashlib.md5(raw.encode("utf‑8")).hexdigest().upper()
def query_page(self, keyword, page, page_size=20):
timestamp = str(int(time.time() * 1000))
params = {
"appKey": self.app_key,
"method": "vip.item.search",
"timestamp": timestamp,
"format": "json",
"v": "1.0",
"keyword": keyword,
"pageNo": page,
"pageSize": min(page_size,20)
}
params["sign"] = self.build_sign(params)
try:
resp = self.session.post(self.api_url, data=params, timeout=10)
return resp.json()
except Exception:
return {}
def search_all(self, keyword):
result_list = []
page = 1
while True:
time.sleep(1.2) # 适配平台QPS限制
resp_data = self.query_page(keyword, page)
resp_body = resp_data.get("vip_item_search_response", {})
item_list = resp_body.get("itemList", [])
if not item_list:
break
for item in item_list:
item_id = item.get("itemId")
if not item_id or item_id in self.item_set:
continue
# 过滤已结束特卖的商品
if item.get("isOnSale") != 1:
continue
self.item_set.add(item_id)
result_list.append({
"item_id": item_id,
"title": item.get("title",""),
"sale_price": item.get("salePrice",0),
"original_price": item.get("originalPrice",0),
"brand": item.get("brandName","")
})
total = resp_body.get("total",0)
if page * 20 >= total:
break
page += 1
return {"count": len(result_list), "goods": result_list}
if __name__ == "__main__":
client = VipKeywordSearch("your_app_key","your_app_secret")
res = client.search_all("女士连衣裙")
print("有效在售商品数量:", res["count"])isOnSale状态过滤,剔除专场结束的失效商品;第三,控制请求间隔,避免触发平台限流风控。