小红书笔记API v4 实战：Python 脚本批量获取20条笔记数据与互动指标-平芜编程栈

小红书笔记API v4 实战：Python脚本批量获取20条笔记数据与互动指标

在内容运营和数据分析领域，获取平台数据是开展工作的第一步。小红书作为国内领先的生活方式分享平台，其API接口为开发者提供了高效获取数据的途径。本文将深入探讨如何通过Python脚本调用小红书笔记API v4的GraphQL接口，实现批量获取20条笔记数据及互动指标的技术方案。

1. 环境准备与API基础

在开始编写脚本前，我们需要完成必要的准备工作。首先确保已安装Python 3.7或更高版本，并准备好以下库：

pip install requests pandas python-dotenv

小红书API v4采用GraphQL架构，与传统的REST API相比，它允许客户端精确指定需要返回的字段，避免了数据过度获取或不足的问题。要使用API，你需要：

注册小红书开放平台开发者账号
创建应用并获取API权限
生成访问令牌(Access Token)

提示：将敏感信息如API密钥存储在环境变量中，不要直接硬编码在脚本里

2. 构建GraphQL查询语句

GraphQL的核心是查询语句的构建。以下是一个获取20条笔记基础信息和互动指标的查询示例：

query { notes(page: 1, size: 20) { list { id title content author { id nickname } stats { likes collects comments shares } tags createdAt } totalCount } }

这个查询请求第一页的20条笔记，包含以下字段：

笔记ID、标题和内容
作者ID和昵称
互动数据（点赞、收藏、评论、分享）
标签和创建时间
符合条件的笔记总数

3. Python实现API调用

下面是一个完整的Python实现，包含错误处理和数据处理：

import requests import json from dotenv import load_dotenv import os import pandas as pd # 加载环境变量 load_dotenv() class XiaohongshuAPI: def __init__(self): self.api_url = "https://api.xiaohongshu.com/v4/graphql" self.access_token = os.getenv("XHS_ACCESS_TOKEN") self.headers = { "Authorization": f"Bearer {self.access_token}", "Content-Type": "application/json" } def fetch_notes(self, page=1, size=20): query = """ query GetNotes($page: Int!, $size: Int!) { notes(page: $page, size: $size) { list { id title content author { id nickname } stats { likes collects comments shares } tags createdAt } totalCount } } """ variables = {"page": page, "size": size} try: response = requests.post( self.api_url, headers=self.headers, json={"query": query, "variables": variables} ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"API请求失败: {e}") return None def process_data(self, raw_data): if not raw_data or "data" not in raw_data: return None notes = raw_data["data"]["notes"]["list"] processed = [] for note in notes: processed.append({ "笔记ID": note["id"], "标题": note["title"], "内容摘要": note["content"][:100] + "..." if note["content"] else "", "作者": note["author"]["nickname"], "作者ID": note["author"]["id"], "点赞数": note["stats"]["likes"], "收藏数": note["stats"]["collects"], "评论数": note["stats"]["comments"], "分享数": note["stats"]["shares"], "标签": ", ".join(note["tags"]), "创建时间": note["createdAt"] }) return pd.DataFrame(processed) # 使用示例 if __name__ == "__main__": api = XiaohongshuAPI() data = api.fetch_notes() if data: df = api.process_data(data) print(df.head()) df.to_csv("xiaohongshu_notes.csv", index=False, encoding="utf_8_sig")

4. 数据处理与分析技巧

获取到原始数据后，我们可以进行多种分析。以下是几个实用的数据处理示例：

4.1 互动率计算

互动率是衡量内容表现的重要指标，可以通过以下公式计算：

df["互动率"] = (df["点赞数"] + df["收藏数"] + df["评论数"] + df["分享数"]) / df["点赞数"].max() * 100

4.2 热门标签分析

分析哪些标签最常出现：

from collections import Counter all_tags = [] for tags in df["标签"].str.split(", "): all_tags.extend(tags) tag_counts = Counter(all_tags) top_tags = tag_counts.most_common(10)

4.3 时间序列分析

分析内容发布的时间分布：

df["创建时间"] = pd.to_datetime(df["创建时间"]) df["小时"] = df["创建时间"].dt.hour hourly_dist = df.groupby("小时").size()

5. 高级应用与优化

5.1 分页获取更多数据

要获取超过20条数据，需要实现分页逻辑：

def fetch_all_notes(total=100): all_notes = [] page = 1 size = 20 while len(all_notes) < total: data = fetch_notes(page, size) if not data or not data["data"]["notes"]["list"]: break all_notes.extend(data["data"]["notes"]["list"]) page += 1 # 避免速率限制 time.sleep(1) return all_notes[:total]

5.2 缓存机制

为减少API调用，可以添加简单的缓存：

from datetime import datetime, timedelta class CachedXiaohongshuAPI(XiaohongshuAPI): def __init__(self): super().__init__() self.cache = {} self.cache_expiry = timedelta(hours=1) def fetch_notes(self, page=1, size=20): cache_key = f"notes_{page}_{size}" if cache_key in self.cache: cached_data, timestamp = self.cache[cache_key] if datetime.now() - timestamp < self.cache_expiry: return cached_data data = super().fetch_notes(page, size) if data: self.cache[cache_key] = (data, datetime.now()) return data

5.3 异常处理与重试

增强API调用的健壮性：

from tenacity import retry, stop_after_attempt, wait_exponential class RobustXiaohongshuAPI(XiaohongshuAPI): @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) def fetch_notes(self, page=1, size=20): try: response = requests.post( self.api_url, headers=self.headers, json={"query": query, "variables": variables}, timeout=10 ) if response.status_code == 429: retry_after = int(response.headers.get("Retry-After", 60)) time.sleep(retry_after) raise Exception("Rate limited") response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"请求失败: {e}") raise