小红书数据采集终极指南xhs Python库完整实战教程【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs小红书作为中国领先的生活方式分享平台蕴含着海量的用户行为数据和消费趋势洞察。对于开发者、数据分析师和研究人员而言如何高效、稳定地获取这些数据成为了一个重要课题。xhs库正是为解决这一问题而生的Python请求封装库它提供了完整的小红书Web端API封装解决方案让开发者能够专注于数据分析和业务逻辑而不是繁琐的网络请求和反爬虫对抗。1. 项目背景与价值定位xhs库的诞生源于对小红书数据采集需求的深度理解。传统爬虫开发面临诸多挑战频繁的反爬虫机制更新、复杂的请求签名算法、以及不稳定的API接口。xhs库通过完整的API封装和智能的签名机制为开发者提供了一个稳定可靠的数据采集解决方案。技术优势亮点开箱即用的API封装封装了小红书Web端的所有核心接口智能签名机制自动处理复杂的x-s/x-t签名算法多登录方式支持提供二维码和手机号验证码两种登录方式完善的异常处理内置IP封锁检测、会话过期处理等机制高性能设计支持异步请求和批量处理应用场景广泛开发者能够利用xhs库实现多种业务场景市场趋势分析采集特定品类笔记数据分析用户偏好变化竞品监测定期采集竞品账号的发布内容和用户互动数据内容质量评估通过点赞、收藏、评论等数据评估内容质量用户画像构建结合用户发布内容和互动行为构建精准画像2. 核心架构解析xhs库的核心架构设计体现了对小红书Web端API的深度理解。让我们深入分析其技术实现原理请求签名机制小红书Web端采用了复杂的签名算法来防止恶意爬取。xhs库通过Playwright模拟浏览器环境调用JavaScript签名函数来生成合法的x-s和x-t参数# xhs/help.py中的签名函数 def sign(uri, dataNone, ctimeNone, a1, b1): 生成小红书请求签名 v int(round(time.time() * 1000) if not ctime else ctime) raw_str f{v}test{uri}{json.dumps(data, separators(,, :), ensure_asciiFalse) if isinstance(data, dict) else } md5_str hashlib.md5(raw_str.encode(utf-8)).hexdigest() x_s h(md5_str) # 自定义编码函数 x_t str(v) return {x-s: x_s, x-t: x_t}客户端架构设计XhsClient类采用了模块化设计将不同功能封装为独立的方法# xhs/core.py中的核心客户端类 class XhsClient: def __init__(self, cookieNone, user_agentNone, timeout10, proxiesNone, signNone): self.__session requests.session() self.timeout timeout self.proxies proxies self.sign sign self.cookie cookie def get_note_by_id(self, note_id: str, xsec_token: str, xsec_source: str pc_feed): 根据笔记ID获取笔记详情 # 实现细节... def search_note(self, keyword: str, page: int 1, page_size: int 20, sort: SearchSortType SearchSortType.GENERAL, note_type: SearchNoteType SearchNoteType.ALL): 搜索笔记 # 实现细节...异常处理体系xhs库定义了完整的异常类体系帮助开发者更好地处理各种错误情况# xhs/exception.py中的异常定义 class ErrorEnum(Enum): IP_BLOCK ErrorTuple(300012, 网络连接异常请检查网络设置或重启试试) NOTE_ABNORMAL ErrorTuple(-510001, 笔记状态异常请稍后查看) SIGN_FAULT ErrorTuple(300015, 浏览器异常请尝试关闭/卸载风险插件或重启试试) class DataFetchError(RequestException): 数据获取错误 class IPBlockError(RequestException): IP被封锁错误 class SignError(RequestException): 签名错误3. 快速入门指南环境安装与配置xhs库的安装非常简单但需要一些前置依赖# 安装xhs库 pip install xhs # 安装Playwright用于签名 pip install playwright playwright install # 下载stealth.min.js绕过环境检测 curl -O https://cdn.jsdelivr.net/gh/requireCool/stealth.min.js/stealth.min.js基础使用示例让我们从一个最简单的示例开始展示如何获取笔记详情# example/basic_usage.py中的基础示例 import datetime import json from xhs import XhsClient, help def sign(uri, dataNone, a1, web_session): 签名函数 - 需要根据实际情况实现 # 这里需要实现具体的签名逻辑 pass if __name__ __main__: # 初始化客户端 cookie your_cookie_here # 替换为实际的cookie xhs_client XhsClient(cookie, signsign) # 获取笔记详情 note_id 6505318c000000001f03c5a6 note_info xhs_client.get_note_by_id(note_id, xsec_token) # 提取图片URL img_urls help.get_imgs_url_from_note(note_info) print(f笔记标题: {note_info.get(title, 无标题)}) print(f图片数量: {len(img_urls)}) print(f点赞数: {note_info.get(likes, 0)})二维码登录实现对于需要登录的场景xhs库提供了便捷的二维码登录方式# example/login_qrcode.py中的二维码登录示例 import qrcode from xhs import XhsClient def sign(uri, dataNone, a1, web_session): 签名函数实现 # 实现签名逻辑 pass # 初始化客户端 xhs_client XhsClient(signsign) # 获取二维码信息 qr_res xhs_client.get_qrcode() qr_id qr_res[qr_id] qr_code qr_res[code] # 生成二维码并显示 qr qrcode.QRCode(version1, error_correctionqrcode.ERROR_CORRECT_L, box_size50, border1) qr.add_data(qr_res[url]) qr.make() qr.print_ascii() # 在终端显示二维码 # 轮询检查登录状态 while True: check_qrcode xhs_client.check_qrcode(qr_id, qr_code) if check_qrcode[code_status] 2: # 登录成功 login_info check_qrcode[login_info] print(f登录成功当前cookie: {xhs_client.cookie}) break time.sleep(1)4. 高级功能深度剖析签名服务器部署对于生产环境建议将签名服务部署为独立的服务器以提高稳定性和可维护性# xhs-api/app.py中的签名服务器实现 from flask import Flask, request from playwright.sync_api import sync_playwright app Flask(__name__) global_a1 app.route(/sign, methods[POST]) def sign_api(): 签名API接口 data request.json uri data.get(uri) req_data data.get(data) a1 data.get(a1, ) web_session data.get(web_session, ) # 调用浏览器环境进行签名 encrypt_params context_page.evaluate( ([url, data]) window._webmsxyw(url, data), [uri, req_data] ) return { x-s: encrypt_params[X-s], x-t: str(encrypt_params[X-t]) } if __name__ __main__: # 启动浏览器环境 playwright sync_playwright().start() # ... 浏览器初始化代码 app.run(host0.0.0.0, port5005)使用Docker部署签名服务更加便捷# 使用官方Docker镜像 docker run -it -d -p 5005:5005 reajason/xhs-api:latest客户端集成签名服务客户端可以轻松集成远程签名服务# example/basic_sign_usage.py中的客户端集成 import requests from xhs import XhsClient def sign(uri, dataNone, a1, web_session): 调用远程签名服务 res requests.post(http://localhost:5005/sign, json{ uri: uri, data: data, a1: a1, web_session: web_session }) signs res.json() return { x-s: signs[x-s], x-t: signs[x-t] } # 使用签名服务初始化客户端 cookie your_cookie_here xhs_client XhsClient(cookie, signsign) # 现在可以正常使用所有API note_info xhs_client.get_note_by_id(63db8819000000001a01ead1)数据模型与类型定义xhs库使用了Python的类型提示和枚举类来提供更好的开发体验# xhs/core.py中的枚举定义 class FeedType(Enum): 首页feed类型 RECOMMEND homefeed_recommend # 推荐 FASION homefeed.fashion_v3 # 穿搭 FOOD homefeed.food_v3 # 美食 COSMETICS homefeed.cosmetics_v3 # 彩妆 MOVIE homefeed.movie_and_tv_v3 # 影视 class SearchSortType(Enum): 搜索排序类型 GENERAL general # 综合排序 MOST_POPULAR popularity_descending # 最热 LATEST time_descending # 最新 class Note(NamedTuple): 笔记数据模型 note_id: str title: str desc: str type: str user: dict img_urls: list video_url: str tag_list: list at_user_list: list collected_count: str comment_count: str liked_count: str share_count: str time: int last_update_time: int5. 实战应用场景市场趋势分析系统通过采集特定品类的笔记数据我们可以分析市场趋势和用户偏好from xhs import XhsClient, SearchSortType, SearchNoteType import pandas as pd from datetime import datetime class MarketTrendAnalyzer: def __init__(self, client): self.client client def analyze_category_trend(self, category, days30): 分析品类趋势 trends [] keywords self.get_category_keywords(category) for keyword in keywords: # 搜索相关笔记 results self.client.search_note( keywordkeyword, sortSearchSortType.MOST_POPULAR, page_size50 ) # 分析数据 trend_data { keyword: keyword, total_notes: len(results.get(items, [])), avg_likes: self.calculate_avg_likes(results), avg_comments: self.calculate_avg_comments(results), trend_score: self.calculate_trend_score(results) } trends.append(trend_data) return pd.DataFrame(trends) def calculate_trend_score(self, results): 计算趋势得分 items results.get(items, []) if not items: return 0 # 基于互动率和发布时间计算趋势得分 score 0 for item in items: likes int(item.get(likes, 0)) comments int(item.get(comments, 0)) time_diff datetime.now().timestamp() - item.get(time, 0) # 互动率权重 时间衰减 engagement_rate (likes comments * 2) / 1000 time_weight 1 / (1 time_diff / 86400) # 24小时衰减 score engagement_rate * time_weight return score / len(items) # 使用示例 analyzer MarketTrendAnalyzer(xhs_client) beauty_trend analyzer.analyze_category_trend(美妆) print(beauty_trend.sort_values(trend_score, ascendingFalse))竞品监测平台定期监测竞品账号的数据变化import schedule import time from xhs import XhsClient class CompetitorMonitor: def __init__(self, client, competitor_ids): self.client client self.competitor_ids competitor_ids self.history_data {} def monitor_competitor(self, user_id): 监测单个竞品账号 try: # 获取用户信息 user_info self.client.get_user_info(user_id) # 获取最新笔记 notes self.client.get_user_notes(user_id) # 计算关键指标 metrics { timestamp: time.time(), fans_count: user_info.get(fans_count, 0), notes_count: user_info.get(notes_count, 0), avg_likes: self.calculate_avg_likes(notes), engagement_rate: self.calculate_engagement_rate(user_info, notes), new_notes: len(notes.get(items, [])) } # 存储历史数据 if user_id not in self.history_data: self.history_data[user_id] [] self.history_data[user_id].append(metrics) return metrics except Exception as e: print(f监测竞品 {user_id} 失败: {e}) return None def calculate_engagement_rate(self, user_info, notes): 计算互动率 fans int(user_info.get(fans_count, 1)) items notes.get(items, []) if not items or fans 0: return 0 total_interactions 0 for item in items[:10]: # 只看最近10条 likes int(item.get(likes, 0)) comments int(item.get(comments, 0)) total_interactions likes comments return total_interactions / (fans * 10) * 100 # 百分比 # 定时监测任务 monitor CompetitorMonitor(xhs_client, [user_id_1, user_id_2]) def daily_monitoring(): for user_id in monitor.competitor_ids: metrics monitor.monitor_competitor(user_id) if metrics: print(f竞品 {user_id} 数据: {metrics}) # 每天执行一次 schedule.every().day.at(09:00).do(daily_monitoring) while True: schedule.run_pending() time.sleep(60)内容质量评估工具通过多维度指标评估内容质量from xhs import XhsClient import numpy as np class ContentQualityAssessor: def __init__(self, client): self.client client def assess_note_quality(self, note_id): 评估笔记质量 try: note self.client.get_note_by_id(note_id) # 计算各项得分 engagement_score self.calculate_engagement_score(note) content_score self.calculate_content_score(note) visual_score self.calculate_visual_score(note) timeliness_score self.calculate_timeliness_score(note) # 加权总分 total_score ( engagement_score * 0.4 content_score * 0.3 visual_score * 0.2 timeliness_score * 0.1 ) return { note_id: note_id, engagement_score: engagement_score, content_score: content_score, visual_score: visual_score, timeliness_score: timeliness_score, total_score: total_score, quality_level: self.get_quality_level(total_score) } except Exception as e: print(f评估笔记 {note_id} 失败: {e}) return None def calculate_engagement_score(self, note): 计算互动得分 likes int(note.get(likes, 0)) comments int(note.get(comments, 0)) collects int(note.get(collects, 0)) shares int(note.get(shares, 0)) # 标准化处理 engagement likes comments * 2 collects * 3 shares * 1.5 return min(engagement / 1000, 1.0) # 归一化到0-1 def get_quality_level(self, score): 根据得分确定质量等级 if score 0.8: return 优质 elif score 0.6: return 良好 elif score 0.4: return 一般 else: return 待提升 # 批量评估示例 assessor ContentQualityAssessor(xhs_client) note_ids [note_id_1, note_id_2, note_id_3] for note_id in note_ids: assessment assessor.assess_note_quality(note_id) if assessment: print(f笔记 {note_id} 质量评估: {assessment})6. 性能优化与最佳实践请求频率控制合理控制请求频率是避免被封IP的关键import time import random from functools import wraps def rate_limited(max_per_minute60): 请求频率限制装饰器 min_interval 60.0 / max_per_minute def decorator(func): last_called [0.0] wraps(func) def wrapper(*args, **kwargs): elapsed time.time() - last_called[0] left_to_wait min_interval - elapsed if left_to_wait 0: # 添加随机延迟模拟人类行为 time.sleep(left_to_wait random.uniform(0.1, 0.5)) result func(*args, **kwargs) last_called[0] time.time() return result return wrapper return decorator class OptimizedXhsClient(XhsClient): 优化版的Xhs客户端 rate_limited(max_per_minute30) # 限制每分钟30次请求 def get_note_by_id(self, note_id: str, xsec_token: str, xsec_source: str pc_feed): 带频率限制的笔记获取 return super().get_note_by_id(note_id, xsec_token, xsec_source) rate_limited(max_per_minute20) # 搜索请求限制更严格 def search_note(self, keyword: str, page: int 1, page_size: int 20, sort: SearchSortType SearchSortType.GENERAL, note_type: SearchNoteType SearchNoteType.ALL): 带频率限制的搜索 return super().search_note(keyword, page, page_size, sort, note_type)数据缓存策略对频繁访问的数据进行缓存减少重复请求import pickle import hashlib from datetime import datetime, timedelta from functools import lru_cache class DataCache: 数据缓存管理器 def __init__(self, cache_dir.xhs_cache, ttl_hours24): self.cache_dir cache_dir self.ttl timedelta(hoursttl_hours) def get_cache_key(self, func_name, *args, **kwargs): 生成缓存键 key_str f{func_name}_{args}_{kwargs} return hashlib.md5(key_str.encode()).hexdigest() def get(self, key): 获取缓存数据 cache_file f{self.cache_dir}/{key}.pkl try: with open(cache_file, rb) as f: data, timestamp pickle.load(f) # 检查是否过期 if datetime.now() - timestamp self.ttl: return data except: pass return None def set(self, key, data): 设置缓存数据 import os os.makedirs(self.cache_dir, exist_okTrue) cache_file f{self.cache_dir}/{key}.pkl with open(cache_file, wb) as f: pickle.dump((data, datetime.now()), f) # 使用缓存的客户端 class CachedXhsClient(XhsClient): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.cache DataCache() lru_cache(maxsize100) def get_note_by_id_cached(self, note_id: str, xsec_token: str): 带内存缓存的笔记获取 cache_key self.cache.get_cache_key(get_note_by_id, note_id, xsec_token) # 先检查磁盘缓存 cached_data self.cache.get(cache_key) if cached_data: return cached_data # 没有缓存则请求数据 data self.get_note_by_id(note_id, xsec_token) # 保存到缓存 self.cache.set(cache_key, data) return data错误处理与重试机制完善的错误处理是生产环境应用的必备import time from xhs.exception import DataFetchError, IPBlockError, SignError class RobustXhsClient(XhsClient): 增强错误处理的客户端 def safe_request(self, func, *args, max_retries3, retry_delay2, **kwargs): 安全的请求包装器 for attempt in range(max_retries): try: return func(*args, **kwargs) except IPBlockError as e: print(fIP被封锁等待{retry_delay * (attempt 1)}秒后重试) time.sleep(retry_delay * (attempt 1)) except SignError as e: print(f签名错误第{attempt 1}次重试: {e}) if attempt max_retries - 1: raise time.sleep(retry_delay) except DataFetchError as e: print(f数据获取错误: {e}) if 404 in str(e) or 笔记不存在 in str(e): return None # 笔记不存在不重试 time.sleep(retry_delay) except Exception as e: print(f未知错误: {e}) if attempt max_retries - 1: raise time.sleep(retry_delay) return None def get_note_safe(self, note_id: str, xsec_token: str): 安全的笔记获取 return self.safe_request( super().get_note_by_id, note_id, xsec_token, max_retries3 ) def search_note_safe(self, keyword: str, **kwargs): 安全的搜索 return self.safe_request( super().search_note, keyword, **kwargs, max_retries2 )7. 生态集成与扩展与数据分析工具集成xhs库可以轻松与主流数据分析工具集成import pandas as pd import matplotlib.pyplot as plt from xhs import XhsClient class XhsDataAnalyzer: 小红书数据分析器 def __init__(self, client): self.client client def collect_user_data(self, user_id, max_notes100): 收集用户数据并转换为DataFrame all_notes [] cursor while len(all_notes) max_notes: result self.client.get_user_notes(user_id, cursorcursor) notes result.get(items, []) if not notes: break all_notes.extend(notes) cursor result.get(cursor, ) if not cursor: break # 转换为DataFrame df pd.DataFrame(all_notes) # 数据清洗和转换 if not df.empty: df[likes] pd.to_numeric(df[likes], errorscoerce).fillna(0) df[comments] pd.to_numeric(df[comments], errorscoerce).fillna(0) df[time] pd.to_datetime(df[time], units, errorscoerce) return df def analyze_engagement_trend(self, user_id): 分析用户互动趋势 df self.collect_user_data(user_id) if df.empty: return None # 按时间分组分析 df.set_index(time, inplaceTrue) monthly_stats df.resample(M).agg({ likes: sum, comments: sum, note_id: count }).rename(columns{note_id: note_count}) return monthly_stats def visualize_engagement(self, user_id): 可视化用户互动数据 stats self.analyze_engagement_trend(user_id) if stats is None: print(没有找到用户数据) return fig, axes plt.subplots(2, 1, figsize(12, 8)) # 互动趋势图 stats[[likes, comments]].plot(axaxes[0]) axes[0].set_title(f用户 {user_id} 互动趋势) axes[0].set_ylabel(互动数量) axes[0].legend([点赞, 评论]) # 发布频率图 stats[note_count].plot(kindbar, axaxes[1]) axes[1].set_title(月度发布数量) axes[1].set_ylabel(笔记数量) axes[1].set_xlabel(月份) plt.tight_layout() plt.show() # 使用示例 analyzer XhsDataAnalyzer(xhs_client) user_stats analyzer.analyze_engagement_trend(target_user_id) print(user_stats)与数据库集成将采集的数据存储到数据库中import sqlite3 from datetime import datetime from xhs import XhsClient class XhsDataStorage: 小红书数据存储管理器 def __init__(self, db_pathxhs_data.db): self.db_path db_path self.init_database() def init_database(self): 初始化数据库表结构 conn sqlite3.connect(self.db_path) cursor conn.cursor() # 创建笔记表 cursor.execute( CREATE TABLE IF NOT EXISTS notes ( note_id TEXT PRIMARY KEY, title TEXT, content TEXT, user_id TEXT, user_name TEXT, likes INTEGER, comments INTEGER, collects INTEGER, shares INTEGER, note_type TEXT, tags TEXT, created_time INTEGER, updated_time INTEGER, crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) # 创建用户表 cursor.execute( CREATE TABLE IF NOT EXISTS users ( user_id TEXT PRIMARY KEY, nickname TEXT, fans_count INTEGER, follows_count INTEGER, notes_count INTEGER, liked_count INTEGER, description TEXT, crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() def save_note(self, note_data): 保存笔记数据 conn sqlite3.connect(self.db_path) cursor conn.cursor() try: cursor.execute( INSERT OR REPLACE INTO notes (note_id, title, content, user_id, user_name, likes, comments, collects, shares, note_type, tags, created_time, updated_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( note_data.get(note_id), note_data.get(title, ), note_data.get(desc, ), note_data.get(user, {}).get(user_id, ), note_data.get(user, {}).get(nickname, ), note_data.get(likes, 0), note_data.get(comments, 0), note_data.get(collects, 0), note_data.get(shares, 0), note_data.get(type, normal), ,.join(note_data.get(tag_list, [])), note_data.get(time, 0), note_data.get(last_update_time, 0) )) conn.commit() return True except Exception as e: print(f保存笔记失败: {e}) return False finally: conn.close() def batch_save_notes(self, notes_list): 批量保存笔记数据 success_count 0 for note in notes_list: if self.save_note(note): success_count 1 print(f成功保存 {success_count}/{len(notes_list)} 条笔记) return success_count # 使用示例 storage XhsDataStorage() client XhsClient() # 搜索并保存笔记 search_results client.search_note(keyword美食探店, page_size50) if search_results and items in search_results: storage.batch_save_notes(search_results[items])Web API服务封装将xhs库封装为RESTful API服务from flask import Flask, request, jsonify from flask_restful import Api, Resource from xhs import XhsClient app Flask(__name__) api Api(app) class XhsAPI: def __init__(self): self.client XhsClient() def get_note(self, note_id): 获取笔记详情 try: note self.client.get_note_by_id(note_id) return { success: True, data: note, message: 获取成功 } except Exception as e: return { success: False, data: None, message: str(e) } def search_notes(self, keyword, page1, page_size20): 搜索笔记 try: result self.client.search_note( keywordkeyword, pagepage, page_sizepage_size ) return { success: True, data: result, message: 搜索成功 } except Exception as e: return { success: False, data: None, message: str(e) } xhs_api XhsAPI() class NoteResource(Resource): def get(self, note_id): 获取笔记详情API return jsonify(xhs_api.get_note(note_id)) class SearchResource(Resource): def get(self): 搜索笔记API keyword request.args.get(keyword, ) page int(request.args.get(page, 1)) page_size int(request.args.get(page_size, 20)) return jsonify(xhs_api.search_notes(keyword, page, page_size)) # 注册API路由 api.add_resource(NoteResource, /api/note/string:note_id) api.add_resource(SearchResource, /api/search) if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)8. 未来发展与社区贡献项目发展方向xhs库作为小红书数据采集的重要工具未来有几个重要的发展方向异步支持集成asyncio和aiohttp提供异步API支持Type Hints完善为所有API添加完整的类型提示更多API覆盖支持评论、私信、直播等更多功能性能优化实现请求池和连接复用插件系统支持自定义插件扩展功能社区贡献指南欢迎开发者参与xhs库的开发和改进# 开发环境设置 git clone https://gitcode.com/gh_mirrors/xh/xhs cd xhs pip install -e .[dev] # 安装开发依赖 # 运行测试 pytest tests/ # 代码规范检查 flake8 xhs/ tests/ black xhs/ tests/ --check最佳实践建议基于项目经验我们总结了一些最佳实践合理控制请求频率避免对小红书服务器造成过大压力使用代理IP轮换在大规模采集时使用代理IP实现数据验证对采集的数据进行有效性验证定期更新cookie处理会话过期问题监控采集状态实现采集任务的监控和报警合规使用提醒在使用xhs库进行数据采集时请务必遵守以下原则尊重robots协议遵守网站的访问规则控制采集频率避免影响网站正常服务保护用户隐私不采集敏感个人信息遵守法律法规确保数据使用合法合规明确使用目的仅用于合法合规的数据分析结语xhs库为小红书数据采集提供了一个强大而灵活的工具链。通过本文的详细介绍开发者能够全面了解xhs库的核心功能、高级用法和最佳实践。无论是进行市场研究、竞品分析还是内容监控xhs库都能帮助你高效获取所需数据。记住技术只是手段真正的价值在于如何利用数据创造洞察。在遵守规则的前提下合理使用xhs库开启你的数据探索之旅吧立即开始pip install xhs深入学习查看example目录下的示例代码遇到问题参考xhs/exception.py中的异常处理指南【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
小红书数据采集终极指南：xhs Python库完整实战教程
小红书数据采集终极指南xhs Python库完整实战教程【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs小红书作为中国领先的生活方式分享平台蕴含着海量的用户行为数据和消费趋势洞察。对于开发者、数据分析师和研究人员而言如何高效、稳定地获取这些数据成为了一个重要课题。xhs库正是为解决这一问题而生的Python请求封装库它提供了完整的小红书Web端API封装解决方案让开发者能够专注于数据分析和业务逻辑而不是繁琐的网络请求和反爬虫对抗。1. 项目背景与价值定位xhs库的诞生源于对小红书数据采集需求的深度理解。传统爬虫开发面临诸多挑战频繁的反爬虫机制更新、复杂的请求签名算法、以及不稳定的API接口。xhs库通过完整的API封装和智能的签名机制为开发者提供了一个稳定可靠的数据采集解决方案。技术优势亮点开箱即用的API封装封装了小红书Web端的所有核心接口智能签名机制自动处理复杂的x-s/x-t签名算法多登录方式支持提供二维码和手机号验证码两种登录方式完善的异常处理内置IP封锁检测、会话过期处理等机制高性能设计支持异步请求和批量处理应用场景广泛开发者能够利用xhs库实现多种业务场景市场趋势分析采集特定品类笔记数据分析用户偏好变化竞品监测定期采集竞品账号的发布内容和用户互动数据内容质量评估通过点赞、收藏、评论等数据评估内容质量用户画像构建结合用户发布内容和互动行为构建精准画像2. 核心架构解析xhs库的核心架构设计体现了对小红书Web端API的深度理解。让我们深入分析其技术实现原理请求签名机制小红书Web端采用了复杂的签名算法来防止恶意爬取。xhs库通过Playwright模拟浏览器环境调用JavaScript签名函数来生成合法的x-s和x-t参数# xhs/help.py中的签名函数 def sign(uri, dataNone, ctimeNone, a1, b1): 生成小红书请求签名 v int(round(time.time() * 1000) if not ctime else ctime) raw_str f{v}test{uri}{json.dumps(data, separators(,, :), ensure_asciiFalse) if isinstance(data, dict) else } md5_str hashlib.md5(raw_str.encode(utf-8)).hexdigest() x_s h(md5_str) # 自定义编码函数 x_t str(v) return {x-s: x_s, x-t: x_t}客户端架构设计XhsClient类采用了模块化设计将不同功能封装为独立的方法# xhs/core.py中的核心客户端类 class XhsClient: def __init__(self, cookieNone, user_agentNone, timeout10, proxiesNone, signNone): self.__session requests.session() self.timeout timeout self.proxies proxies self.sign sign self.cookie cookie def get_note_by_id(self, note_id: str, xsec_token: str, xsec_source: str pc_feed): 根据笔记ID获取笔记详情 # 实现细节... def search_note(self, keyword: str, page: int 1, page_size: int 20, sort: SearchSortType SearchSortType.GENERAL, note_type: SearchNoteType SearchNoteType.ALL): 搜索笔记 # 实现细节...异常处理体系xhs库定义了完整的异常类体系帮助开发者更好地处理各种错误情况# xhs/exception.py中的异常定义 class ErrorEnum(Enum): IP_BLOCK ErrorTuple(300012, 网络连接异常请检查网络设置或重启试试) NOTE_ABNORMAL ErrorTuple(-510001, 笔记状态异常请稍后查看) SIGN_FAULT ErrorTuple(300015, 浏览器异常请尝试关闭/卸载风险插件或重启试试) class DataFetchError(RequestException): 数据获取错误 class IPBlockError(RequestException): IP被封锁错误 class SignError(RequestException): 签名错误3. 快速入门指南环境安装与配置xhs库的安装非常简单但需要一些前置依赖# 安装xhs库 pip install xhs # 安装Playwright用于签名 pip install playwright playwright install # 下载stealth.min.js绕过环境检测 curl -O https://cdn.jsdelivr.net/gh/requireCool/stealth.min.js/stealth.min.js基础使用示例让我们从一个最简单的示例开始展示如何获取笔记详情# example/basic_usage.py中的基础示例 import datetime import json from xhs import XhsClient, help def sign(uri, dataNone, a1, web_session): 签名函数 - 需要根据实际情况实现 # 这里需要实现具体的签名逻辑 pass if __name__ __main__: # 初始化客户端 cookie your_cookie_here # 替换为实际的cookie xhs_client XhsClient(cookie, signsign) # 获取笔记详情 note_id 6505318c000000001f03c5a6 note_info xhs_client.get_note_by_id(note_id, xsec_token) # 提取图片URL img_urls help.get_imgs_url_from_note(note_info) print(f笔记标题: {note_info.get(title, 无标题)}) print(f图片数量: {len(img_urls)}) print(f点赞数: {note_info.get(likes, 0)})二维码登录实现对于需要登录的场景xhs库提供了便捷的二维码登录方式# example/login_qrcode.py中的二维码登录示例 import qrcode from xhs import XhsClient def sign(uri, dataNone, a1, web_session): 签名函数实现 # 实现签名逻辑 pass # 初始化客户端 xhs_client XhsClient(signsign) # 获取二维码信息 qr_res xhs_client.get_qrcode() qr_id qr_res[qr_id] qr_code qr_res[code] # 生成二维码并显示 qr qrcode.QRCode(version1, error_correctionqrcode.ERROR_CORRECT_L, box_size50, border1) qr.add_data(qr_res[url]) qr.make() qr.print_ascii() # 在终端显示二维码 # 轮询检查登录状态 while True: check_qrcode xhs_client.check_qrcode(qr_id, qr_code) if check_qrcode[code_status] 2: # 登录成功 login_info check_qrcode[login_info] print(f登录成功当前cookie: {xhs_client.cookie}) break time.sleep(1)4. 高级功能深度剖析签名服务器部署对于生产环境建议将签名服务部署为独立的服务器以提高稳定性和可维护性# xhs-api/app.py中的签名服务器实现 from flask import Flask, request from playwright.sync_api import sync_playwright app Flask(__name__) global_a1 app.route(/sign, methods[POST]) def sign_api(): 签名API接口 data request.json uri data.get(uri) req_data data.get(data) a1 data.get(a1, ) web_session data.get(web_session, ) # 调用浏览器环境进行签名 encrypt_params context_page.evaluate( ([url, data]) window._webmsxyw(url, data), [uri, req_data] ) return { x-s: encrypt_params[X-s], x-t: str(encrypt_params[X-t]) } if __name__ __main__: # 启动浏览器环境 playwright sync_playwright().start() # ... 浏览器初始化代码 app.run(host0.0.0.0, port5005)使用Docker部署签名服务更加便捷# 使用官方Docker镜像 docker run -it -d -p 5005:5005 reajason/xhs-api:latest客户端集成签名服务客户端可以轻松集成远程签名服务# example/basic_sign_usage.py中的客户端集成 import requests from xhs import XhsClient def sign(uri, dataNone, a1, web_session): 调用远程签名服务 res requests.post(http://localhost:5005/sign, json{ uri: uri, data: data, a1: a1, web_session: web_session }) signs res.json() return { x-s: signs[x-s], x-t: signs[x-t] } # 使用签名服务初始化客户端 cookie your_cookie_here xhs_client XhsClient(cookie, signsign) # 现在可以正常使用所有API note_info xhs_client.get_note_by_id(63db8819000000001a01ead1)数据模型与类型定义xhs库使用了Python的类型提示和枚举类来提供更好的开发体验# xhs/core.py中的枚举定义 class FeedType(Enum): 首页feed类型 RECOMMEND homefeed_recommend # 推荐 FASION homefeed.fashion_v3 # 穿搭 FOOD homefeed.food_v3 # 美食 COSMETICS homefeed.cosmetics_v3 # 彩妆 MOVIE homefeed.movie_and_tv_v3 # 影视 class SearchSortType(Enum): 搜索排序类型 GENERAL general # 综合排序 MOST_POPULAR popularity_descending # 最热 LATEST time_descending # 最新 class Note(NamedTuple): 笔记数据模型 note_id: str title: str desc: str type: str user: dict img_urls: list video_url: str tag_list: list at_user_list: list collected_count: str comment_count: str liked_count: str share_count: str time: int last_update_time: int5. 实战应用场景市场趋势分析系统通过采集特定品类的笔记数据我们可以分析市场趋势和用户偏好from xhs import XhsClient, SearchSortType, SearchNoteType import pandas as pd from datetime import datetime class MarketTrendAnalyzer: def __init__(self, client): self.client client def analyze_category_trend(self, category, days30): 分析品类趋势 trends [] keywords self.get_category_keywords(category) for keyword in keywords: # 搜索相关笔记 results self.client.search_note( keywordkeyword, sortSearchSortType.MOST_POPULAR, page_size50 ) # 分析数据 trend_data { keyword: keyword, total_notes: len(results.get(items, [])), avg_likes: self.calculate_avg_likes(results), avg_comments: self.calculate_avg_comments(results), trend_score: self.calculate_trend_score(results) } trends.append(trend_data) return pd.DataFrame(trends) def calculate_trend_score(self, results): 计算趋势得分 items results.get(items, []) if not items: return 0 # 基于互动率和发布时间计算趋势得分 score 0 for item in items: likes int(item.get(likes, 0)) comments int(item.get(comments, 0)) time_diff datetime.now().timestamp() - item.get(time, 0) # 互动率权重 时间衰减 engagement_rate (likes comments * 2) / 1000 time_weight 1 / (1 time_diff / 86400) # 24小时衰减 score engagement_rate * time_weight return score / len(items) # 使用示例 analyzer MarketTrendAnalyzer(xhs_client) beauty_trend analyzer.analyze_category_trend(美妆) print(beauty_trend.sort_values(trend_score, ascendingFalse))竞品监测平台定期监测竞品账号的数据变化import schedule import time from xhs import XhsClient class CompetitorMonitor: def __init__(self, client, competitor_ids): self.client client self.competitor_ids competitor_ids self.history_data {} def monitor_competitor(self, user_id): 监测单个竞品账号 try: # 获取用户信息 user_info self.client.get_user_info(user_id) # 获取最新笔记 notes self.client.get_user_notes(user_id) # 计算关键指标 metrics { timestamp: time.time(), fans_count: user_info.get(fans_count, 0), notes_count: user_info.get(notes_count, 0), avg_likes: self.calculate_avg_likes(notes), engagement_rate: self.calculate_engagement_rate(user_info, notes), new_notes: len(notes.get(items, [])) } # 存储历史数据 if user_id not in self.history_data: self.history_data[user_id] [] self.history_data[user_id].append(metrics) return metrics except Exception as e: print(f监测竞品 {user_id} 失败: {e}) return None def calculate_engagement_rate(self, user_info, notes): 计算互动率 fans int(user_info.get(fans_count, 1)) items notes.get(items, []) if not items or fans 0: return 0 total_interactions 0 for item in items[:10]: # 只看最近10条 likes int(item.get(likes, 0)) comments int(item.get(comments, 0)) total_interactions likes comments return total_interactions / (fans * 10) * 100 # 百分比 # 定时监测任务 monitor CompetitorMonitor(xhs_client, [user_id_1, user_id_2]) def daily_monitoring(): for user_id in monitor.competitor_ids: metrics monitor.monitor_competitor(user_id) if metrics: print(f竞品 {user_id} 数据: {metrics}) # 每天执行一次 schedule.every().day.at(09:00).do(daily_monitoring) while True: schedule.run_pending() time.sleep(60)内容质量评估工具通过多维度指标评估内容质量from xhs import XhsClient import numpy as np class ContentQualityAssessor: def __init__(self, client): self.client client def assess_note_quality(self, note_id): 评估笔记质量 try: note self.client.get_note_by_id(note_id) # 计算各项得分 engagement_score self.calculate_engagement_score(note) content_score self.calculate_content_score(note) visual_score self.calculate_visual_score(note) timeliness_score self.calculate_timeliness_score(note) # 加权总分 total_score ( engagement_score * 0.4 content_score * 0.3 visual_score * 0.2 timeliness_score * 0.1 ) return { note_id: note_id, engagement_score: engagement_score, content_score: content_score, visual_score: visual_score, timeliness_score: timeliness_score, total_score: total_score, quality_level: self.get_quality_level(total_score) } except Exception as e: print(f评估笔记 {note_id} 失败: {e}) return None def calculate_engagement_score(self, note): 计算互动得分 likes int(note.get(likes, 0)) comments int(note.get(comments, 0)) collects int(note.get(collects, 0)) shares int(note.get(shares, 0)) # 标准化处理 engagement likes comments * 2 collects * 3 shares * 1.5 return min(engagement / 1000, 1.0) # 归一化到0-1 def get_quality_level(self, score): 根据得分确定质量等级 if score 0.8: return 优质 elif score 0.6: return 良好 elif score 0.4: return 一般 else: return 待提升 # 批量评估示例 assessor ContentQualityAssessor(xhs_client) note_ids [note_id_1, note_id_2, note_id_3] for note_id in note_ids: assessment assessor.assess_note_quality(note_id) if assessment: print(f笔记 {note_id} 质量评估: {assessment})6. 性能优化与最佳实践请求频率控制合理控制请求频率是避免被封IP的关键import time import random from functools import wraps def rate_limited(max_per_minute60): 请求频率限制装饰器 min_interval 60.0 / max_per_minute def decorator(func): last_called [0.0] wraps(func) def wrapper(*args, **kwargs): elapsed time.time() - last_called[0] left_to_wait min_interval - elapsed if left_to_wait 0: # 添加随机延迟模拟人类行为 time.sleep(left_to_wait random.uniform(0.1, 0.5)) result func(*args, **kwargs) last_called[0] time.time() return result return wrapper return decorator class OptimizedXhsClient(XhsClient): 优化版的Xhs客户端 rate_limited(max_per_minute30) # 限制每分钟30次请求 def get_note_by_id(self, note_id: str, xsec_token: str, xsec_source: str pc_feed): 带频率限制的笔记获取 return super().get_note_by_id(note_id, xsec_token, xsec_source) rate_limited(max_per_minute20) # 搜索请求限制更严格 def search_note(self, keyword: str, page: int 1, page_size: int 20, sort: SearchSortType SearchSortType.GENERAL, note_type: SearchNoteType SearchNoteType.ALL): 带频率限制的搜索 return super().search_note(keyword, page, page_size, sort, note_type)数据缓存策略对频繁访问的数据进行缓存减少重复请求import pickle import hashlib from datetime import datetime, timedelta from functools import lru_cache class DataCache: 数据缓存管理器 def __init__(self, cache_dir.xhs_cache, ttl_hours24): self.cache_dir cache_dir self.ttl timedelta(hoursttl_hours) def get_cache_key(self, func_name, *args, **kwargs): 生成缓存键 key_str f{func_name}_{args}_{kwargs} return hashlib.md5(key_str.encode()).hexdigest() def get(self, key): 获取缓存数据 cache_file f{self.cache_dir}/{key}.pkl try: with open(cache_file, rb) as f: data, timestamp pickle.load(f) # 检查是否过期 if datetime.now() - timestamp self.ttl: return data except: pass return None def set(self, key, data): 设置缓存数据 import os os.makedirs(self.cache_dir, exist_okTrue) cache_file f{self.cache_dir}/{key}.pkl with open(cache_file, wb) as f: pickle.dump((data, datetime.now()), f) # 使用缓存的客户端 class CachedXhsClient(XhsClient): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.cache DataCache() lru_cache(maxsize100) def get_note_by_id_cached(self, note_id: str, xsec_token: str): 带内存缓存的笔记获取 cache_key self.cache.get_cache_key(get_note_by_id, note_id, xsec_token) # 先检查磁盘缓存 cached_data self.cache.get(cache_key) if cached_data: return cached_data # 没有缓存则请求数据 data self.get_note_by_id(note_id, xsec_token) # 保存到缓存 self.cache.set(cache_key, data) return data错误处理与重试机制完善的错误处理是生产环境应用的必备import time from xhs.exception import DataFetchError, IPBlockError, SignError class RobustXhsClient(XhsClient): 增强错误处理的客户端 def safe_request(self, func, *args, max_retries3, retry_delay2, **kwargs): 安全的请求包装器 for attempt in range(max_retries): try: return func(*args, **kwargs) except IPBlockError as e: print(fIP被封锁等待{retry_delay * (attempt 1)}秒后重试) time.sleep(retry_delay * (attempt 1)) except SignError as e: print(f签名错误第{attempt 1}次重试: {e}) if attempt max_retries - 1: raise time.sleep(retry_delay) except DataFetchError as e: print(f数据获取错误: {e}) if 404 in str(e) or 笔记不存在 in str(e): return None # 笔记不存在不重试 time.sleep(retry_delay) except Exception as e: print(f未知错误: {e}) if attempt max_retries - 1: raise time.sleep(retry_delay) return None def get_note_safe(self, note_id: str, xsec_token: str): 安全的笔记获取 return self.safe_request( super().get_note_by_id, note_id, xsec_token, max_retries3 ) def search_note_safe(self, keyword: str, **kwargs): 安全的搜索 return self.safe_request( super().search_note, keyword, **kwargs, max_retries2 )7. 生态集成与扩展与数据分析工具集成xhs库可以轻松与主流数据分析工具集成import pandas as pd import matplotlib.pyplot as plt from xhs import XhsClient class XhsDataAnalyzer: 小红书数据分析器 def __init__(self, client): self.client client def collect_user_data(self, user_id, max_notes100): 收集用户数据并转换为DataFrame all_notes [] cursor while len(all_notes) max_notes: result self.client.get_user_notes(user_id, cursorcursor) notes result.get(items, []) if not notes: break all_notes.extend(notes) cursor result.get(cursor, ) if not cursor: break # 转换为DataFrame df pd.DataFrame(all_notes) # 数据清洗和转换 if not df.empty: df[likes] pd.to_numeric(df[likes], errorscoerce).fillna(0) df[comments] pd.to_numeric(df[comments], errorscoerce).fillna(0) df[time] pd.to_datetime(df[time], units, errorscoerce) return df def analyze_engagement_trend(self, user_id): 分析用户互动趋势 df self.collect_user_data(user_id) if df.empty: return None # 按时间分组分析 df.set_index(time, inplaceTrue) monthly_stats df.resample(M).agg({ likes: sum, comments: sum, note_id: count }).rename(columns{note_id: note_count}) return monthly_stats def visualize_engagement(self, user_id): 可视化用户互动数据 stats self.analyze_engagement_trend(user_id) if stats is None: print(没有找到用户数据) return fig, axes plt.subplots(2, 1, figsize(12, 8)) # 互动趋势图 stats[[likes, comments]].plot(axaxes[0]) axes[0].set_title(f用户 {user_id} 互动趋势) axes[0].set_ylabel(互动数量) axes[0].legend([点赞, 评论]) # 发布频率图 stats[note_count].plot(kindbar, axaxes[1]) axes[1].set_title(月度发布数量) axes[1].set_ylabel(笔记数量) axes[1].set_xlabel(月份) plt.tight_layout() plt.show() # 使用示例 analyzer XhsDataAnalyzer(xhs_client) user_stats analyzer.analyze_engagement_trend(target_user_id) print(user_stats)与数据库集成将采集的数据存储到数据库中import sqlite3 from datetime import datetime from xhs import XhsClient class XhsDataStorage: 小红书数据存储管理器 def __init__(self, db_pathxhs_data.db): self.db_path db_path self.init_database() def init_database(self): 初始化数据库表结构 conn sqlite3.connect(self.db_path) cursor conn.cursor() # 创建笔记表 cursor.execute( CREATE TABLE IF NOT EXISTS notes ( note_id TEXT PRIMARY KEY, title TEXT, content TEXT, user_id TEXT, user_name TEXT, likes INTEGER, comments INTEGER, collects INTEGER, shares INTEGER, note_type TEXT, tags TEXT, created_time INTEGER, updated_time INTEGER, crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) # 创建用户表 cursor.execute( CREATE TABLE IF NOT EXISTS users ( user_id TEXT PRIMARY KEY, nickname TEXT, fans_count INTEGER, follows_count INTEGER, notes_count INTEGER, liked_count INTEGER, description TEXT, crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() def save_note(self, note_data): 保存笔记数据 conn sqlite3.connect(self.db_path) cursor conn.cursor() try: cursor.execute( INSERT OR REPLACE INTO notes (note_id, title, content, user_id, user_name, likes, comments, collects, shares, note_type, tags, created_time, updated_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( note_data.get(note_id), note_data.get(title, ), note_data.get(desc, ), note_data.get(user, {}).get(user_id, ), note_data.get(user, {}).get(nickname, ), note_data.get(likes, 0), note_data.get(comments, 0), note_data.get(collects, 0), note_data.get(shares, 0), note_data.get(type, normal), ,.join(note_data.get(tag_list, [])), note_data.get(time, 0), note_data.get(last_update_time, 0) )) conn.commit() return True except Exception as e: print(f保存笔记失败: {e}) return False finally: conn.close() def batch_save_notes(self, notes_list): 批量保存笔记数据 success_count 0 for note in notes_list: if self.save_note(note): success_count 1 print(f成功保存 {success_count}/{len(notes_list)} 条笔记) return success_count # 使用示例 storage XhsDataStorage() client XhsClient() # 搜索并保存笔记 search_results client.search_note(keyword美食探店, page_size50) if search_results and items in search_results: storage.batch_save_notes(search_results[items])Web API服务封装将xhs库封装为RESTful API服务from flask import Flask, request, jsonify from flask_restful import Api, Resource from xhs import XhsClient app Flask(__name__) api Api(app) class XhsAPI: def __init__(self): self.client XhsClient() def get_note(self, note_id): 获取笔记详情 try: note self.client.get_note_by_id(note_id) return { success: True, data: note, message: 获取成功 } except Exception as e: return { success: False, data: None, message: str(e) } def search_notes(self, keyword, page1, page_size20): 搜索笔记 try: result self.client.search_note( keywordkeyword, pagepage, page_sizepage_size ) return { success: True, data: result, message: 搜索成功 } except Exception as e: return { success: False, data: None, message: str(e) } xhs_api XhsAPI() class NoteResource(Resource): def get(self, note_id): 获取笔记详情API return jsonify(xhs_api.get_note(note_id)) class SearchResource(Resource): def get(self): 搜索笔记API keyword request.args.get(keyword, ) page int(request.args.get(page, 1)) page_size int(request.args.get(page_size, 20)) return jsonify(xhs_api.search_notes(keyword, page, page_size)) # 注册API路由 api.add_resource(NoteResource, /api/note/string:note_id) api.add_resource(SearchResource, /api/search) if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)8. 未来发展与社区贡献项目发展方向xhs库作为小红书数据采集的重要工具未来有几个重要的发展方向异步支持集成asyncio和aiohttp提供异步API支持Type Hints完善为所有API添加完整的类型提示更多API覆盖支持评论、私信、直播等更多功能性能优化实现请求池和连接复用插件系统支持自定义插件扩展功能社区贡献指南欢迎开发者参与xhs库的开发和改进# 开发环境设置 git clone https://gitcode.com/gh_mirrors/xh/xhs cd xhs pip install -e .[dev] # 安装开发依赖 # 运行测试 pytest tests/ # 代码规范检查 flake8 xhs/ tests/ black xhs/ tests/ --check最佳实践建议基于项目经验我们总结了一些最佳实践合理控制请求频率避免对小红书服务器造成过大压力使用代理IP轮换在大规模采集时使用代理IP实现数据验证对采集的数据进行有效性验证定期更新cookie处理会话过期问题监控采集状态实现采集任务的监控和报警合规使用提醒在使用xhs库进行数据采集时请务必遵守以下原则尊重robots协议遵守网站的访问规则控制采集频率避免影响网站正常服务保护用户隐私不采集敏感个人信息遵守法律法规确保数据使用合法合规明确使用目的仅用于合法合规的数据分析结语xhs库为小红书数据采集提供了一个强大而灵活的工具链。通过本文的详细介绍开发者能够全面了解xhs库的核心功能、高级用法和最佳实践。无论是进行市场研究、竞品分析还是内容监控xhs库都能帮助你高效获取所需数据。记住技术只是手段真正的价值在于如何利用数据创造洞察。在遵守规则的前提下合理使用xhs库开启你的数据探索之旅吧立即开始pip install xhs深入学习查看example目录下的示例代码遇到问题参考xhs/exception.py中的异常处理指南【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
