当前位置： 首页 > news >正文

高效小红书数据采集实战指南：xhs工具完全解析

news 2026/7/3 16:41:58

高效小红书数据采集实战指南：xhs工具完全解析

【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs

小红书作为中国领先的生活方式分享平台，蕴藏着海量的用户行为数据和消费趋势信息。对于开发者和数据分析师而言，如何高效、合规地采集这些数据成为了技术挑战。xhs工具作为基于小红书Web端的Python请求封装库，提供了专业的数据采集解决方案。本文将深入探索xhs工具的技术架构、核心功能实现和实际应用场景，揭秘小红书数据采集的高效实践。

🔧 环境配置三步走：快速搭建采集环境

安装部署指南

xhs工具支持Python 3.7及以上版本，通过pip即可一键安装：

pip install xhs

如需获取最新开发版本，可以直接从GitCode仓库克隆安装：

git clone https://gitcode.com/gh_mirrors/xh/xhs cd xhs python setup.py install

依赖环境检查

项目依赖文件requirements.txt中列出了所有必要组件，核心依赖包括：

  • requests：网络请求库，处理HTTP通信
  • pycryptodome：加密处理库，用于请求签名
  • playwright：浏览器自动化工具，处理复杂交互
  • lxml：HTML/XML解析库，提取结构化数据

验证安装结果

安装完成后，可以通过简单的导入测试验证安装是否成功：

from xhs import XhsClient print("xhs工具安装成功，版本：", XhsClient.__version__)

🔐 安全认证机制：多方式登录实战

二维码登录实现

xhs工具提供了便捷的二维码登录功能，通过example/login_qrcode.py可以快速实现自动化登录：

from xhs import XhsClient import qrcode # 初始化客户端 xhs_client = XhsClient(sign=sign) # 获取登录二维码 qr_res = xhs_client.get_qrcode() qr_id = qr_res["qr_id"] qr_code = qr_res["code"] # 生成二维码图片 qr = qrcode.QRCode() qr.add_data(qr_res["url"]) qr.make() qr.print_ascii() # 轮询检查登录状态 while True: check_result = xhs_client.check_qrcode(qr_id, qr_code) if check_result["code_status"] == 2: login_info = check_result["login_info"] print("登录成功！Cookie:", xhs_client.cookie) break

手机号验证码登录

对于需要自动化登录的场景，example/login_phone.py提供了手机号验证码登录方案：

xhs_client = XhsClient() phone = "13800138000" # 获取验证码 token = xhs_client.get_login_code(phone) # 用户输入验证码后登录 verify_code = input("请输入验证码：") login_res = xhs_client.login_code(phone, token, verify_code)

签名机制深度解析

xhs工具的核心安全机制在于请求签名，xhs/core.py中实现了完整的签名逻辑：

def sign(uri, data=None, a1="", web_session=""): """ 生成小红书请求签名 :param uri: 请求URI :param data: 请求数据 :param a1: 认证cookie :param web_session: 会话标识 :return: 签名结果字典 """ # 实际签名逻辑通过浏览器环境执行 encrypt_params = context_page.evaluate( "([url, data]) => window._webmsxyw(url, data)", [uri, data] ) return { "x-s": encrypt_params["X-s"], "x-t": str(encrypt_params["X-t"]) }

📊 核心功能实战：数据采集与解析

笔记内容获取

xhs工具提供了完整的笔记数据获取接口，支持多种内容类型：

from xhs import XhsClient, NoteType # 获取单篇笔记详情 note_id = "6505318c000000001f03c5a6" note = xhs_client.get_note_by_id(note_id) # 解析笔记内容 print(f"标题：{note['title']}") print(f"作者：{note['user']['nickname']}") print(f"点赞数：{note['likes']}") print(f"收藏数：{note['collects']}") # 获取笔记类型 note_type = NoteType.NORMAL if note['type'] == 'normal' else NoteType.VIDEO

内容搜索功能

支持关键词搜索和多种筛选条件，满足不同场景的数据采集需求：

from xhs import SearchSortType, SearchNoteType # 关键词搜索 search_result = xhs_client.get_note_by_keyword( keyword="Python编程", page=1, page_size=20, sort=SearchSortType.GENERAL, # 综合排序 note_type=SearchNoteType.VIDEO # 视频类型 ) # 处理搜索结果 for note in search_result["items"]: print(f"笔记ID：{note['id']}") print(f"标题：{note['title']}") print(f"摘要：{note['desc']}")

用户信息采集

获取用户公开信息和发布内容：

# 获取用户信息 user_id = "5f8c9b1e0000000001000000" user_info = xhs_client.get_user_info(user_id) # 获取用户发布的笔记 user_notes = xhs_client.get_user_notes( user_id=user_id, page=1, page_size=20 ) print(f"用户昵称：{user_info['nickname']}") print(f"粉丝数：{user_info['fans']}") print(f"获赞数：{user_info['likes']}")

⚙️ 高级配置技巧：优化采集策略

请求参数定制

通过调整xhs/core.py中的请求头配置，可以模拟不同设备的访问特征：

# 自定义请求头 custom_headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://www.xiaohongshu.com/", "Accept-Language": "zh-CN,zh;q=0.9" } # 初始化客户端时传入自定义headers xhs_client = XhsClient( cookie=cookie, sign=sign, headers=custom_headers, proxies={"http": "http://127.0.0.1:7890", "https": "http://127.0.0.1:7890"} )

异常处理机制

xhs工具内置了完善的异常处理体系，定义在xhs/exception.py中：

from xhs.exception import DataFetchError, IPBlockError, SignError try: note = xhs_client.get_note_by_id(note_id) except DataFetchError as e: print(f"数据获取失败：{e}") except IPBlockError as e: print(f"IP被限制：{e}") # 建议更换代理或等待一段时间 except SignError as e: print(f"签名失败：{e}") # 需要重新获取签名

数据存储策略

采集到的数据建议采用结构化存储方案：

import json import csv from datetime import datetime def save_note_data(note, format="json"): """保存笔记数据""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") if format == "json": filename = f"note_{note['id']}_{timestamp}.json" with open(filename, 'w', encoding='utf-8') as f: json.dump(note, f, ensure_ascii=False, indent=2) elif format == "csv": filename = f"notes_{timestamp}.csv" fieldnames = ['id', 'title', 'user_id', 'likes', 'collects', 'comments', 'timestamp'] with open(filename, 'a', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writerow({ 'id': note['id'], 'title': note['title'], 'user_id': note['user']['user_id'], 'likes': note['likes'], 'collects': note['collects'], 'comments': note['comments'], 'timestamp': timestamp })

🔄 实战应用场景：从数据采集到业务洞察

市场趋势分析

通过采集特定领域的内容数据，分析市场趋势和用户偏好：

def analyze_trend(keywords, days=30): """分析关键词趋势""" trends = {} for keyword in keywords: # 采集最近30天的相关笔记 notes = xhs_client.get_note_by_keyword( keyword=keyword, page=1, page_size=100 ) # 分析数据趋势 total_likes = sum(note['likes'] for note in notes['items']) avg_likes = total_likes / len(notes['items']) if notes['items'] else 0 trends[keyword] = { 'total_notes': len(notes['items']), 'total_likes': total_likes, 'avg_likes': avg_likes, 'top_authors': get_top_authors(notes['items']) } return trends

竞品监控系统

构建自动化竞品监控系统，实时跟踪竞争对手动态：

class CompetitorMonitor: def __init__(self, competitor_ids): self.competitor_ids = competitor_ids self.xhs_client = XhsClient() def monitor_new_content(self): """监控新发布内容""" new_contents = [] for user_id in self.competitor_ids: latest_notes = self.xhs_client.get_user_notes( user_id=user_id, page=1, page_size=10 ) for note in latest_notes['items']: if self.is_new_content(note): new_contents.append({ 'competitor': user_id, 'note': note, 'timestamp': datetime.now() }) return new_contents def analyze_engagement(self): """分析互动数据""" engagement_stats = {} for user_id in self.competitor_ids: notes = self.xhs_client.get_user_notes(user_id, page=1, page_size=50) stats = self.calculate_engagement(notes['items']) engagement_stats[user_id] = stats return engagement_stats

内容质量评估

基于采集的数据评估内容质量和用户偏好：

def evaluate_content_quality(note_data): """评估内容质量""" quality_score = 0 # 互动指标权重 weights = { 'likes': 0.4, 'collects': 0.3, 'comments': 0.2, 'shares': 0.1 } # 标准化处理 max_values = get_max_values_from_dataset() for metric, weight in weights.items(): if metric in note_data: normalized_value = note_data[metric] / max_values.get(metric, 1) quality_score += normalized_value * weight # 内容长度加分 if 'desc' in note_data and len(note_data['desc']) > 100: quality_score += 0.1 return min(quality_score, 1.0) # 归一化到0-1

🚀 性能优化与最佳实践

请求频率控制

合理控制请求频率，避免触发平台限制：

import time from random import uniform class RateLimitedClient: def __init__(self, base_client, requests_per_minute=30): self.client = base_client self.requests_per_minute = requests_per_minute self.min_interval = 60 / requests_per_minute self.last_request_time = 0 def safe_request(self, method, *args, **kwargs): """安全的请求方法，自动控制频率""" current_time = time.time() elapsed = current_time - self.last_request_time if elapsed < self.min_interval: sleep_time = self.min_interval - elapsed + uniform(0.1, 0.5) time.sleep(sleep_time) result = method(*args, **kwargs) self.last_request_time = time.time() return result

数据缓存策略

实现数据缓存，减少重复请求：

import hashlib import pickle from pathlib import Path class DataCache: def __init__(self, cache_dir=".xhs_cache"): self.cache_dir = Path(cache_dir) self.cache_dir.mkdir(exist_ok=True) 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, max_age_hours=24): """获取缓存数据""" cache_file = self.cache_dir / f"{key}.pkl" if cache_file.exists(): file_age = time.time() - cache_file.stat().st_mtime if file_age < max_age_hours * 3600: with open(cache_file, 'rb') as f: return pickle.load(f) return None def set(self, key, data): """设置缓存数据""" cache_file = self.cache_dir / f"{key}.pkl" with open(cache_file, 'wb') as f: pickle.dump(data, f)

分布式采集架构

对于大规模数据采集需求，可以构建分布式架构：

import redis from multiprocessing import Pool class DistributedCollector: def __init__(self, redis_host='localhost', redis_port=6379): self.redis = redis.Redis(host=redis_host, port=redis_port) self.task_queue = "xhs:collect:tasks" self.result_queue = "xhs:collect:results" def distribute_tasks(self, user_ids, batch_size=100): """分发采集任务""" for i in range(0, len(user_ids), batch_size): batch = user_ids[i:i+batch_size] task_data = { 'batch_id': f"batch_{i//batch_size}", 'user_ids': batch, 'timestamp': time.time() } self.redis.rpush(self.task_queue, json.dumps(task_data)) def worker_process(self, worker_id): """工作进程处理任务""" while True: task_json = self.redis.blpop(self.task_queue, timeout=30) if task_json: task = json.loads(task_json[1]) results = self.process_batch(task['user_ids']) # 存储结果 result_key = f"xhs:results:{task['batch_id']}" self.redis.set(result_key, json.dumps(results)) self.redis.rpush(self.result_queue, result_key)

📈 合规采集与风险控制

遵守平台规则

在使用xhs工具进行数据采集时，务必遵守小红书平台的使用规范：

  1. 尊重robots协议：检查目标网站的robots.txt文件
  2. 控制请求频率：避免对服务器造成过大压力
  3. 仅采集公开数据：不采集用户隐私信息
  4. 遵守使用条款：严格遵守小红书用户协议

数据使用规范

采集的数据应在法律允许的范围内使用：

class EthicalCollector: def __init__(self): self.collected_data = [] self.privacy_fields = ['phone', 'email', 'id_card', 'address'] def filter_sensitive_data(self, data): """过滤敏感信息""" filtered_data = data.copy() for field in self.privacy_fields: if field in filtered_data: del filtered_data[field] return filtered_data def anonymize_user_data(self, user_data): """匿名化用户数据""" if 'user' in user_data: user_data['user']['user_id'] = f"user_{hash(user_data['user']['user_id']) % 10000}" if 'nickname' in user_data['user']: user_data['user']['nickname'] = "匿名用户" return user_data

🎯 总结与展望

xhs工具为小红书数据采集提供了完整的技术解决方案，从环境配置到高级应用，涵盖了数据采集的全流程。通过本文的实战指南，开发者可以快速掌握：

  1. 环境搭建：快速部署采集环境
  2. 认证机制：掌握多种登录方式
  3. 数据采集：实现高效内容获取
  4. 高级应用：构建业务分析系统
  5. 合规实践：确保采集过程合法合规

随着小红书平台的不断发展和数据价值的持续提升，xhs工具将持续优化和更新，为开发者和数据分析师提供更强大、更稳定的数据采集能力。无论是市场研究、竞品分析还是用户行为洞察，xhs工具都能成为您数据驱动决策的重要技术支撑。

在实际应用中，建议结合具体业务需求，灵活运用xhs工具的各项功能，同时始终保持对平台规则的尊重和对用户隐私的保护，实现技术与伦理的平衡发展。

【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

http://www.cnnetsun.cn/news/2155716.html

相关文章：

  • BTW：AI开发工作流管理器，统一配置提升编码效率
  • ASPO算法：解决LLM强化学习中IS比率失衡问题
  • 三步深度解析KKManager：Illusion游戏模组管理实战指南
  • Universal x86 Tuning Utility：开源硬件调优引擎的技术深度解析与实践指南
  • 从‘搬运工’到‘魔术师’：用SeaTunnel和Flink CDC玩转实时数据同步与转换（附避坑配置）
  • 逆向工程AI创业公司Magic的长上下文处理技术
  • 基于大语言模型构建个人AI助手：从智能体架构到实战部署
  • 抖音直播数据采集实战：从网页端API到实时弹幕分析
  • 保姆级教程：在Ubuntu20.04 ROS Noetic上，从零配置laser_scan_matcher搭配GMapping建图（解决csm依赖报错）
  • TranslucentTB在Windows 11更新后无法启动？3步排查+5种修复方案
  • GitHub中文插件：3分钟让GitHub界面全面中文化的终极解决方案
  • ChatGPT平替方案：基于LM Z-Image构建私有化智能对话助手
  • 如何快速解锁你的微信聊天记录：WechatDecrypt本地解密完整指南
  • 智能文献助手Zotero GPT：3大核心功能深度解析与实战指南
  • 多智能体任务编排框架：从原理到实践，构建复杂AI工作流
  • 思源宋体CN：开源专业字体如何改变你的设计工作流？
  • Go微服务高可用实战：基于gobreaker的熔断器与自适应限流深度实践
  • SRWE终极指南：5分钟掌握实时窗口分辨率控制技术
  • Fast-GitHub终极指南：一键解决国内GitHub访问慢的免费浏览器插件
  • 如何在Blender中导入MMD模型：MMD Tools插件完整教程
  • YOLO26-seg分割优化：注意力魔改 | SimAM（无参Attention），一种轻量级的自注意力机制，效果秒杀CBAM、SE
  • 协程泄漏、心跳超时、流式响应中断——Swoole+LLM长连接三大报错全解析，附可落地的监控熔断脚本
  • 为什么你的AI Sandbox永远“半隔离”？——深度拆解Linux命名空间缺陷、GPU共享陷阱与3种绕过检测的隐蔽行为
  • 多模态代码生成技术：从设计草图到可执行代码的自动化实践
  • LLaMA-Factory结合DPO实现偏好对齐（RLHF简化方案）-实战落地指南
  • 2026年权威披露：杭州GEO优化源头服务商怎么挑选？亲测对比AI搜索优化公司避坑攻略
  • Downkyi：5步掌握B站视频下载的终极秘籍
  • 谷歌收录老是不见涨？翻开GSC后台看这几个红柱子，每天200个精准流量这样找回来
  • 【技术应用】PLA技术“点亮”蛋白互作，破解动脉粥样硬化新机制！
  • 深入解析高性能直播录制技术：StreamCap架构设计与实现