网易考拉（现业务整合至天猫国际，仍保留部分独立运营页面）作为跨境电商标杆平台，其商品搜索功能（item_search接口，非官方命名）是获取海外商品列表、分析跨境品类分布的核心工具。尽管平台聚焦美妆、母婴、保健品等进口商品，数据包含关税、产地、进口方式等跨境特有字段，对跨境比价、市场调研等场景具有重要价值。由于无公开官方 API，开发者需通过页面解析实现搜索对接。本文系统讲解接口逻辑、参数解析、技术实现及反爬策略，助你构建稳定的商品列表获取系统。
一、接口基础认知（核心功能与场景）
二、对接前置准备（参数与 URL 结构）
三、接口调用流程（基于页面解析）
四、代码实现示例（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 KaolaSearchApi:
def __init__(self, proxy_pool: List[str] = None):
self.base_url = "https://www.kaola.com/search.html"
self.ua = UserAgent()
self.proxy_pool = proxy_pool # 代理池列表，如["http://ip:port", ...]
# 分类ID映射（简化版，需根据实际页面更新）
self.category_map = {
"美妆": "101",
"母婴": "202",
"保健品": "303",
"奢侈品": "404"
}
# 国家代码映射（简化版）
self.country_map = {
"日本": "jp",
"美国": "us",
"德国": "de",
"澳大利亚": "au"
}
def _get_headers(self) -> Dict[str, str]:
"""生成随机请求头"""
return {
"User-Agent": self.ua.random,
"Referer": "https://www.kaola.com/",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Cookie": "kaola_uuid=xxx; user_id=anonymous; Hm_lvt_xxx=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:
"""清洗销量字符串（提取数字，处理“500+”等格式）"""
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"/product/(\d+)\.html", link).group(1) if link else ""
# 提取关税（如“关税¥30”）
tax_str = item_soup.select_one(".tax-tag")?.text.strip() or ""
tax = self._clean_price(tax_str)
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.kaola.com{link}" if link.startswith("/") else link,
"price": {
"current": self._clean_price(item_soup.select_one(".price-current")?.text.strip() or ""),
"original": self._clean_price(item_soup.select_one(".price-original")?.text.strip() or ""),
"tax": tax
},
"cross_border": {
"origin": item_soup.select_one(".origin-tag")?.text.strip() or "",
"shipping_type": item_soup.select_one(".shipping-tag")?.text.strip() or "" # 保税仓/直邮
},
"sales": {
"monthly": self._clean_sales(item_soup.select_one(".sales-count")?.text.strip() or ""),
"rating": item_soup.select_one(".rating")?.text.strip() or "" # 好评率
},
"brand": item_soup.select_one(".brand-name")?.text.strip() or ""
}
def _parse_page(self, html: str) -> List[Dict]:
"""解析页面的商品列表"""
soup = BeautifulSoup(html, "lxml")
item_list = soup.select("div.product-list > div.product-item")
return [self._parse_item(item) for item in item_list if item]
def item_search(self,
keyword: str = "",
category: str = "",
price_min: float = None,
price_max: float = None,
country: str = "",
shipping_type: str = "",
sort: str = "",
page_limit: int = 5) -> Dict:
"""
搜索网易考拉商品列表
:param keyword: 搜索关键词
:param category: 分类名称（如“美妆”）或分类ID
:param price_min: 最低价格（元，含税费）
:param price_max: 最高价格（元，含税费）
:param country: 产地名称（如“日本”）或国家代码（如“jp”）
:param shipping_type: 进口方式（direct/保税仓 bonded/直邮）
:param sort: 排序方式（sales/price-asc等）
:param page_limit: 最大页数（默认5）
:return: 标准化搜索结果
"""
try:
# 1. 参数预处理
if not keyword and not category:
return {"success": False, "error_msg": "关键词（keyword）和分类（category）至少需提供一个"}
# 转换分类名称为ID
if category in self.category_map:
category_id = self.category_map[category]
else:
category_id = category if category else ""
# 转换国家名称为代码
if country in self.country_map:
country_code = self.country_map[country]
else:
country_code = country if country else ""
# 编码关键词（支持中文）
encoded_keyword = quote(keyword, encoding="utf-8") if keyword else ""
all_items = []
current_page = 1
while current_page <= page_limit:
# 构建参数
params = {
"key": encoded_keyword,
"page": current_page
}
if category_id:
params["category"] = category_id
if price_min is not None:
params["priceMin"] = price_min
if price_max is not None:
params["priceMax"] = price_max
if country_code:
params["country"] = country_code
if shipping_type:
params["shippingType"] = shipping_type
if sort:
params["sort"] = sort
# 发送请求（带随机延迟）
time.sleep(random.uniform(3, 5)) # 跨境平台间隔需更长
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()
items = self._parse_page(response.text)
if not items:
break # 无数据，终止分页
all_items.extend(items)
# 若当前页商品数<30，说明是最后一页
if len(items) < 30:
break
current_page += 1
# 防止超过最大页数（50页）
if current_page > 50:
break
# 去重（基于item_id）
seen_ids = set()
unique_items = []
for item in all_items:
if item["item_id"] not in seen_ids:
seen_ids.add(item["item_id"])
unique_items.append(item)
return {
"success": True,
"total": len(unique_items),
"page_processed": current_page - 1,
"items": unique_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 = KaolaSearchApi(proxy_pool=PROXIES)
# 搜索“美妆”，分类“美妆”，价格300-1000元，日本产，保税仓发货，按销量降序，最多3页
result = search_api.item_search(
keyword="美妆",
category="美妆",
price_min=300,
price_max=1000,
country="日本",
shipping_type="bonded",
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']['current']}（含关税￥{item['price']['tax']}） | 原价：￥{item['price']['original']}")
print(f"产地：{item['cross_border']['origin']} | 发货方式：{item['cross_border']['shipping_type']}")
print(f"品牌：{item['brand']} | 月销：{item['sales']['monthly']}件 | 好评率：{item['sales']['rating']}")
print(f"详情页：{item['url']}")
else:
print(f"搜索失败：{result['error_msg']}（错误码：{result.get('code')}）")