Python小红书数据采集终极指南：xhs库完整使用教程与实战案例

Python小红书数据采集终极指南xhs库完整使用教程与实战案例【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs小红书作为国内领先的生活方式分享平台蕴含着海量的用户生成内容和商业价值数据。小红书数据采集工具xhs正是为开发者量身打造的Python爬虫库通过封装小红书Web端API接口帮助用户高效获取公开内容数据。本指南将为你全面解析从环境搭建到高级应用的全过程让你轻松掌握小红书数据采集的核心技术。 为什么选择xhs进行小红书数据采集在数据驱动的时代小红书平台的数据对于市场分析、竞品研究、内容创作等领域具有重要价值。xhs工具的核心优势在于其高效稳定的API封装和智能签名机制能够帮助开发者市场趋势洞察实时获取热门话题和消费趋势数据竞品策略分析监控竞争对手的产品推广和营销活动内容优化参考分析爆款笔记的特征规律和用户偏好用户画像构建基于互动数据构建精准的用户兴趣标签项目架构与核心模块xhs项目的核心功能模块位于xhs/目录下其中xhs/core.py包含了主要的API封装功能。项目的模块化设计确保了代码的可维护性和扩展性模块名称功能描述重要性core.py核心API封装包含所有数据获取方法⭐⭐⭐⭐⭐help.py辅助功能模块提供数据处理工具⭐⭐⭐⭐exception.py异常处理模块定义各种错误类型⭐⭐⭐example/使用示例目录包含多种应用场景⭐⭐⭐⭐ 5分钟快速部署与基础使用环境安装与配置安装xhs工具仅需一条命令支持多种安装方式# 通过PyPI安装稳定版本 pip install xhs # 从源码安装最新版本 git clone https://gitcode.com/gh_mirrors/xh/xhs cd xhs python setup.py install # 验证安装 python -c import xhs; print(xhs.__version__)系统要求Python 3.8或更高版本requests库自动安装lxml库自动安装playwright用于签名服务获取必要凭证使用xhs工具需要小红书的cookie信息以下是获取步骤浏览器登录在Chrome或Edge浏览器中登录小红书打开开发者工具按F12打开开发者工具查找Cookie切换到Application或Storage标签页复制关键字段找到并复制a1、web_session、webId等字段编写第一个采集脚本创建first_crawl.py文件开始你的数据采集之旅from xhs import XhsClient import json # 初始化客户端请替换为你的实际cookie cookie your_cookie_here client XhsClient(cookiecookie) # 搜索热门笔记 search_results client.search_note( keyword美食探店, page1, page_size10 ) print(f找到 {len(search_results[items])} 条相关笔记) print(json.dumps(search_results[items][0], indent2, ensure_asciiFalse)) 核心功能深度解析与高级配置智能搜索功能详解xhs提供了强大的搜索功能支持多种搜索参数和排序方式# 多维度搜索示例 def advanced_search_example(): 高级搜索功能演示 client XhsClient(cookieyour_cookie) # 1. 按热度排序搜索 hot_notes client.search_note( keyword旅行攻略, sort_typehot, # 按热度排序 page1, page_size15 ) # 2. 按时间排序搜索 new_notes client.search_note( keyword美妆教程, sort_typetime, # 按时间排序 page1, page_size15 ) # 3. 组合搜索条件 filtered_notes client.search_note( keyword健身, page1, page_size20, # 其他可选参数 # note_typevideo, # 仅视频笔记 # sortgeneral, # 综合排序 ) return hot_notes, new_notes, filtered_notes用户数据分析与内容获取获取指定用户的详细信息和发布内容def get_user_insights(user_id): 获取用户深度数据 client XhsClient(cookieyour_cookie) # 用户基本信息 user_info client.get_user_info(user_iduser_id) # 用户发布的笔记 user_notes client.get_user_notes( user_iduser_id, page1, page_size20 ) # 用户收藏的笔记 user_favorites client.get_user_favorites( user_iduser_id, page1, page_size15 ) # 数据分析 total_notes len(user_notes[items]) avg_likes sum(note[likes] for note in user_notes[items]) / total_notes return { user_info: user_info, total_notes: total_notes, avg_likes: avg_likes, notes_sample: user_notes[items][:3] }笔记详情与多媒体内容提取获取单篇笔记的完整信息包括图片、视频等多媒体内容from xhs import help def extract_note_content(note_id, xsec_token): 提取笔记完整内容 client XhsClient(cookieyour_cookie) # 获取笔记详情 note_detail client.get_note_by_id( note_idnote_id, xsec_tokenxsec_token ) # 提取图片链接 image_urls help.get_imgs_url_from_note(note_detail) # 提取视频链接 video_urls help.get_video_url_from_note(note_detail) # 提取文本内容 content note_detail.get(note, {}).get(content, ) return { title: note_detail.get(note, {}).get(title, ), content: content, images: image_urls, videos: video_urls, stats: { likes: note_detail.get(note, {}).get(likes, 0), collects: note_detail.get(note, {}).get(collects, 0), comments: note_detail.get(note, {}).get(comments, 0) } }️ 签名服务配置与高级功能独立签名服务部署为了应对小红书的签名验证机制xhs提供了签名服务方案。参考example/basic_sign_server.py配置独立签名服务# 签名服务配置示例 def setup_sign_server(): 配置独立签名服务 # 1. 安装必要依赖 # pip install playwright # playwright install chromium # 2. 准备stealth.min.js文件 # 可以从项目中获取或自行配置 # 3. 启动签名服务 # 参考example/basic_sign_server.py # 4. 客户端配置 sign_server_url http://localhost:5000/sign def custom_sign(uri, dataNone, a1, web_session): import requests response requests.post( sign_server_url, json{uri: uri, data: data, a1: a1, web_session: web_session} ) return response.json() return custom_sign错误处理与重试机制完善的错误处理是数据采集稳定性的关键import time import random from xhs.exception import DataFetchError, IPBlockError class SafeXhsClient: 带重试机制的Xhs客户端 def __init__(self, cookie, max_retries3): self.client XhsClient(cookiecookie) self.max_retries max_retries def safe_api_call(self, api_func, *args, **kwargs): 安全的API调用 for attempt in range(self.max_retries): try: return api_func(*args, **kwargs) except DataFetchError as e: print(f数据获取失败 (尝试 {attempt1}/{self.max_retries}): {e}) if attempt self.max_retries - 1: wait_time random.uniform(2, 5) print(f等待 {wait_time:.1f} 秒后重试...) time.sleep(wait_time) except IPBlockError: print(IP可能被限制建议更换IP或稍后再试) break except Exception as e: print(f未知错误: {e}) break return None def search_with_retry(self, keyword, **kwargs): 带重试的搜索功能 return self.safe_api_call(self.client.search_note, keyword, **kwargs) 实战应用场景与最佳实践场景一市场调研与竞品分析假设你是一家美妆品牌的市场分析师需要监控竞品推广策略def competitive_analysis(): 竞品分析实战 client SafeXhsClient(cookieyour_cookie) # 1. 收集竞品关键词 competitors [品牌A, 品牌B, 品牌C] keywords [口红, 眼影, 粉底液, 护肤品] # 2. 批量采集数据 all_data [] for brand in competitors: for keyword in keywords: search_term f{brand} {keyword} results client.search_with_retry( keywordsearch_term, page1, page_size20, sort_typehot ) if results: all_data.append({ brand: brand, keyword: keyword, notes: results[items], total: len(results[items]) }) # 3. 数据分析 analysis_results analyze_competition_data(all_data) return analysis_results def analyze_competition_data(data): 分析竞品数据 # 计算各品牌曝光量 brand_exposure {} for item in data: brand item[brand] brand_exposure[brand] brand_exposure.get(brand, 0) item[total] # 分析热门内容类型 content_types {} for item in data: for note in item[notes]: note_type note.get(type, unknown) content_types[note_type] content_types.get(note_type, 0) 1 return { brand_exposure: brand_exposure, content_types: content_types, total_notes: sum(item[total] for item in data) }场景二内容创作与热点追踪内容创作者可以使用xhs工具优化创作策略class ContentStrategyAnalyzer: 内容策略分析器 def __init__(self, cookie): self.client SafeXhsClient(cookie) def analyze_trending_topics(self, category美妆): 分析热门话题趋势 # 获取热门笔记 hot_notes self.client.search_with_retry( keywordcategory, sort_typehot, page1, page_size50 ) # 提取关键词 keywords self.extract_keywords(hot_notes) # 分析发布时间规律 post_times self.analyze_post_time(hot_notes) # 分析内容形式偏好 content_formats self.analyze_content_format(hot_notes) return { trending_keywords: keywords[:10], optimal_post_times: post_times, preferred_formats: content_formats } def extract_keywords(self, notes): 从笔记中提取高频关键词 # 简单的关键词提取逻辑 all_words [] for note in notes[items]: title note.get(title, ) content note.get(content, ) all_words.extend(title.split()) all_words.extend(content.split()) from collections import Counter word_counts Counter(all_words) return [word for word, count in word_counts.most_common(20)]场景三学术研究与数据分析学术研究者可以利用xhs数据进行社交网络分析import networkx as nx import matplotlib.pyplot as plt class SocialNetworkAnalyzer: 社交网络分析器 def __init__(self, cookie): self.client XhsClient(cookiecookie) self.graph nx.Graph() def build_user_network(self, seed_user_id, depth2): 构建用户社交网络 visited set() queue [(seed_user_id, 0)] while queue: user_id, current_depth queue.pop(0) if user_id in visited or current_depth depth: continue visited.add(user_id) # 获取用户关注的用户 try: following self.client.get_user_following(user_iduser_id) for followed_user in following[items]: followed_id followed_user[user_id] self.graph.add_edge(user_id, followed_id) if followed_id not in visited: queue.append((followed_id, current_depth 1)) except Exception as e: print(f获取用户 {user_id} 的关注列表失败: {e}) return self.graph def analyze_network_metrics(self): 分析网络指标 metrics { 节点数: self.graph.number_of_nodes(), 边数: self.graph.number_of_edges(), 平均度: sum(dict(self.graph.degree()).values()) / self.graph.number_of_nodes(), 聚类系数: nx.average_clustering(self.graph), 直径: nx.diameter(self.graph) if nx.is_connected(self.graph) else 不连通 } return metrics⚡ 性能优化与扩展方案并发处理与批量采集对于大规模数据采集建议使用并发处理提高效率import concurrent.futures from typing import List class BatchCollector: 批量数据采集器 def __init__(self, cookie, max_workers5): self.client XhsClient(cookiecookie) self.max_workers max_workers def collect_notes_batch(self, note_ids: List[str]) - List[dict]: 批量采集笔记信息 results [] with concurrent.futures.ThreadPoolExecutor(max_workersself.max_workers) as executor: # 提交所有任务 future_to_note { executor.submit(self.client.get_note_by_id, note_id, ): note_id for note_id in note_ids } # 处理完成的任务 for future in concurrent.futures.as_completed(future_to_note): note_id future_to_note[future] try: result future.result() results.append(result) print(f成功采集笔记: {note_id}) except Exception as e: print(f采集笔记 {note_id} 失败: {e}) return results def search_multiple_keywords(self, keywords: List[str]) - dict: 并发搜索多个关键词 keyword_results {} with concurrent.futures.ThreadPoolExecutor(max_workers3) as executor: future_to_keyword { executor.submit(self.client.search_note, keyword, page1, page_size20): keyword for keyword in keywords } for future in concurrent.futures.as_completed(future_to_keyword): keyword future_to_keyword[future] try: result future.result() keyword_results[keyword] result[items] except Exception as e: print(f搜索关键词 {keyword} 失败: {e}) return keyword_results数据缓存与持久化存储减少重复请求提高数据采集效率import json import os import hashlib from datetime import datetime, timedelta class DataCache: 数据缓存管理器 def __init__(self, cache_dirdata_cache, expiry_hours24): self.cache_dir cache_dir self.expiry_hours expiry_hours os.makedirs(cache_dir, exist_okTrue) def get_cache_key(self, func_name, *args, **kwargs): 生成缓存键 key_str f{func_name}_{args}_{kwargs} return hashlib.md5(key_str.encode()).hexdigest() def is_cache_valid(self, cache_file): 检查缓存是否有效 if not os.path.exists(cache_file): return False with open(cache_file, r, encodingutf-8) as f: cache_data json.load(f) cache_time datetime.fromisoformat(cache_data[timestamp]) return datetime.now() - cache_time timedelta(hoursself.expiry_hours) def cached_call(self, func, *args, **kwargs): 带缓存的函数调用 cache_key self.get_cache_key(func.__name__, *args, **kwargs) cache_file os.path.join(self.cache_dir, f{cache_key}.json) # 检查缓存 if os.path.exists(cache_file) and self.is_cache_valid(cache_file): with open(cache_file, r, encodingutf-8) as f: cache_data json.load(f) print(f使用缓存数据: {cache_key}) return cache_data[data] # 调用函数并缓存结果 result func(*args, **kwargs) cache_data { timestamp: datetime.now().isoformat(), data: result } with open(cache_file, w, encodingutf-8) as f: json.dump(cache_data, f, ensure_asciiFalse, indent2) print(f缓存新数据: {cache_key}) return result数据库存储方案对于大规模数据采集建议使用数据库存储import sqlite3 import pandas as pd from contextlib import contextmanager class DatabaseManager: 数据库管理器 def __init__(self, db_pathxhs_data.db): self.db_path db_path self.init_database() def init_database(self): 初始化数据库表结构 with self.get_connection() as conn: conn.execute( CREATE TABLE IF NOT EXISTS notes ( id TEXT PRIMARY KEY, title TEXT, content TEXT, user_id TEXT, likes INTEGER, collects INTEGER, comments INTEGER, create_time TIMESTAMP, update_time TIMESTAMP, raw_data TEXT ) ) conn.execute( CREATE TABLE IF NOT EXISTS users ( user_id TEXT PRIMARY KEY, nickname TEXT, avatar TEXT, notes_count INTEGER, fans_count INTEGER, following_count INTEGER, update_time TIMESTAMP ) ) contextmanager def get_connection(self): 获取数据库连接 conn sqlite3.connect(self.db_path) try: yield conn conn.commit() finally: conn.close() def save_note(self, note_data): 保存笔记数据 with self.get_connection() as conn: conn.execute( INSERT OR REPLACE INTO notes (id, title, content, user_id, likes, collects, comments, create_time, update_time, raw_data) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( note_data.get(id), note_data.get(title), note_data.get(content), note_data.get(user_id), note_data.get(likes, 0), note_data.get(collects, 0), note_data.get(comments, 0), note_data.get(create_time), datetime.now().isoformat(), json.dumps(note_data, ensure_asciiFalse) )) 故障排除与常见问题Q1: 签名失败怎么办解决方案检查cookie有效性确保cookie未过期重新获取最新cookie配置签名服务参考example/basic_sign_server.py配置独立签名服务调整等待时间在签名函数中适当增加sleep时间更新stealth.min.js确保使用最新版本的stealth脚本Q2: 请求频率过高被限制预防措施添加延迟在请求间添加随机延迟2-5秒使用代理池轮换使用多个IP地址限制并发数控制同时进行的请求数量监控响应检测异常响应并及时调整策略Q3: 如何提高数据采集稳定性最佳实践# 稳定的采集策略示例 class StableCollector: def __init__(self): self.retry_count 0 self.max_retries 5 def collect_with_strategy(self, collect_func): 带策略的数据采集 strategies [ self._try_direct_collect, self._try_with_delay, self._try_with_proxy, self._try_with_rotated_cookie ] for strategy in strategies: result strategy(collect_func) if result: return result return NoneQ4: 数据格式不一致如何处理数据清洗方案def clean_note_data(raw_note): 清洗笔记数据 cleaned { id: raw_note.get(id, ), title: raw_note.get(title, ).strip(), content: raw_note.get(content, ).strip(), user_id: raw_note.get(user, {}).get(user_id, ), stats: { likes: int(raw_note.get(likes, 0) or 0), collects: int(raw_note.get(collects, 0) or 0), comments: int(raw_note.get(comments, 0) or 0) }, timestamp: raw_note.get(time, ), images: raw_note.get(images, []), videos: raw_note.get(videos, []) } # 处理可能的None值 for key in [title, content]: if cleaned[key] is None: cleaned[key] return cleaned 性能对比与优化建议xhs工具性能特点功能模块性能表现优化建议搜索功能⭐⭐⭐⭐ 响应快速使用缓存减少重复搜索用户数据获取⭐⭐⭐⭐ 稳定性好批量获取减少请求次数笔记详情⭐⭐⭐ 受签名影响配置独立签名服务并发处理⭐⭐⭐⭐ 支持良好控制并发数避免限制性能优化技巧请求合并将多个小请求合并为批量请求数据缓存使用Redis或本地缓存存储频繁访问的数据连接池复用HTTP连接减少建立连接的开销异步处理使用asyncio提高I/O密集型任务效率import asyncio import aiohttp class AsyncXhsClient: 异步Xhs客户端 async def fetch_note_async(self, session, note_id): 异步获取笔记 url fhttps://www.xiaohongshu.com/explore/{note_id} async with session.get(url) as response: return await response.text() async def batch_fetch_notes(self, note_ids): 批量异步获取笔记 async with aiohttp.ClientSession() as session: tasks [self.fetch_note_async(session, note_id) for note_id in note_ids] return await asyncio.gather(*tasks, return_exceptionsTrue) 扩展与定制开发自定义数据处理器class CustomDataProcessor: 自定义数据处理器 def __init__(self): self.processors { text: self.process_text, image: self.process_image, video: self.process_video, user: self.process_user } def process_text(self, text_data): 处理文本数据 # 文本清洗、分词、情感分析等 return { cleaned_text: text_data.strip(), word_count: len(text_data.split()), sentiment: self.analyze_sentiment(text_data) } def process_image(self, image_urls): 处理图片数据 # 图片下载、特征提取等 return { urls: image_urls, count: len(image_urls), downloaded: self.download_images(image_urls) } def add_custom_processor(self, data_type, processor_func): 添加自定义处理器 self.processors[data_type] processor_func插件系统设计class PluginSystem: 插件系统 def __init__(self): self.plugins {} def register_plugin(self, name, plugin): 注册插件 self.plugins[name] plugin def process_data(self, data_type, data): 通过插件处理数据 if data_type in self.plugins: return self.plugins[data_type].process(data) return data # 示例插件 class SentimentAnalysisPlugin: def process(self, text): # 情感分析逻辑 return {sentiment: positive, score: 0.85} class ImageAnalysisPlugin: def process(self, images): # 图片分析逻辑 return {has_faces: True, dominant_color: #FF5733} 总结与最佳实践建议核心要点总结环境配置确保Python环境正确安装所有依赖凭证管理妥善保管cookie定期更新签名服务对于生产环境建议部署独立签名服务错误处理实现完善的错误处理和重试机制性能优化使用缓存、并发和批量处理提高效率部署架构建议对于企业级应用建议采用以下架构┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ 数据采集层 │ │ 数据处理层 │ │ 数据存储层 │ │ │ │ │ │ │ │ • 分布式爬虫 │───▶│ • 数据清洗 │───▶│ • 关系数据库 │ │ • 代理池管理 │ │ • 特征提取 │ │ • 时序数据库 │ │ • 签名服务集群 │ │ • 情感分析 │ │ • 对象存储 │ └─────────────────┘ └─────────────────┘ └─────────────────┘ │ │ │ ▼ ▼ ▼ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ 监控告警层 │ │ 数据分析层 │ │ 应用接口层 │ │ │ │ │ │ │ │ • 运行状态监控 │ │ • 趋势分析 │ │ • REST API │ │ • 异常检测 │ │ • 报表生成 │ │ • WebSocket │ │ • 自动恢复 │ │ • 预测模型 │ │ • GraphQL │ └─────────────────┘ └─────────────────┘ └─────────────────┘合规使用指南遵守robots协议尊重网站的爬取规则控制请求频率避免对服务器造成压力数据使用合规仅用于合法用途用户隐私保护不采集个人敏感信息版权尊重合理使用采集到的内容后续学习资源官方文档查阅项目文档了解最新功能示例代码参考example/目录中的完整示例源码学习研究xhs/core.py了解实现原理社区交流参与项目讨论获取帮助通过本指南你已经掌握了xhs工具的核心功能和使用技巧。记住技术是工具合理使用才能发挥最大价值。在享受数据采集带来的便利的同时也要时刻牢记数据伦理和合规要求。立即开始从简单的示例开始逐步构建复杂的数据采集系统。遇到问题时参考故障排除部分或查阅项目文档和社区资源。祝你数据采集顺利【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考