VVIC(搜款网,聚焦服装批发的 B2B 平台)的商品搜索功能(item_search接口,非官方命名)是获取服装批发商品列表的核心入口,数据包含档口货源、批发价、起订规则、实时库存摘要等服装批发特有字段,对采购选款、档口筛选、潮流跟踪等场景具有关键价值。由于平台无公开官方 API,开发者需通过页面解析实现搜索对接。本文系统讲解接口逻辑、参数解析、技术实现及服装批发场景适配策略,助你构建稳定的 VVIC 商品列表获取系统。
一、接口基础认知(核心功能与场景)
二、对接前置准备(参数与 URL 结构)
三、接口调用流程(基于页面解析)
四、代码实现示例(Python)
import requests
import time
import random
import re
import urllib.parse
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
from typing import List, Dict
class VVICSearchApi:
def __init__(self, proxy_pool: List[str] = None, cookie: str = ""):
self.base_url = "https://www.vvic.com/search/"
self.ua = UserAgent()
self.proxy_pool = proxy_pool # 代理池列表,如["http://ip:port", ...]
self.cookie = cookie # 登录态Cookie(用于完整价格和起订规则)
# 分类ID映射(简化版)
self.category_map = {
"女装-连衣裙": "101_305",
"男装-T恤": "201_402",
"童装-外套": "301_503"
}
# 市场映射(档口位置)
self.market_map = {
"广州十三行": "gzshsx",
"杭州四季青": "hzsjq",
"广州白马": "gzbm"
}
def _get_headers(self) -> Dict[str, str]:
"""生成随机请求头"""
headers = {
"User-Agent": self.ua.random,
"Referer": "https://www.vvic.com/category/",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
}
if self.cookie:
headers["Cookie"] = self.cookie
return headers
def _get_proxy(self) -> Dict[str, str]:
"""随机获取代理"""
if self.proxy_pool and len(self.proxy_pool) > 0:
proxy = random.choice(self.proxy_pool)
return {"http": proxy, "https": proxy}
return None
def _clean_price(self, price_str: str) -> float:
"""清洗批发价(提取数字,支持“65元/件”“58元(10件起)”)"""
if not price_str:
return 0.0
price_num = re.search(r"\d+\.?\d*", price_str)
return float(price_num.group()) if price_num else 0.0
def _clean_sales(self, sales_str: str) -> int:
"""清洗30天销量(提取数字,支持“1200+”“已售890”)"""
if not sales_str:
return 0
sales_num = re.search(r"\d+", sales_str)
return int(sales_num.group()) if sales_num else 0
def _clean_batch_num(self, batch_str: str) -> int:
"""清洗起订量(提取数字,支持“3件起批”“混批5件”)"""
if not batch_str:
return 0
batch_num = re.search(r"\d+", batch_str)
return int(batch_num.group()) if batch_num else 0
def _parse_item(self, item_soup) -> Dict[str, str]:
"""解析单条商品数据"""
# 提取商品ID
link = item_soup.select_one("a.goods-title")["href"] if item_soup.select_one("a.goods-title") else ""
goods_id = re.search(r"/goods/(\d+)\.html", link).group(1) if link else ""
# 提取规格摘要(颜色+尺码)
spec_brief = item_soup.select_one(".spec-brief")?.text.strip() or ""
colors = []
sizes = []
if "颜色:" in spec_brief:
colors_part = re.split(r"颜色:", spec_brief)[1].split("|")[0].strip()
colors = [c.strip() for c in colors_part.split("/")]
if "尺码:" in spec_brief:
sizes_part = re.split(r"尺码:", spec_brief)[1].strip()
sizes = [s.strip() for s in sizes_part.split("/")]
return {
"goods_id": goods_id,
"title": item_soup.select_one(".goods-title")?.text.strip() or "",
"main_image": item_soup.select_one(".goods-img img")?.get("src") or "",
"url": f"https://www.vvic.com{link}" if link.startswith("/") else link,
"price": {
"wholesale": self._clean_price(item_soup.select_one(".price")?.text or ""),
"price_str": item_soup.select_one(".price")?.text.strip() or "" # 原始价格文本
},
"trade": {
"batch_rule": item_soup.select_one(".batch-rule")?.text.strip() or "",
"batch_num": self._clean_batch_num(item_soup.select_one(".batch-rule")?.text or ""),
"sales_count": self._clean_sales(item_soup.select_one(".sales-count")?.text or ""),
"sales_str": item_soup.select_one(".sales-count")?.text.strip() or ""
},
"specs": {
"colors": colors,
"sizes": sizes,
"stock_status": item_soup.select_one(".stock-tag")?.text.strip() or ""
},
"shop": {
"name": item_soup.select_one(".shop-name")?.text.strip() or "",
"market": item_soup.select_one(".market")?.text.strip() or "",
"rating": float(item_soup.select_one(".shop-rating")?.text or "0")
}
}
def _parse_page(self, html: str) -> List[Dict]:
"""解析页面的商品列表"""
soup = BeautifulSoup(html, "lxml")
# 商品列表容器(需根据实际页面结构调整)
item_list = soup.select("div.goods-item")
return [self._parse_item(item) for item in item_list if item]
def _get_total_pages(self, html: str) -> int:
"""获取总页数"""
soup = BeautifulSoup(html, "lxml")
page_box = soup.select_one(".pagination")
if not page_box:
return 1
# 提取最后一页页码
last_page = page_box.select("a")[-1].text.strip()
return int(last_page) if last_page.isdigit() else 1
def item_search(self,
kw: str = "",
category: str = "",
price_min: float = None,
price_max: float = None,
style: List[str] = None,
batch_num: int = None,
market: str = "",
stock: int = 0,
sort: str = "",
page_limit: int = 5) -> Dict:
"""
搜索VVIC商品列表
:param kw: 搜索关键词
:param category: 分类名称(如“女装-连衣裙”)或分类ID
:param price_min: 最低批发价(元)
:param price_max: 最高批发价(元)
:param style: 风格列表(如["韩版", "通勤"])
:param batch_num: 最低起订量(件)
:param market: 档口市场(如“广州十三行”)
:param stock: 库存状态(1=现货,2=预售,0=全部)
:param sort: 排序方式(sales/price_asc等)
:param page_limit: 最大页数(默认5)
:return: 标准化搜索结果
"""
try:
# 1. 参数预处理
if not kw and not category:
return {"success": False, "error_msg": "关键词(kw)和分类(category)至少需提供一个"}
# 转换分类名称为ID
if category in self.category_map:
cat_id = self.category_map[category]
else:
cat_id = category if category else ""
# 转换市场名称为标识
market_code = self.market_map.get(market, "") if market else ""
# 处理风格参数(多风格用逗号分隔并编码)
style_str = ""
if style and len(style) > 0:
encoded_styles = [urllib.parse.quote(s, encoding="utf-8") for s in style]
style_str = ",".join(encoded_styles)
# 编码关键词
encoded_kw = urllib.parse.quote(kw, encoding="utf-8") if kw else ""
all_items = []
current_page = 1
while current_page <= page_limit:
# 构建参数
params = {
"page": current_page
}
if encoded_kw:
params["kw"] = encoded_kw
if cat_id:
params["cat_id"] = cat_id
if price_min is not None:
params["price_min"] = price_min
if price_max is not None:
params["price_max"] = price_max
if style_str:
params["style"] = style_str
if batch_num is not None:
params["batch_num"] = batch_num
if market_code:
params["market"] = market_code
if stock in (0, 1, 2):
params["stock"] = stock
if sort:
params["sort"] = sort
# 发送请求(带随机延迟)
time.sleep(random.uniform(2, 4)) # VVIC反爬较严,延迟需适中
headers = self._get_headers()
proxy = self._get_proxy()
response = requests.get(
url=self.base_url,
params=params,
headers=headers,
proxies=proxy,
timeout=10
)
response.raise_for_status()
html = response.text
# 解析当前页商品
items = self._parse_page(html)
if not items:
break # 无数据,终止分页
all_items.extend(items)
# 获取总页数(仅第一页需要)
if current_page == 1:
total_pages = self._get_total_pages(html)
# 修正最大页数(不超过page_limit和50)
total_pages = min(total_pages, page_limit, 50)
if total_pages < current_page:
break
# 若当前页是最后一页,终止
if current_page >= total_pages:
break
current_page += 1
# 去重(基于goods_id)
seen_ids = set()
unique_items = []
for item in all_items:
if item["goods_id"] not in seen_ids:
seen_ids.add(item["goods_id"])
unique_items.append(item)
return {
"success": True,
"total": len(unique_items),
"page_processed": current_page,
"items": unique_items
}
except requests.exceptions.HTTPError as e:
if "403" in str(e):
return {"success": False, "error_msg": "触发反爬,建议更换代理或Cookie", "code": 403}
if "401" in str(e):
return {"success": False, "error_msg": "需要登录,请提供有效Cookie", "code": 401}
return {"success": False, "error_msg": f"HTTP错误: {str(e)}", "code": response.status_code}
except Exception as e:
return {"success": False, "error_msg": f"搜索失败: {str(e)}", "code": -1}
# 使用示例
if __name__ == "__main__":
# 代理池(替换为有效代理)
PROXIES = [
"http://123.45.67.89:8888",
"http://98.76.54.32:8080"
]
# 登录态Cookie(从浏览器获取,用于完整价格)
COOKIE = "userid=xxx; PHPSESSID=xxx; token=xxx"
# 初始化API客户端
search_api = VVICSearchApi(proxy_pool=PROXIES, cookie=COOKIE)
# 搜索“秋季连衣裙”,分类“女装-连衣裙”,价格60-80元,风格韩版,起订量3件,广州十三行,现货,销量降序,最多3页
result = search_api.item_search(
kw="秋季连衣裙",
category="女装-连衣裙",
price_min=60,
price_max=80,
style=["韩版"],
batch_num=3,
market="广州十三行",
stock=1,
sort="sales",
page_limit=3
)
if result["success"]:
print(f"搜索成功:共找到 {result['total']} 件商品,处理 {result['page_processed']} 页")
for i, item in enumerate(result["items"][:5]): # 打印前5条
print(f"\n商品 {i+1}:")
print(f"标题:{item['title'][:50]}...") # 截断长标题
print(f"价格:{item['price']['price_str']} | 起订规则:{item['trade']['batch_rule']}")
print(f"交易:30天销量{item['trade']['sales_count']}件 | 库存状态:{item['specs']['stock_status']}")
print(f"规格:颜色{', '.join(item['specs']['colors'][:3])} | 尺码{', '.join(item['specs']['sizes'][:3])}")
print(f"档口:{item['shop']['name']} | 市场:{item['shop']['market']} | 评分:{item['shop']['rating']}分")
print(f"详情页:{item['url']}")
else:
print(f"搜索失败:{result['error_msg']}(错误码:{result.get('code')})")