本文将详细介绍如何使用 Python 获取闲鱼商品详情数据,涵盖第三方代理 API、网页逆向爬虫两种主流方案,包含完整的代码示例和合规建议。重要提示:闲鱼官方未对外开放商品详情 API,以下方案均为非官方实现,仅适用于技术学习与研究 。
一、方案概述
1. 闲鱼 API 现状
闲鱼作为阿里巴巴旗下的二手交易平台,并未对外开放官方商品详情 API 。开发者获取数据的主要途径:
| 方案 | 适用场景 | 难度 | 稳定性 | 合规风险 |
|---|---|---|---|---|
| 第三方代理 API | 快速验证/个人开发 | 低 | 中 | 中 |
| 网页逆向爬虫 | 技术学习/研究 | 高 | 低 | 高 |
| 官方开放平台 | 企业合作(需资质) | 中 | 高 | 低 |
2. 官方开放平台(受限)
闲鱼开放平台(阿里开放平台旗下)仅对企业开发者开放部分接口权限,个人开发者仅能申请基础查询类接口。核心限制包括:
- 需企业营业执照+法人认证
- 需申请特定权限并通过审核
- 商品详情接口(如
goodfish.item_get)需特殊授权
二、第三方代理 API 方案
2.1 方案特点
第三方数据服务商封装了闲鱼数据接口,无需企业资质即可调用。
核心接口信息:
- 接口地址:
https://api-gw.onebound.cn/goodfish/item_get/ - 请求方式:GET
- 认证方式:AppKey + AppSecret
2.2 完整调用代码
Python
import requestsimport jsonimport hashlibimport timefrom urllib.parse import quoteclass XianyuThirdPartyAPI:
"""第三方闲鱼商品详情 API 封装"""
def __init__(self, api_key: str, api_secret: str, base_url: str = "https://api-gw.onebound.cn/goodfish"):
self.api_key = api_key
self.api_secret = api_secret
self.base_url = base_url
def generate_sign(self, params: dict) -> str:
"""
生成签名(MD5)
规则:参数按key升序排序,拼接成key1value1key2value2...,首尾加AppSecret
"""
sorted_params = sorted(params.items())
sign_str = self.api_secret for key, value in sorted_params:
if key != 'sign' and value is not None:
sign_str += f"{key}{value}"
sign_str += self.api_secret
return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
def get_item_detail(self, num_iid: str) -> dict:
"""
获取商品详情
Args:
num_iid: 闲鱼商品ID(从商品URL中获取)
"""
url = f"{self.base_url}/item_get/"
# 构建请求参数
params = {
'key': self.api_key,
'num_iid': num_iid,
'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
}
# 生成签名
params['sign'] = self.generate_sign(params)
headers = {
"Accept-Encoding": "gzip",
"Connection": "close"
}
try:
response = requests.get(url, params=params, headers=headers, timeout=30)
response.raise_for_status()
data = response.json()
if data.get('error') != '0':
print(f"API错误: {data.get('error', '未知错误')}")
return None
return data.get('item', {})
except requests.exceptions.RequestException as e:
print(f"请求异常: {e}")
return None
def extract_key_fields(self, item_data: dict) -> dict:
"""提取关键字段"""
if not item_data:
return None
return {
'title': item_data.get('title'),
'price': item_data.get('price'),
'original_price': item_data.get('orginal_price'),
'num_iid': item_data.get('num_iid'),
'detail_url': item_data.get('detail_url'),
'pic_url': item_data.get('pic_url'),
'item_imgs': item_data.get('item_imgs', []),
'desc': item_data.get('desc'),
'seller_id': item_data.get('seller_id'),
'seller_nick': item_data.get('seller_nick'),
'credit_level': item_data.get('credit_level'),
'location': item_data.get('location'),
'tags': item_data.get('item_tags', []),
'trade_count': item_data.get('trade_count'),
'collect_count': item_data.get('collect_count'),
'create_time': item_data.get('create_time'),
'status': item_data.get('status'),
}# 使用示例if __name__ == "__main__":
API_KEY = "your_api_key"
API_SECRET = "your_api_secret"
NUM_IID = "750828541223" # 商品ID,从闲鱼商品URL中获取
api = XianyuThirdPartyAPI(API_KEY, API_SECRET)
item_data = api.get_item_detail(NUM_IID)
if item_data:
clean_data = api.extract_key_fields(item_data)
print(json.dumps(clean_data, ensure_ascii=False, indent=2))三、网页逆向爬虫方案
3.1 方案原理
通过分析闲鱼 H5 页面或小程序接口,模拟浏览器请求获取商品详情数据。注意:此方案存在反爬机制,需严格遵守平台规则。
3.2 核心实现
Python
import requestsimport jsonimport reimport timeimport randomfrom urllib.parse import quoteclass XianyuWebCrawler:
"""闲鱼网页爬虫"""
def __init__(self):
self.session = requests.Session()
# 模拟移动端浏览器
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 14_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.0 Mobile/15E148 Safari/604.1',
'Accept': 'application/json, text/plain, */*',
'Accept-Language': 'zh-CN,zh;q=0.9',
'Accept-Encoding': 'gzip, deflate, br',
'Connection': 'keep-alive',
'Referer': 'https://2.taobao.com/',
})
self.rate_limiter = RateLimiter(min_delay=5.0, max_delay=10.0)
def set_cookies(self, cookies_str: str):
"""
设置登录Cookie(必需)
Args:
cookies_str: 从浏览器开发者工具复制的Cookie字符串
"""
cookies = {}
for item in cookies_str.split(';'):
if '=' in item:
key, value = item.strip().split('=', 1)
cookies[key] = value
self.session.cookies.update(cookies)
def get_item_detail(self, item_id: str) -> dict:
"""
获取商品详情(通过H5接口)
Args:
item_id: 闲鱼商品ID
"""
self.rate_limiter.wait()
# H5详情页接口(需抓包获取最新地址)
url = f"https://h5api.m.taobao.com/h5/mtop.taobao.idle.item.detail/1.0/"
params = {
'jsv': '2.4.11',
'appKey': '12574478', # 闲鱼H5固定AppKey
't': str(int(time.time() * 1000)),
'api': 'mtop.taobao.idle.item.detail',
'v': '1.0',
'type': 'jsonp',
'dataType': 'jsonp',
'callback': 'mtopjsonp',
'data': json.dumps({'itemId': item_id}, separators=(',', ':'))
}
try:
response = self.session.get(url, params=params, timeout=10)
response.raise_for_status()
# 解析JSONP响应
json_str = re.search(r'mtopjsonp\((.*)\)', response.text)
if not json_str:
print("无法解析响应数据")
return None
data = json.loads(json_str.group(1))
if data.get('ret', [''])[0] != 'SUCCESS::调用成功':
print(f"接口调用失败: {data.get('ret')}")
return None
return self._parse_item_data(data.get('data', {}))
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
return None
def _parse_item_data(self, data: dict) -> dict:
"""解析商品数据"""
if not data:
return None
item = data.get('item', {})
seller = data.get('seller', {})
return {
'item_id': item.get('itemId'),
'title': item.get('title'),
'price': item.get('price'),
'original_price': item.get('originalPrice'),
'description': item.get('desc'),
'images': item.get('images', []),
'cover_image': item.get('coverImage'),
'category': item.get('categoryName'),
'location': item.get('itemArea', {}).get('areaName'),
'status': item.get('status'), # 在售/已售/下架
'create_time': item.get('createdTime'),
'want_count': item.get('wantNum'), # 想要数
'view_count': item.get('viewCount'),
'seller_id': seller.get('userId'),
'seller_nick': seller.get('nick'),
'seller_credit': seller.get('creditLevel'),
'seller_location': seller.get('location'),
}
def search_items(self, keyword: str, page: int = 1) -> list:
"""
搜索商品列表(基于网页端)
Args:
keyword: 搜索关键词
page: 页码
"""
self.rate_limiter.wait()
# 网页端搜索接口
url = "https://2.taobao.com/item/list.htm"
params = {
'keyword': quote(keyword),
'page': page,
'_input_charset': 'utf8',
}
try:
response = self.session.get(url, params=params, timeout=10)
response.raise_for_status()
# 从HTML中提取商品列表(需使用BeautifulSoup解析)
from bs4 import BeautifulSoup
soup = BeautifulSoup(response.text, 'html.parser')
items = []
# 解析商品卡片(根据实际HTML结构调整选择器)
for card in soup.select('.item-list .item'):
item = {
'item_id': card.get('data-itemid'),
'title': card.select_one('.title').text.strip() if card.select_one('.title') else None,
'price': card.select_one('.price').text.strip() if card.select_one('.price') else None,
'location': card.select_one('.location').text.strip() if card.select_one('.location') else None,
'image': card.select_one('img').get('src') if card.select_one('img') else None,
}
items.append(item)
return items
except Exception as e:
print(f"搜索失败: {e}")
return []class RateLimiter:
"""请求频率限制器"""
def __init__(self, min_delay: float = 5.0, max_delay: float = 10.0):
self.min_delay = min_delay
self.max_delay = max_delay
self.last_request_time = 0
def wait(self):
"""等待随机时间"""
elapsed = time.time() - self.last_request_time if elapsed < self.min_delay:
sleep_time = random.uniform(self.min_delay - elapsed, self.max_delay - elapsed)
time.sleep(max(0, sleep_time))
self.last_request_time = time.time()# 使用示例if __name__ == "__main__":
# 初始化爬虫
crawler = XianyuWebCrawler()
# 设置Cookie(从浏览器登录后复制)
COOKIES = "your_cookies_here"
crawler.set_cookies(COOKIES)
# 获取商品详情
ITEM_ID = "750828541223"
detail = crawler.get_item_detail(ITEM_ID)
if detail:
print(json.dumps(detail, ensure_ascii=False, indent=2))
# 搜索商品
# results = crawler.search_items("二手 iPhone", page=1)
# print(json.dumps(results, ensure_ascii=False, indent=2))四、关键注意事项
4.1 反爬机制应对
闲鱼平台具有严格的反爬机制,包括:
| 机制 | 说明 | 应对策略 |
|---|---|---|
| IP限制 | 高频请求封禁IP | 使用代理池,单IP每分钟≤5次 |
| Cookie验证 | 未登录或过期Cookie返回登录页 | 定期更新Cookie(有效期7-15天) |
| 请求频率 | 超过阈值触发验证码 | 控制间隔≥5秒,随机延迟 |
| 设备指纹 | 检测异常User-Agent | 使用真实移动端UA |
| 行为分析 | 检测非人类操作模式 | 模拟完整浏览行为 |
4.2 数据字段说明
根据第三方接口和网页解析,闲鱼商品详情核心字段包括:
| 字段名 | 类型 | 说明 |
|---|---|---|
title | String | 商品标题 |
price | Float | 商品标价(元) |
original_price | Float | 原价 |
desc | String | 详细描述(含HTML) |
images | Array | 图片链接列表 |
seller_nick | String | 卖家昵称 |
credit_level | String | 信用等级 |
location | String | 所在地 |
want_count | Int | 想要数 |
view_count | Int | 浏览量 |
create_time | String | 发布时间 |
status | String | 商品状态(在售/已售/下架) |
4.3 合规与法律风险
重要提示 :
- 遵守平台规则:严格遵守《闲鱼平台用户服务协议》,不得用于恶意爬取、商业滥用
- 频率控制:建议单IP每分钟请求不超过5次,使用多个合规账号轮换Cookie
- 隐私保护:部分卖家隐私数据(如手机号、精确地址)会被平台隐藏,接口无法获取也不应尝试获取
- 法律风险:高频抓取或破坏平台技术措施可能违反《反不正当竞争法》,面临法律诉讼
- 数据使用:仅限个人学习和研究,严禁将爬取的数据用于商业用途
五、数据存储示例
Python
import sqlite3import jsonfrom datetime import datetimeclass XianyuDataStorage:
"""闲鱼商品数据存储"""
def __init__(self, db_path: str = "xianyu_items.db"):
self.conn = sqlite3.connect(db_path)
self._init_tables()
def _init_tables(self):
"""初始化数据表"""
self.conn.execute("""
CREATE TABLE IF NOT EXISTS xianyu_items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
item_id TEXT UNIQUE NOT NULL,
title TEXT,
price REAL,
original_price REAL,
description TEXT,
seller_id TEXT,
seller_nick TEXT,
credit_level TEXT,
location TEXT,
status TEXT,
want_count INTEGER DEFAULT 0,
view_count INTEGER DEFAULT 0,
images TEXT, -- JSON格式存储
raw_data TEXT, -- 原始完整数据
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# 价格历史表
self.conn.execute("""
CREATE TABLE IF NOT EXISTS price_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
item_id TEXT NOT NULL,
price REAL,
recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (item_id) REFERENCES xianyu_items(item_id)
)
""")
self.conn.commit()
def save_item(self, item: dict):
"""保存商品信息"""
try:
self.conn.execute("""
INSERT INTO xianyu_items
(item_id, title, price, original_price, description, seller_id, seller_nick,
credit_level, location, status, want_count, view_count, images, raw_data)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(item_id) DO UPDATE SET
title=excluded.title,
price=excluded.price,
original_price=excluded.original_price,
description=excluded.description,
seller_nick=excluded.seller_nick,
credit_level=excluded.credit_level,
location=excluded.location,
status=excluded.status,
want_count=excluded.want_count,
view_count=excluded.view_count,
images=excluded.images,
raw_data=excluded.raw_data,
updated_at=CURRENT_TIMESTAMP
""", (
item.get('item_id') or item.get('num_iid'),
item.get('title'),
item.get('price'),
item.get('original_price'),
item.get('description') or item.get('desc'),
item.get('seller_id'),
item.get('seller_nick'),
item.get('credit_level'),
item.get('location'),
item.get('status'),
item.get('want_count', 0),
item.get('view_count', 0),
json.dumps(item.get('images', []), ensure_ascii=False),
json.dumps(item, ensure_ascii=False)
))
self.conn.commit()
# 记录价格历史
if item.get('price'):
self.conn.execute("""
INSERT INTO price_history (item_id, price)
VALUES (?, ?)
""", (item.get('item_id') or item.get('num_iid'), item.get('price')))
self.conn.commit()
except Exception as e:
print(f"保存失败: {e}")
self.conn.rollback()
def get_item_history(self, item_id: str, days: int = 30):
"""获取商品价格历史"""
cursor = self.conn.execute("""
SELECT price, recorded_at
FROM price_history
WHERE item_id = ?
AND recorded_at > datetime('now', '-{} days')
ORDER BY recorded_at ASC
""".format(days), (item_id,))
return [{
'price': row[0],
'date': row[1]
} for row in cursor.fetchall()]六、总结
闲鱼商品详情数据的获取相比其他电商平台更具挑战性,主要原因在于:
- 严格的反爬机制:作为二手交易平台,闲鱼对数据保护更为严格
- 权限门槛高:官方开放平台仅对企业开发者开放高级接口