小红书数据采集终极指南：Python SDK快速上手与实战应用
·
小红书数据采集终极指南：Python SDK快速上手与实战应用
项目地址: https://gitcode.com/gh_mirrors/xh/xhs 小红书作为国内领先的生活方式分享平台，汇聚了海量用户生成内容，对于数据分析师、营销从业者和开发者来说，小红书数据采集已成为获取市场洞察的重要途径。xhs 是一个基于小红书 Web 端请求封装的 Python SDK，提供了完整的小红书数据采集解决方案，让开发者能够高效、稳定地获取平台公开数据。
项目概述与核心价值
xhs 项目是一个专门为小红书平台设计的 Python 数据采集工具包，它封装了复杂的网络请求和签名逻辑，提供了简洁易用的 API 接口。通过这个工具，开发者可以轻松获取小红书笔记内容、用户信息、搜索数据等公开信息，为内容分析、竞品研究和市场趋势预测提供数据支持。
核心优势对比
| 特性 | xhs SDK | 传统爬虫 | 官方API |
|---|---|---|---|
| 易用性 | ⭐⭐⭐⭐⭐ | ⭐⭐ | ⭐⭐⭐ |
| 稳定性 | ⭐⭐⭐⭐ | ⭐⭐ | ⭐⭐⭐⭐⭐ |
| 功能完整性 | ⭐⭐⭐⭐ | ⭐ | ⭐⭐⭐ |
| 维护成本 | 低 | 高 | 低 |
| 合规性 | 中等 | 低 | 高 |
核心功能深度解析
1. 笔记数据获取功能
xhs 提供了完整的笔记数据获取能力，支持多种类型的笔记内容提取：
from xhs import XhsClient
# 初始化客户端
cookie = "your_cookie_here"
xhs_client = XhsClient(cookie)
# 获取指定笔记详情
note_id = "6505318c000000001f03c5a6"
note = xhs_client.get_note_by_id(note_id)
print(f"笔记标题：{note['title']}")
print(f"作者：{note['user']['nickname']}")
print(f"点赞数：{note['likes']}")
print(f"收藏数：{note['collects']}")
2. 搜索功能支持
支持多种搜索条件和排序方式，满足不同场景的数据采集需求：
from xhs import SearchSortType, SearchNoteType
# 关键词搜索
search_results = xhs_client.search(
keyword="Python编程",
sort=SearchSortType.GENERAL,
note_type=SearchNoteType.VIDEO
)
# 获取搜索结果详情
for result in search_results['items']:
print(f"笔记ID：{result['id']}")
print(f"标题：{result['title']}")
print(f"封面图：{result['images_list'][0]['url']}")
3. 内容分类浏览
支持按内容分类获取推荐流，覆盖小红书主要内容领域：
from xhs import FeedType
# 获取美食分类内容
food_notes = xhs_client.get_home_feed(feed_type=FeedType.FOOD)
# 获取穿搭分类内容
fashion_notes = xhs_client.get_home_feed(feed_type=FeedType.FASION)
# 获取旅行分类内容
travel_notes = xhs_client.get_home_feed(feed_type=FeedType.TRAVEL)
实战应用场景展示
场景一：竞品内容监控系统
对于品牌营销团队，监控竞品在小红书上的表现至关重要。xhs 可以帮助构建自动化监控系统：
import schedule
import time
from datetime import datetime
class CompetitorMonitor:
def __init__(self, competitors, xhs_client):
self.competitors = competitors
self.xhs_client = xhs_client
def monitor_competitor_posts(self, competitor_name):
"""监控竞品发布内容"""
print(f"开始监控 {competitor_name} - {datetime.now()}")
# 搜索竞品相关内容
results = self.xhs_client.search(
keyword=competitor_name,
sort=SearchSortType.TIME_DESC
)
# 分析最新内容
latest_posts = results['items'][:5]
for post in latest_posts:
self.analyze_post_engagement(post)
def analyze_post_engagement(self, post_data):
"""分析帖子互动数据"""
engagement_rate = (post_data['likes'] + post_data['collects']) / post_data['views']
print(f"互动率：{engagement_rate:.2%}")
# 提取关键词
keywords = self.extract_keywords(post_data['title'])
print(f"关键词：{', '.join(keywords)}")
场景二：内容趋势分析平台
通过 xhs 采集的数据，可以构建内容趋势分析平台：
import pandas as pd
from collections import Counter
class ContentTrendAnalyzer:
def __init__(self, xhs_client):
self.xhs_client = xhs_client
def analyze_trending_topics(self, category, days=7):
"""分析热门话题趋势"""
trends_data = []
for feed_type in self.get_category_feed_types(category):
notes = self.xhs_client.get_home_feed(feed_type=feed_type)
for note in notes['items'][:20]: # 取前20条
trends_data.append({
'title': note['title'],
'likes': note['likes'],
'keywords': self.extract_keywords(note['title']),
'category': category,
'timestamp': datetime.now()
})
# 创建DataFrame进行分析
df = pd.DataFrame(trends_data)
# 分析热门关键词
all_keywords = []
for keywords in df['keywords']:
all_keywords.extend(keywords)
keyword_counts = Counter(all_keywords)
trending_keywords = keyword_counts.most_common(10)
return {
'trending_keywords': trending_keywords,
'avg_engagement': df['likes'].mean(),
'top_posts': df.nlargest(5, 'likes')[['title', 'likes']].to_dict('records')
}
性能优化与最佳实践
1. 请求频率控制策略
为了避免被平台限制，需要合理控制请求频率：
import time
from functools import wraps
def rate_limit(max_calls=5, period=60):
"""请求频率限制装饰器"""
calls = []
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
now = time.time()
# 清理过期记录
calls[:] = [t for t in calls if t > now - period]
if len(calls) >= max_calls:
sleep_time = period - (now - calls[0])
print(f"达到频率限制，等待 {sleep_time:.1f} 秒")
time.sleep(sleep_time + 1)
calls.append(time.time())
return func(*args, **kwargs)
return wrapper
return decorator
# 使用频率限制
@rate_limit(max_calls=3, period=60)
def safe_get_note(note_id):
"""安全的笔记获取函数"""
return xhs_client.get_note_by_id(note_id)
2. 错误处理与重试机制
健壮的错误处理是数据采集系统的关键：
import random
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class RobustXhsClient:
def __init__(self, cookie, max_retries=3):
self.xhs_client = XhsClient(cookie)
self.max_retries = max_retries
def create_retry_session(self):
"""创建带重试机制的会话"""
session = requests.Session()
retry = Retry(
total=self.max_retries,
read=self.max_retries,
connect=self.max_retries,
backoff_factor=0.5,
status_forcelist=(500, 502, 503, 504),
)
adapter = HTTPAdapter(max_retries=retry)
session.mount('http://', adapter)
session.mount('https://', adapter)
return session
def get_note_with_retry(self, note_id, max_attempts=3):
"""带重试的笔记获取"""
for attempt in range(max_attempts):
try:
note = self.xhs_client.get_note_by_id(note_id)
return note
except Exception as e:
if attempt == max_attempts - 1:
raise
wait_time = (2 ** attempt) + random.random()
print(f"第{attempt+1}次尝试失败，等待{wait_time:.1f}秒后重试")
time.sleep(wait_time)
3. 数据存储优化方案
import sqlite3
import json
from datetime import datetime
class XhsDataManager:
def __init__(self, db_path="xhs_data.db"):
self.db_path = db_path
self.init_database()
def init_database(self):
"""初始化数据库结构"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# 创建笔记表
cursor.execute('''
CREATE TABLE IF NOT EXISTS notes (
id TEXT PRIMARY KEY,
title TEXT,
content TEXT,
user_id TEXT,
likes INTEGER,
collects INTEGER,
comments INTEGER,
publish_time DATETIME,
raw_data TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# 创建用户表
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
user_id TEXT PRIMARY KEY,
nickname TEXT,
avatar TEXT,
notes_count INTEGER,
fans_count INTEGER,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
conn.commit()
conn.close()
def save_note(self, note_data):
"""保存笔记数据"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
INSERT OR REPLACE INTO notes
(id, title, content, user_id, likes, collects, comments, publish_time, raw_data)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (
note_data['id'],
note_data['title'],
note_data.get('desc', ''),
note_data['user']['user_id'],
note_data['likes'],
note_data['collects'],
note_data['comments'],
datetime.fromtimestamp(note_data['time']/1000),
json.dumps(note_data, ensure_ascii=False)
))
conn.commit()
conn.close()
常见问题与解决方案
问题一：签名验证失败
症状：请求返回签名错误或验证失败
解决方案：
- 检查 cookie 是否有效
- 更新签名函数逻辑
- 添加适当的延迟避免频繁请求
# 优化后的签名函数示例
def enhanced_sign(uri, data=None, a1="", web_session=""):
"""增强版签名函数"""
import time
from playwright.sync_api import sync_playwright
for retry in range(3):
try:
with sync_playwright() as playwright:
browser = playwright.chromium.launch(headless=True)
context = browser.new_context()
page = context.new_page()
# 添加必要的初始化脚本
page.goto("https://www.xiaohongshu.com")
# 设置cookie
context.add_cookies([
{'name': 'a1', 'value': a1, 'domain': ".xiaohongshu.com", 'path': "/"}
])
page.reload()
time.sleep(2) # 增加等待时间
# 执行签名
encrypt_params = page.evaluate(
"([url, data]) => window._webmsxyw(url, data)",
[uri, data]
)
browser.close()
return {
"x-s": encrypt_params["X-s"],
"x-t": str(encrypt_params["X-t"])
}
except Exception as e:
if retry == 2:
raise Exception(f"签名失败：{str(e)}")
time.sleep(retry * 2 + 1) # 指数退避
问题二：IP被封禁
症状：请求返回403或连接被拒绝
解决方案：
- 使用代理IP池
- 降低请求频率
- 实现IP自动切换机制
class ProxyManager:
def __init__(self, proxy_list):
self.proxy_list = proxy_list
self.current_index = 0
def get_proxy(self):
"""获取当前代理"""
proxy = self.proxy_list[self.current_index]
self.current_index = (self.current_index + 1) % len(self.proxy_list)
return proxy
def mark_failed(self, proxy):
"""标记代理失效"""
if proxy in self.proxy_list:
self.proxy_list.remove(proxy)
print(f"移除失效代理：{proxy}")
问题三：数据解析异常
症状：返回数据格式变化导致解析失败
解决方案：
- 添加数据验证逻辑
- 实现兼容性解析
- 建立数据格式监控
def safe_parse_note_data(note_response):
"""安全解析笔记数据"""
try:
# 尝试多种可能的字段名
note_data = note_response.get('data', {}) or note_response
# 提取标题（兼容不同字段名）
title = (
note_data.get('title') or
note_data.get('note_title') or
note_data.get('desc', '')
)
# 提取用户信息
user_info = note_data.get('user', {}) or note_data.get('author', {})
return {
'id': note_data.get('id') or note_data.get('note_id'),
'title': title[:100] if title else '', # 限制长度
'user': {
'user_id': user_info.get('user_id'),
'nickname': user_info.get('nickname', '未知用户')
},
'likes': note_data.get('likes', 0) or note_data.get('like_count', 0),
'collects': note_data.get('collects', 0) or note_data.get('collect_count', 0),
'comments': note_data.get('comments', 0) or note_data.get('comment_count', 0)
}
except Exception as e:
print(f"数据解析失败：{str(e)}")
return None
未来发展与社区贡献
1. 功能扩展方向
xhs 项目在现有基础上可以进一步扩展以下功能：
- 实时数据流支持：添加 WebSocket 支持，实现实时内容监控
- 批量处理优化：提升批量数据采集的效率和稳定性
- 数据导出格式：支持更多数据导出格式（CSV、JSON、Excel、数据库）
- 可视化分析：集成数据可视化组件，提供开箱即用的分析报告
2. 性能优化计划
| 优化方向 | 目标 | 预计收益 |
|---|---|---|
| 异步请求支持 | 支持 asyncio 异步请求 | 提升并发性能300% |
| 缓存机制 | 实现多级缓存策略 | 减少重复请求50% |
| 连接池优化 | 优化HTTP连接复用 | 降低延迟30% |
| 内存管理 | 优化大数据集处理 | 减少内存占用40% |
3. 社区贡献指南
欢迎开发者参与 xhs 项目的改进和扩展：
- 问题反馈：在项目仓库提交 Issue，描述遇到的问题和复现步骤
- 功能建议：提出新功能需求或改进建议
- 代码贡献：遵循项目代码规范，提交 Pull Request
- 文档完善：帮助改进文档和示例代码
4. 安全与合规建议
在使用 xhs 进行数据采集时，请遵守以下原则：
- 尊重平台规则：遵守小红书平台的使用条款和服务协议
- 合理使用数据：仅用于学习和研究目的，不用于商业竞争
- 控制请求频率：避免对服务器造成过大压力
- 保护用户隐私：对采集的数据进行匿名化处理
- 注明数据来源：在分析报告中注明数据来源
快速开始指南
安装与配置
# 安装最新版本
pip install xhs
# 或者从源码安装
git clone https://gitcode.com/gh_mirrors/xh/xhs
cd xhs
pip install -e .
基础使用示例
参考示例代码：example/basic_usage.py
from xhs import XhsClient
# 初始化客户端
cookie = "your_cookie_string"
xhs_client = XhsClient(cookie)
# 获取笔记详情
note = xhs_client.get_note_by_id("笔记ID")
# 搜索内容
results = xhs_client.search(keyword="Python编程")
# 获取推荐流
recommendations = xhs_client.get_home_feed()
高级配置
参考核心源码：xhs/core.py 了解所有可用配置选项
通过本文的介绍，您已经掌握了使用 xhs 进行小红书数据采集的核心技术。记住，技术是工具，合规使用是关键。合理运用这些方法，将为您的数据分析项目提供强有力的支持，帮助您在小红书内容生态中获得有价值的洞察。
项目文档：docs/source/xhs.rst 配置示例：example/ 核心源码：xhs/
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
5
0- 0
已为社区贡献45条内容
所有评论(0)