Python小红书数据采集：5个核心技巧构建高效爬虫系统

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

在社交媒体数据分析领域，小红书作为中国领先的生活方式分享平台，蕴含着巨大的商业价值。对于数据分析师、市场研究员和内容运营者来说，如何高效、稳定地采集小红书数据成为了关键需求。xhs项目提供了一个完整的Python解决方案，让你能够轻松获取小红书笔记、用户信息和搜索数据，为数据驱动决策提供支持。

项目概述与价值主张

xhs是一个基于小红书Web端API封装的Python库，专门用于小红书数据采集和分析。相比于传统的爬虫方案，xhs提供了更稳定、更高效的接口调用方式，避免了频繁的页面解析和反爬虫对抗。

核心价值点：

• 🚀 官方API级别的稳定性
• 🔒 内置签名验证机制
• 📊 支持多种数据类型的获取
• ⚡ 异步请求和批量处理能力
• 🛡️ 完善的错误处理和重试机制

核心技术架构解析

xhs的核心架构基于请求签名机制，这是小红书API安全验证的关键环节。让我们深入了解其技术实现：

签名机制的工作原理

签名验证流程图

xhs使用Playwright模拟浏览器环境来获取签名参数，这是绕过小红书反爬虫系统的关键。签名流程包含以下步骤：

1. 环境初始化：启动无头浏览器并加载stealth.min.js绕过检测
2. Cookie注入：将用户cookie注入浏览器环境
3. 签名计算：调用JavaScript函数生成x-s和x-t参数
4. 请求构造：将签名参数附加到API请求中

核心模块功能对比

模块名称	主要功能	使用场景	性能特点
XhsClient	主客户端类	所有API调用	线程安全，支持代理
help.py	工具函数集合	数据处理和转换	纯Python实现，无依赖
xhs-api	签名服务器	生产环境部署	Docker支持，多实例负载均衡

主要API接口

from xhs import XhsClient, FeedType, SearchSortType

# 初始化客户端
client = XhsClient(cookie="your_cookie", sign=sign_func)

# 获取笔记详情
note = client.get_note_by_id("6505318c000000001f03c5a6")

# 搜索用户
users = client.get_user_by_keyword("Python", page=1)

# 获取推荐feed
feed = client.get_home_feed(FeedType.RECOMMEND)

快速上手指南

环境配置步骤

首先，安装必要的依赖包：

# 安装xhs库
pip install xhs

# 安装Playwright（用于签名）
pip install playwright
playwright install chromium

# 下载反检测脚本
curl -O https://cdn.jsdelivr.net/gh/requireCool/stealth.min.js/stealth.min.js

Cookie获取方法

获取小红书cookie是使用xhs库的前提。你可以通过以下方式获取：

1. 浏览器开发者工具：登录小红书后，在Network标签页复制cookie
2. 关键字段：确保cookie中包含a1、web_session和webId三个必需字段
3. 有效期：小红书cookie通常有较长的有效期，但建议定期更新

基础使用示例

import json
from xhs import XhsClient

def sign_func(uri, data=None, a1="", web_session=""):
    """签名函数实现"""
    # 这里需要实现签名逻辑
    # 可以使用xhs-api服务或本地Playwright
    return {"x-s": "signature", "x-t": "timestamp"}

# 初始化客户端
cookie = "your_xhs_cookie_here"
client = XhsClient(cookie=cookie, sign=sign_func)

# 获取笔记信息
note_id = "65682d4500000000380339a5"
note_data = client.get_note_by_id(note_id)

# 打印笔记信息
print(f"笔记标题: {note_data.get('title', 'N/A')}")
print(f"作者: {note_data.get('user', {}).get('nickname', 'N/A')}")
print(f"点赞数: {note_data.get('liked_count', 0)}")
print(f"收藏数: {note_data.get('collected_count', 0)}")

高级功能实现

1. 多账号管理策略

在生产环境中，你可能需要管理多个小红书账号。xhs支持多账号并发处理：

from concurrent.futures import ThreadPoolExecutor
from xhs import XhsClient

class MultiAccountManager:
    def __init__(self, accounts_config):
        self.accounts = []
        for config in accounts_config:
            client = XhsClient(
                cookie=config['cookie'],
                sign=config['sign_func'],
                proxies=config.get('proxies')
            )
            self.accounts.append(client)
    
    def batch_get_notes(self, note_ids, max_workers=5):
        """批量获取笔记信息"""
        results = {}
        
        def fetch_note(account_idx, note_id):
            try:
                return note_id, self.accounts[account_idx].get_note_by_id(note_id)
            except Exception as e:
                return note_id, {"error": str(e)}
        
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = []
            for i, note_id in enumerate(note_ids):
                account_idx = i % len(self.accounts)  # 轮询使用账号
                futures.append(executor.submit(fetch_note, account_idx, note_id))
            
            for future in futures:
                note_id, result = future.result()
                results[note_id] = result
        
        return results

# 使用示例
accounts_config = [
    {'cookie': 'cookie1', 'sign_func': sign_func1},
    {'cookie': 'cookie2', 'sign_func': sign_func2},
]

manager = MultiAccountManager(accounts_config)
notes = manager.batch_get_notes(['note_id1', 'note_id2', 'note_id3'])

2. 数据存储与处理管道

构建完整的数据处理流程：

import sqlite3
import pandas as pd
from datetime import datetime

class XhsDataPipeline:
    def __init__(self, db_path="xhs_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 (
            note_id TEXT PRIMARY KEY,
            title TEXT,
            desc TEXT,
            type TEXT,
            user_id TEXT,
            liked_count INTEGER,
            collected_count INTEGER,
            comment_count INTEGER,
            share_count INTEGER,
            publish_time INTEGER,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        """)
        
        # 用户表
        cursor.execute("""
        CREATE TABLE IF NOT EXISTS users (
            user_id TEXT PRIMARY KEY,
            nickname TEXT,
            avatar TEXT,
            fans_count INTEGER,
            follows_count INTEGER,
            interaction_count INTEGER,
            ip_location TEXT,
            tags TEXT,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        """)
        
        self.conn.commit()
    
    def save_note(self, note_data):
        """保存笔记数据"""
        cursor = self.conn.cursor()
        cursor.execute("""
        INSERT OR REPLACE INTO notes 
        (note_id, title, desc, type, user_id, liked_count, collected_count, 
         comment_count, share_count, publish_time)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            note_data['note_id'],
            note_data.get('title', ''),
            note_data.get('desc', ''),
            note_data.get('type', ''),
            note_data.get('user', {}).get('user_id', ''),
            int(note_data.get('liked_count', 0)),
            int(note_data.get('collected_count', 0)),
            int(note_data.get('comment_count', 0)),
            int(note_data.get('share_count', 0)),
            note_data.get('time', 0)
        ))
        self.conn.commit()
    
    def export_to_csv(self, output_file="xhs_data_analysis.csv"):
        """导出数据分析报告"""
        # 读取数据
        notes_df = pd.read_sql("SELECT * FROM notes", self.conn)
        users_df = pd.read_sql("SELECT * FROM users", self.conn)
        
        # 分析笔记互动数据
        analysis = notes_df.describe()
        
        # 保存分析结果
        with pd.ExcelWriter(output_file) as writer:
            notes_df.to_excel(writer, sheet_name='笔记数据', index=False)
            users_df.to_excel(writer, sheet_name='用户数据', index=False)
            analysis.to_excel(writer, sheet_name='数据分析')
        
        return analysis

3. 内容搜索与过滤

xhs提供了强大的搜索功能，支持多种过滤条件：

from xhs import SearchSortType, SearchNoteType

def advanced_search(client, keyword, filters=None):
    """高级搜索功能"""
    params = {
        "keyword": keyword,
        "page": 1,
        "page_size": 20,
        "sort": SearchSortType.GENERAL,
        "note_type": SearchNoteType.ALL
    }
    
    if filters:
        params.update(filters)
    
    # 执行搜索
    results = client.get_note_by_keyword(**params)
    
    # 结果过滤
    if filters and 'min_likes' in filters:
        results['items'] = [
            item for item in results['items'] 
            if item.get('liked_count', 0) >= filters['min_likes']
        ]
    
    return results

# 使用示例：搜索Python相关的高质量笔记
filters = {
    'sort': SearchSortType.MOST_POPULAR,  # 按热度排序
    'note_type': SearchNoteType.ALL,
    'min_likes': 1000  # 至少1000点赞
}

python_notes = advanced_search(client, "Python编程", filters)
print(f"找到{len(python_notes['items'])}篇高质量Python笔记")

性能优化技巧

1. 请求优化策略

优化策略	实施方法	效果提升
连接复用	使用Session对象	减少30%的请求时间
批量处理	合并相似请求	减少50%的API调用
缓存机制	本地缓存热点数据	减少80%的重复请求
异步处理	使用asyncio并发	提升3-5倍吞吐量

2. 异步并发实现

import asyncio
import aiohttp
from xhs import XhsClient

class AsyncXhsClient:
    def __init__(self, cookie, sign_func, max_concurrent=10):
        self.cookie = cookie
        self.sign_func = sign_func
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def get_note_async(self, session, note_id):
        """异步获取笔记"""
        async with self.semaphore:
            # 构建请求URL和参数
            url = f"https://edith.xiaohongshu.com/api/sns/web/v1/feed"
            params = {"note_id": note_id}
            
            # 获取签名
            signature = self.sign_func(url, params)
            
            # 发送异步请求
            headers = {
                "Cookie": self.cookie,
                "x-s": signature["x-s"],
                "x-t": signature["x-t"]
            }
            
            async with session.get(url, params=params, headers=headers) as response:
                if response.status == 200:
                    return await response.json()
                else:
                    raise Exception(f"请求失败: {response.status}")
    
    async def batch_fetch_notes(self, note_ids):
        """批量获取多个笔记"""
        async with aiohttp.ClientSession() as session:
            tasks = [self.get_note_async(session, note_id) for note_id in note_ids]
            results = await asyncio.gather(*tasks, return_exceptions=True)
            return results

# 使用示例
async def main():
    client = AsyncXhsClient(cookie, sign_func)
    note_ids = ["id1", "id2", "id3", "id4", "id5"]
    results = await client.batch_fetch_notes(note_ids)
    
    for note_id, result in zip(note_ids, results):
        if isinstance(result, Exception):
            print(f"获取{note_id}失败: {result}")
        else:
            print(f"获取{note_id}成功")

# asyncio.run(main())

3. 错误处理与重试机制

import time
from functools import wraps
from xhs.exception import DataFetchError, IPBlockError

def retry_on_failure(max_retries=3, delay=1, backoff=2):
    """失败重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except (DataFetchError, IPBlockError) as e:
                    last_exception = e
                    if attempt < max_retries - 1:
                        sleep_time = delay * (backoff ** attempt)
                        print(f"请求失败，{sleep_time}秒后重试 (尝试 {attempt + 1}/{max_retries})")
                        time.sleep(sleep_time)
                        continue
            raise last_exception
        return wrapper
    return decorator

class RobustXhsClient:
    def __init__(self, cookie, sign_func):
        self.client = XhsClient(cookie=cookie, sign=sign_func)
    
    @retry_on_failure(max_retries=3, delay=2)
    def safe_get_note(self, note_id):
        """安全的笔记获取方法"""
        return self.client.get_note_by_id(note_id)
    
    @retry_on_failure(max_retries=5, delay=3)
    def safe_search(self, keyword, page=1):
        """安全的搜索方法"""
        return self.client.get_note_by_keyword(keyword, page=page)
    
    def health_check(self):
        """健康检查"""
        try:
            # 尝试获取一个公开笔记
            test_note = self.client.get_note_by_id_from_html("646837b9000000001300a4c3")
            return True, "服务正常"
        except Exception as e:
            return False, f"服务异常: {str(e)}"

实际应用场景

1. 竞品分析系统

使用xhs构建竞品监控系统：

class CompetitorAnalyzer:
    def __init__(self, client, competitor_ids):
        self.client = client
        self.competitor_ids = competitor_ids
    
    def analyze_content_strategy(self, days=30):
        """分析竞品内容策略"""
        import datetime
        
        end_time = int(datetime.datetime.now().timestamp() * 1000)
        start_time = end_time - (days * 24 * 60 * 60 * 1000)
        
        all_notes = []
        for user_id in self.competitor_ids:
            # 获取用户发布的笔记
            user_notes = self.client.get_user_notes(user_id)
            
            # 过滤时间范围
            recent_notes = [
                note for note in user_notes 
                if start_time <= note.get('time', 0) <= end_time
            ]
            
            all_notes.extend(recent_notes)
        
        # 分析发布频率
        publish_stats = self._calculate_publish_frequency(all_notes)
        
        # 分析互动数据
        engagement_stats = self._calculate_engagement_rate(all_notes)
        
        # 分析内容主题
        topic_stats = self._analyze_content_topics(all_notes)
        
        return {
            'publish_stats': publish_stats,
            'engagement_stats': engagement_stats,
            'topic_stats': topic_stats,
            'total_notes': len(all_notes)
        }
    
    def _calculate_publish_frequency(self, notes):
        """计算发布频率"""
        from collections import Counter
        import datetime
        
        # 按日期统计
        date_counts = Counter()
        for note in notes:
            publish_time = note.get('time', 0) / 1000
            date = datetime.datetime.fromtimestamp(publish_time).strftime('%Y-%m-%d')
            date_counts[date] += 1
        
        return {
            'daily_avg': len(notes) / 30,
            'most_active_day': date_counts.most_common(1)[0] if date_counts else None,
            'date_distribution': dict(date_counts)
        }

2. 内容趋势预测

import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import KMeans

class TrendPredictor:
    def __init__(self, client):
        self.client = client
        self.vectorizer = TfidfVectorizer(max_features=1000)
    
    def predict_trending_topics(self, keyword, days=7):
        """预测趋势话题"""
        # 收集相关笔记
        notes = []
        for page in range(1, 6):  # 前5页
            result = self.client.get_note_by_keyword(keyword, page=page)
            notes.extend(result.get('items', []))
        
        # 提取文本内容
        texts = [note.get('desc', '') + ' ' + note.get('title', '') for note in notes]
        
        # 文本向量化
        X = self.vectorizer.fit_transform(texts)
        
        # 聚类分析
        kmeans = KMeans(n_clusters=5, random_state=42)
        clusters = kmeans.fit_predict(X)
        
        # 分析每个簇的主题
        feature_names = self.vectorizer.get_feature_names_out()
        trending_topics = []
        
        for i in range(5):
            # 获取簇中心的关键词
            center = kmeans.cluster_centers_[i]
            top_indices = center.argsort()[-10:][::-1]
            top_keywords = [feature_names[idx] for idx in top_indices]
            
            # 计算簇的互动率
            cluster_notes = [notes[j] for j in range(len(notes)) if clusters[j] == i]
            avg_engagement = self._calculate_avg_engagement(cluster_notes)
            
            trending_topics.append({
                'cluster_id': i,
                'keywords': top_keywords,
                'note_count': len(cluster_notes),
                'avg_engagement': avg_engagement,
                'growth_potential': self._estimate_growth(cluster_notes)
            })
        
        # 按增长潜力排序
        trending_topics.sort(key=lambda x: x['growth_potential'], reverse=True)
        
        return trending_topics
    
    def _calculate_avg_engagement(self, notes):
        """计算平均互动率"""
        if not notes:
            return 0
        
        total_engagement = 0
        for note in notes:
            engagement = (
                int(note.get('liked_count', 0)) +
                int(note.get('collected_count', 0)) +
                int(note.get('comment_count', 0))
            )
            total_engagement += engagement
        
        return total_engagement / len(notes)

常见问题解答

Q1: 如何解决签名失败问题？

A: 签名失败通常由以下原因导致：

1. Cookie过期：重新获取小红书cookie，确保a1、web_session字段有效
2. 环境检测：确保使用了stealth.min.js绕过浏览器检测
3. 签名服务异常：检查xhs-api服务是否正常运行

# 诊断签名问题
def diagnose_signature_issue():
    # 检查cookie格式
    cookie_parts = cookie.split('; ')
    required_fields = ['a1', 'web_session', 'webId']
    missing_fields = [field for field in required_fields 
                     if not any(field in part for part in cookie_parts)]
    
    if missing_fields:
        print(f"缺少必要字段: {missing_fields}")
        return False
    
    # 测试签名服务
    test_url = "https://edith.xiaohongshu.com/api/sns/web/v1/feed"
    test_data = {"note_id": "test_note_id"}
    
    try:
        signature = sign_func(test_url, test_data)
        if 'x-s' in signature and 'x-t' in signature:
            print("签名服务正常")
            return True
        else:
            print("签名返回格式错误")
            return False
    except Exception as e:
        print(f"签名服务异常: {e}")
        return False

Q2: 如何处理IP被封禁问题？

A: IP被封禁是常见问题，可以通过以下策略解决：

1. 使用代理IP池：轮换不同地区的IP地址
2. 降低请求频率：增加请求间隔时间
3. 模拟真实用户行为：添加随机延迟和User-Agent轮换

import random
import time
from fake_useragent import UserAgent

class IPRotationManager:
    def __init__(self, proxy_list=None):
        self.proxy_list = proxy_list or []
        self.current_proxy_idx = 0
        self.ua = UserAgent()
        
    def get_proxy(self):
        """获取代理IP"""
        if not self.proxy_list:
            return None
        
        proxy = self.proxy_list[self.current_proxy_idx]
        self.current_proxy_idx = (self.current_proxy_idx + 1) % len(self.proxy_list)
        return {
            'http': proxy,
            'https': proxy
        }
    
    def get_random_delay(self, min_delay=1, max_delay=3):
        """获取随机延迟"""
        return random.uniform(min_delay, max_delay)
    
    def get_random_headers(self):
        """获取随机请求头"""
        return {
            'User-Agent': self.ua.random,
            'Accept': 'application/json, text/plain, */*',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
            'Referer': 'https://www.xiaohongshu.com/',
            'Origin': 'https://www.xiaohongshu.com'
        }

Q3: 如何提高数据采集效率？

A: 提高效率的关键策略：

策略	实施方法	预期效果
并发采集	使用多线程/异步IO	提升3-5倍速度
数据缓存	缓存已采集的数据	减少重复请求
批量处理	合并API调用	减少网络开销
智能调度	根据成功率动态调整	提高稳定性

Q4: 数据采集的合规性如何保障？

A: 确保合规性的关键措施：

1. 遵守robots.txt：尊重网站的爬虫协议
2. 设置合理频率：避免对服务器造成压力
3. 仅采集公开数据：不获取用户隐私信息
4. 数据使用声明：明确标注数据来源和用途
5. 商业用途授权：如需商业使用，获取相应授权

下一步建议

1. 深入学习xhs核心源码：查看xhs/core.py了解API调用细节
2. 探索高级功能：研究xhs/help.py中的工具函数
3. 部署生产环境：使用xhs-api/Dockerfile部署签名服务
4. 参考示例代码：查看example/目录中的使用示例
5. 性能调优：根据实际需求调整并发数和请求间隔

通过xhs项目，你可以构建强大的小红书数据采集系统，为内容分析、市场研究和竞品监控提供数据支持。记住，技术是工具，合理使用才能创造最大价值。开始你的小红书数据探索之旅吧！

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