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

小红书数据采集终极指南快速获取公开数据的完整方案【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs在当今社交媒体分析领域小红书已经成为内容创作者和数据分析师不可或缺的平台。xhs项目是一个基于小红书Web端请求封装的Python库让你能够以编程方式高效获取笔记、用户、搜索等公开数据无需手动操作网页。无论你是技术开发者、数据分析师还是内容运营人员这个工具都能为你提供强大的数据采集能力。核心关键词小红书数据采集、Python爬虫、内容分析长尾关键词小红书API封装、笔记数据获取、用户信息分析、内容分类浏览、数据采集合规为什么你需要小红书数据采集工具传统的网页爬虫开发面临诸多挑战复杂的反爬机制、频繁的接口变更、数据格式不统一。xhs工具通过封装官方接口为你解决了这些痛点挑战传统方案xhs解决方案反爬机制需要处理验证码、动态签名内置自动签名验证接口维护频繁调整代码适配接口变更封装稳定接口层数据解析手动解析HTML结构易变提供结构化数据返回开发效率数天到数周开发周期几分钟上手使用合规风险容易触发频率限制内置请求间隔控制重要提示本工具仅用于学习和研究目的请遵守小红书平台的使用条款不要对网站造成压力或进行未经授权的商业活动。5分钟快速入门安装与配置首先确保你已安装Python 3.7或更高版本然后通过pip一键安装pip install xhs如果需要最新功能可以直接从Git仓库安装pip install githttps://gitcode.com/gh_mirrors/xh/xhs基础认证准备要使用xhs工具你需要准备两个关键信息Cookie- 从小红书网站获取的登录凭证签名函数- 处理请求签名的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_interval2)3. 关键词搜索笔记# 搜索美妆教程相关笔记 search_results xhs_client.get_note_by_keyword( keyword美妆教程, page1, sortgeneral # 可选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, days30): 分析多个竞品账号的内容表现 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, cursorcursor) notes response.get(notes, []) has_more response.get(has_more, False) cursor response.get(cursor, ) # 过滤最近30天的笔记 cutoff_date datetime.now() - timedelta(daysdays) 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, days7, max_pages5): 追踪话题热度变化趋势 trend_data [] for day_offset in range(days): date datetime.now() - timedelta(daysday_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( keywordkeyword, pagepage, sorttime ) 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, keylambda 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_prefixnotes): 保存笔记数据到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, encodingutf-8) as f: json.dump(notes, f, ensure_asciiFalse, indent2) # 保存为CSV csv_filename f{filename_prefix}_{timestamp}.csv if notes: df pd.DataFrame(notes) df.to_csv(csv_filename, indexFalse, encodingutf-8-sig) return json_filename, csv_filename性能优化技巧1. 请求优化策略import time from xhs.exception import DataFetchError, IPBlockError def safe_request(func, *args, max_retries3, **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_size10): 批量采集笔记数据 results [] for i in tqdm(range(0, len(note_ids), batch_size), desc采集进度): batch note_ids[i:ibatch_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 results3. 数据清洗与格式化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(fIP被限制: {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 True3. 配置管理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, encodingutf-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, encodingutf-8) as f: yaml.dump(self.__dict__, f, allow_unicodeTrue)合规使用与道德考量可以做的 ✅采集公开笔记数据进行学术研究用于个人学习和小规模项目分析公开的用户行为模式遵守平台的robots.txt规则禁止做的 ❌大规模商业数据采集未经授权侵犯用户隐私和个人信息对服务器造成压力或攻击违反平台用户协议和法律法规最佳实践建议频率控制合理设置请求间隔避免对服务器造成压力数据最小化只采集必要的数据字段用户尊重不采集敏感个人信息透明使用明确告知数据用途如用于研究定期审查定期检查采集行为是否符合平台政策资源与支持核心模块说明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小红书数据采集工具为你提供了一个强大而灵活的数据获取方案。通过这个工具你可以快速上手几分钟内开始采集小红书公开数据深度分析获取完整的笔记、用户、搜索数据高效处理内置的数据清洗和格式化功能安全合规遵循最佳实践降低合规风险记住技术只是工具如何使用它才是关键。始终将合规性和道德考量放在首位用技术创造价值而不是问题。希望这个工具能帮助你在小红书数据分析的道路上走得更远、更稳如果你在使用过程中遇到问题或者有改进建议欢迎参与项目的开发和讨论。开源社区的力量能让这个工具变得更加强大和完善。【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考