小红书数据采集终极指南：快速获取公开数据的完整方案

小红书数据采集终极指南：快速获取公开数据的完整方案

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

在当今社交媒体分析领域，小红书已经成为内容创作者和数据分析师不可或缺的平台。xhs项目是一个基于小红书Web端请求封装的Python库，让你能够以编程方式高效获取笔记、用户、搜索等公开数据，无需手动操作网页。无论你是技术开发者、数据分析师还是内容运营人员，这个工具都能为你提供强大的数据采集能力。

核心关键词：小红书数据采集、Python爬虫、内容分析长尾关键词：小红书API封装、笔记数据获取、用户信息分析、内容分类浏览、数据采集合规

为什么你需要小红书数据采集工具？

传统的网页爬虫开发面临诸多挑战：复杂的反爬机制、频繁的接口变更、数据格式不统一。xhs工具通过封装官方接口，为你解决了这些痛点：

重要提示：本工具仅用于学习和研究目的，请遵守小红书平台的使用条款，不要对网站造成压力或进行未经授权的商业活动。

5分钟快速入门

安装与配置

首先确保你已安装Python 3.7或更高版本，然后通过pip一键安装：

pip install xhs

如果需要最新功能，可以直接从Git仓库安装：

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

基础认证准备

要使用xhs工具，你需要准备两个关键信息：

1. Cookie- 从小红书网站获取的登录凭证
2. 签名函数- 处理请求签名的JavaScript函数

获取Cookie的简单方法：登录小红书网页版后，按F12打开开发者工具，在Network标签中复制任意请求的Cookie字段。

你的第一个采集脚本

from xhs import XhsClient # 初始化客户端 cookie = "your_xiaohongshu_cookie_here" xhs_client = XhsClient(cookie) # 获取笔记详情 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('comments', 0)}")

核心功能深度解析

笔记数据采集的三种方式

1. 按ID获取单篇笔记

# 获取指定ID的笔记详情 note = xhs_client.get_note_by_id("笔记ID") # 提取笔记中的图片URL from xhs import help image_urls = help.get_imgs_url_from_note(note) video_url = help.get_video_url_from_note(note)

2. 获取用户所有笔记

# 分页获取用户发布的笔记 user_notes = xhs_client.get_user_notes("用户ID", cursor="") # 批量获取用户所有笔记（自动分页） all_notes = xhs_client.get_user_all_notes("用户ID", crawl_interval=2)

3. 关键词搜索笔记

# 搜索"美妆教程"相关笔记 search_results = xhs_client.get_note_by_keyword( keyword="美妆教程", page=1, sort="general" # 可选：general(综合)、time(时间) )

用户数据分析

深入了解创作者信息对于内容分析至关重要：

# 获取用户基本信息 user_info = xhs_client.get_user_info("用户ID") print(f"用户名：{user_info['nickname']}") print(f"粉丝数：{user_info['fans']}") print(f"获赞数：{user_info['likes']}") print(f"笔记数：{user_info['notes']}") # 搜索用户 search_users = xhs_client.get_user_by_keyword("美妆博主")

内容分类浏览

xhs工具内置了丰富的内容分类，让你可以按兴趣领域获取内容：

from xhs import FeedType # 获取穿搭类热门内容 fashion_notes = xhs_client.get_home_feed(FeedType.FASION) # 获取美食类内容 food_notes = xhs_client.get_home_feed(FeedType.FOOD) # 获取旅行攻略 travel_notes = xhs_client.get_home_feed(FeedType.TRAVEL)

支持的内容分类包括：

• 穿搭（FASION） - 时尚潮流内容
• 美食（FOOD） - 餐饮探店分享
• 彩妆（COSMETICS） - 美妆教程
• 影视（MOVIE） - 电影电视剧评
• 职场（CAREER） - 职场经验分享
• 情感（EMOTION） - 情感故事
• 家居（HOURSE） - 家居装饰
• 游戏（GAME） - 游戏攻略
• 旅行（TRAVEL） - 旅行攻略
• 健身（FITNESS） - 健身教程

实战案例：竞品内容监控系统

案例1：美妆品牌竞品分析

import pandas as pd from datetime import datetime, timedelta def analyze_competitor_performance(competitor_ids, days=30): """分析多个竞品账号的内容表现""" competitor_data = {} for competitor_id in competitor_ids: all_notes = [] has_more = True cursor = "" # 获取最近30天的笔记 while has_more: response = xhs_client.get_user_notes(competitor_id, cursor=cursor) notes = response.get("notes", []) has_more = response.get("has_more", False) cursor = response.get("cursor", "") # 过滤最近30天的笔记 cutoff_date = datetime.now() - timedelta(days=days) recent_notes = [ note for note in notes if datetime.fromtimestamp(note['time']/1000) > cutoff_date ] all_notes.extend(recent_notes) if not has_more or len(recent_notes) == 0: break # 计算关键指标 if all_notes: df = pd.DataFrame(all_notes) competitor_data[competitor_id] = { "total_notes": len(df), "avg_likes": df['likes'].mean(), "avg_comments": df['comments'].mean(), "avg_shares": df['shares'].mean(), "best_note": df.loc[df['likes'].idxmax()].to_dict() } return competitor_data

案例2：热门话题趋势追踪

def track_topic_trend(keyword, days=7, max_pages=5): """追踪话题热度变化趋势""" trend_data = [] for day_offset in range(days): date = datetime.now() - timedelta(days=day_offset) date_str = date.strftime("%Y-%m-%d") daily_notes = [] for page in range(1, max_pages + 1): try: results = xhs_client.get_note_by_keyword( keyword=keyword, page=page, sort="time" ) daily_notes.extend(results) time.sleep(1) # 避免请求过快 except Exception as e: print(f"第{page}页请求失败: {e}") break if daily_notes: avg_likes = sum(n.get('likes', 0) for n in daily_notes) / len(daily_notes) avg_comments = sum(n.get('comments', 0) for n in daily_notes) / len(daily_notes) trend_data.append({ "date": date_str, "total_notes": len(daily_notes), "avg_likes": round(avg_likes, 2), "avg_comments": round(avg_comments, 2), "top_note": max(daily_notes, key=lambda x: x.get('likes', 0)) }) return trend_data

避坑指南与最佳实践

常见问题解决方案

Q1: 遇到"签名失败"错误怎么办？A: 签名失败通常是因为签名函数配置问题。项目提供了完整的签名示例，你可以参考example/basic_sign_usage.py中的实现。确保你的签名函数正确处理了URI和data参数。

Q2: 如何避免被限制访问？A: 建议采取以下措施：

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

Q3: 数据采集的合法性边界在哪里？A: 请务必注意：

• 仅采集公开数据，不访问私有内容
• 不要用于商业侵权用途
• 尊重用户隐私和版权
• 遵守小红书平台的使用条款

Q4: 如何高效保存采集的数据？A: 建议使用结构化存储：

import json import csv from datetime import datetime def save_notes_data(notes, filename_prefix="notes"): """保存笔记数据到JSON和CSV格式""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") # 保存为JSON json_filename = f"{filename_prefix}_{timestamp}.json" with open(json_filename, 'w', encoding='utf-8') as f: json.dump(notes, f, ensure_ascii=False, indent=2) # 保存为CSV csv_filename = f"{filename_prefix}_{timestamp}.csv" if notes: df = pd.DataFrame(notes) df.to_csv(csv_filename, index=False, encoding='utf-8-sig') return json_filename, csv_filename

性能优化技巧

1. 请求优化策略

import time from xhs.exception import DataFetchError, IPBlockError def safe_request(func, *args, max_retries=3, **kwargs): """安全的请求函数，带重试机制""" for attempt in range(max_retries): try: return func(*args, **kwargs) except (DataFetchError, IPBlockError) as e: if attempt == max_retries - 1: raise wait_time = 2 ** attempt # 指数退避策略 print(f"请求失败，{wait_time}秒后重试...") time.sleep(wait_time)

2. 批量处理与进度监控

from tqdm import tqdm def batch_collect_notes(note_ids, batch_size=10): """批量采集笔记数据""" results = [] for i in tqdm(range(0, len(note_ids), batch_size), desc="采集进度"): batch = note_ids[i:i+batch_size] for note_id in batch: try: note = safe_request(xhs_client.get_note_by_id, note_id) results.append(note) time.sleep(1) # 控制请求频率 except Exception as e: print(f"采集笔记 {note_id} 失败: {e}") return results

3. 数据清洗与格式化

def clean_and_format_note(raw_note): """清洗和格式化笔记数据""" cleaned = { "note_id": raw_note.get("id", ""), "title": raw_note.get("title", "").strip(), "content": raw_note.get("desc", ""), "author_id": raw_note.get("user", {}).get("user_id", ""), "author_name": raw_note.get("user", {}).get("nickname", ""), "publish_time": datetime.fromtimestamp( raw_note.get("time", 0) / 1000 ).strftime("%Y-%m-%d %H:%M:%S"), "likes": raw_note.get("likes", 0), "comments": raw_note.get("comments", 0), "shares": raw_note.get("shares", 0), "collects": raw_note.get("collects", 0), "tags": [tag.get("name", "") for tag in raw_note.get("tag_list", [])], "image_count": len(raw_note.get("images", [])), "video_url": raw_note.get("video", {}).get("url", "") if raw_note.get("video") else "" } # 计算互动率 total_interactions = cleaned["likes"] + cleaned["comments"] + cleaned["shares"] cleaned["interaction_rate"] = round(total_interactions / max(cleaned["likes"], 1), 4) return cleaned

进阶技巧与高级用法

1. 错误处理与监控

import logging from functools import wraps def log_errors(func): """错误处理装饰器""" @wraps(func) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except DataFetchError as e: logging.error(f"数据获取失败: {e}") raise except IPBlockError as e: logging.critical(f"IP被限制: {e}") # 这里可以添加IP切换逻辑 raise except Exception as e: logging.exception(f"未知错误: {e}") raise return wrapper @log_errors def safe_get_user_info(user_id): """安全获取用户信息""" return xhs_client.get_user_info(user_id)

2. 数据质量验证

def validate_note_data(note_data): """验证笔记数据的完整性""" required_fields = ['id', 'title', 'user', 'time'] missing_fields = [field for field in required_fields if field not in note_data] if missing_fields: raise ValueError(f"笔记数据缺失必要字段: {missing_fields}") # 验证时间戳格式 if not isinstance(note_data.get('time'), (int, float)): raise ValueError("时间戳格式不正确") # 验证用户信息 user_info = note_data.get('user', {}) if not user_info.get('user_id'): raise ValueError("用户信息不完整") return True

3. 配置管理

import yaml from dataclasses import dataclass from typing import Optional @dataclass class XhsConfig: """小红书采集配置""" cookie: str request_interval: float = 2.0 max_retries: int = 3 timeout: int = 30 proxy: Optional[str] = None @classmethod def from_yaml(cls, config_path): """从YAML文件加载配置""" with open(config_path, 'r', encoding='utf-8') as f: config_data = yaml.safe_load(f) return cls(**config_data) def to_yaml(self, config_path): """保存配置到YAML文件""" with open(config_path, 'w', encoding='utf-8') as f: yaml.dump(self.__dict__, f, allow_unicode=True)

合规使用与道德考量

可以做的 ✅

• 采集公开笔记数据进行学术研究
• 用于个人学习和小规模项目
• 分析公开的用户行为模式
• 遵守平台的robots.txt规则

禁止做的 ❌

• 大规模商业数据采集未经授权
• 侵犯用户隐私和个人信息
• 对服务器造成压力或攻击
• 违反平台用户协议和法律法规

最佳实践建议

1. 频率控制：合理设置请求间隔，避免对服务器造成压力
2. 数据最小化：只采集必要的数据字段
3. 用户尊重：不采集敏感个人信息
4. 透明使用：明确告知数据用途（如用于研究）
5. 定期审查：定期检查采集行为是否符合平台政策

资源与支持

核心模块说明

• xhs/core.py- 主要功能实现，包含所有API方法
• xhs/help.py- 辅助函数，提供数据处理工具
• xhs/exception.py- 异常处理类定义

示例代码参考

项目提供了丰富的示例代码，帮助你快速上手：

• example/basic_usage.py- 基础使用示例
• example/login_qrcode.py- 二维码登录示例
• example/basic_sign_server.py- 签名服务器示例
• example/basic_sign_usage.py- 签名使用示例

测试用例学习

通过测试用例了解工具的正确用法：

• tests/test_xhs.py- 主要功能测试
• tests/test_help.py- 辅助函数测试
• tests/utils.py- 测试工具函数

总结

xhs小红书数据采集工具为你提供了一个强大而灵活的数据获取方案。通过这个工具，你可以：

1. 快速上手：几分钟内开始采集小红书公开数据
2. 深度分析：获取完整的笔记、用户、搜索数据
3. 高效处理：内置的数据清洗和格式化功能
4. 安全合规：遵循最佳实践，降低合规风险

记住，技术只是工具，如何使用它才是关键。始终将合规性和道德考量放在首位，用技术创造价值而不是问题。希望这个工具能帮助你在小红书数据分析的道路上走得更远、更稳！

如果你在使用过程中遇到问题，或者有改进建议，欢迎参与项目的开发和讨论。开源社区的力量能让这个工具变得更加强大和完善。

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