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

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