小红书数据采集终极指南Python爬虫快速上手实战【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs小红书作为国内领先的社交电商平台蕴含着海量的用户行为数据和内容趋势信息。对于数据分析师和开发者来说如何高效、稳定地获取这些数据成为了技术挑战。xhs工具正是为解决这一问题而生的Python爬虫库它封装了小红书Web端API让数据采集变得简单高效。 为什么xhs是小红书数据采集的最佳选择在众多数据采集方案中xhs以其专业性和易用性脱颖而出特性xhs工具传统爬虫方案优势对比认证机制内置签名算法手动处理Cookie自动化程度高稳定性强反爬应对集成PlaywrightStealth.js需要自行实现开箱即用维护成本低功能完整性50个API方法零散功能实现功能全面持续更新开发效率几行代码完成采集数百行代码开发开发周期缩短90%核心优势解析签名算法自动化小红书的反爬机制核心在于x-s签名xhs通过Playwright模拟浏览器环境自动生成签名无需手动破解算法。多登录方式支持支持二维码登录、手机验证码登录、Cookie登录等多种认证方式适应不同使用场景。完整的数据接口从笔记搜索到用户信息从评论数据到创作者功能覆盖小红书90%以上的公开数据接口。 5分钟快速部署指南环境准备与安装首先确保你的Python环境版本≥3.8然后执行以下命令# 安装xhs核心库 pip install xhs # 安装浏览器模拟依赖 pip install playwright playwright install # 下载反检测脚本 curl -O https://cdn.jsdelivr.net/gh/requireCool/stealth.min.js/stealth.min.js基础使用从零到第一个数据请求from xhs import XhsClient import json # 初始化客户端需要先获取有效的Cookie cookie your_xhs_cookie_here xhs_client XhsClient(cookie) # 获取当前登录用户信息 user_info xhs_client.get_self_info() print(f用户昵称{user_info[nickname]}) print(f粉丝数量{user_info[fans]}) print(f获赞总数{user_info[likes]})获取有效Cookie的两种方法方法一手动获取适合临时使用登录小红书网页版打开浏览器开发者工具F12进入Network标签页刷新页面找到任意请求复制Request Headers中的Cookie值方法二二维码自动登录适合自动化from xhs import XhsClient # 生成登录二维码 client XhsClient() qr_info client.get_qrcode() print(f请扫描二维码{qr_info[qrcode_url]}) # 轮询检查登录状态 while True: check_result client.check_qrcode(qr_info[qr_id], qr_info[code]) if check_result[code_status] 2: # 登录成功 print(登录成功) break 实战应用四大核心数据采集场景场景一笔记搜索与趋势分析# 搜索热门美食笔记 search_results xhs_client.get_note_by_keyword( keyword美食探店, page1, page_size20, sorthot # 按热度排序 ) # 分析热门笔记特征 hot_notes [] for note in search_results[items]: note_data { title: note[title], author: note[user][nickname], likes: note[like_count], collects: note[collect_count], comments: note[comment_count], publish_time: note[time] } hot_notes.append(note_data) print(f 热门笔记{note_data[title]}点赞{note_data[likes]})场景二用户画像分析# 获取用户基本信息 user_id 目标用户ID user_info xhs_client.get_user_info(user_id) # 获取用户所有笔记 all_notes xhs_client.get_user_all_notes( user_iduser_id, crawl_interval2 # 爬取间隔2秒避免频率过高 ) # 分析用户内容偏好 note_categories {} for note in all_notes: # 提取关键词或分类标签 # 这里可以根据笔记内容进行分类分析 category analyze_note_category(note) note_categories[category] note_categories.get(category, 0) 1 print(f用户内容分布{note_categories})场景三评论情感分析# 获取笔记所有评论 note_id 目标笔记ID all_comments xhs_client.get_note_all_comments( note_idnote_id, crawl_interval1 # 每秒获取一次 ) # 分析评论情感倾向 positive_comments [] negative_comments [] neutral_comments [] for comment in all_comments: sentiment analyze_sentiment(comment[content]) if sentiment positive: positive_comments.append(comment) elif sentiment negative: negative_comments.append(comment) else: neutral_comments.append(comment) print(f 正面评论{len(positive_comments)}条) print(f 负面评论{len(negative_comments)}条) print(f 中性评论{len(neutral_comments)}条)场景四竞品监控系统import schedule import time from datetime import datetime def monitor_competitor(user_id): 监控竞争对手的最新动态 try: # 获取用户最新笔记 user_notes xhs_client.get_user_notes(user_id) latest_note user_notes[0] if user_notes else None if latest_note: # 检查是否为新发布的笔记 note_time datetime.fromtimestamp(latest_note[time]/1000) if (datetime.now() - note_time).days 1: print(f 竞争对手发布新笔记{latest_note[title]}) print(f 发布时间{note_time.strftime(%Y-%m-%d %H:%M:%S)}) print(f 互动数据{latest_note[like_count]} ⭐{latest_note[collect_count]}) # 发送通知邮件/钉钉/微信等 send_notification(latest_note) except Exception as e: print(f监控失败{e}) # 定时监控每小时执行一次 schedule.every().hour.do(monitor_competitor, 竞争对手用户ID) while True: schedule.run_pending() time.sleep(1)⚡ 高级技巧提升采集效率与稳定性签名服务部署方案对于大规模数据采集需求建议部署独立的签名服务# 参考示例example/basic_sign_server.py # 将签名服务部署为独立的Flask应用 # 支持多客户端并发请求提高效率部署命令# 使用Docker快速部署 docker run -it -d -p 5005:5005 reajason/xhs-api:latest智能请求频率控制import random import time from functools import wraps def rate_limit(min_delay1, max_delay3): 请求频率限制装饰器 def decorator(func): wraps(func) def wrapper(*args, **kwargs): # 随机延迟模拟人类操作 delay random.uniform(min_delay, max_delay) time.sleep(delay) try: return func(*args, **kwargs) except Exception as e: # 请求失败时增加等待时间 print(f请求失败等待5秒后重试{e}) time.sleep(5) return func(*args, **kwargs) return wrapper return decorator # 使用装饰器 rate_limit(min_delay1, max_delay3) def safe_search(keyword, page1): return xhs_client.get_note_by_keyword(keyword, pagepage)数据持久化存储方案import sqlite3 import json from datetime import datetime 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, like_count INTEGER, collect_count INTEGER, comment_count INTEGER, publish_time TIMESTAMP, category TEXT, tags TEXT, raw_data TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) # 用户数据表 cursor.execute( CREATE TABLE IF NOT EXISTS users ( user_id TEXT PRIMARY KEY, nickname TEXT, fans_count INTEGER, likes_count INTEGER, notes_count INTEGER, collected_count INTEGER, raw_data TEXT, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() def save_note(self, note_data): 保存笔记数据 conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( INSERT OR REPLACE INTO notes VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( note_data[note_id], note_data.get(title, ), note_data.get(desc, ), note_data[user][user_id], note_data.get(like_count, 0), note_data.get(collect_count, 0), note_data.get(comment_count, 0), datetime.fromtimestamp(note_data[time]/1000), note_data.get(category, ), ,.join(note_data.get(tags, [])), json.dumps(note_data), datetime.now() )) conn.commit() conn.close() 企业级应用架构设计分布式采集系统架构┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ 任务调度中心 │ │ 签名服务集群 │ │ 数据存储层 │ │ (Celery) │───▶│ (Flask) │───▶│ (MySQL/ES) │ └─────────────────┘ └─────────────────┘ └─────────────────┘ │ │ │ ▼ ▼ ▼ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ 采集工作节点 │ │ 代理IP池 │ │ 数据清洗 │ │ (Worker) │ │ (Proxy) │ │ (ETL) │ └─────────────────┘ └─────────────────┘ └─────────────────┘监控与告警系统import logging from prometheus_client import Counter, Histogram from datetime import datetime # 定义监控指标 REQUEST_COUNT Counter(xhs_request_total, Total requests) REQUEST_DURATION Histogram(xhs_request_duration_seconds, Request duration) ERROR_COUNT Counter(xhs_error_total, Total errors) class MonitoredXhsClient: 带监控的xhs客户端 def __init__(self, cookie): self.client XhsClient(cookie) self.logger logging.getLogger(__name__) REQUEST_DURATION.time() def get_note_by_keyword(self, keyword, **kwargs): 监控包装的搜索方法 REQUEST_COUNT.inc() try: start_time datetime.now() result self.client.get_note_by_keyword(keyword, **kwargs) duration (datetime.now() - start_time).total_seconds() self.logger.info(f搜索成功{keyword}耗时{duration:.2f}s) return result except Exception as e: ERROR_COUNT.inc() self.logger.error(f搜索失败{keyword}错误{e}) raise❓ 常见问题与解决方案Q1签名失败怎么办A确保Cookie中的a1字段有效并检查stealth.min.js是否正确加载。可以尝试更新Cookie值增加签名前的等待时间检查浏览器环境是否正常Q2请求频率过高被限制A实现智能延迟策略# 在请求间添加随机延迟 import time import random def smart_request(func, *args, **kwargs): time.sleep(random.uniform(1, 5)) # 1-5秒随机延迟 return func(*args, **kwargs)Q3如何获取用户所有笔记A使用get_user_all_notes方法注意设置合理的爬取间隔all_notes xhs_client.get_user_all_notes( user_id用户ID, crawl_interval2 # 每2秒获取一次避免过快 )Q4数据如何存储和分析A建议使用数据库存储原始数据再进行分析SQLite适合小规模数据MySQL/PostgreSQL适合企业级应用Elasticsearch适合全文搜索和实时分析 学习资源与进阶指南官方文档与示例基础使用文档docs/basic.rst爬虫进阶指南docs/crawl.rst创作者功能说明docs/creator.rst完整示例代码example/目录下的各种使用示例项目源码结构解析xhs/ ├── core.py # 核心API实现包含50个数据接口 ├── help.py # 辅助函数提供数据处理工具 ├── exception.py # 异常处理机制 └── __init__.py # 模块导出和初始化 example/ # 实战示例 ├── basic_usage.py # 基础使用示例 ├── basic_sign_server.py # 签名服务部署 ├── login_phone.py # 手机登录示例 └── login_qrcode.py # 二维码登录示例最佳实践建议遵守平台规则控制请求频率避免对服务器造成压力数据合规使用仅用于合法用途尊重用户隐私定期更新维护关注项目更新及时适配API变化错误处理完善实现完整的重试和降级机制 开始你的数据探索之旅xhs工具为小红书数据采集提供了完整的技术解决方案。无论你是进行市场研究、竞品分析还是开发数据驱动的应用这个工具都能为你节省大量开发时间。立即开始安装xhs工具pip install xhs参考示例代码example/basic_usage.py构建你的第一个数据采集脚本探索更多高级功能记住技术只是工具合理、合规地使用数据才是关键。在享受技术便利的同时请始终遵守相关法律法规和平台规则。提示建议先从简单的数据采集任务开始逐步扩展到复杂的业务场景。遇到问题时可以参考项目文档或社区讨论。祝你数据采集顺利【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考