小红书数据采集终极指南Python爬虫库xhs完全手册【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs你是否正在寻找一种高效、稳定的小红书数据采集解决方案xhs Python库正是你需要的工具这个基于小红书Web端API封装的爬虫库为开发者提供了完整的公开数据获取能力。无论你是数据分析师、市场研究员还是内容创作者都能通过xhs轻松获取小红书平台上的笔记、用户、搜索等关键数据助力你的业务决策和内容分析。 技术原理解密xhs如何绕过小红书安全机制核心架构解析xhs的核心功能集中在xhs/core.py模块中它通过精心设计的请求封装机制模拟真实用户行为访问小红书Web端API。这个库的最大亮点在于其智能签名系统——通过动态生成请求签名有效避免了被平台检测和限制。关键工作机制Cookie认证管理自动处理用户会话状态请求签名生成动态计算X-Sec-Token等安全参数智能重试机制应对网络波动和临时限制数据解析引擎高效提取结构化信息安全策略深度剖析xhs采用分层安全策略确保采集过程的稳定性请求伪装层完全模拟浏览器行为签名计算层实时生成合法签名频率控制层智能控制请求间隔异常处理层自动识别和处理各种错误状态⚡ 快速上手攻略5分钟开启数据采集之旅环境配置一步到位安装xhs库只需要一条简单的命令pip install xhs或者从源码安装最新版本git clone https://gitcode.com/gh_mirrors/xh/xhs cd xhs python setup.py install基础配置三步曲获取必要凭证从浏览器中提取小红书Cookie信息初始化客户端创建XhsClient实例发起首次请求验证配置是否正确from xhs import XhsClient # 最简单的初始化方式 client XhsClient(cookie你的cookie信息) # 验证连接 try: user_info client.get_user_info(user_idsample_user) print(连接成功) except Exception as e: print(f连接失败{e})签名服务配置高级对于需要更高稳定性的场景xhs提供了独立的签名服务方案。参考example/basic_sign_server.py和example/basic_sign_usage.py配置你的签名服务端和客户端。 实战场景应用从数据采集到商业洞察场景一竞品监控与分析假设你是一家美妆品牌的营销经理需要监控竞争对手在小红书上的推广策略# 搜索竞品相关笔记 competitor_notes client.search_note( keyword竞品品牌名, sort_typehot, page1, page_size50 ) # 分析笔记特征 for note in competitor_notes[items]: print(f标题{note[title]}) print(f点赞数{note[likes_count]}) print(f收藏数{note[collect_count]}) print(f发布时间{note[time]}) print(- * 40)场景二内容趋势发现内容创作者可以通过xhs发现平台热点趋势# 获取热门话题 hot_topics client.get_hot_search() # 分析趋势变化 for topic in hot_topics: trend 上升 if topic[heat_increase] 0 else 下降 print(f话题{topic[word]} | 热度{topic[hot_value]} | 趋势{trend})场景三用户行为研究研究人员可以分析用户互动模式# 获取用户历史笔记 user_notes client.get_user_notes( user_id目标用户ID, cursor0, page_size30 ) # 分析内容偏好 content_types {} for note in user_notes[notes]: note_type note[type] content_types[note_type] content_types.get(note_type, 0) 1 print(用户内容类型分布, content_types) 进阶技巧分享提升采集效率与稳定性并发处理优化大规模数据采集时合理使用并发可以显著提升效率import concurrent.futures from xhs.exception import DataFetchError def batch_collect_users(user_ids, max_workers5): 批量获取用户信息 results [] with concurrent.futures.ThreadPoolExecutor(max_workersmax_workers) as executor: future_to_user { executor.submit(client.get_user_info, uid): uid for uid in user_ids } for future in concurrent.futures.as_completed(future_to_user): user_id future_to_user[future] try: user_data future.result() results.append(user_data) except DataFetchError as e: print(f用户 {user_id} 数据获取失败{e}) return results智能缓存策略减少重复请求保护API配额import json import hashlib from datetime import datetime, timedelta class SmartCache: def __init__(self, cache_dir.xhs_cache, ttl_hours6): self.cache_dir cache_dir self.ttl timedelta(hoursttl_hours) def get_cache_key(self, func_name, **kwargs): 生成缓存键 key_str f{func_name}_{json.dumps(kwargs, sort_keysTrue)} return hashlib.md5(key_str.encode()).hexdigest() def get(self, key): 获取缓存数据 cache_file f{self.cache_dir}/{key}.json if os.path.exists(cache_file): with open(cache_file, r) as f: data json.load(f) cache_time datetime.fromisoformat(data[timestamp]) if datetime.now() - cache_time self.ttl: return data[data] return None def set(self, key, data): 设置缓存数据 os.makedirs(self.cache_dir, exist_okTrue) cache_file f{self.cache_dir}/{key}.json cache_data { timestamp: datetime.now().isoformat(), data: data } with open(cache_file, w) as f: json.dump(cache_data, f, indent2)错误处理最佳实践import time import random from xhs.exception import IPBlockError, SignatureError class ResilientClient: def __init__(self, base_client, max_retries3): self.client base_client self.max_retries max_retries def safe_call(self, func, *args, **kwargs): 带重试机制的API调用 for attempt in range(self.max_retries): try: return func(*args, **kwargs) except IPBlockError: print(⚠️ IP被限制建议更换IP或等待) break except SignatureError: print( 签名错误检查签名服务配置) break except Exception as e: print(f❌ 第{attempt1}次尝试失败{e}) if attempt self.max_retries - 1: wait random.uniform(2, 5) * (attempt 1) print(f⏳ 等待{wait:.1f}秒后重试...) time.sleep(wait) return None 问题诊断手册常见问题与解决方案Q1: 获取不到数据或返回空结果可能原因Cookie已过期或无效签名服务未正确运行请求频率过高被限制目标内容已删除或设为私密解决方案检查Cookie有效性重新获取验证签名服务配置参考example/basic_sign_server.py降低请求频率增加随机延迟使用client.get_note_by_id()验证单个笔记是否可访问Q2: 签名服务部署问题部署步骤确保Node.js环境已安装下载stealth.min.js到正确位置启动签名服务python basic_sign_server.py在客户端配置签名服务地址验证方法# 测试签名服务 from xhs import XhsClient client XhsClient( cookie你的cookie, sign_urlhttp://localhost:5000/sign ) # 简单请求测试 test_result client.get_user_info(test_user) if test_result: print(✅ 签名服务运行正常) else: print(❌ 签名服务异常)Q3: 性能优化建议采集效率提升使用连接池管理HTTP连接实现请求批处理减少网络开销合理设置并发数量建议3-5个线程启用GZIP压缩减少数据传输量内存管理及时清理不再使用的数据对象使用生成器处理大量数据分批保存结果避免内存溢出Q4: 数据解析异常处理当遇到数据格式变化时def safe_parse_note(note_data): 安全解析笔记数据 try: # 尝试多种可能的字段名 title note_data.get(title) or note_data.get(note_title) or likes note_data.get(likes_count) or note_data.get(like_count) or 0 return { title: title, likes: likes, user: note_data.get(user, {}).get(nickname, 未知用户) } except Exception as e: print(f数据解析异常{e}) print(f原始数据{note_data}) return None 生态整合方案xhs与其他工具的无缝对接与数据分析工具集成Pandas数据处理import pandas as pd from xhs import XhsClient # 采集数据并转为DataFrame client XhsClient(cookieyour_cookie) notes_data client.search_note(keyword美食, page1, page_size100) # 创建DataFrame进行分析 df pd.DataFrame(notes_data[items]) print(df[[title, likes_count, collect_count]].describe()) # 保存为CSV df.to_csv(xiaohongshu_notes.csv, indexFalse, encodingutf-8-sig)数据库存储方案import sqlite3 from datetime import datetime def save_to_database(notes_data, db_pathxhs_data.db): 保存数据到SQLite数据库 conn sqlite3.connect(db_path) cursor conn.cursor() # 创建表 cursor.execute( CREATE TABLE IF NOT EXISTS notes ( id TEXT PRIMARY KEY, title TEXT, likes INTEGER, collects INTEGER, user_id TEXT, created_at TIMESTAMP, collected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) # 插入数据 for note in notes_data: cursor.execute( INSERT OR REPLACE INTO notes (id, title, likes, collects, user_id, created_at) VALUES (?, ?, ?, ?, ?, ?) , ( note[id], note[title], note[likes_count], note[collect_count], note[user][user_id], datetime.fromtimestamp(note[time]) )) conn.commit() conn.close()与可视化工具结合使用Matplotlib生成图表import matplotlib.pyplot as plt import numpy as np def visualize_note_metrics(notes_data): 可视化笔记指标 likes [n[likes_count] for n in notes_data] collects [n[collect_count] for n in notes_data] fig, (ax1, ax2) plt.subplots(1, 2, figsize(12, 5)) # 点赞数分布 ax1.hist(likes, bins20, alpha0.7, colorskyblue) ax1.set_xlabel(点赞数) ax1.set_ylabel(笔记数量) ax1.set_title(点赞数分布) # 收藏数分布 ax2.hist(collects, bins20, alpha0.7, colorlightcoral) ax2.set_xlabel(收藏数) ax2.set_ylabel(笔记数量) ax2.set_title(收藏数分布) plt.tight_layout() plt.savefig(note_metrics.png, dpi150) plt.show()与自动化工作流集成使用Airflow调度定期采集# airflow_dag.py from airflow import DAG from airflow.operators.python_operator import PythonOperator from datetime import datetime, timedelta from xhs import XhsClient def collect_xhs_data(**context): 采集小红书数据任务 client XhsClient(cookiecontext[params][cookie]) # 执行采集逻辑 keywords [美妆, 穿搭, 美食, 旅行] all_data [] for keyword in keywords: data client.search_note(keywordkeyword, page1, page_size50) all_data.extend(data[items]) # 保存结果 save_to_database(all_data) return len(all_data) # 定义DAG default_args { owner: data_team, depends_on_past: False, start_date: datetime(2024, 1, 1), retries: 3, retry_delay: timedelta(minutes5) } dag DAG( xhs_data_pipeline, default_argsdefault_args, description小红书数据定期采集管道, schedule_interval0 2 * * *, # 每天凌晨2点执行 catchupFalse ) collect_task PythonOperator( task_idcollect_xhs_data, python_callablecollect_xhs_data, params{cookie: your_cookie_here}, dagdag ) 最佳实践总结采集策略建议分时段采集避免在高峰时段集中请求关键词轮换使用不同的搜索关键词组合增量更新只采集新增或更新的内容数据验证定期检查数据质量和完整性合规使用指南仅采集公开可访问的数据尊重用户隐私和版权遵守小红书平台的使用条款合理控制请求频率避免对服务器造成压力持续学习资源官方文档docs/source/xhs.rst示例代码example/核心源码xhs/core.py测试用例tests/test_xhs.py通过本指南你已经掌握了xhs Python库的核心功能和使用技巧。记住技术是工具合理使用才能发挥最大价值。开始你的小红书数据采集之旅吧让数据驱动你的决策提示建议定期查看项目的更新日志CHANGELOG.md了解最新的功能改进和注意事项。遇到问题时可以参考tests/目录下的测试用例这些是官方验证过的使用示例。【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
小红书数据采集终极指南：Python爬虫库xhs完全手册
小红书数据采集终极指南Python爬虫库xhs完全手册【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs你是否正在寻找一种高效、稳定的小红书数据采集解决方案xhs Python库正是你需要的工具这个基于小红书Web端API封装的爬虫库为开发者提供了完整的公开数据获取能力。无论你是数据分析师、市场研究员还是内容创作者都能通过xhs轻松获取小红书平台上的笔记、用户、搜索等关键数据助力你的业务决策和内容分析。 技术原理解密xhs如何绕过小红书安全机制核心架构解析xhs的核心功能集中在xhs/core.py模块中它通过精心设计的请求封装机制模拟真实用户行为访问小红书Web端API。这个库的最大亮点在于其智能签名系统——通过动态生成请求签名有效避免了被平台检测和限制。关键工作机制Cookie认证管理自动处理用户会话状态请求签名生成动态计算X-Sec-Token等安全参数智能重试机制应对网络波动和临时限制数据解析引擎高效提取结构化信息安全策略深度剖析xhs采用分层安全策略确保采集过程的稳定性请求伪装层完全模拟浏览器行为签名计算层实时生成合法签名频率控制层智能控制请求间隔异常处理层自动识别和处理各种错误状态⚡ 快速上手攻略5分钟开启数据采集之旅环境配置一步到位安装xhs库只需要一条简单的命令pip install xhs或者从源码安装最新版本git clone https://gitcode.com/gh_mirrors/xh/xhs cd xhs python setup.py install基础配置三步曲获取必要凭证从浏览器中提取小红书Cookie信息初始化客户端创建XhsClient实例发起首次请求验证配置是否正确from xhs import XhsClient # 最简单的初始化方式 client XhsClient(cookie你的cookie信息) # 验证连接 try: user_info client.get_user_info(user_idsample_user) print(连接成功) except Exception as e: print(f连接失败{e})签名服务配置高级对于需要更高稳定性的场景xhs提供了独立的签名服务方案。参考example/basic_sign_server.py和example/basic_sign_usage.py配置你的签名服务端和客户端。 实战场景应用从数据采集到商业洞察场景一竞品监控与分析假设你是一家美妆品牌的营销经理需要监控竞争对手在小红书上的推广策略# 搜索竞品相关笔记 competitor_notes client.search_note( keyword竞品品牌名, sort_typehot, page1, page_size50 ) # 分析笔记特征 for note in competitor_notes[items]: print(f标题{note[title]}) print(f点赞数{note[likes_count]}) print(f收藏数{note[collect_count]}) print(f发布时间{note[time]}) print(- * 40)场景二内容趋势发现内容创作者可以通过xhs发现平台热点趋势# 获取热门话题 hot_topics client.get_hot_search() # 分析趋势变化 for topic in hot_topics: trend 上升 if topic[heat_increase] 0 else 下降 print(f话题{topic[word]} | 热度{topic[hot_value]} | 趋势{trend})场景三用户行为研究研究人员可以分析用户互动模式# 获取用户历史笔记 user_notes client.get_user_notes( user_id目标用户ID, cursor0, page_size30 ) # 分析内容偏好 content_types {} for note in user_notes[notes]: note_type note[type] content_types[note_type] content_types.get(note_type, 0) 1 print(用户内容类型分布, content_types) 进阶技巧分享提升采集效率与稳定性并发处理优化大规模数据采集时合理使用并发可以显著提升效率import concurrent.futures from xhs.exception import DataFetchError def batch_collect_users(user_ids, max_workers5): 批量获取用户信息 results [] with concurrent.futures.ThreadPoolExecutor(max_workersmax_workers) as executor: future_to_user { executor.submit(client.get_user_info, uid): uid for uid in user_ids } for future in concurrent.futures.as_completed(future_to_user): user_id future_to_user[future] try: user_data future.result() results.append(user_data) except DataFetchError as e: print(f用户 {user_id} 数据获取失败{e}) return results智能缓存策略减少重复请求保护API配额import json import hashlib from datetime import datetime, timedelta class SmartCache: def __init__(self, cache_dir.xhs_cache, ttl_hours6): self.cache_dir cache_dir self.ttl timedelta(hoursttl_hours) def get_cache_key(self, func_name, **kwargs): 生成缓存键 key_str f{func_name}_{json.dumps(kwargs, sort_keysTrue)} return hashlib.md5(key_str.encode()).hexdigest() def get(self, key): 获取缓存数据 cache_file f{self.cache_dir}/{key}.json if os.path.exists(cache_file): with open(cache_file, r) as f: data json.load(f) cache_time datetime.fromisoformat(data[timestamp]) if datetime.now() - cache_time self.ttl: return data[data] return None def set(self, key, data): 设置缓存数据 os.makedirs(self.cache_dir, exist_okTrue) cache_file f{self.cache_dir}/{key}.json cache_data { timestamp: datetime.now().isoformat(), data: data } with open(cache_file, w) as f: json.dump(cache_data, f, indent2)错误处理最佳实践import time import random from xhs.exception import IPBlockError, SignatureError class ResilientClient: def __init__(self, base_client, max_retries3): self.client base_client self.max_retries max_retries def safe_call(self, func, *args, **kwargs): 带重试机制的API调用 for attempt in range(self.max_retries): try: return func(*args, **kwargs) except IPBlockError: print(⚠️ IP被限制建议更换IP或等待) break except SignatureError: print( 签名错误检查签名服务配置) break except Exception as e: print(f❌ 第{attempt1}次尝试失败{e}) if attempt self.max_retries - 1: wait random.uniform(2, 5) * (attempt 1) print(f⏳ 等待{wait:.1f}秒后重试...) time.sleep(wait) return None 问题诊断手册常见问题与解决方案Q1: 获取不到数据或返回空结果可能原因Cookie已过期或无效签名服务未正确运行请求频率过高被限制目标内容已删除或设为私密解决方案检查Cookie有效性重新获取验证签名服务配置参考example/basic_sign_server.py降低请求频率增加随机延迟使用client.get_note_by_id()验证单个笔记是否可访问Q2: 签名服务部署问题部署步骤确保Node.js环境已安装下载stealth.min.js到正确位置启动签名服务python basic_sign_server.py在客户端配置签名服务地址验证方法# 测试签名服务 from xhs import XhsClient client XhsClient( cookie你的cookie, sign_urlhttp://localhost:5000/sign ) # 简单请求测试 test_result client.get_user_info(test_user) if test_result: print(✅ 签名服务运行正常) else: print(❌ 签名服务异常)Q3: 性能优化建议采集效率提升使用连接池管理HTTP连接实现请求批处理减少网络开销合理设置并发数量建议3-5个线程启用GZIP压缩减少数据传输量内存管理及时清理不再使用的数据对象使用生成器处理大量数据分批保存结果避免内存溢出Q4: 数据解析异常处理当遇到数据格式变化时def safe_parse_note(note_data): 安全解析笔记数据 try: # 尝试多种可能的字段名 title note_data.get(title) or note_data.get(note_title) or likes note_data.get(likes_count) or note_data.get(like_count) or 0 return { title: title, likes: likes, user: note_data.get(user, {}).get(nickname, 未知用户) } except Exception as e: print(f数据解析异常{e}) print(f原始数据{note_data}) return None 生态整合方案xhs与其他工具的无缝对接与数据分析工具集成Pandas数据处理import pandas as pd from xhs import XhsClient # 采集数据并转为DataFrame client XhsClient(cookieyour_cookie) notes_data client.search_note(keyword美食, page1, page_size100) # 创建DataFrame进行分析 df pd.DataFrame(notes_data[items]) print(df[[title, likes_count, collect_count]].describe()) # 保存为CSV df.to_csv(xiaohongshu_notes.csv, indexFalse, encodingutf-8-sig)数据库存储方案import sqlite3 from datetime import datetime def save_to_database(notes_data, db_pathxhs_data.db): 保存数据到SQLite数据库 conn sqlite3.connect(db_path) cursor conn.cursor() # 创建表 cursor.execute( CREATE TABLE IF NOT EXISTS notes ( id TEXT PRIMARY KEY, title TEXT, likes INTEGER, collects INTEGER, user_id TEXT, created_at TIMESTAMP, collected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) # 插入数据 for note in notes_data: cursor.execute( INSERT OR REPLACE INTO notes (id, title, likes, collects, user_id, created_at) VALUES (?, ?, ?, ?, ?, ?) , ( note[id], note[title], note[likes_count], note[collect_count], note[user][user_id], datetime.fromtimestamp(note[time]) )) conn.commit() conn.close()与可视化工具结合使用Matplotlib生成图表import matplotlib.pyplot as plt import numpy as np def visualize_note_metrics(notes_data): 可视化笔记指标 likes [n[likes_count] for n in notes_data] collects [n[collect_count] for n in notes_data] fig, (ax1, ax2) plt.subplots(1, 2, figsize(12, 5)) # 点赞数分布 ax1.hist(likes, bins20, alpha0.7, colorskyblue) ax1.set_xlabel(点赞数) ax1.set_ylabel(笔记数量) ax1.set_title(点赞数分布) # 收藏数分布 ax2.hist(collects, bins20, alpha0.7, colorlightcoral) ax2.set_xlabel(收藏数) ax2.set_ylabel(笔记数量) ax2.set_title(收藏数分布) plt.tight_layout() plt.savefig(note_metrics.png, dpi150) plt.show()与自动化工作流集成使用Airflow调度定期采集# airflow_dag.py from airflow import DAG from airflow.operators.python_operator import PythonOperator from datetime import datetime, timedelta from xhs import XhsClient def collect_xhs_data(**context): 采集小红书数据任务 client XhsClient(cookiecontext[params][cookie]) # 执行采集逻辑 keywords [美妆, 穿搭, 美食, 旅行] all_data [] for keyword in keywords: data client.search_note(keywordkeyword, page1, page_size50) all_data.extend(data[items]) # 保存结果 save_to_database(all_data) return len(all_data) # 定义DAG default_args { owner: data_team, depends_on_past: False, start_date: datetime(2024, 1, 1), retries: 3, retry_delay: timedelta(minutes5) } dag DAG( xhs_data_pipeline, default_argsdefault_args, description小红书数据定期采集管道, schedule_interval0 2 * * *, # 每天凌晨2点执行 catchupFalse ) collect_task PythonOperator( task_idcollect_xhs_data, python_callablecollect_xhs_data, params{cookie: your_cookie_here}, dagdag ) 最佳实践总结采集策略建议分时段采集避免在高峰时段集中请求关键词轮换使用不同的搜索关键词组合增量更新只采集新增或更新的内容数据验证定期检查数据质量和完整性合规使用指南仅采集公开可访问的数据尊重用户隐私和版权遵守小红书平台的使用条款合理控制请求频率避免对服务器造成压力持续学习资源官方文档docs/source/xhs.rst示例代码example/核心源码xhs/core.py测试用例tests/test_xhs.py通过本指南你已经掌握了xhs Python库的核心功能和使用技巧。记住技术是工具合理使用才能发挥最大价值。开始你的小红书数据采集之旅吧让数据驱动你的决策提示建议定期查看项目的更新日志CHANGELOG.md了解最新的功能改进和注意事项。遇到问题时可以参考tests/目录下的测试用例这些是官方验证过的使用示例。【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
