小红书数据采集架构深度解析：基于Python请求封装的Web端数据采集解决方案
【免费下载链接】xhs 基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/ 
项目地址: https://gitcode.com/gh_mirrors/xh/xhs
在当今数据驱动的互联网时代，小红书作为中国最具影响力的生活方式分享平台，积累了海量的用户生成内容和社交互动数据。xhs项目作为一款基于小红书Web端的高级Python请求封装库，为开发者和数据分析师提供了稳定、高效且合规的数据采集解决方案。本文将从技术架构、核心实现、应用场景等多个维度，深入解析xhs项目的设计理念和技术实现细节。
技术架构解析：多层封装与反爬虫对抗机制
xhs项目采用了多层架构设计，通过抽象化网络请求、签名验证和数据处理流程，实现了对小红书Web API的高效封装。整个架构分为四个核心层次：请求管理层、签名验证层、数据解析层和异常处理层。
请求管理层架构设计
请求管理层位于架构最底层，负责处理HTTP请求的发送和响应。xhs项目基于Python的requests库构建，通过Session对象管理cookie和连接池，实现了高效的网络通信。以下是核心请求管理类的初始化实现：
class XhsClient:
def __init__(
self, cookie=None, user_agent=None, timeout=10, proxies=None, sign=None
):
"""构造函数"""
self.proxies = proxies
self.__session: requests.Session = requests.session()
self.timeout = timeout
# 设置默认请求头
self.__session.headers.update({
"User-Agent": user_agent or "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Accept-Encoding": "gzip, deflate, br",
"Content-Type": "application/json;charset=UTF-8",
})
# 设置cookie
if cookie:
update_session_cookies_from_cookie(self.__session, cookie)
self.sign = sign # 签名函数
该设计采用了依赖注入模式，将签名函数作为可配置参数，支持自定义签名实现。这种设计模式使得系统具有高度可扩展性，开发者可以根据实际需求替换签名算法或集成第三方签名服务。
签名验证层实现机制
签名验证是xhs项目的核心技术难点。小红书采用了复杂的JavaScript混淆和动态签名算法，xhs通过模拟浏览器环境实现了签名生成。签名层的核心函数位于xhs/help.py文件中：
def sign(uri, data=None, ctime=None, a1="", b1=""):
"""
签名函数：接收URI、数据字典和可选的时间戳参数
返回包含"x-s"和"x-t"的字典
"""
v = int(round(time.time() * 1000) if not ctime else ctime)
raw_str = f"{v}test{uri}{json.dumps(data, separators=(',', ':'), ensure_ascii=False) if isinstance(data, dict) else ''}"
md5_str = hashlib.md5(raw_str.encode('utf-8')).hexdigest()
x_s = h(md5_str) # 自定义编码函数
x_t = str(v)
# 构建公共参数
common = {
"s0": 5, # 平台代码
"s1": "",
"x0": "1", # localStorage.getItem("b1b1")
"x1": "3.2.0", # 版本号
"x2": "Windows",
"x3": "xhs-pc-web",
"x4": "2.3.1",
"x5": a1, # a1 cookie
"x6": x_t,
"x7": x_s,
"x8": b1, # localStorage.getItem("b1")
"x9": mrc(x_t + x_s), # 二次加密
"x10": 1, # 签名计数
}
encodeStr = encodeUtf8(json.dumps(common, separators=(',', ':')))
x_s_common = b64Encode(encodeStr)
return {
"x-s": x_s,
"x-t": x_t,
"x-s-common": x_s_common,
}
签名算法采用了多层加密策略：首先对原始数据进行MD5哈希，然后通过自定义编码函数进行Base64变种编码，最后构建包含平台信息、时间戳和加密参数的完整签名包。
数据解析层设计原理
数据解析层负责将API返回的JSON数据转换为结构化的Python对象。xhs项目使用了Python的NamedTuple和Enum类型来定义数据结构，确保类型安全和代码可读性：
from enum import Enum
from typing import NamedTuple
class FeedType(Enum):
"""主页feed类型枚举"""
RECOMMEND = "homefeed_recommend" # 推荐
FASHION = "homefeed.fashion_v3" # 穿搭
FOOD = "homefeed.food_v3" # 美食
COSMETICS = "homefeed.cosmetics_v3" # 彩妆
MOVIE = "homefeed.movie_and_tv_v3" # 影视
CAREER = "homefeed.career_v3" # 职场
EMOTION = "homefeed.love_v3" # 情感
HOUSE = "homefeed.household_product_v3" # 家居
GAME = "homefeed.gaming_v3" # 游戏
TRAVEL = "homefeed.travel_v3" # 旅行
FITNESS = "homefeed.fitness_v3" # 健身
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: int
这种设计使得数据访问更加直观和安全，开发者可以通过属性访问方式获取数据，避免了字典键名拼写错误的风险。
核心模块详解：API接口与功能实现
xhs项目提供了完整的API接口覆盖，包括用户信息获取、内容搜索、互动操作等多个功能模块。每个模块都经过精心设计，确保接口的一致性和易用性。
用户信息管理模块
用户信息管理模块提供了获取用户基本资料、笔记列表、收藏和点赞信息等功能。以下是核心方法的实现：
def get_user_info(self, user_id: str) -> dict:
"""获取用户信息"""
uri = "/api/sns/web/v1/user/otherinfo"
data = {
"target_user_id": user_id
}
return self.__request(uri, data)
def get_user_notes(self, user_id: str, cursor: str = "") -> dict:
"""获取用户发布的笔记列表"""
uri = "/api/sns/web/v1/user_posted"
data = {
"user_id": user_id,
"cursor": cursor,
"num": 30 # 每页数量
}
return self.__request(uri, data)
def get_user_collect_notes(self, user_id: str, cursor: str = "") -> dict:
"""获取用户收藏的笔记列表"""
uri = "/api/sns/web/v1/note/collect/page"
data = {
"user_id": user_id,
"cursor": cursor,
"num": 30
}
return self.__request(uri, data)
这些方法采用了统一的参数命名规范，支持分页加载（通过cursor参数），确保了大数据量场景下的稳定性和性能。
内容搜索与发现模块
内容搜索模块支持关键词搜索、分类筛选和排序功能，为数据采集提供了强大的灵活性：
class SearchSortType(Enum):
"""搜索排序类型枚举"""
GENERAL = "general" # 默认排序
MOST_POPULAR = "popularity_descending" # 最热门
LATEST = "time_descending" # 最新发布
class SearchNoteType(Enum):
"""搜索笔记类型枚举"""
ALL = 0 # 全部类型
VIDEO = 1 # 仅视频
IMAGE = 2 # 仅图片
def get_note_by_keyword(
self,
keyword: str,
page: int = 1,
page_size: int = 20,
sort: SearchSortType = SearchSortType.GENERAL,
note_type: SearchNoteType = SearchNoteType.ALL
) -> dict:
"""根据关键词搜索笔记"""
uri = "/api/sns/web/v1/search/notes"
data = {
"keyword": keyword,
"page": page,
"page_size": page_size,
"sort": sort.value,
"note_type": note_type.value,
"search_id": get_search_id() # 生成搜索ID
}
return self.__request(uri, data)
搜索功能支持多种筛选条件，包括内容类型（图文/视频）、排序方式（热门/最新）等，满足了不同场景下的数据采集需求。
互动操作模块
互动操作模块实现了点赞、收藏、评论等用户行为模拟，为自动化测试和数据分析提供了便利：
def like_note(self, note_id: str) -> dict:
"""点赞笔记"""
uri = "/api/sns/web/v1/note/like"
data = {
"note_id": note_id,
"note_type": 1 # 点赞类型
}
return self.__request(uri, data)
def collect_note(self, note_id: str) -> dict:
"""收藏笔记"""
uri = "/api/sns/web/v1/note/collect"
data = {
"note_id": note_id,
"note_type": 1
}
return self.__request(uri, data)
def comment_note(self, note_id: str, content: str) -> dict:
"""评论笔记"""
uri = "/api/sns/web/v1/comment/post"
data = {
"note_id": note_id,
"content": content,
"at_users": []
}
return self.__request(uri, data)
应用场景实践：企业级数据采集解决方案
xhs项目在实际业务中具有广泛的应用场景，从市场研究到竞品分析，从用户行为分析到内容策略制定，都能提供有力的数据支持。
市场趋势分析系统
通过定期采集特定品类的热门笔记数据，可以构建市场趋势分析系统。以下是一个完整的趋势分析示例：
import asyncio
import aiohttp
from datetime import datetime, timedelta
from typing import List, Dict
import pandas as pd
from xhs import XhsClient, FeedType
class MarketTrendAnalyzer:
"""市场趋势分析器"""
def __init__(self, cookie: str):
self.client = XhsClient(cookie=cookie)
self.categories = [
FeedType.FASHION, # 穿搭
FeedType.FOOD, # 美食
FeedType.COSMETICS, # 彩妆
FeedType.TRAVEL, # 旅行
FeedType.FITNESS # 健身
]
async def collect_category_data(self, category: FeedType, days: int = 7) -> List[Dict]:
"""采集指定分类的历史数据"""
end_time = datetime.now()
start_time = end_time – timedelta(days=days)
all_notes = []
cursor = ""
while True:
try:
response = self.client.get_home_feed(category, cursor=cursor)
notes = response.get("notes", [])
if not notes:
break
# 过滤时间范围内的笔记
filtered_notes = [
note for note in notes
if start_time.timestamp() <= note.get("time", 0) <= end_time.timestamp()
]
all_notes.extend(filtered_notes)
# 检查是否还有更多数据
if not response.get("has_more", False):
break
cursor = response.get("cursor", "")
except Exception as e:
print(f"采集{category.value}数据时出错: {e}")
break
return all_notes
def analyze_trend(self, notes: List[Dict]) -> pd.DataFrame:
"""分析趋势数据"""
df = pd.DataFrame(notes)
# 数据清洗和转换
df['time'] = pd.to_datetime(df['time'], unit='s')
df['liked_count'] = pd.to_numeric(df['liked_count'], errors='coerce')
df['collected_count'] = pd.to_numeric(df['collected_count'], errors='coerce')
df['comment_count'] = pd.to_numeric(df['comment_count'], errors='coerce')
# 计算互动指标
df['engagement_rate'] = (
df['liked_count'] + df['collected_count'] * 2 + df['comment_count'] * 3
) / df['liked_count'].max()
# 按时间分组分析
daily_stats = df.groupby(df['time'].dt.date).agg({
'engagement_rate': 'mean',
'note_id': 'count',
'liked_count': 'sum',
'collected_count': 'sum',
'comment_count': 'sum'
}).rename(columns={'note_id': 'note_count'})
return daily_stats
async def run_analysis(self):
"""运行完整分析流程"""
results = {}
async with aiohttp.ClientSession() as session:
tasks = []
for category in self.categories:
task = asyncio.create_task(
self.collect_category_data(category)
)
tasks.append(task)
category_results = await asyncio.gather(*tasks)
for category, notes in zip(self.categories, category_results):
if notes:
trend_data = self.analyze_trend(notes)
results[category.value] = trend_data
return results
# 使用示例
async def main():
analyzer = MarketTrendAnalyzer(cookie="your_cookie_here")
trends = await analyzer.run_analysis()
# 输出分析结果
for category, data in trends.items():
print(f"\\n{category} 品类趋势分析:")
print(data.tail()) # 显示最近几天的数据
if __name__ == "__main__":
asyncio.run(main())
竞品监测与内容策略优化
竞品监测系统可以帮助企业跟踪竞争对手的内容策略和用户互动情况：
import schedule
import time
from typing import Dict, List
from dataclasses import dataclass
from datetime import datetime
import json
@dataclass
class CompetitorProfile:
"""竞品账号配置"""
user_id: str
competitor_name: str
industry: str
monitoring_frequency: int # 监测频率（小时）
key_metrics: List[str] # 关键指标
class CompetitorMonitor:
"""竞品监测系统"""
def __init__(self, cookie: str, competitors: List[CompetitorProfile]):
self.client = XhsClient(cookie=cookie)
self.competitors = competitors
self.history_data = {}
def monitor_single_competitor(self, profile: CompetitorProfile) -> Dict:
"""监测单个竞品账号"""
try:
# 获取用户基本信息
user_info = self.client.get_user_info(profile.user_id)
# 获取用户最新笔记
notes_response = self.client.get_user_notes(profile.user_id)
recent_notes = notes_response.get("notes", [])[:10] # 最近10条笔记
# 计算关键指标
metrics = {
"timestamp": datetime.now().isoformat(),
"follower_count": user_info.get("fans", 0),
"note_count": user_info.get("notes", 0),
"avg_likes": 0,
"avg_collects": 0,
"avg_comments": 0,
"recent_engagement": []
}
if recent_notes:
total_likes = sum(int(note.get("liked_count", 0)) for note in recent_notes)
total_collects = sum(int(note.get("collected_count", 0)) for note in recent_notes)
total_comments = sum(int(note.get("comment_count", 0)) for note in recent_notes)
metrics["avg_likes"] = total_likes / len(recent_notes)
metrics["avg_collects"] = total_collects / len(recent_notes)
metrics["avg_comments"] = total_comments / len(recent_notes)
# 记录每条笔记的互动数据
for note in recent_notes:
metrics["recent_engagement"].append({
"note_id": note.get("note_id"),
"title": note.get("title", "")[:50],
"likes": int(note.get("liked_count", 0)),
"collects": int(note.get("collected_count", 0)),
"comments": int(note.get("comment_count", 0)),
"time": note.get("time", 0)
})
return metrics
except Exception as e:
print(f"监测竞品 {profile.competitor_name} 时出错: {e}")
return {}
def run_monitoring(self):
"""执行监测任务"""
print(f"开始竞品监测，时间: {datetime.now()}")
for competitor in self.competitors:
print(f"正在监测: {competitor.competitor_name}")
metrics = self.monitor_single_competitor(competitor)
if metrics:
# 保存历史数据
if competitor.user_id not in self.history_data:
self.history_data[competitor.user_id] = []
self.history_data[competitor.user_id].append(metrics)
# 保存到文件
self.save_to_file(competitor, metrics)
print(f" ✓ 粉丝数: {metrics.get('follower_count', 0)}")
print(f" ✓ 平均点赞: {metrics.get('avg_likes', 0):.1f}")
print(f" ✓ 平均收藏: {metrics.get('avg_collects', 0):.1f}")
def save_to_file(self, competitor: CompetitorProfile, metrics: Dict):
"""保存监测数据到文件"""
filename = f"competitor_{competitor.competitor_name}_{datetime.now().strftime('%Y%m')}.json"
try:
# 读取现有数据
try:
with open(filename, 'r', encoding='utf-8') as f:
existing_data = json.load(f)
except FileNotFoundError:
existing_data = []
# 添加新数据
existing_data.append(metrics)
# 保存数据
with open(filename, 'w', encoding='utf-8') as f:
json.dump(existing_data, f, ensure_ascii=False, indent=2)
except Exception as e:
print(f"保存数据到文件 {filename} 时出错: {e}")
def start_scheduled_monitoring(self):
"""启动定时监测"""
for competitor in self.competitors:
schedule.every(competitor.monitoring_frequency).hours.do(
lambda c=competitor: self.monitor_single_competitor(c)
)
print("竞品监测系统已启动，按Ctrl+C停止")
while True:
schedule.run_pending()
time.sleep(60)
# 配置竞品监测
competitors = [
CompetitorProfile(
user_id="user_id_1",
competitor_name="品牌A",
industry="美妆",
monitoring_frequency=6,
key_metrics=["follower_count", "avg_likes", "avg_comments"]
),
CompetitorProfile(
user_id="user_id_2",
competitor_name="品牌B",
industry="时尚",
monitoring_frequency=12,
key_metrics=["note_count", "engagement_rate"]
)
]
# 启动监测系统
monitor = CompetitorMonitor(cookie="your_cookie_here", competitors=competitors)
monitor.start_scheduled_monitoring()
性能优化指南：高级配置与调优策略
xhs项目在性能优化方面提供了多种配置选项，开发者可以根据具体需求进行调整，以获得最佳的数据采集性能。
连接池与超时优化
通过合理配置连接池参数和超时设置，可以显著提升数据采集效率：
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class OptimizedXhsClient(XhsClient):
"""优化版的Xhs客户端"""
def __init__(self, cookie=None, user_agent=None, timeout=10,
proxies=None, sign=None, max_retries=3, pool_connections=10,
pool_maxsize=10):
super().__init__(cookie, user_agent, timeout, proxies, sign)
# 配置重试策略
retry_strategy = Retry(
total=max_retries,
backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET", "POST"]
)
# 配置适配器
adapter = HTTPAdapter(
max_retries=retry_strategy,
pool_connections=pool_connections,
pool_maxsize=pool_maxsize
)
# 应用到session
self.__session.mount("https://", adapter)
self.__session.mount("http://", adapter)
def batch_get_notes(self, note_ids: List[str], batch_size: int = 10) -> List[Dict]:
"""批量获取笔记信息"""
import concurrent.futures
from functools import partial
def get_note_safe(note_id: str):
"""安全的笔记获取函数"""
try:
return self.get_note_by_id(note_id)
except Exception as e:
print(f"获取笔记 {note_id} 失败: {e}")
return None
notes = []
# 使用线程池并行处理
with concurrent.futures.ThreadPoolExecutor(max_workers=batch_size) as executor:
# 分批处理
for i in range(0, len(note_ids), batch_size):
batch = note_ids[i:i+batch_size]
# 提交批量任务
future_to_note = {
executor.submit(get_note_safe, note_id): note_id
for note_id in batch
}
# 收集结果
for future in concurrent.futures.as_completed(future_to_note):
note_id = future_to_note[future]
try:
result = future.result()
if result:
notes.append(result)
except Exception as e:
print(f"处理笔记 {note_id} 时出错: {e}")
# 批次间延迟，避免请求过快
time.sleep(1)
return notes
缓存策略优化
对于频繁访问的数据，实现缓存机制可以显著减少网络请求：
import pickle
from functools import lru_cache
from datetime import datetime, timedelta
class CachedXhsClient(XhsClient):
"""带缓存功能的Xhs客户端"""
def __init__(self, *args, cache_dir=".xhs_cache", cache_ttl=3600, **kwargs):
super().__init__(*args, **kwargs)
self.cache_dir = cache_dir
self.cache_ttl = cache_ttl # 缓存过期时间（秒）
# 确保缓存目录存在
import os
os.makedirs(cache_dir, exist_ok=True)
def _get_cache_key(self, func_name: str, *args, **kwargs) -> str:
"""生成缓存键"""
import hashlib
import json
# 将参数序列化为字符串
params_str = json.dumps({
"func": func_name,
"args": args,
"kwargs": kwargs
}, sort_keys=True)
# 生成MD5哈希作为缓存键
return hashlib.md5(params_str.encode()).hexdigest()
def _get_cache_path(self, cache_key: str) -> str:
"""获取缓存文件路径"""
return f"{self.cache_dir}/{cache_key}.pkl"
def _is_cache_valid(self, cache_path: str) -> bool:
"""检查缓存是否有效"""
import os
from datetime import datetime, timedelta
if not os.path.exists(cache_path):
return False
# 检查文件修改时间
mtime = datetime.fromtimestamp(os.path.getmtime(cache_path))
now = datetime.now()
return (now – mtime).seconds < self.cache_ttl
@lru_cache(maxsize=100)
def get_note_by_id_cached(self, note_id: str) -> Dict:
"""带缓存的笔记获取"""
cache_key = self._get_cache_key("get_note_by_id", note_id)
cache_path = self._get_cache_path(cache_key)
# 检查缓存是否有效
if self._is_cache_valid(cache_path):
try:
with open(cache_path, 'rb') as f:
return pickle.load(f)
except Exception:
pass # 缓存读取失败，继续从API获取
# 从API获取数据
data = self.get_note_by_id(note_id)
# 保存到缓存
try:
with open(cache_path, 'wb') as f:
pickle.dump(data, f)
except Exception as e:
print(f"保存缓存失败: {e}")
return data
def get_user_info_cached(self, user_id: str) -> Dict:
"""带缓存的用户信息获取"""
cache_key = self._get_cache_key("get_user_info", user_id)
cache_path = self._get_cache_path(cache_key)
if self._is_cache_valid(cache_path):
try:
with open(cache_path, 'rb') as f:
return pickle.load(f)
except Exception:
pass
data = self.get_user_info(user_id)
try:
with open(cache_path, 'wb') as f:
pickle.dump(data, f)
except Exception as e:
print(f"保存缓存失败: {e}")
return data
代理与IP轮换策略
在大规模数据采集场景下，合理的代理配置和IP轮换策略至关重要：
import random
from typing import List, Dict
import time
class ProxyManager:
"""代理管理器"""
def __init__(self, proxy_list: List[Dict]):
"""
初始化代理管理器
Args:
proxy_list: 代理列表，格式为 [{"http": "http://proxy1:port", "https": "https://proxy1:port"}, …]
"""
self.proxy_list = proxy_list
self.current_index = 0
self.proxy_stats = {} # 代理使用统计
self.blacklist = set() # 黑名单
def get_proxy(self) -> Dict:
"""获取下一个可用代理"""
if not self.proxy_list:
return {}
# 移除黑名单中的代理
available_proxies = [
proxy for i, proxy in enumerate(self.proxy_list)
if i not in self.blacklist
]
if not available_proxies:
# 如果没有可用代理，清空黑名单并重试
self.blacklist.clear()
available_proxies = self.proxy_list
# 使用轮询策略选择代理
proxy = available_proxies[self.current_index % len(available_proxies)]
self.current_index += 1
# 记录使用统计
proxy_key = str(proxy)
if proxy_key not in self.proxy_stats:
self.proxy_stats[proxy_key] = {
"success_count": 0,
"failure_count": 0,
"last_used": time.time(),
"total_time": 0
}
return proxy
def mark_success(self, proxy: Dict, response_time: float):
"""标记代理使用成功"""
proxy_key = str(proxy)
if proxy_key in self.proxy_stats:
self.proxy_stats[proxy_key]["success_count"] += 1
self.proxy_stats[proxy_key]["total_time"] += response_time
self.proxy_stats[proxy_key]["last_used"] = time.time()
def mark_failure(self, proxy: Dict):
"""标记代理使用失败"""
proxy_key = str(proxy)
if proxy_key in self.proxy_stats:
self.proxy_stats[proxy_key]["failure_count"] += 1
# 如果失败次数过多，加入黑名单
stats = self.proxy_stats[proxy_key]
total_attempts = stats["success_count"] + stats["failure_count"]
failure_rate = stats["failure_count"] / total_attempts if total_attempts > 0 else 0
if failure_rate > 0.7: # 失败率超过70%
self.blacklist.add(list(self.proxy_list).index(proxy))
print(f"代理 {proxy} 已被加入黑名单，失败率: {failure_rate:.2%}")
def get_stats(self) -> Dict:
"""获取代理使用统计"""
return self.proxy_stats
class RotatingProxyXhsClient(XhsClient):
"""支持代理轮换的Xhs客户端"""
def __init__(self, cookie=None, user_agent=None, timeout=10,
proxy_manager=None, sign=None):
super().__init__(cookie, user_agent, timeout, None, sign)
self.proxy_manager = proxy_manager
self.request_count = 0
self.rotate_threshold = 50 # 每50次请求轮换代理
def __request(self, uri, data=None):
"""重写请求方法，支持代理轮换"""
self.request_count += 1
# 每rotate_threshold次请求轮换代理
if self.proxy_manager and self.request_count % self.rotate_threshold == 0:
proxy = self.proxy_manager.get_proxy()
self.proxies = proxy
print(f"轮换代理: {proxy}")
# 记录请求开始时间
start_time = time.time()
try:
# 调用父类的请求方法
response = super().__request(uri, data)
# 计算响应时间
response_time = time.time() – start_time
# 标记代理成功
if self.proxy_manager and self.proxies:
self.proxy_manager.mark_success(self.proxies, response_time)
return response
except Exception as e:
# 标记代理失败
if self.proxy_manager and self.proxies:
self.proxy_manager.mark_failure(self.proxies)
# 重新抛出异常
raise e
生态集成方案：与其他技术栈的无缝对接
xhs项目设计时考虑了与其他Python生态系统的集成，可以轻松与数据分析、Web框架和任务调度系统结合使用。
与Pandas和NumPy的数据分析集成
import pandas as pd
import numpy as np
from typing import List, Dict
import matplotlib.pyplot as plt
import seaborn as sns
class XhsDataAnalyzer:
"""小红书数据分析器"""
def __init__(self, client: XhsClient):
self.client = client
def collect_user_metrics(self, user_ids: List[str]) -> pd.DataFrame:
"""收集多个用户的指标数据"""
metrics_list = []
for user_id in user_ids:
try:
# 获取用户信息
user_info = self.client.get_user_info(user_id)
# 获取用户笔记
notes_response = self.client.get_user_notes(user_id)
notes = notes_response.get("notes", [])
if notes:
# 计算互动指标
likes = [int(note.get("liked_count", 0)) for note in notes]
collects = [int(note.get("collected_count", 0)) for note in notes]
comments = [int(note.get("comment_count", 0)) for note in notes]
metrics = {
"user_id": user_id,
"username": user_info.get("nickname", ""),
"follower_count": user_info.get("fans", 0),
"note_count": len(notes),
"avg_likes": np.mean(likes),
"avg_collects": np.mean(collects),
"avg_comments": np.mean(comments),
"max_likes": np.max(likes) if likes else 0,
"total_likes": np.sum(likes),
"engagement_rate": (np.sum(likes) + np.sum(collects) * 2) /
max(user_info.get("fans", 1), 1)
}
metrics_list.append(metrics)
except Exception as e:
print(f"收集用户 {user_id} 数据时出错: {e}")
continue
return pd.DataFrame(metrics_list)
def analyze_content_patterns(self, keyword: str, limit: int = 100) -> Dict:
"""分析内容模式"""
# 搜索相关笔记
search_results = []
page = 1
while len(search_results) < limit:
try:
response = self.client.get_note_by_keyword(
keyword,
page=page,
page_size=20
)
notes = response.get("notes", [])
if not notes:
break
search_results.extend(notes)
page += 1
if len(search_results) >= limit:
break
except Exception as e:
print(f"搜索第{page}页时出错: {e}")
break
# 转换为DataFrame
df = pd.DataFrame(search_results[:limit])
# 数据清洗
df['liked_count'] = pd.to_numeric(df['liked_count'], errors='coerce').fillna(0)
df['collected_count'] = pd.to_numeric(df['collected_count'], errors='coerce').fillna(0)
df['comment_count'] = pd.to_numeric(df['comment_count'], errors='coerce').fillna(0)
# 分析结果
analysis = {
"total_notes": len(df),
"avg_likes": df['liked_count'].mean(),
"avg_collects": df['collected_count'].mean(),
"avg_comments": df['comment_count'].mean(),
"top_tags": self._extract_top_tags(df),
"content_length_stats": self._analyze_content_length(df),
"time_distribution": self._analyze_time_distribution(df)
}
return analysis
def _extract_top_tags(self, df: pd.DataFrame, top_n: int = 10) -> List[Dict]:
"""提取热门标签"""
all_tags = []
for tags in df['tag_list'].dropna():
if isinstance(tags, list):
all_tags.extend([tag.get('name', '') for tag in tags])
from collections import Counter
tag_counts = Counter(all_tags)
return [
{"tag": tag, "count": count}
for tag, count in tag_counts.most_common(top_n)
]
def _analyze_content_length(self, df: pd.DataFrame) -> Dict:
"""分析内容长度"""
df['title_length'] = df['title'].apply(lambda x: len(str(x)))
df['desc_length'] = df['desc'].apply(lambda x: len(str(x)))
return {
"title_avg_length": df['title_length'].mean(),
"title_median_length": df['title_length'].median(),
"desc_avg_length": df['desc_length'].mean(),
"desc_median_length": df['desc_length'].median(),
"correlation_with_likes": df[['title_length', 'liked_count']].corr().iloc[0, 1]
}
def _analyze_time_distribution(self, df: pd.DataFrame) -> Dict:
"""分析时间分布"""
if 'time' in df.columns:
df['post_time'] = pd.to_datetime(df['time'], unit='s')
df['hour'] = df['post_time'].dt.hour
df['weekday'] = df['post_time'].dt.weekday
hourly_dist = df.groupby('hour').size().to_dict()
weekday_dist = df.groupby('weekday').size().to_dict()
return {
"hourly_distribution": hourly_dist,
"weekday_distribution": weekday_dist,
"best_posting_hour": df.groupby('hour')['liked_count'].mean().idxmax(),
"best_posting_weekday": df.groupby('weekday')['liked_count'].mean().idxmax()
}
return {}
def visualize_metrics(self, df: pd.DataFrame, output_path: str = "metrics_visualization.png"):
"""可视化指标数据"""
fig, axes = plt.subplots(2, 2, figsize=(15, 12))
# 1. 粉丝数与互动率散点图
axes[0, 0].scatter(df['follower_count'], df['engagement_rate'], alpha=0.6)
axes[0, 0].set_xlabel('粉丝数')
axes[0, 0].set_ylabel('互动率')
axes[0, 0].set_title('粉丝数与互动率关系')
axes[0, 0].grid(True, alpha=0.3)
# 2. 点赞、收藏、评论的箱线图
metrics_data = [df['avg_likes'], df['avg_collects'], df['avg_comments']]
axes[0, 1].boxplot(metrics_data, labels=['点赞', '收藏', '评论'])
axes[0, 1].set_ylabel('平均数量')
axes[0, 1].set_title('互动指标分布')
axes[0, 1].grid(True, alpha=0.3)
# 3. 笔记数量分布直方图
axes[1, 0].hist(df['note_count'], bins=20, alpha=0.7, edgecolor='black')
axes[1, 0].set_xlabel('笔记数量')
axes[1, 0].set_ylabel('用户数')
axes[1, 0].set_title('用户笔记数量分布')
axes[1, 0].grid(True, alpha=0.3)
# 4. 热力图：各指标相关性
corr_matrix = df[['follower_count', 'avg_likes', 'avg_collects',
'avg_comments', 'engagement_rate']].corr()
sns.heatmap(corr_matrix, annot=True, fmt='.2f', cmap='coolwarm',
ax=axes[1, 1], cbar_kws={'label': '相关系数'})
axes[1, 1].set_title('指标相关性热力图')
plt.tight_layout()
plt.savefig(output_path, dpi=300, bbox_inches='tight')
plt.close()
print(f"可视化图表已保存到: {output_path}")
# 使用示例
def analyze_competitors():
"""分析竞品账号"""
client = XhsClient(cookie="your_cookie_here")
analyzer = XhsDataAnalyzer(client)
# 竞品账号ID列表
competitor_ids = [
"user_id_1", "user_id_2", "user_id_3",
"user_id_4", "user_id_5"
]
# 收集数据
metrics_df = analyzer.collect_user_metrics(competitor_ids)
# 分析内容模式
content_analysis = analyzer.analyze_content_patterns("美妆教程", limit=50)
# 可视化结果
analyzer.visualize_metrics(metrics_df, "competitor_analysis.png")
# 输出分析报告
print("竞品分析报告")
print("=" * 50)
print(f"分析账号数量: {len(metrics_df)}")
print(f"平均粉丝数: {metrics_df['follower_count'].mean():.0f}")
print(f"平均笔记数: {metrics_df['note_count'].mean():.0f}")
print(f"平均点赞数: {metrics_df['avg_likes'].mean():.1f}")
print(f"平均互动率: {metrics_df['engagement_rate'].mean():.3f}")
print("\\n内容分析结果")
print("=" * 50)
print(f"分析笔记数量: {content_analysis['total_notes']}")
print(f"平均点赞: {content_analysis['avg_likes']:.1f}")
print(f"平均收藏: {content_analysis['avg_collects']:.1f}")
print("\\n热门标签:")
for tag_info in content_analysis['top_tags'][:5]:
print(f" {tag_info['tag']}: {tag_info['count']}次")
return metrics_df, content_analysis
与FastAPI的Web服务集成
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Optional
import uvicorn
from datetime import datetime
app = FastAPI(title="小红书数据采集API", version="1.0.0")
class NoteRequest(BaseModel):
"""笔记请求模型"""
note_id: str
include_comments: bool = False
include_user_info: bool = False
class SearchRequest(BaseModel):
"""搜索请求模型"""
keyword: str
page: int = 1
page_size: int = 20
sort: str = "general"
note_type: str = "all"
class UserAnalysisRequest(BaseModel):
"""用户分析请求模型"""
user_ids: List[str]
metrics: List[str] = ["follower_count", "note_count", "engagement_rate"]
class XhsAPIService:
"""小红书API服务层"""
def __init__(self):
self.client = None
def init_client(self, cookie: str):
"""初始化客户端"""
self.client = XhsClient(cookie=cookie)
async def get_note_details(self, request: NoteRequest) -> dict:
"""获取笔记详情"""
if not self.client:
raise HTTPException(status_code=500, detail="客户端未初始化")
try:
# 获取笔记基本信息
note = self.client.get_note_by_id(request.note_id)
result = {"note": note}
# 如果需要评论信息
if request.include_comments:
comments = self.client.get_note_comments(request.note_id)
result["comments"] = comments
# 如果需要用户信息
if request.include_user_info and note.get("user"):
user_id = note["user"].get("user_id")
if user_id:
user_info = self.client.get_user_info(user_id)
result["user_info"] = user_info
return result
except Exception as e:
raise HTTPException(status_code=500, detail=f"获取笔记失败: {str(e)}")
async def search_notes(self, request: SearchRequest) -> dict:
"""搜索笔记"""
if not self.client:
raise HTTPException(status_code=500, detail="客户端未初始化")
try:
# 映射排序类型
sort_mapping = {
"general": SearchSortType.GENERAL,
"popularity": SearchSortType.MOST_POPULAR,
"latest": SearchSortType.LATEST
}
# 映射笔记类型
type_mapping = {
"all": SearchNoteType.ALL,
"video": SearchNoteType.VIDEO,
"image": SearchNoteType.IMAGE
}
result = self.client.get_note_by_keyword(
keyword=request.keyword,
page=request.page,
page_size=request.page_size,
sort=sort_mapping.get(request.sort, SearchSortType.GENERAL),
note_type=type_mapping.get(request.note_type, SearchNoteType.ALL)
)
return {
"keyword": request.keyword,
"page": request.page,
"total": result.get("total", 0),
"notes": result.get("notes", []),
"has_more": result.get("has_more", False)
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"搜索失败: {str(e)}")
async def analyze_users(self, request: UserAnalysisRequest) -> dict:
"""分析用户数据"""
if not self.client:
raise HTTPException(status_code=500, detail="客户端未初始化")
try:
analyzer = XhsDataAnalyzer(self.client)
metrics_df = analyzer.collect_user_metrics(request.user_ids)
# 根据请求的指标筛选数据
available_metrics = [
col for col in metrics_df.columns
if col in request.metrics or col in ["user_id", "username"]
]
result_df = metrics_df[available_metrics]
# 计算统计信息
stats = {}
for metric in request.metrics:
if metric in metrics_df.columns:
stats[f"{metric}_mean"] = metrics_df[metric].mean()
stats[f"{metric}_median"] = metrics_df[metric].median()
stats[f"{metric}_std"] = metrics_df[metric].std()
stats[f"{metric}_max"] = metrics_df[metric].max()
stats[f"{metric}_min"] = metrics_df[metric].min()
return {
"analysis_time": datetime.now().isoformat(),
"user_count": len(metrics_df),
"metrics": result_df.to_dict(orient="records"),
"statistics": stats,
"top_performers": self._get_top_performers(metrics_df, request.metrics)
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"分析失败: {str(e)}")
def _get_top_performers(self, df, metrics, top_n=3):
"""获取表现最佳的用户"""
top_performers = {}
for metric in metrics:
if metric in df.columns:
sorted_df = df.sort_values(by=metric, ascending=False).head(top_n)
top_performers[metric] = sorted_df[["user_id", "username", metric]].to_dict(orient="records")
return top_performers
# 依赖注入
def get_xhs_service():
"""获取小红书服务实例"""
service = XhsAPIService()
# 这里应该从配置或环境变量中获取cookie
service.init_client(cookie="your_cookie_here")
return service
# API路由
@app.post("/api/note")
async def get_note(request: NoteRequest, service: XhsAPIService = Depends(get_xhs_service)):
"""获取笔记详情API"""
return await service.get_note_details(request)
@app.post("/api/search")
async def search(request: SearchRequest, service: XhsAPIService = Depends(get_xhs_service)):
"""搜索笔记API"""
return await service.search_notes(request)
@app.post("/api/analyze/users")
async def analyze_users(request: UserAnalysisRequest, service: XhsAPIService = Depends(get_xhs_service)):
"""分析用户数据API"""
return await service.analyze_users(request)
@app.get("/api/health")
async def health_check():
"""健康检查端点"""
return {"status": "healthy", "timestamp": datetime.now().isoformat()}
# 启动服务
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
未来路线图：技术演进与生态扩展
xhs项目作为小红书数据采集的解决方案，未来将在以下几个方向进行技术演进和功能扩展：
1. 异步IO支持与性能优化
计划引入asyncio和aiohttp支持，实现真正的异步数据采集，大幅提升大规模数据采集的效率。通过协程和连接池优化，预计可以将数据采集速度提升3-5倍。
2. 分布式采集架构
开发分布式采集框架，支持多节点协同工作、负载均衡和故障转移。通过Redis或RabbitMQ实现任务队列管理，支持水平扩展，满足企业级大数据采集需求。
3. 智能反爬虫策略
集成机器学习算法，自动识别和应对网站的反爬虫机制。包括请求频率自适应调整、用户行为模拟优化、验证码自动识别等功能。
4. 数据管道与ETL集成
提供完整的数据管道解决方案，支持与Apache Airflow、Apache NiFi等ETL工具集成。实现数据采集、清洗、转换、加载的全流程自动化。
5. 实时数据处理
支持Kafka、RabbitMQ等消息队列，实现实时数据流处理。结合Apache Flink或Apache Spark Streaming，提供实时数据分析能力。
6. 云原生部署
提供Docker容器化部署方案，支持Kubernetes集群部署。实现自动扩缩容、服务发现和配置管理，降低运维复杂度。
7. 生态系统扩展
开发与其他数据分析工具的插件和扩展，包括：
- Tableau/Power BI数据连接器
- Elasticsearch数据索引插件
- Grafana监控仪表板
- Jupyter Notebook集成
8. 合规与安全增强
加强数据采集的合规性检查，提供数据脱敏、隐私保护功能。建立完善的审计日志和访问控制机制，确保数据使用的合法合规。
通过持续的技术创新和生态建设，xhs项目将发展成为小红书数据采集领域的技术标杆，为开发者和企业提供更加完善、高效、安全的数据服务解决方案。
【免费下载链接】xhs 基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/ 项目地址: https://gitcode.com/gh_mirrors/xh/xhs
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
