小红书数据采集：用Python轻松解锁公开内容分析

魏秦任

412人浏览 · 2026-07-10 10:24:39

魏秦任 · 2026-07-10 10:24:39 发布

小红书数据采集：用Python轻松解锁公开内容分析

【免费下载链接】xhs 基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/ 项目地址: https://gitcode.com/gh_mirrors/xh/xhs

还在为获取小红书公开数据而烦恼吗？xhs工具让Python开发者能够轻松访问小红书平台的公开内容，为市场分析、内容研究和用户洞察提供强大支持。这个开源项目通过封装小红书Web端API，将复杂的网络请求简化为简单的Python调用，让你专注于数据分析而不是网络爬虫技术。

为什么你需要关注小红书数据？

小红书作为国内领先的生活方式分享平台，汇聚了海量的用户生成内容和消费洞察。无论是品牌营销、市场研究还是内容创作，了解小红书上的趋势和用户行为都至关重要。然而，手动收集数据效率低下，而传统爬虫技术又面临诸多技术挑战。

常见的数据获取痛点

技术门槛高：需要处理复杂的网络请求和反爬机制
维护成本大：平台接口变化时需要频繁调整代码
数据质量不稳定：自建爬虫容易出现数据缺失或格式不一致
合规风险：不当的数据获取方式可能违反平台规则

xhs工具：你的小红书数据助手

xhs工具正是为解决这些问题而生。它通过Python封装小红书官方接口，提供稳定可靠的数据获取能力，同时确保使用方式符合平台规范。

核心功能一览

功能模块	主要用途	适用场景
笔记搜索	按关键词查找相关笔记	热点追踪、话题分析
用户信息	获取用户基本资料和笔记列表	竞品分析、KOL研究
笔记详情	获取单篇笔记完整信息	内容分析、数据采集
推荐流	获取首页推荐内容	趋势发现、内容推荐

快速开始：三步上手

第一步：环境准备

确保你的Python环境已就绪，然后通过PyPI安装：

pip install xhs

第二步：获取访问凭证

使用浏览器登录小红书，从开发者工具中复制cookie信息。这是访问数据的关键凭证。

第三步：编写第一个脚本

from xhs import XhsClient

# 初始化客户端
client = XhsClient(cookie="你的cookie信息")

# 搜索热门内容
results = client.search_note(keyword="Python编程", page=1, page_size=10)

for note in results['items']:
    print(f"标题: {note.get('title', '无标题')}")
    print(f"作者: {note['user']['nickname']}")
    print(f"互动数据: 👍{note['like_count']} ❤️{note['collect_count']}")

实战应用场景展示

场景一：市场趋势分析

假设你想了解"健康饮食"相关话题在小红书上的热度变化：

def analyze_trend(keyword, days=7):
    """分析关键词的热度趋势"""
    trend_data = []
    for i in range(days):
        # 模拟按时间获取数据
        results = client.search_note(
            keyword=keyword, 
            page=1, 
            page_size=20,
            sort_type="hot"  # 按热度排序
        )
        daily_stats = {
            "date": f"第{i+1}天",
            "total_notes": len(results['items']),
            "avg_engagement": calculate_avg_engagement(results)
        }
        trend_data.append(daily_stats)
    return trend_data

场景二：内容创作者分析

对于内容运营人员，了解头部创作者的策略至关重要：

def analyze_creator_content(user_id):
    """分析创作者的内容策略"""
    user_info = client.get_user_info(user_id)
    user_notes = client.get_user_notes(user_id)
    
    analysis = {
        "创作者信息": user_info,
        "内容数量": len(user_notes),
        "内容类型分布": categorize_content_types(user_notes),
        "互动表现": calculate_engagement_metrics(user_notes)
    }
    return analysis

功能速查卡：快速找到你需要的方法

核心API方法

方法名称	功能描述	关键参数
`search_note()`	搜索笔记	keyword, page, page_size, sort_type
`get_note_by_id()`	获取笔记详情	note_id, xsec_token
`get_user_info()`	获取用户信息	user_id
`get_user_notes()`	获取用户笔记列表	user_id, page
`get_home_feed()`	获取首页推荐流	feed_type, cursor

实用工具函数

from xhs import help

# 从笔记中提取图片URL
image_urls = help.get_imgs_url_from_note(note_data)

# 处理笔记内容格式
formatted_content = help.format_note_content(note_data)

常见误区与避坑指南

❌ 错误做法

频繁无限制请求：可能导致IP被封禁
使用过期cookie：无法获取最新数据
忽略错误处理：程序崩溃时无恢复机制
存储敏感信息：将cookie硬编码在代码中

✅ 最佳实践

合理控制请求频率：添加适当的延迟
实现自动重试机制：处理临时性错误
使用环境变量：安全存储敏感信息
定期更新cookie：确保访问权限有效

import time
import random
from xhs.exception import DataFetchError, IPBlockError

def safe_api_call(api_func, *args, max_retries=3):
    """安全的API调用包装器"""
    for attempt in range(max_retries):
        try:
            # 随机延迟，避免规律性请求
            time.sleep(random.uniform(1, 3))
            return api_func(*args)
        except IPBlockError as e:
            print(f"IP被封禁: {e}")
            time.sleep(60)  # 等待更长时间
        except DataFetchError as e:
            if attempt < max_retries - 1:
                print(f"第{attempt+1}次失败，重试中...")
                time.sleep(2 ** attempt)  # 指数退避
            else:
                raise

进阶配置：提升稳定性和性能

签名服务配置

对于需要更高稳定性的生产环境，xhs工具支持签名服务：

def custom_sign(uri, data=None, a1="", web_session=""):
    """自定义签名函数"""
    # 这里可以实现你的签名逻辑
    # 或者调用远程签名服务
    return {
        "x-s": "生成的签名",
        "x-t": "时间戳"
    }

# 使用签名服务的客户端
client = XhsClient(
    cookie="your_cookie",
    sign=custom_sign  # 传入签名函数
)

异步处理优化

对于大规模数据采集，可以考虑使用异步处理：

import asyncio
import aiohttp

async def fetch_multiple_notes(note_ids):
    """异步获取多个笔记"""
    async with aiohttp.ClientSession() as session:
        tasks = []
        for note_id in note_ids:
            task = asyncio.create_task(
                fetch_note_async(session, note_id)
            )
            tasks.append(task)
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results