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

小红书数据采集终极指南：Python xhs库的完整使用教程

news 2026/7/17 10:45:54

小红书数据采集终极指南：Python xhs库的完整使用教程

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

在小红书内容营销和数据分析日益重要的今天，你是否还在为手动采集数据而苦恼？xhs工具作为专为小红书数据采集设计的Python利器，让数据获取变得前所未有的简单高效，即使是技术新手也能轻松上手。这款小红书数据采集神器基于小红书Web端进行请求封装，提供了完整的API接口，让你能够轻松获取用户信息、笔记详情、评论数据等宝贵资源。

🚀 为什么选择xhs进行小红书数据采集？

传统爬虫开发的痛点

传统的小红书爬虫开发面临多重挑战：JavaScript逆向工程复杂、反爬虫机制频繁更新、签名算法维护成本高。对于非专业开发者来说，技术门槛过高，而xhs库完美解决了这些问题。

xhs的核心优势

智能签名机制：内置动态签名算法，自动适应平台更新
多账号统一管理：支持多账号统一签名服务，确保长期稳定运行
完整API覆盖：从用户信息到笔记详情，再到评论互动，一应俱全
简单易用：几行代码即可实现复杂的数据采集功能

📦 快速安装与环境配置

基础环境准备

确保系统已安装Python 3.8+环境，推荐使用虚拟环境管理依赖：

# 创建虚拟环境 python -m venv xhs_env # 激活虚拟环境（Linux/Mac） source xhs_env/bin/activate # 激活虚拟环境（Windows） xhs_env\Scripts\activate # 安装xhs库 pip install xhs

签名服务配置

xhs库需要签名服务来绕过小红书的反爬虫机制。你可以选择以下两种方式：

方式一：本地启动签名服务

# 安装playwright和stealth.min.js pip install playwright playwright install curl -O https://cdn.jsdelivr.net/gh/requireCool/stealth.min.js/stealth.min.js

方式二：Docker一键部署

docker run -it -d -p 5005:5005 reajason/xhs-api:latest

🔧 核心功能实战演示

1. 基础数据采集

获取小红书笔记详细信息变得异常简单：

from xhs import XhsClient import json # 初始化客户端（需要配置签名函数） cookie = "your_cookie_here" xhs_client = XhsClient(cookie, sign=sign_function) # 获取笔记详情 note = xhs_client.get_note_by_id("笔记ID", "xsec_token") print(json.dumps(note, indent=4))

2. 用户信息获取

# 获取用户基本信息 user_info = xhs_client.get_user_info("用户ID") # 搜索用户 search_results = xhs_client.get_user_by_keyword("关键词", page=1, page_size=20)

3. 内容批量采集

# 获取用户所有笔记 all_notes = xhs_client.get_user_all_notes("用户ID", crawl_interval=1) # 搜索相关笔记 search_notes = xhs_client.get_note_by_keyword( keyword="美食", page=1, page_size=20, sort="general" )

4. 评论数据提取

# 获取笔记评论 comments = xhs_client.get_note_comments("笔记ID", cursor="", xsec_token="") # 获取所有评论（包括子评论） all_comments = xhs_client.get_note_all_comments("笔记ID", crawl_interval=1)

🎯 高级功能与应用场景

品牌营销监控

通过设置相关关键词和筛选条件，实时追踪品牌提及笔记：

# 监控品牌关键词 brand_keywords = ["品牌A", "品牌B", "竞品C"] for keyword in brand_keywords: notes = xhs_client.get_note_by_keyword(keyword, page=1, page_size=50) # 分析笔记内容、发布时间、互动数据等

竞品分析对比

同时监控多个竞品账号，对比关键指标：

competitor_ids = ["用户ID1", "用户ID2", "用户ID3"] for user_id in competitor_ids: user_info = xhs_client.get_user_info(user_id) user_notes = xhs_client.get_user_all_notes(user_id) # 分析粉丝增长、内容策略、互动效果

内容策略优化

分析热门笔记的特征，为内容创作提供数据支撑：

# 分析热门笔记特征 hot_notes = xhs_client.get_note_by_keyword("热门话题", sort="hot") for note in hot_notes: # 分析标题、内容、标签、发布时间、互动数据 analyze_note_features(note)

⚙️ 实用配置技巧与最佳实践

稳定性保障策略

合理请求间隔：建议设置不少于2秒的采集间隔，避免触发反爬机制
异常重试机制：网络波动时自动重试，确保数据完整性
数据质量检查：自动验证关键字段完整性，排除异常数据

性能优化建议

# 并发控制示例 import concurrent.futures def fetch_note_data(note_id): return xhs_client.get_note_by_id(note_id, "xsec_token") note_ids = ["id1", "id2", "id3", "id4"] with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: results = list(executor.map(fetch_note_data, note_ids))

Cookie管理与维护

正确的Cookie配置是稳定运行的关键：

# Cookie格式要求 cookie = """ a1=your_a1_value; web_session=your_web_session; webId=your_web_id; 其他cookie字段... """ # 验证Cookie有效性 try: self_info = xhs_client.get_self_info() print("Cookie验证成功") except Exception as e: print(f"Cookie无效或已过期: {e}")

🔍 项目结构与源码解析

核心模块说明

xhs/core.py：主客户端类XhsClient的实现，包含所有API方法
xhs/help.py：辅助函数模块，提供签名、URL转换等工具函数
xhs/exception.py：自定义异常处理类
example/：丰富的使用示例代码
tests/：完整的测试用例

签名机制深度解析

xhs库的签名机制是其核心优势：

# 签名函数示例（简化版） def sign(uri, data=None, a1="", web_session=""): """ 通过playwright模拟浏览器环境获取签名 uri: 请求的URI data: 请求数据 a1: cookie中的a1字段 web_session: cookie中的web_session字段 """ # 使用playwright调用浏览器环境中的签名函数 # 返回x-s和x-t签名参数 return { "x-s": encrypt_params["X-s"], "x-t": str(encrypt_params["X-t"]) }

🛡️ 合规使用与风险控制

合法合规采集原则

仅采集公开数据：只获取公开可访问的用户和内容信息
尊重平台规则：遵守小红书的robots.txt和服务条款
合理请求频率：避免对服务器造成过大负担
数据使用规范：仅用于合法合规的分析和研究目的

错误处理与日志记录

import logging from xhs import DataFetchError # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) try: # 数据采集操作 data = xhs_client.get_note_by_id(note_id, xsec_token) except DataFetchError as e: logger.error(f"数据获取失败: {e}") # 实现重试逻辑 except Exception as e: logger.error(f"未知错误: {e}")

📊 数据存储与处理建议

数据格式标准化

import pandas as pd from datetime import datetime def process_note_data(note): """标准化笔记数据结构""" return { "note_id": note.get("id"), "title": note.get("title"), "content": note.get("desc"), "user_id": note.get("user", {}).get("user_id"), "likes": note.get("likes", 0), "collects": note.get("collects", 0), "comments": note.get("comments", 0), "publish_time": datetime.fromtimestamp(note.get("time", 0)), "tags": [tag.get("name") for tag in note.get("tag_list", [])] } # 转换为DataFrame notes_data = [process_note_data(note) for note in notes_list] df = pd.DataFrame(notes_data)

数据持久化方案

import json import csv import sqlite3 # JSON格式存储 with open('notes_data.json', 'w', encoding='utf-8') as f: json.dump(notes_data, f, ensure_ascii=False, indent=2) # CSV格式存储 df.to_csv('notes_data.csv', index=False, encoding='utf-8-sig') # SQLite数据库存储 conn = sqlite3.connect('xhs_data.db') df.to_sql('notes', conn, if_exists='replace', index=False)

🚀 进阶应用与扩展开发

自定义数据管道

class XhsDataPipeline: def __init__(self, cookie, sign_func): self.client = XhsClient(cookie, sign=sign_func) self.data_buffer = [] def collect_user_insights(self, user_id, max_notes=100): """收集用户洞察数据""" notes = self.client.get_user_all_notes(user_id) insights = self.analyze_user_pattern(notes[:max_notes]) return insights def monitor_topic_trends(self, topic, days=7): """监控话题趋势""" # 实现话题趋势分析逻辑 pass def export_report(self, format='excel'): """导出分析报告""" # 实现数据导出功能 pass

集成到现有系统

# 与Django/Flask集成示例 from flask import Flask, jsonify, request from xhs import XhsClient app = Flask(__name__) @app.route('/api/xhs/notes/<note_id>', methods=['GET']) def get_note_details(note_id): """API接口：获取笔记详情""" try: note = xhs_client.get_note_by_id(note_id, request.args.get('xsec_token')) return jsonify({"success": True, "data": note}) except Exception as e: return jsonify({"success": False, "error": str(e)}), 500

💡 常见问题与解决方案

Q1: 签名失败怎么办？

解决方案：

检查Cookie中的a1字段是否正确
确保stealth.min.js文件路径正确
适当增加playwright的等待时间
尝试重启签名服务

Q2: 如何获取有效的Cookie？

获取方法：

登录小红书网页版
使用浏览器开发者工具获取Cookie
确保包含a1、web_session、webId等关键字段
定期更新Cookie（通常有效期为7-30天）

Q3: 请求频率限制如何处理？

优化策略：

实现请求间隔控制（建议≥2秒）
使用代理IP轮换
实现指数退避重试机制
监控响应状态码，适时暂停采集

📈 性能监控与优化

监控指标设计

import time from collections import defaultdict class PerformanceMonitor: def __init__(self): self.metrics = defaultdict(list) def track_request(self, endpoint, duration, success): """跟踪请求性能""" self.metrics[endpoint].append({ "timestamp": time.time(), "duration": duration, "success": success }) def get_performance_report(self): """生成性能报告""" report = {} for endpoint, records in self.metrics.items(): success_rate = sum(1 for r in records if r["success"]) / len(records) avg_duration = sum(r["duration"] for r in records) / len(records) report[endpoint] = { "total_requests": len(records), "success_rate": success_rate, "avg_duration": avg_duration } return report