Python开发者必看：3步掌握小红书数据采集神器xhs

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

在当今社交媒体数据驱动的时代，小红书作为国内领先的生活方式分享平台，蕴含着丰富的用户行为数据和内容趋势。xhs是一个基于小红书Web端请求封装的Python工具库，专为开发者提供高效、稳定的数据采集解决方案。无论你是内容创作者、市场分析师还是研究人员，掌握这款工具都能让你轻松获取小红书平台上的笔记、用户信息和互动数据，为数据分析、内容挖掘和商业决策提供有力支持。

🎯 项目核心特性：为什么选择xhs？

1. 高度封装的API接口

xhs最大的优势在于将小红书复杂的Web端请求进行了完整封装，开发者无需深入了解平台的反爬机制和签名算法。核心功能全部集中在xhs/core.py文件中，提供了简洁易用的Python接口。

2. 多重认证方式支持

支持二维码登录和Cookie认证两种方式，适应不同使用场景。二维码登录适合个人用户，而Cookie认证则更适合自动化脚本和批量处理。

3. 完善的异常处理机制

内置了丰富的异常类型，包括DataFetchError、IPBlockError、NeedVerifyError等，帮助开发者快速定位问题并进行相应处理。

4. 模块化设计

项目结构清晰，核心功能分离明确：

• xhs/core.py - 主要数据采集功能
• xhs/help.py - 辅助工具函数
• xhs/exception.py - 自定义异常类
• example/ - 丰富的使用示例

🚀 快速部署：3分钟搭建采集环境

环境要求与安装

确保你的Python版本在3.8及以上，然后执行以下命令：

# 安装xhs包
pip install xhs

# 安装Playwright浏览器环境
pip install playwright
playwright install

# 安装stealth.min.js绕过检测
curl -O https://cdn.jsdelivr.net/gh/requireCool/stealth.min.js/stealth.min.js

配置签名服务（可选）

对于需要高性能签名的场景，可以使用Docker快速部署签名服务：

# 使用Docker部署签名服务
docker run -it -d -p 5005:5005 reajason/xhs-api:latest

或者在本机启动Flask服务：

# 安装Flask依赖
pip install flask gevent requests

# 运行签名服务
python example/basic_sign_server.py

💡 实战演练：从零开始的数据采集

场景一：快速获取单篇笔记详情

from xhs import XhsClient

# 初始化客户端（使用签名服务）
def sign(uri, data=None, a1="", web_session=""):
    # 签名函数实现
    pass

cookie = "your_cookie_here"
xhs_client = XhsClient(cookie, sign=sign)

# 获取笔记详情
note = xhs_client.get_note_by_id("6505318c000000001f03c5a6", "xsec_token")
print(f"笔记标题：{note.get('title', '')}")
print(f"作者：{note.get('user', {}).get('nickname', '')}")
print(f"点赞数：{note.get('like_count', 0)}")

场景二：二维码登录获取Cookie

import qrcode
from xhs import XhsClient

xhs_client = XhsClient(sign=sign)

# 获取登录二维码
qr_res = xhs_client.get_qrcode()
qr_id = qr_res["qr_id"]
qr_code = qr_res["code"]

# 生成二维码图片
qr = qrcode.QRCode(version=1, error_correction=qrcode.ERROR_CORRECT_L,
                   box_size=50, border=1)
qr.add_data(qr_res["url"])
qr.make()
qr.print_ascii()  # 在终端显示二维码

# 轮询检查登录状态
while True:
    check_result = xhs_client.check_qrcode(qr_id, qr_code)
    if check_result["code_status"] == 2:  # 登录成功
        print(f"登录成功！当前Cookie：{xhs_client.cookie}")
        break

场景三：批量采集搜索结果的笔记

import time
from xhs import XhsClient

def batch_collect_notes(keyword, max_pages=5, delay=2):
    """批量采集指定关键词的笔记"""
    client = XhsClient(cookie, sign=sign)
    all_notes = []
    
    for page in range(1, max_pages + 1):
        try:
            notes = client.search_notes(
                keyword=keyword,
                page=page,
                count=20  # 每页数量
            )
            all_notes.extend(notes)
            print(f"第{page}页采集完成，共{len(notes)}条笔记")
            
            # 添加延迟避免被封
            time.sleep(delay)
            
        except Exception as e:
            print(f"第{page}页采集失败：{e}")
            break
    
    return all_notes

# 使用示例
travel_notes = batch_collect_notes("旅行攻略", max_pages=3)
print(f"共采集到{len(travel_notes)}条旅行相关笔记")

📊 数据采集最佳实践

1. 合理的请求频率控制

import random
import time

def smart_delay(base_delay=2, jitter=1.5):
    """智能延迟，增加随机性"""
    delay = base_delay + random.uniform(0, jitter)
    time.sleep(delay)

2. 错误重试机制

from xhs.exception import DataFetchError, IPBlockError

def safe_fetch_with_retry(client, func, *args, max_retries=3, **kwargs):
    """带重试的安全获取函数"""
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except DataFetchError as e:
            if attempt == max_retries - 1:
                raise
            print(f"第{attempt+1}次尝试失败，等待重试...")
            time.sleep(2 ** attempt)  # 指数退避
        except IPBlockError:
            print("IP被封禁，请更换IP或等待")
            raise

3. 数据存储策略

import json
import csv
from datetime import datetime

class DataStorage:
    def __init__(self, base_path="./data"):
        self.base_path = base_path
    
    def save_as_json(self, data, filename_prefix):
        """保存为JSON格式"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"{self.base_path}/{filename_prefix}_{timestamp}.json"
        
        with open(filename, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        
        print(f"数据已保存到：{filename}")
    
    def save_as_csv(self, notes, filename_prefix):
        """保存为CSV格式"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"{self.base_path}/{filename_prefix}_{timestamp}.csv"
        
        # 提取关键字段
        rows = []
        for note in notes:
            row = {
                "title": note.get("title", ""),
                "author": note.get("user", {}).get("nickname", ""),
                "likes": note.get("like_count", 0),
                "collects": note.get("collect_count", 0),
                "comments": note.get("comment_count", 0),
                "note_id": note.get("note_id", ""),
                "timestamp": datetime.now().isoformat()
            }
            rows.append(row)
        
        # 写入CSV
        with open(filename, "w", newline="", encoding="utf-8-sig") as f:
            writer = csv.DictWriter(f, fieldnames=rows[0].keys())
            writer.writeheader()
            writer.writerows(rows)
        
        print(f"CSV数据已保存到：{filename}")

⚠️ 常见问题与解决方案

Q1: 签名失败怎么办？

问题现象：出现"重试了这么多次还是无法签名成功"错误

解决方案：

1. 检查stealth.min.js文件是否正确下载
2. 确保Playwright浏览器环境已正确安装
3. 增加签名函数中的sleep时间
4. 尝试将headless参数设置为False查看浏览器状态

# 在sign函数中调整
browser = chromium.launch(headless=False)  # 改为False查看浏览器
sleep(3)  # 增加等待时间

Q2: Cookie失效如何处理？

问题现象：请求返回登录验证错误

解决方案：

1. 重新获取Cookie：使用二维码登录方式获取新的Cookie
2. 检查Cookie字段：确保a1、web_session和webId三个字段完整
3. 使用Cookie刷新机制：定期自动刷新Cookie

Q3: 采集速度太慢怎么优化？

优化建议：

1. 使用签名服务模式，避免每次请求都启动浏览器
2. 合理设置延迟时间，避免触发反爬机制
3. 使用多线程/异步请求（注意频率控制）
4. 考虑使用代理IP池

Q4: 数据格式不一致如何处理？

处理方案：

def normalize_note_data(note):
    """标准化笔记数据格式"""
    return {
        "id": note.get("note_id", ""),
        "title": note.get("title", note.get("desc", "")),
        "author_id": note.get("user", {}).get("user_id", ""),
        "author_name": note.get("user", {}).get("nickname", ""),
        "likes": note.get("like_count", 0),
        "collects": note.get("collect_count", 0),
        "comments": note.get("comment_count", 0),
        "images": note.get("image_list", []),
        "video": note.get("video", {}),
        "timestamp": note.get("time", ""),
        "tags": note.get("tag_list", [])
    }

🚀 进阶应用场景

1. 竞品分析系统

利用xhs采集竞品账号的内容策略、互动数据和粉丝增长情况，构建竞品监控仪表盘。

2. 内容趋势预测

通过大量笔记数据的采集分析，识别热门话题、流行标签和内容趋势，为内容创作提供数据支持。

3. 用户行为研究

分析用户互动模式（点赞、收藏、评论），研究不同内容类型的用户偏好和传播规律。

4. 营销效果评估

跟踪品牌营销活动在小红书上的表现，量化ROI和用户参与度。

📈 性能优化指南

内存优化

import gc

def memory_efficient_collection(client, keyword, total_pages):
    """内存友好的批量采集"""
    for page in range(1, total_pages + 1):
        notes = client.search_notes(keyword=keyword, page=page)
        
        # 立即处理数据并释放内存
        process_and_save(notes)
        del notes
        gc.collect()  # 手动触发垃圾回收

连接池管理

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_robust_session():
    """创建健壮的请求会话"""
    session = requests.Session()
    
    # 配置重试策略
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504]
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    
    return session

🎯 开始你的数据采集之旅

现在你已经掌握了xhs工具的核心使用方法和最佳实践。建议从以下步骤开始：

1. 环境搭建：按照快速部署章节完成基础环境配置
2. 示例学习：仔细阅读example目录下的代码示例
3. 小规模测试：先用少量数据进行功能验证
4. 逐步扩展：根据需求逐步增加采集规模和复杂度

记住，数据采集的核心原则是合法合规、尊重平台规则。合理控制请求频率，避免对目标网站造成过大压力。

项目提供了丰富的示例代码和文档支持，遇到问题时可以：

• 查看example目录下的使用示例
• 参考docs目录中的详细文档
• 查阅CHANGELOG.md了解最新更新

开始探索小红书的数据世界吧！用数据驱动决策，让内容创作和商业分析更加精准高效。

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