×

Python 爬虫获取闲鱼商品详情 API 接口实战指南

admin admin 发表于2026-04-19 16:43:51 浏览12 评论0

抢沙发发表评论

本文将详细介绍如何使用 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 数据字段说明

根据第三方接口和网页解析,闲鱼商品详情核心字段包括
表格
字段名类型说明
titleString商品标题
priceFloat商品标价(元)
original_priceFloat原价
descString详细描述(含HTML)
imagesArray图片链接列表
seller_nickString卖家昵称
credit_levelString信用等级
locationString所在地
want_countInt想要数
view_countInt浏览量
create_timeString发布时间
statusString商品状态(在售/已售/下架)

4.3 合规与法律风险

重要提示
  1. 遵守平台规则:严格遵守《闲鱼平台用户服务协议》,不得用于恶意爬取、商业滥用
  2. 频率控制:建议单IP每分钟请求不超过5次,使用多个合规账号轮换Cookie
  3. 隐私保护:部分卖家隐私数据(如手机号、精确地址)会被平台隐藏,接口无法获取也不应尝试获取
  4. 法律风险:高频抓取或破坏平台技术措施可能违反《反不正当竞争法》,面临法律诉讼
  5. 数据使用:仅限个人学习和研究,严禁将爬取的数据用于商业用途

五、数据存储示例

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()]

六、总结

闲鱼商品详情数据的获取相比其他电商平台更具挑战性,主要原因在于:
  1. 无官方开放 API:闲鱼未像淘宝/京东那样提供标准化的商品详情接口
  2. 严格的反爬机制:作为二手交易平台,闲鱼对数据保护更为严格
  3. 权限门槛高:官方开放平台仅对企业开发者开放高级接口


群贤毕至

访客