聚美优品的商品搜索功能（对应item_search接口，非官方命名）是通过关键词、分类、价格等条件批量获取商品列表的核心工具，广泛应用于比价平台、导购系统、市场分析等场景。由于聚美优品无公开官方 API，开发者需通过合规的页面解析或第三方服务实现对接。本文将系统讲解item_search接口的对接逻辑、参数解析、技术实现及反爬应对，帮助开发者构建稳定高效的商品搜索数据获取系统。
一、接口基础认知（核心功能与场景）
二、对接前置准备（环境与参数解析）
三、接口调用流程（基于页面解析）
四、代码实现示例（Python）
import requests
import time
import random
import re
from urllib.parse import quote
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
from typing import List, Dict
class JumeiSearchApi:
def __init__(self, proxy_pool: List[str] = None):
self.base_url = "https://search.jumei.com/search"
self.ajax_url = "https://search.jumei.com/ajax/search" # 动态分页接口
self.ua = UserAgent()
self.proxy_pool = proxy_pool # 代理池列表，如["http://ip:port", ...]
self.cat_id_map = self._load_category_map() # 分类ID映射表
def _load_category_map(self) -> Dict[str, str]:
"""加载分类名称-ID映射表（简化版）"""
return {
"美妆": "1005",
"护肤": "1026",
"个人护理": "1032",
"零食": "1112"
}
def _get_headers(self) -> Dict[str, str]:
"""生成随机请求头"""
return {
"User-Agent": self.ua.random,
"Referer": "https://www.jumei.com/",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Cookie": "uid=anonymous; sid=xxx; _jme_tuid=xxx" # 替换为实际Cookie
}
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:
"""清洗价格字符串（去除¥、逗号等）"""
if not price_str:
return 0.0
price_str = re.sub(r"[^\d.]", "", price_str)
return float(price_str) if price_str else 0.0
def _clean_sales(self, sales_str: str) -> int:
"""清洗销量字符串（提取数字）"""
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 _parse_item(self, item_soup) -> Dict[str, str]:
"""解析单条商品数据"""
# 提取商品ID
link = item_soup.select_one("a.product-link")["href"]
item_id = re.search(r"/item/(\d+)\.html", link).group(1) if link else ""
# 提取核心字段
return {
"item_id": item_id,
"title": item_soup.select_one(".product-title")?.text.strip() or "",
"main_image": item_soup.select_one(".product-img img")?.get("src") or "",
"url": f"https://www.jumei.com{link}" if link.startswith("/") else link,
"price": {
"original": self._clean_price(item_soup.select_one(".original-price")?.text.strip() or ""),
"current": self._clean_price(item_soup.select_one(".current-price")?.text.strip() or "")
},
"sales": self._clean_sales(item_soup.select_one(".sales-count")?.text.strip() or ""),
"stock_status": item_soup.select_one(".stock-status")?.text.strip() or "",
"promotion_tag": item_soup.select_one(".promotion-tag")?.text.strip() or ""
}
def _parse_static_page(self, html: str) -> List[Dict]:
"""解析静态HTML页面的商品列表"""
soup = BeautifulSoup(html, "lxml")
item_list = soup.select("ul.products-list > li")
return [self._parse_item(item) for item in item_list if item]
def _fetch_ajax_page(self, params: Dict, headers: Dict, proxy: Dict) -> List[Dict]:
"""调用动态接口获取分页商品"""
try:
response = requests.get(
url=self.ajax_url,
params=params,
headers=headers,
proxies=proxy,
timeout=10
)
response.raise_for_status()
data = response.json()
# 动态接口返回的HTML片段解析
html = data.get("data", {}).get("html", "")
return self._parse_static_page(html)
except Exception as e:
print(f"动态接口获取失败: {str(e)}")
return []
def item_search(self,
keyword: str = "",
cat: str = "",
start_price: float = None,
end_price: float = None,
sort: str = "default",
page_limit: int = 5,
is_self: int = 0) -> Dict:
"""
搜索聚美优品商品列表
:param keyword: 搜索关键词（与cat二选一）
:param cat: 分类名称（如“美妆”，需在cat_id_map中存在）或分类ID
:param start_price: 起始价格（元）
:param end_price: 结束价格（元）
:param sort: 排序方式（default/sale_desc/price_asc等）
:param page_limit: 最大页数（避免过度爬取，默认5）
:param is_self: 是否自营（0-全部，1-自营）
:return: 标准化搜索结果
"""
try:
# 1. 参数预处理
if not keyword and not cat:
return {"success": False, "error_msg": "关键词（keyword）和分类（cat）至少需提供一个"}
# 转换分类名称为ID
if cat in self.cat_id_map:
cat_id = self.cat_id_map[cat]
else:
cat_id = cat # 若为ID则直接使用
# 编码关键词（支持中文）
encoded_keyword = quote(keyword, encoding="utf-8") if keyword else ""
all_items = []
current_page = 1
while current_page <= page_limit:
# 构建参数
params = {
"keyword": encoded_keyword,
"sort": sort,
"page": current_page,
"is_self": is_self
}
if cat_id:
params["cat"] = cat_id
if start_price is not None:
params["startPrice"] = start_price
if end_price is not None:
params["endPrice"] = end_price
# 发送请求（带随机延迟）
time.sleep(random.uniform(3, 6)) # 间隔3-6秒，避免反爬
headers = self._get_headers()
proxy = self._get_proxy()
# 分页≤5用静态页面，>5用动态接口（根据实际抓包调整）
if current_page <= 5:
response = requests.get(
url=self.base_url,
params=params,
headers=headers,
proxies=proxy,
timeout=10
)
response.raise_for_status()
items = self._parse_static_page(response.text)
else:
items = self._fetch_ajax_page(params, headers, proxy)
if not items:
break # 无数据，终止分页
all_items.extend(items)
current_page += 1
return {
"success": True,
"total": len(all_items),
"page_processed": current_page - 1,
"items": all_items
}
except requests.exceptions.HTTPError as e:
if "403" in str(e):
return {"success": False, "error_msg": "触发反爬，建议更换代理或Cookie", "code": 403}
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"
]
# 初始化API客户端
search_api = JumeiSearchApi(proxy_pool=PROXIES)
# 搜索“面膜”，价格50-200元，按销量降序，最多3页
result = search_api.item_search(
keyword="面膜",
start_price=50,
end_price=200,
sort="sale_desc",
page_limit=3,
is_self=0 # 包含非自营
)
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']['original']} 元 → 折扣价 {item['price']['current']} 元")
print(f"销量：{item['sales']} 件 | 库存状态：{item['stock_status']}")
if item['promotion_tag']:
print(f"促销：{item['promotion_tag']}")
print(f"详情页：{item['url']}")
else:
print(f"搜索失败：{result['error_msg']}（错误码：{result.get('code')}）")