当前位置：首页 > news >正文

某鱼平台二手商品搜索接口开发实战：个人闲置与商家转让数据获取方案

news 2025/8/26 11:20:57

在二手电商系统开发和闲置物品交易分析场景中，某鱼平台作为国内领先的 C2C 二手交易平台，其商品数据包含丰富的个人闲置物品信息、价格趋势和交易特征。本文将将篇将详细介绍某鱼平台 API 的调用方法，包括认证机制、搜索参数配置、反爬特有的数据解析及反爬策略，并提供可直接复用的 Python 代码实现，帮助开发者合规获取二手商品数据。

一、某鱼平台接口基础信息

某鱼平台提供的开放接口主要包括商品搜索、详情查询和卖家信息获取等功能，其中/api/v2/search/items是获取二手商品列表的核心接口，特别适用于个人闲置物品的检索。

接口特点：

采用 Cookie+Token 的复合认证机制，部分接口需要登录状态
支持按物品类型、新旧程度、价格区间、地理位置等多维度筛选
包含二手商品特有的闲置原因、使用时长、瑕疵描述等字段
提供卖家信用、粉丝数、交易评价等社交化交易数据

接口端点：https://api.mouyu.com/api/v2/search/items

点击获取key和secret

二、认证机制与核心参数

1. 认证方式

某鱼接口采用较为严格的认证机制：

基础接口可通过公开 Cookie 访问，高级接口需要登录态 Token
登录态获取需通过手机验证码或扫码登录，生成sessionid和token
请求头需包含User-Agent、Referer和Origin等字段，模拟浏览器环境

2. 核心搜索参数

keyword：搜索关键字（物品名称、品牌等，必填）
category_id：分类 ID（3C 数码 / 家居用品 / 服装鞋帽等，可选）
item_status：物品状态（0 - 全新，1 - 几乎全新，2 - 九成新，3 - 八成新，4 - 七成新及以下）
price_min/price_max：价格区间（可选）
distance：距离范围（单位 km，1-10km，可选）
city：城市名称（按地理位置筛选，可选）
sort_type：排序方式（0 - 综合，1 - 价格低到高，2 - 价格高到低，3 - 最新发布）
page：页码（默认 1）
limit：每页条数（1-20，默认 10）
is_promotion：是否仅显示促销商品（true/false，可选）

3. 响应数据结构

total_count：总结果数
has_more：是否有更多数据
items：商品列表数组
filters：可用筛选条件
request_id：请求 ID（用于问题排查）

三、完整代码实现

以下是 Python 实现的某鱼平台商品搜索功能，包含认证处理、搜索参数构建、数据解析和反爬策略：

import requests
import time
import random
import json
from typing import Dict, List, Optional, Any, Tuple
from urllib.parse import urlencode
import logging
from fake_useragent import UserAgent# 配置日志
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('mouyu_api')class MouyuSearchAPI:def __init__(self, cookie: str = "", proxy_pool: List[str] = None, use_login: bool = False):"""初始化某鱼搜索API客户端:param cookie: 平台Cookie:param proxy_pool: 代理IP池列表:param use_login: 是否使用登录态（获取更多数据）"""self.base_url = "https://api.mouyu.com"self.search_endpoint = "/api/v2/search/items"self.detail_endpoint = "/api/v2/items/detail"self.cookie = cookieself.use_login = use_loginself.proxy_pool = proxy_pool or []self.ua = UserAgent()self.session = self._init_session()self.token = self._extract_token()def _init_session(self) -> requests.Session:"""初始化请求会话，配置持久连接和基础头信息"""session = requests.Session()# 设置基础请求头base_headers = {"Accept": "application/json, text/plain, */*","Accept-Encoding": "gzip, deflate, br","Accept-Language": "zh-CN,zh;q=0.9","Connection": "keep-alive","Origin": "https://m.mouyu.com","Referer": "https://m.mouyu.com/","Cookie": self.cookie}session.headers.update(base_headers)return sessiondef _extract_token(self) -> Optional[str]:"""从Cookie中提取token（如果有）"""if not self.cookie:return Nonefor cookie in self.cookie.split(';'):key_value = cookie.strip().split('=')if len(key_value) == 2 and key_value[0] == 'token':return key_value[1]return Nonedef _get_random_headers(self) -> Dict[str, str]:"""生成随机请求头，降低反爬风险"""headers = {"User-Agent": self.ua.random,"X-Requested-With": "XMLHttpRequest","X-Csrf-Token": self.token if self.token else "","Timestamp": str(int(time.time() * 1000))}# 登录态额外头信息if self.use_login and self.token:headers["Authorization"] = f"Bearer {self.token}"return headersdef _get_proxy(self) -> Optional[Dict[str, str]]:"""从代理池获取随机代理"""if self.proxy_pool:proxy = random.choice(self.proxy_pool)return {"http": proxy, "https": proxy}return Nonedef _add_random_delay(self, base: float = 1.0, range: float = 1.0) -> None:"""添加随机延迟，模拟人类浏览行为"""delay = base + random.random() * rangetime.sleep(delay)logger.debug(f"延迟 {delay:.2f} 秒")def search_items(self,keyword: str,category_id: Optional[str] = None,item_status: Optional[int] = None,price_min: Optional[float] = None,price_max: Optional[float] = None,city: Optional[str] = None,sort_type: int = 0,page: int = 1,limit: int = 10) -> Dict[str, Any]:"""搜索某鱼平台商品:param keyword: 搜索关键字:param category_id: 分类ID:param item_status: 物品状态:param price_min: 最低价格:param price_max: 最高价格:param city: 城市名称:param sort_type: 排序方式:param page: 页码:param limit: 每页条数:return: 搜索结果"""# 限制每页最大条数limit = min(limit, 20)# 构建查询参数params: Dict[str, Any] = {"keyword": keyword,"sort_type": sort_type,"page": page,"limit": limit,"version": "2.3.0"}# 添加可选参数if category_id:params["category_id"] = category_idif item_status is not None:params["item_status"] = item_statusif price_min is not None:params["price_min"] = price_minif price_max is not None:params["price_max"] = price_maxif city:params["city"] = city# 准备请求配置headers = self._get_random_headers()proxy = self._get_proxy()url = f"{self.base_url}{self.search_endpoint}?{urlencode(params)}"try:# 随机延迟，避免被识别为爬虫self._add_random_delay()# 发送请求logger.info(f"搜索商品: {keyword}, 页码: {page}")response = self.session.get(url,headers=headers,proxies=proxy,timeout=15)# 检查响应状态if response.status_code == 401:return {"success": False, "error_msg": "认证失败，请检查Cookie或登录状态"}if response.status_code == 429:return {"success": False, "error_msg": "请求过于频繁，已被临时限制"}response.raise_for_status()# 解析响应result = response.json()# 处理API错误if result.get("code") != 0:logger.error(f"API错误: {result.get('message')}")return {"success": False,"error_code": result.get("code"),"error_msg": result.get("message")}# 解析搜索结果return self._parse_search_result(result.get("data", {}))except requests.exceptions.RequestException as e:logger.error(f"请求异常: {str(e)}")return {"success": False,"error_msg": f"请求异常: {str(e)}"}except Exception as e:logger.error(f"处理响应失败: {str(e)}")return {"success": False,"error_msg": f"处理响应失败: {str(e)}"}def _parse_search_result(self, raw_data: Dict[str, Any]) -> Dict[str, Any]:"""解析搜索结果为结构化数据"""# 分页信息pagination = {"total_count": raw_data.get("total_count", 0),"page": raw_data.get("page", 1),"limit": raw_data.get("limit", 10),"has_more": raw_data.get("has_more", False)}# 解析商品列表items = []for item in raw_data.get("items", []):# 处理二手商品特有信息used_info = {"status": item.get("item_status"),  # 新旧程度"status_text": self._get_status_text(item.get("item_status")),"used_time": item.get("used_time"),  # 使用时长"defect_desc": item.get("defect_desc")  # 瑕疵描述}# 卖家信息seller = {"id": item.get("seller_id"),"nickname": item.get("seller_nickname"),"credit": item.get("seller_credit"),  # 信用等级"fans_count": item.get("seller_fans_count"),  # 粉丝数"location": item.get("seller_location")  # 所在地}items.append({"item_id": item.get("item_id"),"title": item.get("title"),"price": float(item.get("price", 0)),"original_price": float(item.get("original_price", 0)),"discount": round(float(item.get("price", 0)) / float(item.get("original_price", 1)) * 100, 1) if item.get("original_price") else 0,"category": {"id": item.get("category_id"),"name": item.get("category_name")},"used_info": used_info,"seller": seller,"sales_status": item.get("sales_status"),  # 在售/已售"publish_time": item.get("publish_time"),  # 发布时间"view_count": item.get("view_count"),  # 浏览数"comment_count": item.get("comment_count"),  # 评论数"images": {"main": item.get("main_image"),"thumbnails": item.get("thumbnail_images", [])},"url": f"https://m.mouyu.com/item/{item.get('item_id')}.html","tags": item.get("tags", [])})# 解析可用筛选条件filters = self._parse_filters(raw_data.get("filters", {}))return {"success": True,"pagination": pagination,"items": items,"filters": filters,"request_id": raw_data.get("request_id")}def _get_status_text(self, status_code: Optional[int]) -> str:"""将物品状态代码转换为文本描述"""status_map = {0: "全新",1: "几乎全新",2: "九成新",3: "八成新",4: "七成新及以下"}return status_map.get(status_code, "未知")def _parse_filters(self, raw_filters: Dict[str, Any]) -> Dict[str, Any]:"""解析筛选条件"""filters = {}# 分类筛选if "categories" in raw_filters:filters["categories"] = [{"id": item.get("id"), "name": item.get("name"), "count": item.get("count")}for item in raw_filters["categories"]]# 物品状态筛选if "item_status" in raw_filters:filters["item_status"] = [{"code": item.get("code"), "name": item.get("name"), "count": item.get("count")}for item in raw_filters["item_status"]]# 价格区间筛选if "price_ranges" in raw_filters:filters["price_ranges"] = raw_filters["price_ranges"]return filtersdef batch_search(self,keyword: str,max_pages: int = 3,**kwargs) -> Dict[str, Any]:"""批量获取多页搜索结果:param keyword: 搜索关键字:param max_pages: 最大获取页数:param**kwargs: 其他搜索参数:return: 合并的搜索结果"""all_items = []current_page = 1has_more = Truewhile current_page <= max_pages and has_more:# 搜索当前页result = self.search_items(keyword=keyword,page=current_page,**kwargs)if not result.get("success"):return result# 收集商品数据all_items.extend(result.get("items", []))# 更新分页信息pagination = result.get("pagination", {})has_more = pagination.get("has_more", False)# 准备下一页current_page += 1return {"success": True,"total_items": len(all_items),"items": all_items,"summary": {"total_available": pagination.get("total_count", 0),"fetched_pages": current_page - 1}}# 使用示例
if __name__ == "__main__":# 替换为你的Cookie（可选，部分接口需要）COOKIE = "your_cookie_here"# 代理配置（可选）PROXY_POOL = [# "http://ip1:port",# "http://ip2:port"]# 初始化API客户端mouyu_api = MouyuSearchAPI(cookie=COOKIE,proxy_pool=PROXY_POOL,use_login=False  # 如需获取更多数据可设为True（需有效Cookie）)# 示例1：搜索二手手机phone_result = mouyu_api.search_items(keyword="iPhone 13",category_id="1001",  # 假设1001是手机分类item_status=1,  # 几乎全新price_min=3000,price_max=5000,city="上海",sort_type=1,  # 价格从低到高page=1,limit=10)if phone_result["success"]:print(f"搜索结果: 找到 {phone_result['pagination']['total_count']} 件相关商品")if phone_result["items"]:item = phone_result["items"][0]print(f"标题: {item['title']}")print(f"价格: {item['price']}元 (原价: {item['original_price']}元, {item['discount']}折)")print(f"新旧程度: {item['used_info']['status_text']}")print(f"卖家: {item['seller']['nickname']} ({item['seller']['location']})")print(f"发布时间: {item['publish_time']}")# 示例2：批量搜索笔记本电脑# laptop_result = mouyu_api.batch_search(#     keyword="笔记本电脑",#     item_status=2,  # 九成新#     max_pages=2# )# # if laptop_result["success"]:#     print(f"\n批量搜索: 共获取 {laptop_result['total_items']} 件商品")