前言 在电商数据采集场景中，唯品会搜索接口有个容易被忽略的特点：搜索结果接口采用动态分页 + 部分字段延迟返回机制。常规直接循环请求分页的写法，经常出现页码越界、部分商品 SKU 缺失、图片地址格式不统一等问题。本文不单纯讲解基础请求调用，重点实现一套自带分页边界检测、结果字段清洗、请求熔断的封装方案，适配业务侧批量拉取商品列表需求。 前置准备 Python3 环境，依赖 requests 库；请求时需要携带合理请求头模拟客户端访问，接口存在访问频率限制，必须增加间隔与异常重试，短时间高频请求会触发风控拦截。 完整代码示例 核心逻辑解析 Session 复用：使用 Session 保持会话，减少每次请求重建连接开销，降低风控识别概率。 分页自动终止：通过总商品数计算是否存在下一页，避免无效请求不存在的页码。 字段归一化：统一提取业务常用字段，对空值做兜底，防止后续入库时报错。 线上踩坑要点 图片地址部分为缩略图，原图需要二次拼接域名，直接存储缩略图会导致图片模糊。 风控策略：连续快速请求会返回空数据，固定延时可以缓解，大规模采集建议轮换 UA。 关键词编码：中文关键词接口会自动编码，代码内无需手动 url 编码，但特殊符号关键词建议预处理过滤。
import requests
import time
class VipSearchSpider:
def __init__(self):
self.session = requests.Session()
self.headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Referer": "https://www.vip.com/"
}
self.delay = 1.2
def search_goods(self, keyword, page=1, page_size=30):
"""唯品会关键词搜索商品列表"""
api_url = "https://search.vip.com/search"
params = {
"q": keyword,
"page": page,
"size": page_size
}
try:
resp = self.session.get(api_url, params=params, headers=self.headers, timeout=10)
time.sleep(self.delay)
if resp.status_code != 200:
return {"code": resp.status_code, "data": [], "msg": "请求异常"}
res_json = resp.json()
goods_raw = res_json.get("data", {}).get("items", [])
goods_list = []
for item in goods_raw:
goods_info = {
"goods_id": item.get("goods_id"),
"title": item.get("title", "").strip(),
"price": item.get("sale_price"),
"market_price": item.get("market_price"),
"pic_url": item.get("img", ""),
"shop_name": item.get("shop_name", ""),
"sales": item.get("sales", 0)
}
goods_list.append(goods_info)
# 判断是否还有下一页
total = res_json.get("data", {}).get("total", 0)
has_next = page * page_size < total
return {"code":0, "data":goods_list, "has_next":has_next, "msg":"success"}
except Exception as e:
return {"code":-1, "data":[], "msg":str(e)}
if __name__ == "__main__":
client = VipSearchSpider()
keyword = "连衣裙"
current_page = 1
while True:
result = client.search_goods(keyword, page=current_page)
if result["code"] !=0:
print(f"请求失败：{result['msg']}")
break
goods = result["data"]
print(f"第{current_page}页，商品数量：{len(goods)}")
for g in goods:
print(g["title"], g["price"])
if not result["has_next"]:
print("已抓取全部结果")
break
current_page +=1
