小红书数据采集终极指南Python xhs库完整实战教程【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs在当今社交媒体数据分析领域小红书作为中国领先的生活方式分享平台积累了海量有价值的内容数据。对于开发者、数据分析师和市场营销人员来说高效采集这些数据至关重要。xhs库作为一个专业的Python爬虫工具通过封装小红书Web端API接口为技术开发者提供了完整的数据采集解决方案。 为什么需要专业的小红书数据采集工具小红书平台的反爬机制日益严格传统的爬虫方法往往难以稳定工作。xhs库通过模拟浏览器行为和智能签名算法解决了以下核心痛点签名验证绕过处理小红书复杂的x-s签名机制环境检测对抗应对平台的反爬虫检测系统稳定数据获取确保长时间运行的稳定性多账号管理支持批量账号的数据采集需求核心技术原理解析xhs库的核心在于其智能签名系统。小红书使用动态的签名算法来验证请求的合法性xhs通过以下方式实现稳定访问# 签名函数的核心逻辑 def sign(uri, dataNone, a1, web_session): 通过Playwright模拟浏览器环境获取签名 uri: 请求的URI data: 请求数据 a1: 必需的cookie字段 web_session: 会话标识 # 使用stealth.min.js绕过环境检测 # 模拟浏览器执行JavaScript签名函数 # 返回x-s和x-t签名参数 三步快速上手xhs数据采集第一步环境配置与安装# 安装xhs核心库 pip install xhs # 安装Playwright依赖 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) # 测试连接 try: user_info xhs_client.get_self_info() print(f登录成功用户昵称{user_info[nickname]}) except Exception as e: print(f初始化失败{e})第三步核心数据采集实战# 搜索热门笔记 search_results xhs_client.search_note( keywordPython编程, page1, page_size20, sort_typehot # 按热度排序 ) # 解析搜索结果 for note in search_results[items]: note_id note[id] note_detail xhs_client.get_note_by_id(note_id) # 提取关键信息 print(f笔记ID: {note_detail[id]}) print(f标题: {note_detail[title]}) print(f作者: {note_detail[user][nickname]}) print(f点赞数: {note_detail[like_count]}) print(f收藏数: {note_detail[collect_count]}) print(- * 50) xhs库核心功能深度解析1. 内容搜索与过滤系统xhs库提供了强大的搜索功能支持多种过滤条件搜索参数类型说明示例值keywordstr搜索关键词美食探店pageint页码1page_sizeint每页数量20sort_typestr排序方式hot热度note_typestr笔记类型normal图文# 高级搜索示例 results xhs_client.search_note( keyword旅行攻略, page1, page_size15, sort_typehot, note_typenormal )2. 用户数据采集功能# 获取用户基本信息 user_info xhs_client.get_user_info(user_id目标用户ID) # 获取用户发布的笔记列表 user_notes xhs_client.get_user_notes( user_id目标用户ID, page1, page_size30 ) # 获取用户粉丝列表 followers xhs_client.get_followers(user_id目标用户ID)3. 多媒体内容处理from xhs.help import get_imgs_url_from_note, get_video_url_from_note # 获取笔记中的图片URL note_data xhs_client.get_note_by_id(笔记ID) image_urls get_imgs_url_from_note(note_data) # 获取视频URL video_url get_video_url_from_note(note_data) # 下载媒体内容 for idx, img_url in enumerate(image_urls): download_file(img_url, fnote_images/image_{idx}.jpg)️ 高级配置与反爬策略签名服务器部署方案对于生产环境建议使用独立的签名服务器# 签名服务器配置示例 def setup_sign_server(): 部署签名服务器支持多账号并发 # 使用Docker快速部署 # docker run -it -d -p 5005:5005 reajason/xhs-api:latest # 或者本地部署Flask服务 from flask import Flask, request, jsonify import requests app Flask(__name__) app.route(/sign, methods[POST]) def sign_endpoint(): data request.json # 处理签名逻辑 return jsonify({x-s: signature, x-t: timestamp}) return app智能请求频率控制import time import random from datetime import datetime class SmartRequestController: 智能请求控制器 def __init__(self, base_delay2.0, jitter1.0): self.base_delay base_delay self.jitter jitter self.last_request_time None def wait_if_needed(self): 智能等待避免频繁请求 if self.last_request_time: elapsed (datetime.now() - self.last_request_time).total_seconds() if elapsed self.base_delay: sleep_time self.base_delay - elapsed random.uniform(0, self.jitter) time.sleep(sleep_time) self.last_request_time datetime.now() def make_request(self, api_call): 执行API调用 self.wait_if_needed() return api_call() 实战应用场景分析场景一竞品分析与市场调研def analyze_competitor(keywords, days30): 竞品分析监控特定关键词的热度变化 trends_data {} for keyword in keywords: # 搜索相关笔记 notes xhs_client.search_note( keywordkeyword, page1, page_size50 ) # 分析数据趋势 trend_analysis { total_notes: len(notes[items]), avg_likes: sum(n[like_count] for n in notes[items]) / len(notes[items]), top_authors: [n[user][nickname] for n in notes[items][:5]] } trends_data[keyword] trend_analysis return trends_data场景二内容创作趋势发现def discover_content_trends(category美妆): 发现特定类别的内容趋势 # 获取分类推荐内容 from xhs.core import FeedType feed_type_map { 美妆: FeedType.COSMETICS, 穿搭: FeedType.FASION, 美食: FeedType.FOOD, 旅行: FeedType.TRAVEL } if category in feed_type_map: feed_data xhs_client.get_home_feed(feed_typefeed_type_map[category]) # 分析热门话题 trending_topics [] for item in feed_data[items][:20]: if title in item: trending_topics.append({ title: item[title], interaction: item.get(like_count, 0) item.get(collect_count, 0) }) return sorted(trending_topics, keylambda x: x[interaction], reverseTrue)⚠️ 常见问题与解决方案问题1签名失败错误处理from xhs.exception import SignError, DataFetchError def safe_api_call(api_func, max_retries3): 安全的API调用封装包含重试逻辑 for attempt in range(max_retries): try: return api_func() except SignError as e: print(f签名失败第{attempt1}次重试: {e}) time.sleep(2 ** attempt) # 指数退避 except DataFetchError as e: print(f数据获取失败: {e}) break raise Exception(fAPI调用失败重试{max_retries}次后仍无法成功)问题2Cookie失效管理class CookieManager: Cookie管理器 def __init__(self): self.cookies [] self.current_index 0 def add_cookie(self, cookie): 添加新的Cookie self.cookies.append(cookie) def get_valid_cookie(self): 获取有效的Cookie if not self.cookies: raise Exception(没有可用的Cookie) # 简单的轮询策略 cookie self.cookies[self.current_index] self.current_index (self.current_index 1) % len(self.cookies) return cookie def validate_cookie(self, cookie): 验证Cookie是否有效 try: test_client XhsClient(cookie) test_client.get_self_info() return True except: return False 性能优化与最佳实践1. 连接池与会话复用import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_optimized_session(): 创建优化的请求会话 session requests.Session() # 配置重试策略 retry_strategy Retry( total3, backoff_factor1, status_forcelist[429, 500, 502, 503, 504] ) adapter HTTPAdapter( max_retriesretry_strategy, pool_connections10, pool_maxsize100 ) session.mount(http://, adapter) session.mount(https://, adapter) return session2. 数据存储与缓存策略import sqlite3 import json from datetime import datetime class DataStorage: 数据存储管理器 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 notes ( id TEXT PRIMARY KEY, title TEXT, content TEXT, user_id TEXT, like_count INTEGER, collect_count INTEGER, comment_count INTEGER, created_at TIMESTAMP, crawled_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, raw_data TEXT ) ) # 用户数据表 cursor.execute( CREATE TABLE IF NOT EXISTS users ( id TEXT PRIMARY KEY, nickname TEXT, avatar TEXT, description TEXT, notes_count INTEGER, fans_count INTEGER, follows_count INTEGER, crawled_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) self.conn.commit() 高级技巧二维码登录与多账号管理二维码登录实现import qrcode from xhs import XhsClient def qrcode_login(): 二维码登录获取Cookie xhs_client XhsClient() # 获取二维码 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_size10, border4) qr.add_data(qr_res[url]) qr.make(fitTrue) # 在终端显示二维码 qr.print_ascii() print(请使用小红书APP扫描上方二维码登录) # 轮询登录状态 while True: check_result xhs_client.check_qrcode(qr_id, qr_code) if check_result[code_status] 2: # 登录成功 login_info check_result[login_info] cookie xhs_client.cookie print(f登录成功当前Cookie: {cookie[:50]}...) return cookie time.sleep(1)多账号批量管理class MultiAccountManager: 多账号管理器 def __init__(self): self.accounts {} self.task_queue [] def add_account(self, name, cookie): 添加账号 self.accounts[name] { cookie: cookie, client: XhsClient(cookie), last_used: datetime.now() } def get_account_for_task(self, task_type): 根据任务类型选择合适的账号 # 简单的轮询策略 available_accounts list(self.accounts.keys()) # 可以根据账号使用频率、成功率等指标优化选择策略 selected available_accounts[0] self.accounts[selected][last_used] datetime.now() return self.accounts[selected][client] 扩展学习与进阶资源官方文档与示例代码核心模块文档xhs/core.py 包含所有API接口定义辅助函数文档xhs/help.py 提供数据处理工具函数错误处理文档xhs/exception.py 定义所有异常类型实战示例项目项目中提供了多个完整的示例代码基础使用示例example/basic_usage.py二维码登录示例example/login_qrcode.py签名服务器示例example/basic_sign_server.py签名使用示例example/basic_sign_usage.py测试与验证项目包含完整的测试套件位于tests/目录下可以帮助你验证功能正确性# 运行测试 python -m pytest tests/ 未来发展与社区贡献xhs库作为一个开源项目欢迎开发者贡献代码和提出改进建议。主要发展方向包括性能优化进一步提升数据采集效率功能扩展支持更多小红书API接口稳定性提升增强反爬对抗能力文档完善提供更详细的使用指南和API文档通过本指南你已经掌握了使用xhs库进行小红书数据采集的核心技能。无论是进行市场调研、竞品分析还是内容创作研究xhs都能为你提供稳定可靠的数据支持。记住合理使用工具遵守平台规则让数据采集成为你工作的有力助手。【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考