小红书数据采集Python库从入门到实战的完整指南【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs在小红书内容生态日益繁荣的今天如何高效获取和分析平台数据成为了众多开发者和数据分析师关注的焦点。xhs 是一个基于 Python 的小红书数据采集工具它封装了小红书 Web 端的请求接口为开发者提供了稳定、易用的数据采集解决方案。无论你是进行社交媒体分析、竞品研究还是内容监控这个开源库都能帮助你快速获取所需的小红书公开数据。项目概览为什么选择 xhs 进行数据采集xhs 库通过封装小红书官方接口解决了传统爬虫开发中的诸多痛点。与直接编写爬虫相比使用 xhs 具有以下显著优势特性xhs 库的优势传统爬虫的挑战接口稳定性基于官方接口封装稳定性高需要频繁应对接口变更反爬处理内置签名验证机制手动处理验证码、加密参数开发效率几分钟即可上手需要数天到数周开发时间数据完整性提供完整的数据结构可能缺失关键字段维护成本开源社区共同维护需要持续投入维护资源安装与基础配置安装 xhs 库非常简单只需一行命令pip install xhs如果你需要最新版本可以直接从 Git 仓库安装pip install githttps://gitcode.com/gh_mirrors/xh/xhs核心能力展示解锁小红书数据采集的完整功能1. 客户端初始化与认证xhs 库支持多种认证方式包括 Cookie 认证和二维码登录from xhs import XhsClient import datetime # 方式一使用 Cookie 初始化 def sign(uri, dataNone, a1, web_session): 签名函数实现 # 这里需要实现具体的签名逻辑 return {x-s: signature, x-t: str(int(datetime.datetime.now().timestamp() * 1000))} cookie your_xhs_cookie_here xhs_client XhsClient(cookiecookie, signsign) # 方式二二维码登录 from xhs import XhsClient import qrcode xhs_client_qr XhsClient(signsign) qr_res xhs_client_qr.get_qrcode() print(f请扫描二维码登录{qr_res[url]}) # 生成二维码图片 qr qrcode.QRCode(version1, error_correctionqrcode.ERROR_CORRECT_L, box_size10, border4) qr.add_data(qr_res[url]) qr.make(fitTrue) img qr.make_image(fill_colorblack, back_colorwhite) 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, page1) 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_id5ff0e6410000000001008400, page1, page_size20 ) 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, days30): 分析竞品最近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, pagepage) if not notes: break # 过滤最近N天的笔记 cutoff_date datetime.now() - timedelta(daysdays) 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(days7) 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, days7, max_pages5): 追踪话题热度趋势 trend_data {} for day_offset in range(days): # 模拟按时间搜索实际可能需要更复杂的逻辑 search_results self.client.get_note_by_keyword( keywordkeyword, page1, sortgeneral, note_type0 ) date_str (datetime.now() - timedelta(daysday_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, keylambda 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编程, days5) 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_retries3, retry_delay1): 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_retries3) 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_okTrue) def _get_cache_key(self, func_name: str, *args, **kwargs) - str: 生成缓存键 params_str json.dumps({ args: args, kwargs: kwargs }, sort_keysTrue, ensure_asciiFalse) 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, encodingutf-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, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent2) 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. 性能优化配置表优化策略实施方法预期效果请求合并批量获取多个笔记数据减少30%请求次数数据缓存本地缓存已获取数据降低50%重复请求并发控制合理设置并发线程数提升2-3倍采集速度错误恢复实现断点续采机制避免重复工作智能延迟动态调整请求间隔平衡速度与稳定性常见问题快速解答Q1: 如何获取有效的 CookieA: 登录小红书网页版后按 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_diroutput): self.output_dir Path(output_dir) self.output_dir.mkdir(exist_okTrue) def export_to_json(self, data, filename): 导出为JSON格式 filepath self.output_dir / f{filename}.json with open(filepath, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent2) 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, indexFalse, encodingutf-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_notes50): 分析用户互动数据 notes_data [] page 1 while len(notes_data) num_notes: notes self.client.get_user_notes(user_id, pagepage) 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(lower1) # 生成可视化报告 fig, axes plt.subplots(2, 2, figsize(12, 10)) # 点赞分布 axes[0, 0].hist(df[likes], bins20, edgecolorblack) axes[0, 0].set_title(点赞数分布) axes[0, 0].set_xlabel(点赞数) axes[0, 0].set_ylabel(频次) # 发布时间分析 df[publish_hour] pd.to_datetime(df[time], unitms).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], alpha0.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, dpi300, bbox_inchestight) return df, fig2. 构建监控告警系统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, threshold1000): 监控关键词趋势 current_data self.client.get_note_by_keyword(keyword, page1) 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, page1) 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编程, threshold50) # 监控竞品每天 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 库作为一个功能完善的小红书数据采集工具为开发者和数据分析师提供了强大的数据获取能力。通过本文的介绍你应该已经掌握了基础使用如何安装、配置和使用 xhs 库进行数据采集核心功能笔记获取、用户信息、内容搜索、分类浏览等核心功能实战应用竞品分析、趋势追踪、质量评估等实际业务场景最佳实践错误处理、性能优化、数据缓存等工程化实践生态集成如何将 xhs 集成到现有的数据分析工作流中未来发展方向随着小红书平台的不断发展和数据需求的增加xhs 库可以考虑以下方向的扩展异步支持添加异步请求支持提升大规模数据采集效率数据管道集成到数据管道工具如 Apache Airflow、Prefect实时监控支持实时数据流处理和监控告警机器学习集成与机器学习框架结合进行内容分类、情感分析等可视化仪表板提供内置的数据可视化界面合规使用建议最后再次强调使用 xhs 库进行数据采集时请务必 遵守小红书平台的使用条款和服务协议 尊重用户隐私不采集非公开数据⚖️ 合理控制请求频率不对平台服务器造成压力 仅将数据用于合法的研究和分析目的️ 妥善保管获取的数据遵守相关数据保护法规通过合理、合规地使用 xhs 库你可以高效地获取小红书平台的公开数据为你的数据分析、市场研究或内容运营工作提供有力支持。项目的完整文档和示例代码可以在项目目录中查看包括 xhs/core.py 的核心实现和 example/ 目录下的丰富示例。记住技术是工具如何使用它才是关键。希望 xhs 库能成为你在小红书数据分析道路上的得力助手【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考