闲鱼 item_search 接口是按关键字检索二手商品列表的核心入口,支持通过关键词、类目、价格区间、成色等多维度筛选商品,返回分页商品基础信息(含 ID、标题、价格、主图等),可联动 item_get 接口获取商品详情。该接口基于阿里开放平台网关,采用 HMAC-SHA256 签名认证,权限分级严格,高频调用易触发风控。本攻略从接口认知、权限获取、实操对接、调试排错到生产级优化,提供结构化全链路指导,适配二手商品聚合、比价、运营分析等场景。
一、接口核心认知:功能与适配场景
1. 接口定位与核心价值
2. 核心参数与返回字段
(1)请求参数(公共参数 + 私有参数,POST 方式提交)
(2)返回核心字段(按业务场景分类)
3. 接口限制与注意事项
二、对接前准备:权限与环境搭建
1. 获取接口权限(官方唯一合规路径)
2. 技术环境准备
(1)支持语言与协议
(2)必备工具与依赖
三、实操步骤:接口对接全流程(Python 示例)
步骤 1:理解签名认证规则(核心,必掌握)
步骤 2:完整代码实现(含签名 + 调用 + 数据标准化)
(1)依赖安装
(2)Python 代码实现
import requests
import hmac
import hashlib
import time
import pandas as pd
import logging
from urllib.parse import urlencode
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# 日志配置
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
handlers=[logging.FileHandler("xianyu_item_search.log"), logging.StreamHandler()]
)
# 配置信息(替换为你的阿里开放平台信息)
CONFIG = {
"app_key": "你的app_key",
"app_secret": "你的app_secret",
"api_url": "https://gw.api.taobao.com/router/rest", # 闲鱼接口网关
"version": "v2"
}
def generate_sign(params: dict, app_secret: str) -> str:
"""生成闲鱼接口HMAC-SHA256签名"""
# 1. 排除sign字段,筛选非空参数
filtered_params = {k: v for k, v in params.items() if v and k != "sign"}
# 2. 按参数名ASCII升序排序
sorted_params = sorted(filtered_params.items(), key=lambda x: x[0])
# 3. 拼接参数字符串(UTF-8编码)
param_str = urlencode(sorted_params, encoding="utf-8") + f"&app_secret={app_secret}"
# 4. HMAC-SHA256加密,生成小写签名
sign = hmac.new(
app_secret.encode("utf-8"),
param_str.encode("utf-8"),
hashlib.sha256
).hexdigest().lower()
return sign
def standardize_search_data(raw_item: dict, keyword: str) -> dict:
"""标准化搜索结果数据,统一输出格式"""
# 成色映射
condition_map = {
"new": "全新",
"like_new": "九成新",
"very_good": "八成新",
"good": "七成新",
"acceptable": "六成新及以下"
}
condition = condition_map.get(raw_item.get("condition", ""), "未知成色")
# 上架时间转换
onsale_time = raw_item.get("onsale_time", 0)
onsale_time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(onsale_time/1000)) if onsale_time else ""
return {
"搜索关键字": keyword,
"商品ID": raw_item.get("item_id", ""),
"商品标题": raw_item.get("title", ""),
"商品主图": raw_item.get("pic_url", ""),
"商品标价(元)": float(raw_item.get("price", 0)),
"商品成色": condition,
"商品类目": raw_item.get("category", ""),
"已售数量": int(raw_item.get("sales", 0)),
"发货地": raw_item.get("location", ""),
"物流方式": raw_item.get("logistics", ""),
"上架时间": onsale_time_str,
"卖家昵称": raw_item.get("seller_nick", ""),
"卖家等级": raw_item.get("seller_level", ""),
"请求时间": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
}
def xianyu_item_search(
keyword: str,
category_id: str = None,
start_price: float = None,
end_price: float = None,
condition: str = None,
sort: str = "relevance",
page_no: int = 1,
page_size: int = 20
) -> dict:
"""调用闲鱼item_search接口按关键字搜索商品"""
# 1. 构建请求参数
params = {
"app_key": CONFIG["app_key"],
"method": "xianyu.item.search",
"timestamp": str(int(time.time() * 1000)),
"version": CONFIG["version"],
"q": keyword,
"sort": sort,
"page_no": page_no,
"page_size": min(page_size, 50) # 单页最大50条
}
# 补充分筛参数
if category_id:
params["category_id"] = category_id
if start_price:
params["start_price"] = start_price
if end_price:
params["end_price"] = end_price
if condition:
params["condition"] = condition
# 2. 生成签名
params["sign"] = generate_sign(params, CONFIG["app_secret"])
try:
# 3. 发送POST请求(闲鱼接口推荐POST)
response = requests.post(
url=CONFIG["api_url"],
data=params,
headers={"Content-Type": "application/x-www-form-urlencoded; charset=utf-8"},
timeout=10,
verify=True
)
response.raise_for_status()
result = response.json()
# 4. 解析响应结果
if result.get("error_response"):
error_msg = f"{result['error_response']['code']}: {result['error_response']['msg']}"
logging.error(f"接口调用失败(关键字:{keyword}):{error_msg}")
return {
"success": False,
"error_msg": error_msg,
"data": [],
"pagination": {}
}
search_response = result.get("item_search_response", {})
raw_items = search_response.get("items", {}).get("item", [])
if not raw_items:
logging.warning(f"无商品数据返回(关键字:{keyword})")
return {
"success": False,
"error_msg": "无商品数据",
"data": [],
"pagination": {}
}
# 5. 标准化数据
standard_items = [standardize_search_data(item, keyword) for item in raw_items]
# 解析分页信息
pagination = {
"total_results": int(search_response.get("total_results", 0)),
"page_no": page_no,
"page_size": page_size,
"has_more": search_response.get("has_more", False)
}
return {
"success": True,
"data": standard_items,
"pagination": pagination,
"error_msg": ""
}
except requests.exceptions.RequestException as e:
logging.error(f"网络请求异常(关键字:{keyword}):{str(e)}")
return {
"success": False,
"error_msg": f"网络异常:{str(e)}",
"data": [],
"pagination": {}
}
except Exception as e:
logging.error(f"数据解析异常(关键字:{keyword}):{str(e)}")
return {
"success": False,
"error_msg": f"解析异常:{str(e)}",
"data": [],
"pagination": {}
}
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# 调用示例
if __name__ == "__main__":
# 搜索参数配置
keyword = "iPhone 14 256GB 国行"
condition = "like_new" # 九成新
sort = "price_asc" # 价格升序
page_no = 1
page_size = 20
# 调用接口
result = xianyu_item_search(
keyword=keyword,
condition=condition,
sort=sort,
page_no=page_no,
page_size=page_size
)
if result["success"]:
print(f"搜索成功:共找到 {result['pagination']['total_results']} 条商品")
print(f"第 {page_no} 页获取到 {len(result['data'])} 条数据")
# 打印前5条数据
for item in result["data"][:5]:
print(f"商品ID:{item['商品ID']} | 标题:{item['商品标题']} | 价格:{item['商品标价(元)']} 元")
# 保存为Excel
df = pd.DataFrame(result["data"])
df.to_excel(f"xianyu_search_result_{keyword}.xlsx", index=False)
# 翻页示例:获取下一页
if result["pagination"]["has_more"]:
next_page_result = xianyu_item_search(
keyword=keyword,
condition=condition,
sort=sort,
page_no=page_no + 1,
page_size=page_size
)
print(f"下一页获取到 {len(next_page_result['data'])} 条数据")
else:
print(f"搜索失败:{result['error_msg']}")