Python小红书数据采集实战指南xhs库高效方案深度解析【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs小红书作为中国领先的社交电商平台每天产生海量用户生成内容为市场研究、竞品分析和内容创作提供了宝贵的数据资源。xhs库作为专业的Python小红书数据采集工具通过封装小红书Web端API为开发者和数据分析师提供了高效、合规的数据获取方案。本文将深度解析xhs库的核心功能与实战应用帮助你快速掌握小红书数据采集技术。 为什么选择xhs库进行小红书数据采集在众多数据采集方案中xhs库以其独特优势脱颖而出技术优势对比开发效率相比传统爬虫开发xhs库提供完整的API封装大幅降低开发门槛维护成本官方API接口相对稳定减少因网站改版导致的维护工作量合规性保障基于官方接口的数据采集更加符合平台规则功能完整性覆盖搜索、用户、互动等小红书核心功能适用场景市场分析趋势监控、竞品研究、用户行为分析内容创作热点发现、内容策略制定、效果评估学术研究社交媒体分析、用户画像构建商业智能品牌监测、口碑分析、营销效果评估 环境搭建与快速开始安装部署方案基础安装pip install xhs源码安装获取最新特性git clone https://gitcode.com/gh_mirrors/xh/xhs cd xhs pip install -e .Docker快速部署docker run -d -p 5005:5005 reajason/xhs-api:latest核心配置要点Cookie获取指南登录小红书Web版www.xiaohongshu.com按F12打开开发者工具进入Network标签页并刷新页面查找任意请求复制Request Headers中的Cookie值基础使用示例from xhs import XhsClient import json # 初始化客户端 client XhsClient(cookieyour_cookie_here) # 搜索热门笔记 hot_notes client.get_note_by_keyword( keyword美妆教程, page1, page_size20, sortgeneral ) # 数据解析处理 for note in hot_notes[items]: print(f标题: {note[title]}) print(f作者: {note[user][nickname]}) print(f互动数据 - 点赞: {note[like_count]}, 收藏: {note[collect_count]}) 核心功能深度解析内容搜索与分析模块智能搜索功能# 多维度搜索配置 search_params { keyword: 健身教程, page: 1, page_size: 30, sort: hot, # 支持hot、time、general等排序方式 note_type: all # 支持all、video、normal筛选 } # 执行搜索 results client.get_note_by_keyword(**search_params) # 搜索结果分析 def analyze_search_results(results): total_notes len(results[items]) avg_likes sum(n[like_count] for n in results[items]) / total_notes video_ratio sum(1 for n in results[items] if n[type] video) / total_notes return { total_results: total_notes, average_likes: avg_likes, video_content_ratio: f{video_ratio:.1%} }分类内容获取from xhs import FeedType # 获取不同分类的首页推荐 feed_types { 穿搭推荐: FeedType.FASION, 美食分享: FeedType.FOOD, 美妆教程: FeedType.COSMETICS, 旅行攻略: FeedType.TRAVEL } for category_name, feed_type in feed_types.items(): feed_data client.get_home_feed(feed_type) print(f{category_name}: 获取到{len(feed_data[items])}条内容)用户数据采集系统用户信息获取def get_user_insights(user_id): 获取用户深度洞察数据 # 基础信息 user_info client.get_user_info(user_id) # 笔记列表 user_notes client.get_user_notes(user_id) # 收藏列表 collected_notes client.get_user_collect_notes(user_id) # 点赞列表 liked_notes client.get_user_like_notes(user_id) return { user_profile: user_info, content_metrics: { total_notes: len(user_notes[items]), total_collections: len(collected_notes[items]), total_likes: len(liked_notes[items]) }, content_analysis: analyze_user_content_pattern(user_notes[items]) }用户行为分析def analyze_user_engagement(user_id, time_period7d): 分析用户互动行为模式 notes client.get_user_all_notes(user_id) engagement_data [] for note in notes: engagement_rate (note[like_count] note[collect_count] note[comment_count]) / note[view_count] engagement_data.append({ note_id: note[id], engagement_rate: engagement_rate, publish_time: note[time], content_type: note[type] }) return calculate_engagement_trends(engagement_data)互动功能实现方案评论管理系统class CommentManager: def __init__(self, client): self.client client def get_note_comments_analysis(self, note_id, xsec_token): 获取笔记评论并进行情感分析 comments client.get_note_all_comments(note_id, xsec_tokenxsec_token) analysis_results { total_comments: len(comments), positive_count: 0, neutral_count: 0, negative_count: 0, top_commentators: [], comment_timeline: [] } # 评论情感分析逻辑 for comment in comments: sentiment analyze_comment_sentiment(comment[content]) analysis_results[f{sentiment}_count] 1 return analysis_results def post_comment_with_strategy(self, note_id, content_template, strategyengagement): 根据策略发布评论 optimized_content optimize_comment_content(content_template, strategy) return client.comment_note(note_id, optimized_content)关注与互动管理def manage_user_relationships(target_user_id, strategygradual): 智能用户关系管理 user_info client.get_user_info(target_user_id) if should_follow_user(user_info): client.follow_user(target_user_id) print(f已关注用户: {user_info[nickname]}) # 获取用户最新笔记 latest_notes client.get_user_notes(target_user_id, cursor) for note in latest_notes[items][:3]: # 只处理最近3篇 if should_interact_with_note(note): client.like_note(note[id]) client.collect_note(note[id]) print(f已互动笔记: {note[title]})️ 高级功能与稳定性保障签名服务部署方案本地签名服务实现# example/basic_sign_server.py 核心逻辑 def setup_signature_service(): 配置签名服务确保请求稳定性 # 浏览器自动化签名 from playwright.sync_api import sync_playwright def sign_request(uri, dataNone, a1, web_session): with sync_playwright() as playwright: browser playwright.chromium.launch(headlessTrue) context browser.new_context() page context.new_page() # 加载小红书页面 page.goto(https://www.xiaohongshu.com) # 设置cookie context.add_cookies([ {name: a1, value: a1, domain: .xiaohongshu.com, path: /} ]) page.reload() time.sleep(1) # 等待页面加载 # 执行签名 encrypt_params page.evaluate( ([url, data]) window._webmsxyw(url, data), [uri, data] ) browser.close() return { x-s: encrypt_params[X-s], x-t: str(encrypt_params[X-t]) } return sign_request # 使用签名服务 sign_func setup_signature_service() client XhsClient(cookieyour_cookie, signsign_func)分布式签名服务架构class DistributedSignService: 分布式签名服务管理 def __init__(self, service_urls): self.service_urls service_urls self.current_index 0 def get_signature(self, uri, dataNone): 轮询获取签名 for _ in range(len(self.service_urls)): try: service_url self.service_urls[self.current_index] response requests.post( f{service_url}/sign, json{uri: uri, data: data}, timeout5 ) self.current_index (self.current_index 1) % len(self.service_urls) return response.json() except: self.current_index (self.current_index 1) % len(self.service_urls) raise Exception(所有签名服务均不可用)错误处理与重试机制智能重试策略import time import random from functools import wraps def retry_with_exponential_backoff( max_retries5, initial_delay1, exponential_base2, jitterTrue ): 指数退避重试装饰器 def decorator(func): wraps(func) def wrapper(*args, **kwargs): delay initial_delay for attempt in range(max_retries): try: return func(*args, **kwargs) except Exception as e: if attempt max_retries - 1: raise # 计算延迟时间 delay * exponential_base ** attempt if jitter: # 添加随机抖动避免同步重试 delay random.uniform(0, 0.1 * delay) time.sleep(delay) return None return wrapper return decorator retry_with_exponential_backoff(max_retries3) def safe_api_call(api_method, *args, **kwargs): 安全的API调用封装 return api_method(*args, **kwargs)请求频率控制class RateLimiter: 智能请求频率控制器 def __init__(self, requests_per_minute60): self.requests_per_minute requests_per_minute self.request_times [] def wait_if_needed(self): 根据历史请求频率决定是否需要等待 current_time time.time() # 清理一分钟前的记录 self.request_times [ t for t in self.request_times if current_time - t 60 ] if len(self.request_times) self.requests_per_minute: # 计算需要等待的时间 oldest_request self.request_times[0] wait_time 60 - (current_time - oldest_request) if wait_time 0: time.sleep(wait_time random.uniform(0.5, 1.5)) self.request_times.append(current_time) # 使用示例 limiter RateLimiter(requests_per_minute50) def make_limited_request(client, method, *args, **kwargs): limiter.wait_if_needed() return method(*args, **kwargs) 数据存储与分析方案结构化数据存储SQLite数据库设计import sqlite3 from datetime import datetime import json class XhsDataStorage: 小红书数据存储管理器 def __init__(self, db_pathxhs_data.db): self.conn sqlite3.connect(db_path) self.create_tables() def create_tables(self): 创建数据表结构 cursor self.conn.cursor() # 用户表 cursor.execute( CREATE TABLE IF NOT EXISTS users ( user_id TEXT PRIMARY KEY, nickname TEXT, avatar TEXT, gender INTEGER, location TEXT, red_id TEXT, description TEXT, ip_location TEXT, college TEXT, tags TEXT, like_count INTEGER, collect_count INTEGER, fans INTEGER, follows INTEGER, collected_at TIMESTAMP ) ) # 笔记表 cursor.execute( CREATE TABLE IF NOT EXISTS notes ( note_id TEXT PRIMARY KEY, user_id TEXT, title TEXT, desc TEXT, type TEXT, like_count INTEGER, collect_count INTEGER, comment_count INTEGER, share_count INTEGER, time TIMESTAMP, last_update_time TIMESTAMP, image_list TEXT, video_info TEXT, tag_list TEXT, at_user_list TEXT, collected_at TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users (user_id) ) ) # 互动记录表 cursor.execute( CREATE TABLE IF NOT EXISTS interactions ( id INTEGER PRIMARY KEY AUTOINCREMENT, note_id TEXT, user_id TEXT, interaction_type TEXT, interaction_time TIMESTAMP, content TEXT, FOREIGN KEY (note_id) REFERENCES notes (note_id), FOREIGN KEY (user_id) REFERENCES users (user_id) ) ) self.conn.commit() def save_note_data(self, note_data): 保存笔记数据 cursor self.conn.cursor() # 保存用户信息 user_info note_data.get(user, {}) cursor.execute( INSERT OR REPLACE INTO users VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( user_info.get(user_id), user_info.get(nickname), user_info.get(avatar), user_info.get(gender), user_info.get(location), user_info.get(red_id), user_info.get(desc), user_info.get(ip_location), user_info.get(college), json.dumps(user_info.get(tags, []), ensure_asciiFalse), user_info.get(likes), user_info.get(collects), user_info.get(fans), user_info.get(follows), datetime.now() )) # 保存笔记信息 cursor.execute( INSERT OR REPLACE INTO notes VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( note_data.get(id), user_info.get(user_id), note_data.get(title), note_data.get(desc), note_data.get(type), note_data.get(like_count, 0), note_data.get(collect_count, 0), note_data.get(comment_count, 0), note_data.get(share_count, 0), note_data.get(time), note_data.get(last_update_time), json.dumps(note_data.get(image_list, []), ensure_asciiFalse), json.dumps(note_data.get(video, {}), ensure_asciiFalse), json.dumps(note_data.get(tag_list, []), ensure_asciiFalse), json.dumps(note_data.get(at_user_list, []), ensure_asciiFalse), datetime.now() )) self.conn.commit()数据分析与可视化数据聚合分析import pandas as pd import matplotlib.pyplot as plt class XhsDataAnalyzer: 小红书数据分析器 def __init__(self, storage): self.storage storage def analyze_content_trends(self, time_period7d): 分析内容趋势 query SELECT DATE(time) as date, COUNT(*) as note_count, AVG(like_count) as avg_likes, AVG(collect_count) as avg_collects, AVG(comment_count) as avg_comments FROM notes WHERE time date(now, ?) GROUP BY DATE(time) ORDER BY date df pd.read_sql_query(query, self.storage.conn, params(f-{time_period},)) # 生成趋势图表 fig, axes plt.subplots(2, 2, figsize(12, 8)) axes[0, 0].plot(df[date], df[note_count], markero) axes[0, 0].set_title(每日笔记发布量) axes[0, 0].set_xlabel(日期) axes[0, 0].set_ylabel(数量) axes[0, 1].plot(df[date], df[avg_likes], markers, colororange) axes[0, 1].set_title(平均点赞数趋势) axes[0, 1].set_xlabel(日期) axes[0, 1].set_ylabel(点赞数) axes[1, 0].plot(df[date], df[avg_collects], marker^, colorgreen) axes[1, 0].set_title(平均收藏数趋势) axes[1, 0].set_xlabel(日期) axes[1, 0].set_ylabel(收藏数) axes[1, 1].plot(df[date], df[avg_comments], markerd, colorred) axes[1, 1].set_title(平均评论数趋势) axes[1, 1].set_xlabel(日期) axes[1, 1].set_ylabel(评论数) plt.tight_layout() return fig, df 实战应用场景场景一竞品监控系统class CompetitorMonitor: 竞品监控系统 def __init__(self, client, competitor_ids): self.client client self.competitor_ids competitor_ids def monitor_competitor_activity(self): 监控竞品活动 competitor_data {} for competitor_id in self.competitor_ids: # 获取竞品最新内容 latest_notes self.client.get_user_notes(competitor_id) # 分析内容策略 content_analysis self.analyze_content_strategy(latest_notes[items]) # 分析互动表现 engagement_analysis self.analyze_engagement_pattern(latest_notes[items]) competitor_data[competitor_id] { recent_content: latest_notes[items][:5], # 最近5篇 content_strategy: content_analysis, engagement_pattern: engagement_analysis, update_time: datetime.now() } return competitor_data def generate_competitor_report(self, competitor_data): 生成竞品分析报告 report { summary: { total_competitors: len(competitor_data), monitoring_period: 7天, report_date: datetime.now().strftime(%Y-%m-%d) }, detailed_analysis: {}, recommendations: [] } for competitor_id, data in competitor_data.items(): report[detailed_analysis][competitor_id] { content_frequency: len(data[recent_content]), avg_engagement_rate: self.calculate_avg_engagement(data[recent_content]), content_types: self.analyze_content_types(data[recent_content]), posting_schedule: self.analyze_posting_schedule(data[recent_content]) } # 生成策略建议 report[recommendations] self.generate_strategy_recommendations(report[detailed_analysis]) return report场景二内容创作助手class ContentCreationAssistant: 内容创作智能助手 def __init__(self, client): self.client client def find_trending_topics(self, categorybeauty): 发现热门话题 feed_type_mapping { beauty: FeedType.COSMETICS, fashion: FeedType.FASION, food: FeedType.FOOD, travel: FeedType.TRAVEL } feed_type feed_type_mapping.get(category, FeedType.RECOMMEND) feed_data self.client.get_home_feed(feed_type) # 分析热门内容特征 trending_features self.analyze_trending_features(feed_data[items]) return { trending_topics: self.extract_topics(feed_data[items]), successful_patterns: trending_features, recommended_keywords: self.generate_keyword_suggestions(feed_data[items]) } def optimize_content_strategy(self, historical_data): 基于历史数据优化内容策略 analysis_results { best_posting_times: self.analyze_optimal_posting_times(historical_data), content_type_performance: self.analyze_content_type_performance(historical_data), keyword_effectiveness: self.analyze_keyword_effectiveness(historical_data), engagement_patterns: self.identify_engagement_patterns(historical_data) } return { analysis: analysis_results, actionable_insights: self.generate_actionable_insights(analysis_results) } 安全与合规指南合规使用原则数据采集伦理尊重用户隐私仅采集公开数据不获取用户隐私信息控制采集频率避免对服务器造成过大压力遵守平台规则严格遵守小红书用户协议和服务条款合法使用数据确保数据使用符合相关法律法规技术合规措施class EthicalDataCollector: 伦理数据采集器 def __init__(self, client, config): self.client client self.config config self.request_log [] def ethical_collect(self, collection_task, max_items100): 伦理数据采集主函数 collected_data [] item_count 0 while item_count max_items: try: # 添加请求延迟 self.respect_rate_limit() # 执行采集任务 data_batch collection_task() # 过滤敏感信息 filtered_data self.filter_sensitive_info(data_batch) collected_data.extend(filtered_data) item_count len(filtered_data) # 记录采集日志 self.log_collection_activity(filtered_data) if len(filtered_data) 0: break except Exception as e: self.handle_collection_error(e) break return collected_data def respect_rate_limit(self): 遵守请求频率限制 time.sleep(self.config.get(request_interval, 2)) # 检查每日请求限额 today_requests self.count_today_requests() if today_requests self.config.get(daily_limit, 1000): raise Exception(已达到每日请求限额) def filter_sensitive_info(self, data): 过滤敏感信息 filtered_data [] for item in data: filtered_item item.copy() # 移除可能包含的个人信息 sensitive_fields [phone, email, id_card, address] for field in sensitive_fields: if field in filtered_item: del filtered_item[field] filtered_data.append(filtered_item) return filtered_data 部署与优化建议生产环境部署容器化部署方案# Dockerfile示例 FROM python:3.9-slim WORKDIR /app # 安装依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 创建非root用户 RUN useradd -m -u 1000 appuser chown -R appuser:appuser /app USER appuser # 启动应用 CMD [python, app.py]监控与告警配置class MonitoringSystem: 采集系统监控 def __init__(self): self.metrics { total_requests: 0, successful_requests: 0, failed_requests: 0, last_error: None, average_response_time: 0 } def monitor_request(self, func): 请求监控装饰器 wraps(func) def wrapper(*args, **kwargs): start_time time.time() try: result func(*args, **kwargs) self.metrics[successful_requests] 1 return result except Exception as e: self.metrics[failed_requests] 1 self.metrics[last_error] str(e) raise finally: self.metrics[total_requests] 1 response_time time.time() - start_time # 更新平均响应时间 total_time self.metrics[average_response_time] * (self.metrics[total_requests] - 1) self.metrics[average_response_time] (total_time response_time) / self.metrics[total_requests] # 检查是否需要告警 self.check_alerts() return wrapper def check_alerts(self): 检查告警条件 error_rate self.metrics[failed_requests] / max(self.metrics[total_requests], 1) if error_rate 0.1: # 错误率超过10% self.send_alert(f高错误率告警: {error_rate:.1%}) if self.metrics[average_response_time] 5: # 平均响应时间超过5秒 self.send_alert(f响应时间过长: {self.metrics[average_response_time]:.2f}秒)性能优化策略缓存机制实现import redis from functools import lru_cache class CacheManager: 数据缓存管理器 def __init__(self, redis_hostlocalhost, redis_port6379): self.redis_client redis.Redis(hostredis_host, portredis_port, decode_responsesTrue) self.local_cache {} def get_cached_data(self, cache_key, ttl3600): 获取缓存数据 # 首先检查本地缓存 if cache_key in self.local_cache: return self.local_cache[cache_key] # 检查Redis缓存 cached_data self.redis_client.get(cache_key) if cached_data: data json.loads(cached_data) self.local_cache[cache_key] data return data return None def set_cached_data(self, cache_key, data, ttl3600): 设置缓存数据 # 设置本地缓存 self.local_cache[cache_key] data # 设置Redis缓存 self.redis_client.setex( cache_key, ttl, json.dumps(data, ensure_asciiFalse) ) lru_cache(maxsize128) def get_user_info_cached(self, user_id): 带缓存的用户信息获取 cache_key fuser_info:{user_id} cached_data self.get_cached_data(cache_key) if cached_data: return cached_data # 从API获取数据 user_info self.client.get_user_info(user_id) # 缓存数据 self.set_cached_data(cache_key, user_info, ttl1800) return user_info 学习路径与资源渐进式学习计划第一阶段基础掌握学习xhs库基本安装与配置掌握Cookie获取与客户端初始化实现基础数据采集功能第二阶段进阶应用学习签名服务部署与配置掌握错误处理与重试机制实现数据存储与处理流程第三阶段高级优化学习分布式部署方案掌握性能监控与优化技巧实现生产级数据采集系统核心资源参考项目结构概览核心模块xhs/core.py - 主要API接口实现工具函数xhs/help.py - 数据处理与转换工具示例代码example/ - 各类使用场景示例测试用例tests/ - 功能测试与验证关键配置文件requirements.txt - 项目依赖包列表setup.cfg - 项目打包配置Dockerfile - 容器化部署配置 最佳实践总结技术实践要点环境隔离使用虚拟环境管理Python依赖配置管理使用环境变量管理敏感信息日志记录实现完整的日志系统便于问题排查版本控制使用Git进行代码版本管理持续集成配置自动化测试与部署流程项目管理建议需求明确明确数据采集的具体需求和目标渐进开发从简单功能开始逐步增加复杂度测试驱动为每个功能编写测试用例文档完善保持代码注释和文档的及时更新监控维护建立系统监控和定期维护机制风险控制策略合规审查定期检查数据采集的合规性备份机制实现数据备份和恢复方案应急预案制定系统故障的应急处理流程性能监控持续监控系统性能和稳定性安全审计定期进行安全漏洞扫描和修复通过本文的深度解析你应该已经掌握了使用xhs库进行小红书数据采集的核心技术和最佳实践。记住技术只是工具合理、合规地使用这些工具才能让数据真正为业务创造价值。开始你的小红书数据采集之旅用数据驱动决策创造更大的商业价值【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Python小红书数据采集实战指南：xhs库高效方案深度解析
发布时间：2026/7/5 7:16:12
Python小红书数据采集实战指南xhs库高效方案深度解析【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs小红书作为中国领先的社交电商平台每天产生海量用户生成内容为市场研究、竞品分析和内容创作提供了宝贵的数据资源。xhs库作为专业的Python小红书数据采集工具通过封装小红书Web端API为开发者和数据分析师提供了高效、合规的数据获取方案。本文将深度解析xhs库的核心功能与实战应用帮助你快速掌握小红书数据采集技术。 为什么选择xhs库进行小红书数据采集在众多数据采集方案中xhs库以其独特优势脱颖而出技术优势对比开发效率相比传统爬虫开发xhs库提供完整的API封装大幅降低开发门槛维护成本官方API接口相对稳定减少因网站改版导致的维护工作量合规性保障基于官方接口的数据采集更加符合平台规则功能完整性覆盖搜索、用户、互动等小红书核心功能适用场景市场分析趋势监控、竞品研究、用户行为分析内容创作热点发现、内容策略制定、效果评估学术研究社交媒体分析、用户画像构建商业智能品牌监测、口碑分析、营销效果评估 环境搭建与快速开始安装部署方案基础安装pip install xhs源码安装获取最新特性git clone https://gitcode.com/gh_mirrors/xh/xhs cd xhs pip install -e .Docker快速部署docker run -d -p 5005:5005 reajason/xhs-api:latest核心配置要点Cookie获取指南登录小红书Web版www.xiaohongshu.com按F12打开开发者工具进入Network标签页并刷新页面查找任意请求复制Request Headers中的Cookie值基础使用示例from xhs import XhsClient import json # 初始化客户端 client XhsClient(cookieyour_cookie_here) # 搜索热门笔记 hot_notes client.get_note_by_keyword( keyword美妆教程, page1, page_size20, sortgeneral ) # 数据解析处理 for note in hot_notes[items]: print(f标题: {note[title]}) print(f作者: {note[user][nickname]}) print(f互动数据 - 点赞: {note[like_count]}, 收藏: {note[collect_count]}) 核心功能深度解析内容搜索与分析模块智能搜索功能# 多维度搜索配置 search_params { keyword: 健身教程, page: 1, page_size: 30, sort: hot, # 支持hot、time、general等排序方式 note_type: all # 支持all、video、normal筛选 } # 执行搜索 results client.get_note_by_keyword(**search_params) # 搜索结果分析 def analyze_search_results(results): total_notes len(results[items]) avg_likes sum(n[like_count] for n in results[items]) / total_notes video_ratio sum(1 for n in results[items] if n[type] video) / total_notes return { total_results: total_notes, average_likes: avg_likes, video_content_ratio: f{video_ratio:.1%} }分类内容获取from xhs import FeedType # 获取不同分类的首页推荐 feed_types { 穿搭推荐: FeedType.FASION, 美食分享: FeedType.FOOD, 美妆教程: FeedType.COSMETICS, 旅行攻略: FeedType.TRAVEL } for category_name, feed_type in feed_types.items(): feed_data client.get_home_feed(feed_type) print(f{category_name}: 获取到{len(feed_data[items])}条内容)用户数据采集系统用户信息获取def get_user_insights(user_id): 获取用户深度洞察数据 # 基础信息 user_info client.get_user_info(user_id) # 笔记列表 user_notes client.get_user_notes(user_id) # 收藏列表 collected_notes client.get_user_collect_notes(user_id) # 点赞列表 liked_notes client.get_user_like_notes(user_id) return { user_profile: user_info, content_metrics: { total_notes: len(user_notes[items]), total_collections: len(collected_notes[items]), total_likes: len(liked_notes[items]) }, content_analysis: analyze_user_content_pattern(user_notes[items]) }用户行为分析def analyze_user_engagement(user_id, time_period7d): 分析用户互动行为模式 notes client.get_user_all_notes(user_id) engagement_data [] for note in notes: engagement_rate (note[like_count] note[collect_count] note[comment_count]) / note[view_count] engagement_data.append({ note_id: note[id], engagement_rate: engagement_rate, publish_time: note[time], content_type: note[type] }) return calculate_engagement_trends(engagement_data)互动功能实现方案评论管理系统class CommentManager: def __init__(self, client): self.client client def get_note_comments_analysis(self, note_id, xsec_token): 获取笔记评论并进行情感分析 comments client.get_note_all_comments(note_id, xsec_tokenxsec_token) analysis_results { total_comments: len(comments), positive_count: 0, neutral_count: 0, negative_count: 0, top_commentators: [], comment_timeline: [] } # 评论情感分析逻辑 for comment in comments: sentiment analyze_comment_sentiment(comment[content]) analysis_results[f{sentiment}_count] 1 return analysis_results def post_comment_with_strategy(self, note_id, content_template, strategyengagement): 根据策略发布评论 optimized_content optimize_comment_content(content_template, strategy) return client.comment_note(note_id, optimized_content)关注与互动管理def manage_user_relationships(target_user_id, strategygradual): 智能用户关系管理 user_info client.get_user_info(target_user_id) if should_follow_user(user_info): client.follow_user(target_user_id) print(f已关注用户: {user_info[nickname]}) # 获取用户最新笔记 latest_notes client.get_user_notes(target_user_id, cursor) for note in latest_notes[items][:3]: # 只处理最近3篇 if should_interact_with_note(note): client.like_note(note[id]) client.collect_note(note[id]) print(f已互动笔记: {note[title]})️ 高级功能与稳定性保障签名服务部署方案本地签名服务实现# example/basic_sign_server.py 核心逻辑 def setup_signature_service(): 配置签名服务确保请求稳定性 # 浏览器自动化签名 from playwright.sync_api import sync_playwright def sign_request(uri, dataNone, a1, web_session): with sync_playwright() as playwright: browser playwright.chromium.launch(headlessTrue) context browser.new_context() page context.new_page() # 加载小红书页面 page.goto(https://www.xiaohongshu.com) # 设置cookie context.add_cookies([ {name: a1, value: a1, domain: .xiaohongshu.com, path: /} ]) page.reload() time.sleep(1) # 等待页面加载 # 执行签名 encrypt_params page.evaluate( ([url, data]) window._webmsxyw(url, data), [uri, data] ) browser.close() return { x-s: encrypt_params[X-s], x-t: str(encrypt_params[X-t]) } return sign_request # 使用签名服务 sign_func setup_signature_service() client XhsClient(cookieyour_cookie, signsign_func)分布式签名服务架构class DistributedSignService: 分布式签名服务管理 def __init__(self, service_urls): self.service_urls service_urls self.current_index 0 def get_signature(self, uri, dataNone): 轮询获取签名 for _ in range(len(self.service_urls)): try: service_url self.service_urls[self.current_index] response requests.post( f{service_url}/sign, json{uri: uri, data: data}, timeout5 ) self.current_index (self.current_index 1) % len(self.service_urls) return response.json() except: self.current_index (self.current_index 1) % len(self.service_urls) raise Exception(所有签名服务均不可用)错误处理与重试机制智能重试策略import time import random from functools import wraps def retry_with_exponential_backoff( max_retries5, initial_delay1, exponential_base2, jitterTrue ): 指数退避重试装饰器 def decorator(func): wraps(func) def wrapper(*args, **kwargs): delay initial_delay for attempt in range(max_retries): try: return func(*args, **kwargs) except Exception as e: if attempt max_retries - 1: raise # 计算延迟时间 delay * exponential_base ** attempt if jitter: # 添加随机抖动避免同步重试 delay random.uniform(0, 0.1 * delay) time.sleep(delay) return None return wrapper return decorator retry_with_exponential_backoff(max_retries3) def safe_api_call(api_method, *args, **kwargs): 安全的API调用封装 return api_method(*args, **kwargs)请求频率控制class RateLimiter: 智能请求频率控制器 def __init__(self, requests_per_minute60): self.requests_per_minute requests_per_minute self.request_times [] def wait_if_needed(self): 根据历史请求频率决定是否需要等待 current_time time.time() # 清理一分钟前的记录 self.request_times [ t for t in self.request_times if current_time - t 60 ] if len(self.request_times) self.requests_per_minute: # 计算需要等待的时间 oldest_request self.request_times[0] wait_time 60 - (current_time - oldest_request) if wait_time 0: time.sleep(wait_time random.uniform(0.5, 1.5)) self.request_times.append(current_time) # 使用示例 limiter RateLimiter(requests_per_minute50) def make_limited_request(client, method, *args, **kwargs): limiter.wait_if_needed() return method(*args, **kwargs) 数据存储与分析方案结构化数据存储SQLite数据库设计import sqlite3 from datetime import datetime import json class XhsDataStorage: 小红书数据存储管理器 def __init__(self, db_pathxhs_data.db): self.conn sqlite3.connect(db_path) self.create_tables() def create_tables(self): 创建数据表结构 cursor self.conn.cursor() # 用户表 cursor.execute( CREATE TABLE IF NOT EXISTS users ( user_id TEXT PRIMARY KEY, nickname TEXT, avatar TEXT, gender INTEGER, location TEXT, red_id TEXT, description TEXT, ip_location TEXT, college TEXT, tags TEXT, like_count INTEGER, collect_count INTEGER, fans INTEGER, follows INTEGER, collected_at TIMESTAMP ) ) # 笔记表 cursor.execute( CREATE TABLE IF NOT EXISTS notes ( note_id TEXT PRIMARY KEY, user_id TEXT, title TEXT, desc TEXT, type TEXT, like_count INTEGER, collect_count INTEGER, comment_count INTEGER, share_count INTEGER, time TIMESTAMP, last_update_time TIMESTAMP, image_list TEXT, video_info TEXT, tag_list TEXT, at_user_list TEXT, collected_at TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users (user_id) ) ) # 互动记录表 cursor.execute( CREATE TABLE IF NOT EXISTS interactions ( id INTEGER PRIMARY KEY AUTOINCREMENT, note_id TEXT, user_id TEXT, interaction_type TEXT, interaction_time TIMESTAMP, content TEXT, FOREIGN KEY (note_id) REFERENCES notes (note_id), FOREIGN KEY (user_id) REFERENCES users (user_id) ) ) self.conn.commit() def save_note_data(self, note_data): 保存笔记数据 cursor self.conn.cursor() # 保存用户信息 user_info note_data.get(user, {}) cursor.execute( INSERT OR REPLACE INTO users VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( user_info.get(user_id), user_info.get(nickname), user_info.get(avatar), user_info.get(gender), user_info.get(location), user_info.get(red_id), user_info.get(desc), user_info.get(ip_location), user_info.get(college), json.dumps(user_info.get(tags, []), ensure_asciiFalse), user_info.get(likes), user_info.get(collects), user_info.get(fans), user_info.get(follows), datetime.now() )) # 保存笔记信息 cursor.execute( INSERT OR REPLACE INTO notes VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( note_data.get(id), user_info.get(user_id), note_data.get(title), note_data.get(desc), note_data.get(type), note_data.get(like_count, 0), note_data.get(collect_count, 0), note_data.get(comment_count, 0), note_data.get(share_count, 0), note_data.get(time), note_data.get(last_update_time), json.dumps(note_data.get(image_list, []), ensure_asciiFalse), json.dumps(note_data.get(video, {}), ensure_asciiFalse), json.dumps(note_data.get(tag_list, []), ensure_asciiFalse), json.dumps(note_data.get(at_user_list, []), ensure_asciiFalse), datetime.now() )) self.conn.commit()数据分析与可视化数据聚合分析import pandas as pd import matplotlib.pyplot as plt class XhsDataAnalyzer: 小红书数据分析器 def __init__(self, storage): self.storage storage def analyze_content_trends(self, time_period7d): 分析内容趋势 query SELECT DATE(time) as date, COUNT(*) as note_count, AVG(like_count) as avg_likes, AVG(collect_count) as avg_collects, AVG(comment_count) as avg_comments FROM notes WHERE time date(now, ?) GROUP BY DATE(time) ORDER BY date df pd.read_sql_query(query, self.storage.conn, params(f-{time_period},)) # 生成趋势图表 fig, axes plt.subplots(2, 2, figsize(12, 8)) axes[0, 0].plot(df[date], df[note_count], markero) axes[0, 0].set_title(每日笔记发布量) axes[0, 0].set_xlabel(日期) axes[0, 0].set_ylabel(数量) axes[0, 1].plot(df[date], df[avg_likes], markers, colororange) axes[0, 1].set_title(平均点赞数趋势) axes[0, 1].set_xlabel(日期) axes[0, 1].set_ylabel(点赞数) axes[1, 0].plot(df[date], df[avg_collects], marker^, colorgreen) axes[1, 0].set_title(平均收藏数趋势) axes[1, 0].set_xlabel(日期) axes[1, 0].set_ylabel(收藏数) axes[1, 1].plot(df[date], df[avg_comments], markerd, colorred) axes[1, 1].set_title(平均评论数趋势) axes[1, 1].set_xlabel(日期) axes[1, 1].set_ylabel(评论数) plt.tight_layout() return fig, df 实战应用场景场景一竞品监控系统class CompetitorMonitor: 竞品监控系统 def __init__(self, client, competitor_ids): self.client client self.competitor_ids competitor_ids def monitor_competitor_activity(self): 监控竞品活动 competitor_data {} for competitor_id in self.competitor_ids: # 获取竞品最新内容 latest_notes self.client.get_user_notes(competitor_id) # 分析内容策略 content_analysis self.analyze_content_strategy(latest_notes[items]) # 分析互动表现 engagement_analysis self.analyze_engagement_pattern(latest_notes[items]) competitor_data[competitor_id] { recent_content: latest_notes[items][:5], # 最近5篇 content_strategy: content_analysis, engagement_pattern: engagement_analysis, update_time: datetime.now() } return competitor_data def generate_competitor_report(self, competitor_data): 生成竞品分析报告 report { summary: { total_competitors: len(competitor_data), monitoring_period: 7天, report_date: datetime.now().strftime(%Y-%m-%d) }, detailed_analysis: {}, recommendations: [] } for competitor_id, data in competitor_data.items(): report[detailed_analysis][competitor_id] { content_frequency: len(data[recent_content]), avg_engagement_rate: self.calculate_avg_engagement(data[recent_content]), content_types: self.analyze_content_types(data[recent_content]), posting_schedule: self.analyze_posting_schedule(data[recent_content]) } # 生成策略建议 report[recommendations] self.generate_strategy_recommendations(report[detailed_analysis]) return report场景二内容创作助手class ContentCreationAssistant: 内容创作智能助手 def __init__(self, client): self.client client def find_trending_topics(self, categorybeauty): 发现热门话题 feed_type_mapping { beauty: FeedType.COSMETICS, fashion: FeedType.FASION, food: FeedType.FOOD, travel: FeedType.TRAVEL } feed_type feed_type_mapping.get(category, FeedType.RECOMMEND) feed_data self.client.get_home_feed(feed_type) # 分析热门内容特征 trending_features self.analyze_trending_features(feed_data[items]) return { trending_topics: self.extract_topics(feed_data[items]), successful_patterns: trending_features, recommended_keywords: self.generate_keyword_suggestions(feed_data[items]) } def optimize_content_strategy(self, historical_data): 基于历史数据优化内容策略 analysis_results { best_posting_times: self.analyze_optimal_posting_times(historical_data), content_type_performance: self.analyze_content_type_performance(historical_data), keyword_effectiveness: self.analyze_keyword_effectiveness(historical_data), engagement_patterns: self.identify_engagement_patterns(historical_data) } return { analysis: analysis_results, actionable_insights: self.generate_actionable_insights(analysis_results) } 安全与合规指南合规使用原则数据采集伦理尊重用户隐私仅采集公开数据不获取用户隐私信息控制采集频率避免对服务器造成过大压力遵守平台规则严格遵守小红书用户协议和服务条款合法使用数据确保数据使用符合相关法律法规技术合规措施class EthicalDataCollector: 伦理数据采集器 def __init__(self, client, config): self.client client self.config config self.request_log [] def ethical_collect(self, collection_task, max_items100): 伦理数据采集主函数 collected_data [] item_count 0 while item_count max_items: try: # 添加请求延迟 self.respect_rate_limit() # 执行采集任务 data_batch collection_task() # 过滤敏感信息 filtered_data self.filter_sensitive_info(data_batch) collected_data.extend(filtered_data) item_count len(filtered_data) # 记录采集日志 self.log_collection_activity(filtered_data) if len(filtered_data) 0: break except Exception as e: self.handle_collection_error(e) break return collected_data def respect_rate_limit(self): 遵守请求频率限制 time.sleep(self.config.get(request_interval, 2)) # 检查每日请求限额 today_requests self.count_today_requests() if today_requests self.config.get(daily_limit, 1000): raise Exception(已达到每日请求限额) def filter_sensitive_info(self, data): 过滤敏感信息 filtered_data [] for item in data: filtered_item item.copy() # 移除可能包含的个人信息 sensitive_fields [phone, email, id_card, address] for field in sensitive_fields: if field in filtered_item: del filtered_item[field] filtered_data.append(filtered_item) return filtered_data 部署与优化建议生产环境部署容器化部署方案# Dockerfile示例 FROM python:3.9-slim WORKDIR /app # 安装依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 创建非root用户 RUN useradd -m -u 1000 appuser chown -R appuser:appuser /app USER appuser # 启动应用 CMD [python, app.py]监控与告警配置class MonitoringSystem: 采集系统监控 def __init__(self): self.metrics { total_requests: 0, successful_requests: 0, failed_requests: 0, last_error: None, average_response_time: 0 } def monitor_request(self, func): 请求监控装饰器 wraps(func) def wrapper(*args, **kwargs): start_time time.time() try: result func(*args, **kwargs) self.metrics[successful_requests] 1 return result except Exception as e: self.metrics[failed_requests] 1 self.metrics[last_error] str(e) raise finally: self.metrics[total_requests] 1 response_time time.time() - start_time # 更新平均响应时间 total_time self.metrics[average_response_time] * (self.metrics[total_requests] - 1) self.metrics[average_response_time] (total_time response_time) / self.metrics[total_requests] # 检查是否需要告警 self.check_alerts() return wrapper def check_alerts(self): 检查告警条件 error_rate self.metrics[failed_requests] / max(self.metrics[total_requests], 1) if error_rate 0.1: # 错误率超过10% self.send_alert(f高错误率告警: {error_rate:.1%}) if self.metrics[average_response_time] 5: # 平均响应时间超过5秒 self.send_alert(f响应时间过长: {self.metrics[average_response_time]:.2f}秒)性能优化策略缓存机制实现import redis from functools import lru_cache class CacheManager: 数据缓存管理器 def __init__(self, redis_hostlocalhost, redis_port6379): self.redis_client redis.Redis(hostredis_host, portredis_port, decode_responsesTrue) self.local_cache {} def get_cached_data(self, cache_key, ttl3600): 获取缓存数据 # 首先检查本地缓存 if cache_key in self.local_cache: return self.local_cache[cache_key] # 检查Redis缓存 cached_data self.redis_client.get(cache_key) if cached_data: data json.loads(cached_data) self.local_cache[cache_key] data return data return None def set_cached_data(self, cache_key, data, ttl3600): 设置缓存数据 # 设置本地缓存 self.local_cache[cache_key] data # 设置Redis缓存 self.redis_client.setex( cache_key, ttl, json.dumps(data, ensure_asciiFalse) ) lru_cache(maxsize128) def get_user_info_cached(self, user_id): 带缓存的用户信息获取 cache_key fuser_info:{user_id} cached_data self.get_cached_data(cache_key) if cached_data: return cached_data # 从API获取数据 user_info self.client.get_user_info(user_id) # 缓存数据 self.set_cached_data(cache_key, user_info, ttl1800) return user_info 学习路径与资源渐进式学习计划第一阶段基础掌握学习xhs库基本安装与配置掌握Cookie获取与客户端初始化实现基础数据采集功能第二阶段进阶应用学习签名服务部署与配置掌握错误处理与重试机制实现数据存储与处理流程第三阶段高级优化学习分布式部署方案掌握性能监控与优化技巧实现生产级数据采集系统核心资源参考项目结构概览核心模块xhs/core.py - 主要API接口实现工具函数xhs/help.py - 数据处理与转换工具示例代码example/ - 各类使用场景示例测试用例tests/ - 功能测试与验证关键配置文件requirements.txt - 项目依赖包列表setup.cfg - 项目打包配置Dockerfile - 容器化部署配置 最佳实践总结技术实践要点环境隔离使用虚拟环境管理Python依赖配置管理使用环境变量管理敏感信息日志记录实现完整的日志系统便于问题排查版本控制使用Git进行代码版本管理持续集成配置自动化测试与部署流程项目管理建议需求明确明确数据采集的具体需求和目标渐进开发从简单功能开始逐步增加复杂度测试驱动为每个功能编写测试用例文档完善保持代码注释和文档的及时更新监控维护建立系统监控和定期维护机制风险控制策略合规审查定期检查数据采集的合规性备份机制实现数据备份和恢复方案应急预案制定系统故障的应急处理流程性能监控持续监控系统性能和稳定性安全审计定期进行安全漏洞扫描和修复通过本文的深度解析你应该已经掌握了使用xhs库进行小红书数据采集的核心技术和最佳实践。记住技术只是工具合理、合规地使用这些工具才能让数据真正为业务创造价值。开始你的小红书数据采集之旅用数据驱动决策创造更大的商业价值【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
