小红书数据采集Python库：从入门到实战的完整指南

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

在小红书内容生态日益繁荣的今天，如何高效获取和分析平台数据成为了众多开发者和数据分析师关注的焦点。xhs 是一个基于 Python 的小红书数据采集工具，它封装了小红书 Web 端的请求接口，为开发者提供了稳定、易用的数据采集解决方案。无论你是进行社交媒体分析、竞品研究还是内容监控，这个开源库都能帮助你快速获取所需的小红书公开数据。

项目概览：为什么选择 xhs 进行数据采集？

xhs 库通过封装小红书官方接口，解决了传统爬虫开发中的诸多痛点。与直接编写爬虫相比，使用 xhs 具有以下显著优势：

安装与基础配置

安装 xhs 库非常简单，只需一行命令：

pip install xhs

如果你需要最新版本，可以直接从 Git 仓库安装：

pip install git+https://gitcode.com/gh_mirrors/xh/xhs

核心能力展示：解锁小红书数据采集的完整功能

1. 客户端初始化与认证

xhs 库支持多种认证方式，包括 Cookie 认证和二维码登录：

from xhs import XhsClient import datetime # 方式一：使用 Cookie 初始化 def sign(uri, data=None, a1="", web_session=""): """签名函数实现""" # 这里需要实现具体的签名逻辑 return {"x-s": "signature", "x-t": str(int(datetime.datetime.now().timestamp() * 1000))} cookie = "your_xhs_cookie_here" xhs_client = XhsClient(cookie=cookie, sign=sign) # 方式二：二维码登录 from xhs import XhsClient import qrcode xhs_client_qr = XhsClient(sign=sign) qr_res = xhs_client_qr.get_qrcode() print(f"请扫描二维码登录：{qr_res['url']}") # 生成二维码图片 qr = qrcode.QRCode(version=1, error_correction=qrcode.ERROR_CORRECT_L, box_size=10, border=4) qr.add_data(qr_res["url"]) qr.make(fit=True) img = qr.make_image(fill_color="black", back_color="white") img.save("xhs_qrcode.png")

2. 笔记数据获取与分析

xhs 库提供了丰富的笔记数据获取接口：

# 获取单篇笔记详情 note_id = "6505318c000000001f03c5a6" note_data = xhs_client.get_note_by_id(note_id) print(f"笔记标题：{note_data.get('title', '无标题')}") print(f"作者：{note_data.get('user', {}).get('nickname', '匿名')}") print(f"点赞数：{note_data.get('likes', 0)}") print(f"收藏数：{note_data.get('collects', 0)}") print(f"发布时间：{datetime.datetime.fromtimestamp(note_data.get('time', 0)/1000)}") # 获取笔记中的图片和视频 from xhs import help image_urls = help.get_imgs_url_from_note(note_data) video_url = help.get_video_url_from_note(note_data) print(f"包含 {len(image_urls)} 张图片") if video_url: print(f"视频地址：{video_url}")

3. 用户信息与内容搜索

# 搜索用户 search_results = xhs_client.get_user_by_keyword("Python", page=1) for user in search_results["users"]: print(f"用户ID：{user['user_id']}，昵称：{user['nickname']}，粉丝数：{user['fans']}") # 获取用户详细信息 user_info = xhs_client.get_user_info("5ff0e6410000000001008400") print(f"用户详情：{user_info}") # 获取用户发布的笔记 user_notes = xhs_client.get_user_notes( user_id="5ff0e6410000000001008400", page=1, page_size=20 ) print(f"用户共发布 {len(user_notes)} 篇笔记")

4. 内容分类与推荐流

xhs 库内置了多种内容分类，方便按领域获取内容：

from xhs import FeedType # 获取穿搭类内容推荐 fashion_notes = xhs_client.get_home_feed(FeedType.FASION) print(f"获取到 {len(fashion_notes)} 篇穿搭类笔记") # 获取美食类内容 food_notes = xhs_client.get_home_feed(FeedType.FOOD) # 获取旅行类内容 travel_notes = xhs_client.get_home_feed(FeedType.TRAVEL) # 支持的所有分类 feed_types = [ FeedType.RECOMMEND, # 推荐 FeedType.FASION, # 穿搭 FeedType.FOOD, # 美食 FeedType.COSMETICS, # 彩妆 FeedType.MOVIE, # 影视 FeedType.CAREER, # 职场 FeedType.EMOTION, # 情感 FeedType.HOURSE, # 家居 FeedType.GAME, # 游戏 FeedType.TRAVEL, # 旅行 FeedType.FITNESS # 健身 ]

实战应用场景：解决真实业务问题

场景一：竞品内容监控与分析

对于品牌营销团队，监控竞品在小红书上的表现至关重要：

import pandas as pd from datetime import datetime, timedelta class CompetitorAnalyzer: def __init__(self, xhs_client, competitor_ids): self.client = xhs_client self.competitor_ids = competitor_ids def analyze_content_performance(self, days=30): """分析竞品最近N天的内容表现""" results = [] for competitor_id in self.competitor_ids: competitor_data = { "user_id": competitor_id, "notes": [], "metrics": {} } # 获取用户信息 user_info = self.client.get_user_info(competitor_id) competitor_data["user_info"] = user_info # 获取最近发布的笔记 page = 1 while True: notes = self.client.get_user_notes(competitor_id, page=page) if not notes: break # 过滤最近N天的笔记 cutoff_date = datetime.now() - timedelta(days=days) recent_notes = [ note for note in notes if datetime.fromtimestamp(note['time']/1000) > cutoff_date ] competitor_data["notes"].extend(recent_notes) # 如果笔记时间早于截止日期，停止翻页 if notes and datetime.fromtimestamp(notes[-1]['time']/1000) < cutoff_date: break page += 1 time.sleep(1) # 避免请求过快 # 计算关键指标 if competitor_data["notes"]: df = pd.DataFrame(competitor_data["notes"]) competitor_data["metrics"] = { "total_notes": len(df), "avg_likes": df['likes'].mean(), "avg_comments": df['comments'].mean(), "avg_shares": df['shares'].mean(), "max_likes": df['likes'].max(), "best_note": df.loc[df['likes'].idxmax()] if not df.empty else None } results.append(competitor_data) return results # 使用示例 analyzer = CompetitorAnalyzer(xhs_client, ["user_id_1", "user_id_2", "user_id_3"]) competitor_data = analyzer.analyze_content_performance(days=7) for data in competitor_data: print(f"用户 {data['user_info']['nickname']}:") print(f" 发布笔记数: {data['metrics'].get('total_notes', 0)}") print(f" 平均点赞: {data['metrics'].get('avg_likes', 0):.1f}")

场景二：热门话题趋势追踪

class TopicTracker: def __init__(self, xhs_client): self.client = xhs_client def track_topic_trend(self, keyword, days=7, max_pages=5): """追踪话题热度趋势""" trend_data = {} for day_offset in range(days): # 模拟按时间搜索（实际可能需要更复杂的逻辑） search_results = self.client.get_note_by_keyword( keyword=keyword, page=1, sort="general", note_type=0 ) date_str = (datetime.now() - timedelta(days=day_offset)).strftime("%Y-%m-%d") if search_results and "notes" in search_results: notes = search_results["notes"] trend_data[date_str] = { "total_notes": len(notes), "total_likes": sum(n.get('likes', 0) for n in notes), "total_comments": sum(n.get('comments', 0) for n in notes), "avg_engagement": sum(n.get('likes', 0) + n.get('comments', 0) for n in notes) / max(len(notes), 1), "top_note": max(notes, key=lambda x: x.get('likes', 0)) if notes else None } time.sleep(0.5) # 请求间隔 return trend_data # 使用示例 tracker = TopicTracker(xhs_client) trend_data = tracker.track_topic_trend("Python编程", days=5) for date, data in trend_data.items(): print(f"{date}: {data['total_notes']} 篇笔记，" f"总点赞 {data['total_likes']}，" f"平均互动率 {data['avg_engagement']:.2f}")

场景三：内容质量评估系统

class ContentQualityAnalyzer: def __init__(self): self.metrics_weights = { 'likes': 0.4, 'comments': 0.3, 'shares': 0.2, 'collects': 0.1 } def calculate_quality_score(self, note_data): """计算内容质量分数""" score = 0 for metric, weight in self.metrics_weights.items(): value = note_data.get(metric, 0) # 归一化处理（假设最大值为10000） normalized_value = min(value / 10000, 1.0) score += normalized_value * weight # 考虑发布时间权重（越新权重越高） publish_time = datetime.fromtimestamp(note_data.get('time', 0) / 1000) days_old = (datetime.now() - publish_time).days time_weight = max(0, 1 - days_old / 30) # 30天内有效 return score * time_weight * 100 # 转换为百分制 def analyze_content_patterns(self, notes_data): """分析内容模式""" patterns = { 'high_performing': [], 'trending_topics': {}, 'best_post_times': [] } for note in notes_data: score = self.calculate_quality_score(note) if score > 70: # 高质量内容 patterns['high_performing'].append({ 'note_id': note.get('id'), 'score': score, 'title': note.get('title', ''), 'tags': [tag.get('name', '') for tag in note.get('tag_list', [])] }) # 分析标签 for tag in note.get('tag_list', []): tag_name = tag.get('name', '') if tag_name: patterns['trending_topics'][tag_name] = patterns['trending_topics'].get(tag_name, 0) + 1 # 分析发布时间 publish_hour = datetime.fromtimestamp(note.get('time', 0) / 1000).hour patterns['best_post_times'].append(publish_hour) return patterns

最佳实践与性能优化指南

1. 错误处理与重试机制

from xhs.exception import DataFetchError, IPBlockError, SignError import time from typing import Optional, Any class SafeXhsClient: def __init__(self, base_client, max_retries=3, retry_delay=1): self.client = base_client self.max_retries = max_retries self.retry_delay = retry_delay def safe_request(self, func, *args, **kwargs) -> Optional[Any]: """安全请求包装器，包含重试机制""" for attempt in range(self.max_retries): try: return func(*args, **kwargs) except (DataFetchError, IPBlockError, SignError) as e: if attempt == self.max_retries - 1: raise wait_time = self.retry_delay * (2 ** attempt) # 指数退避 print(f"请求失败 ({type(e).__name__})，{wait_time}秒后重试 (第{attempt + 1}次)...") time.sleep(wait_time) except Exception as e: print(f"未知错误: {e}") raise return None def get_note_safe(self, note_id: str, **kwargs): """安全获取笔记""" return self.safe_request(self.client.get_note_by_id, note_id, **kwargs) def search_safe(self, keyword: str, **kwargs): """安全搜索""" return self.safe_request(self.client.get_note_by_keyword, keyword, **kwargs) # 使用示例 safe_client = SafeXhsClient(xhs_client, max_retries=3) note_data = safe_client.get_note_safe("6505318c000000001f03c5a6")

2. 数据缓存与批量处理

import json import hashlib from pathlib import Path from typing import Dict, List class XhsDataCache: def __init__(self, cache_dir=".xhs_cache"): self.cache_dir = Path(cache_dir) self.cache_dir.mkdir(exist_ok=True) def _get_cache_key(self, func_name: str, *args, **kwargs) -> str: """生成缓存键""" params_str = json.dumps({ 'args': args, 'kwargs': kwargs }, sort_keys=True, ensure_ascii=False) key = f"{func_name}_{hashlib.md5(params_str.encode()).hexdigest()}" return key def get_cached(self, func_name: str, *args, **kwargs): """获取缓存数据""" cache_key = self._get_cache_key(func_name, *args, **kwargs) cache_file = self.cache_dir / f"{cache_key}.json" if cache_file.exists(): with open(cache_file, 'r', encoding='utf-8') as f: return json.load(f) return None def set_cache(self, func_name: str, data, *args, **kwargs): """设置缓存""" cache_key = self._get_cache_key(func_name, *args, **kwargs) cache_file = self.cache_dir / f"{cache_key}.json" with open(cache_file, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) def cached_request(self, func, *args, **kwargs): """带缓存的请求""" func_name = func.__name__ # 尝试从缓存获取 cached_data = self.get_cached(func_name, *args, **kwargs) if cached_data is not None: return cached_data # 执行请求并缓存结果 result = func(*args, **kwargs) self.set_cache(func_name, result, *args, **kwargs) return result # 使用示例 cache = XhsDataCache() cached_note = cache.cached_request(xhs_client.get_note_by_id, "6505318c000000001f03c5a6")

3. 性能优化配置表

常见问题快速解答

Q1: 如何获取有效的 Cookie？

A: 登录小红书网页版后，按 F12 打开开发者工具，在 Network 标签中找到任意请求，复制 Request Headers 中的 Cookie 字段即可。或者使用二维码登录功能自动获取。

Q2: 遇到"签名失败"错误怎么办？

A: 签名失败通常是因为签名函数配置问题。可以参考项目中的示例代码 example/basic_usage.py 和 example/basic_sign_server.py 实现正确的签名逻辑。

Q3: 请求频率有限制吗？

A: 为避免对小红书服务器造成压力，建议：

• 添加适当的请求间隔（1-3秒）
• 避免短时间内大量请求同一接口
• 使用代理IP轮换（如果需要大量采集）
• 遵守平台的 robots.txt 协议

Q4: 数据采集的合法性如何？

A: 请务必注意合规使用：

• ✅ 仅采集公开数据用于研究分析
• ✅ 用于个人学习和小规模项目
• ✅ 遵守平台用户协议
• ❌ 禁止大规模商业数据采集
• ❌ 禁止侵犯用户隐私
• ❌ 禁止对服务器造成压力

Q5: 如何保存和处理采集的数据？

A: 建议使用结构化格式保存：

import pandas as pd import json from datetime import datetime class DataExporter: def __init__(self, output_dir="output"): self.output_dir = Path(output_dir) self.output_dir.mkdir(exist_ok=True) def export_to_json(self, data, filename): """导出为JSON格式""" filepath = self.output_dir / f"{filename}.json" with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) def export_to_csv(self, notes_data, filename): """导出为CSV格式""" # 清洗和格式化数据 cleaned_data = [] for note in notes_data: cleaned_note = { "note_id": note.get("id", ""), "title": note.get("title", ""), "author": note.get("user", {}).get("nickname", ""), "publish_time": datetime.fromtimestamp( note.get("time", 0) / 1000 ).strftime("%Y-%m-%d %H:%M:%S"), "likes": note.get("likes", 0), "comments": note.get("comments", 0), "shares": note.get("shares", 0), "collects": note.get("collects", 0), "tags": ",".join(tag.get("name", "") for tag in note.get("tag_list", [])), "image_count": len(note.get("images", [])), "has_video": bool(note.get("video")) } cleaned_data.append(cleaned_note) df = pd.DataFrame(cleaned_data) filepath = self.output_dir / f"{filename}.csv" df.to_csv(filepath, index=False, encoding='utf-8-sig')

生态集成与扩展

1. 与数据分析工具集成

xhs 库可以轻松集成到现有的数据分析工作流中：

import pandas as pd import matplotlib.pyplot as plt from xhs import XhsClient class XhsDataAnalyzer: def __init__(self, xhs_client): self.client = xhs_client def analyze_user_engagement(self, user_id, num_notes=50): """分析用户互动数据""" notes_data = [] page = 1 while len(notes_data) < num_notes: notes = self.client.get_user_notes(user_id, page=page) if not notes: break notes_data.extend(notes) page += 1 time.sleep(1) # 创建DataFrame分析 df = pd.DataFrame(notes_data[:num_notes]) # 计算互动率 df['interaction_rate'] = (df['likes'] + df['comments'] + df['shares']) / df['likes'].clip(lower=1) # 生成可视化报告 fig, axes = plt.subplots(2, 2, figsize=(12, 10)) # 点赞分布 axes[0, 0].hist(df['likes'], bins=20, edgecolor='black') axes[0, 0].set_title('点赞数分布') axes[0, 0].set_xlabel('点赞数') axes[0, 0].set_ylabel('频次') # 发布时间分析 df['publish_hour'] = pd.to_datetime(df['time'], unit='ms').dt.hour hour_counts = df['publish_hour'].value_counts().sort_index() axes[0, 1].bar(hour_counts.index, hour_counts.values) axes[0, 1].set_title('发布时间分布') axes[0, 1].set_xlabel('小时') axes[0, 1].set_ylabel('发布数量') # 互动率趋势 df_sorted = df.sort_values('time') axes[1, 0].plot(df_sorted['time'], df_sorted['interaction_rate']) axes[1, 0].set_title('互动率趋势') axes[1, 0].set_xlabel('时间') axes[1, 0].set_ylabel('互动率') # 多维度散点图 axes[1, 1].scatter(df['likes'], df['comments'], alpha=0.6) axes[1, 1].set_title('点赞 vs 评论') axes[1, 1].set_xlabel('点赞数') axes[1, 1].set_ylabel('评论数') plt.tight_layout() plt.savefig('user_engagement_analysis.png', dpi=300, bbox_inches='tight') return df, fig

2. 构建监控告警系统

import schedule import time from datetime import datetime import smtplib from email.mime.text import MIMEText class XhsMonitor: def __init__(self, xhs_client, config): self.client = xhs_client self.config = config self.previous_data = {} def monitor_keyword_trend(self, keyword, threshold=1000): """监控关键词趋势""" current_data = self.client.get_note_by_keyword(keyword, page=1) if keyword in self.previous_data: prev_count = self.previous_data[keyword].get('total_notes', 0) curr_count = current_data.get('total_notes', 0) if current_data else 0 growth_rate = (curr_count - prev_count) / max(prev_count, 1) * 100 if growth_rate > threshold: self.send_alert( f"关键词 '{keyword}' 热度激增！", f"增长率: {growth_rate:.1f}%\n" f"前一次: {prev_count} 篇笔记\n" f"当前: {curr_count} 篇笔记" ) self.previous_data[keyword] = current_data def monitor_competitor(self, user_id, metrics_thresholds): """监控竞品表现""" user_info = self.client.get_user_info(user_id) user_notes = self.client.get_user_notes(user_id, page=1) alerts = [] # 检查粉丝增长 if 'fans' in user_info: current_fans = user_info['fans'] # 这里可以添加与历史数据的比较逻辑 # 检查笔记互动率 if user_notes: avg_likes = sum(n.get('likes', 0) for n in user_notes) / len(user_notes) if avg_likes > metrics_thresholds.get('avg_likes', 1000): alerts.append(f"平均点赞数异常: {avg_likes:.0f}") if alerts: self.send_alert( f"竞品 {user_info.get('nickname', user_id)} 异常", "\n".join(alerts) ) def send_alert(self, subject, message): """发送告警""" # 这里可以实现邮件、钉钉、微信等告警方式 print(f"[ALERT] {datetime.now()}: {subject}\n{message}") def start_monitoring(self): """启动监控""" # 监控关键词趋势（每小时） schedule.every().hour.do(self.monitor_keyword_trend, "Python编程", threshold=50) # 监控竞品（每天） schedule.every().day.at("09:00").do( self.monitor_competitor, "competitor_id", {"avg_likes": 1000} ) print("监控系统已启动...") while True: schedule.run_pending() time.sleep(60)

总结与展望：小红书数据采集的未来

xhs 库作为一个功能完善的小红书数据采集工具，为开发者和数据分析师提供了强大的数据获取能力。通过本文的介绍，你应该已经掌握了：

1. 基础使用：如何安装、配置和使用 xhs 库进行数据采集
2. 核心功能：笔记获取、用户信息、内容搜索、分类浏览等核心功能
3. 实战应用：竞品分析、趋势追踪、质量评估等实际业务场景
4. 最佳实践：错误处理、性能优化、数据缓存等工程化实践
5. 生态集成：如何将 xhs 集成到现有的数据分析工作流中

未来发展方向

随着小红书平台的不断发展和数据需求的增加，xhs 库可以考虑以下方向的扩展：

1. 异步支持：添加异步请求支持，提升大规模数据采集效率
2. 数据管道：集成到数据管道工具（如 Apache Airflow、Prefect）
3. 实时监控：支持实时数据流处理和监控告警
4. 机器学习集成：与机器学习框架结合，进行内容分类、情感分析等
5. 可视化仪表板：提供内置的数据可视化界面

合规使用建议

最后再次强调，使用 xhs 库进行数据采集时，请务必：

• 📋 遵守小红书平台的使用条款和服务协议
• 🔒 尊重用户隐私，不采集非公开数据
• ⚖️ 合理控制请求频率，不对平台服务器造成压力
• 📊 仅将数据用于合法的研究和分析目的
• 🛡️ 妥善保管获取的数据，遵守相关数据保护法规

通过合理、合规地使用 xhs 库，你可以高效地获取小红书平台的公开数据，为你的数据分析、市场研究或内容运营工作提供有力支持。项目的完整文档和示例代码可以在项目目录中查看，包括 xhs/core.py 的核心实现和 example/ 目录下的丰富示例。

记住，技术是工具，如何使用它才是关键。希望 xhs 库能成为你在小红书数据分析道路上的得力助手！

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