小红书爬虫实战:三源采集接口解析与x-s签名风控方案
2026/9/14 2:06:58 网站建设 项目流程

简介:这是一份面向高校计算机相关专业学生与爬虫初学者的课程实训资源,围绕小红书笔记、主页、搜索三大核心场景,系统演示如何深度采集小红书用户与笔记数据,并内置详细文档说明,可直接用于Python爬虫课设、作业或项目初期立项演示。资源压缩包共4个文件,包含1个Python爬虫脚本、2个CSV数据文件及1个Markdown说明文档;其中脚本负责请求发送、页面解析与数据清洗,两个CSV分别存储用户基本信息与笔记详情,MD文档则覆盖环境配置、运行步骤、常见异常排错思路,整体包体仅5KB,轻量透明、便于逐行研读和二次修改。目前已有63人学习下载,适合需要快速抓取社交媒体公开数据的开发人员。借助该资源,读者既能掌握从URL分析、参数构造到结果落盘的完整爬虫链路,也能通过替换目标参数,将同一套逻辑迁移到其他内容平台,是一份实用性强、完成度较高的实训样例。

1. 小红书爬虫的三个数据入口:笔记、主页、搜索到底在爬什么

很多人第一次接触小红书爬虫时会误以为难点在于「怎么写 requests 抓 HTML」,真正上手才发现页面结构里全是动态渲染的数据接口,直接分析 DOM 反而绕了远路。这个标题把所有场景拆得很直白:笔记数据、主页数据、搜索数据,三类入口指向的是同一套底层 API 体系,只是请求参数和翻页游标不同。做用户数据深度挖掘的基本盘就是把这三种数据源全部打通,再按用户维度重新组织。

这种爬虫适合谁?简单说就是需要做竞品调研、投放分析、消费者洞察、舆情监测、本地生活热点追踪的人。你不需要把全站数据拉下来,只需要在给定关键词、用户主页和笔记 ID 的前提下,把结构化字段稳定地落进数据库。本文会顺着「三个数据入口的真实形态 → 最小采集骨架 → 签名与风控 → 数据落库」这个链路走一遍,所有示例都会把参数含义写清楚,方便你直接改造成自己的采集工具。

2. 小红书三源采集的技术选型与端点规律:从短链接到正式 API

2.1 为什么选 Web 端协议而不是 App 接口

小红书有 Web 端、App 端、小程序端三套数据通道。常见做法是优先梳理 Web 端接口,原因很现实:Web 端返回 JSON 结构最规整,部分接口甚至不需要高版本签名就能访问,调试和二次开发的成本最低。App 端虽然能拿到更多字段,但需要处理 protobuf 编解码、设备指纹上报、双向证书校验,很多中小型采集项目根本不需要那么深的字段,投入产出比不划算。

还有一个被低估的点:Web 端的数据字段与数据仓库里常见的「商业地理、舆情监测」类需求字段匹配度更高。比如笔记搜索结果里有note_card.avatarnote_card.nicknamenote_card.display_titlenote_card.interact_info.liked_count这类结构化字段,直接响应到字典里就能用。App 端的字段分散在二进制流中,解析成本反而高。

2.2 三种数据入口的真实端点与参数对照

经过社区反复验证和被多家采集服务商的批量请求格式比对,目前 Web 端核心端点可以归纳为下面这批:

数据入口端点路径核心参数翻页机制
搜索笔记/api/sns/web/v1/search/noteskeywordpagepage_sizesearch_id页码游标
用户主页/api/sns/web/v1/user_postedsec_uidcursornum游标 cursor
笔记详情/api/sns/web/v1/feedidsource单篇拉取
用户详情/api/sns/web/v1/user/otherinfotarget_user_idtarget_user_sec_id无翻页

这些端点是 Web 端流量最大的几条数据链路,和你在网页上点击搜索按钮后触发的网络请求一一对应。搜索接口的page参数是简单数字页码,但实际爬取时要注意search_id—— 这个参数是服务器下发的会话标识,你进入搜索页面时会先拿到一个search_id,后续翻页带上它会让风控判定更友好。

用户主页接口是最容易踩坑的。它的翻页用的是cursor游标,而且是字节流经过 URL 编码后的字符串,每次请求返回的 JSON 里会带着下一个cursor值。正确姿势是循环:请求 → 取 JSON 中的cursor→ 再次请求,直到has_more字段变成false

2.3 把分享短链接解析成真实 ID:小红书 id 解析网址在线方案

用户给你一个http://xhslink.com/m/xxxx短链,直接丢进请求函数会被服务器打回。因为短链是跳转中介,真正能作为搜索依据的是笔记 ID 和用户sec_uid。常见做法是先用 requests 请求短链,开启allow_redirects=True,让服务端返回 302 跳转到完整页面 URL,再从跳转后的链接里正则提取 ID。

import re import requests def resolve_short_url(short_url: str) -> str: # 必须关闭自动解压,保持原始 302 响应头 session = requests.Session() resp = session.get(short_url, allow_redirects=True, timeout=10) final_url = resp.url print(f"[resolve] redirect to: {final_url}") note_match = re.search(r"/explore/([a-f0-9]+)", final_url) user_match = re.search(r"/user/profile/([a-f0-9]+)", final_url) if note_match: return f"note_id: {note_match.group(1)}" if user_match: # 真正的 sec_uid 还需要再请求一次用户页面去解析 return f"sec_uid: {user_match.group(1)}" raise ValueError(f"无法从短链中解析ID: {final_url}")

final_url是 302 落地后的详情页链接,里面包含笔记 ID 或用户 ID;如果是用户主页,还需要用sec_uid去请求用户详情接口补齐整个映射。另一种做法是直接访问xhslink.com短链后取落地页 HTML 里的<title>或 meta 标签,但正则匹配 URL 路径是最稳的,因为 title 经常带表情符号和格式符,编码后不好处理。

网页端所有分享链接最终都指向/explore//user/profile/这两个路径前缀,记住这个规律,做批量解析时可以把几百条短链丢进线程池一起解,基本不会触发限流。

3. 用 requests 搭一个小红书三源采集的最小骨架

3.1 统一的会话保持与请求函数

爬虫工程里最容易被忽视的是会话上下文。用requests.Session()保持 Cookie 一致、默认 Header 一致,能显著降低被判定为新访问者的概率。我在实际项目里一般会构造一个BaseCollector类,把请求头、超时、重试、加密参数注入都封装到_request_with_sign()方法里,避免后期每个入口各写一套请求逻辑。

import time import random import requests class XHSCollector: def __init__(self, cookie: str, sign_func=None): self.session = requests.Session() self.session.headers.update({ "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/124.0.0.0 Safari/537.36" ), "Accept": "application/json", "Origin": "https://www.xiaohongshu.com", "Referer": "https://www.xiaohongshu.com/", }) self.session.cookies.set("cookie", cookie, domain=".xiaohongshu.com") self.sign_func = sign_func or (lambda url, data: {}) def _get_json(self, url: str, params: dict): # 签名注入必须在 params 拼装完成后进行,因为签名依赖 query 原文 for attempt in range(3): try: headers = self.sign_func(url, params) resp = self.session.get(url, params=params, headers=headers, timeout=10) resp.raise_for_status() data = resp.json() if data.get("code") == 0: return data.get("data") or {} if data.get("code") == 461: print(f"[461] 签名校验失败, 请求参数: {params}") time.sleep(2 + random.random()) except requests.RequestException as exc: print(f"[retry {attempt}] {exc}") time.sleep(3 * (attempt + 1)) return {}

逻辑说明:_get_json是整个采集器的唯一网络出入口,sign_func是一个注入签名的回调函数,返回带x-sx-t等头部的字典。重试 3 次的策略是应对网络抖动,不是应对风控封禁,风控需要更长的冷却时间。code == 461是小红书签名校验失败的典型返回码,遇到这个说明签名失效,只管打印请求参数方便定位是哪个环节算错了。

3.2 搜索翻页与主页游标:两种翻页模型的实现差异

搜索接口的翻页是显式页码,而主页是游标字符串,两者不能共用同一套循环逻辑。搜索接口最简单,循环请求直到某页返回空列表;主页则要解析返回体中的cursorhas_more字段。

def fetch_search_notes(self, keyword: str, max_pages: int = 5): all_notes = [] for page in range(1, max_pages + 1): params = { "keyword": keyword, "page": page, "page_size": 20, "search_id": self._get_search_id(keyword), "scope": "all", } data = self._get_json("/api/sns/web/v1/search/notes", params) items = (data.get("items") or []) if data else [] if not items: break for item in items: note = item.get("note_card", {}) all_notes.append(self._extract_note_fields(note)) time.sleep(1 + random.random()) return all_notes def fetch_user_notes(self, sec_uid: str, max_notes: int = 50): notes = [] cursor = "" while len(notes) < max_notes: params = { "sec_uid": sec_uid, "cursor": cursor, "num": 30, } data = self._get_json("/api/sns/web/v1/user_posted", params) if not data: break items = data.get("notes") or [] for item in items: notes.append(self._extract_note_fields(item)) if not data.get("has_more"): break cursor = data.get("cursor", "") time.sleep(1 + random.random() * 2) return notes

fetch_user_notes里的cursor初始为空字符串代表第一页,之后每一次把上次响应体里的cursor原样传回。num是单页数量,官方端最大为 30,调大没用。搜索接口的_get_search_id一般是先请求一次搜索首页拿 search_id,再开始翻页,更省事的做法是直接写死一个历史 search_id,在部分账号下也能跑通,但久了一样失效,建议每次采集开始时重新获取。

3.3 并发设计:爬虫 并发设计 到底哪个好

小红书不是一个「快就是好」的目标站,并发必须克制。经验判断:单账号 10 个并发请求内比较安全,超过 15 个会快速触发风控,表现为返回空列表、验证码、或code == -1。主流方案是ThreadPoolExecutor+ 信号量限速,而不是异步框架,因为瓶颈不在 IO,而在对目标站的礼貌度。

from concurrent.futures import ThreadPoolExecutor, as_completed import threading def batch_fetch_user_notes(self, sec_uids: list[str], max_workers=8): results = {} lock = threading.Lock() def worker(sec_uid): notes = self.fetch_user_notes(sec_uid, max_notes=50) with lock: results[sec_uid] = notes print(f"[done] {sec_uid} -> {len(notes)} notes") with ThreadPoolExecutor(max_workers=max_workers) as pool: futures = [pool.submit(worker, uid) for uid in sec_uids] for future in as_completed(futures): # 避免任务内部异常导致整体中断 try: future.result() except Exception as exc: print(f"[task error] {exc}") return results

线程池之间的停顿比线程数量更重要。max_workers=8只是控制同时存活的请求数,但每个任务内部还有time.sleep兜底,两层限速叠加才能稳定跑批。如果想做真正的分布式爬虫,把sec_uids换成 Redis 队列的消费任务即可,核心请求逻辑不变。

4. 小红书 x-s 签名与风控策略:被 461 拦下之后怎么办

4.1 签名在请求链中的位置与常见算法形态

小红书的大多数 Web API 都要带x-sx-t头,缺少或过期都会直接返回 461。这套签名是从页面中动态生成的:x-t是 13 位毫秒级时间戳,x-s则是基于请求目标、路径、参数和时间戳共同计算出来的密文。社区常见称呼是「x-s 签名」,实际算出来是一串由大小写字母和数字组成的变长字符串。

签名参数的长相大致是这样:

Header 名示例是否必须说明
x-sXYW1B2h8KhCmA...签名主体,每次请求不同
x-t1718409478123毫秒时间戳
x-b3-traceid0a0b0c...链路追踪 ID,部分接口缺失会降级
x-b3-spanid0a0b0c...同上

签名算法在不同的请求类型上有差异:搜索接口和主页接口的签名因子可能包含keywordpage参数文本,笔记详情接口会把idsource也拼进去;它们都是把「请求方法 + 路径 + query 原文 + body 原文」做拼接,再经过哈希和加密后生成x-s。这就是为什么必须在 params 组装好之后再调用签名函数,顺序反了会导致签名错乱。

4.2 注入签名的可运行方案:本地计算还是中控服务

对于个人采集脚本,最常见的方案是通过execjs调用一段和小红书核心签名逻辑等价的 JS。把生成的x-sx-t并入请求头,先用主页接口验证签名可用性,再开放到全部入口:

import execjs import time # 假设 signature.js 中实现了 create_x_s(url, params, ts) 函数 with open("signature.js", "r", encoding="utf-8") as f: ctx = execjs.compile(f.read()) def sign_request(url, params): ts = str(int(time.time() * 1000)) query_string = "&".join(f"{k}={v}" for k, v in params.items()) x_s_value = ctx.call("create_x_s", url, query_string, ts) return { "x-s": x_s_value, "x-t": ts, "x-b3-traceid": ctx.call("gen_trace_id"), "x-b3-spanid": ctx.call("gen_span_id"), } collector = XHSCollector(cookie="your_cookie", sign_func=sign_request) notes = collector.fetch_search_notes("Python 爬虫", max_pages=2)

execjs方案的优势是本地计算,零网络依赖,单人使用足够稳定;缺点是和浏览器环境里的原生实现存在细微差异,如果签名逻辑里混入了 DOM 状态或随机指纹,execjs 计算结果会和页面端不一致。更工程化的做法是把签名模块独立成一个本地 HTTP 服务,采集进程通过requests.post("http://127.0.0.1:3000/sign", json={...})远程拿签名头。好处是分布式节点和多语言框架共用同一套签名逻辑,签名模块升级不需要改采集端代码。

4.3 频控预热、IP 代理与多账号预案

签名只是第一道门槛,真正的长效运行靠的是频控设计。启动采集任务前先做一次「预热」:用只带时间戳的请求轻量访问笔记详情接口 10 次,再逐步进入搜索和主页循环。预热的好处是让风控系统认为这个 IP 是正常浏览者,而不是刚开机就疯狂请求的机器人。

IP 代理的选型直接决定能跑多久。机房 IP 段被检测后大概率直接拒绝连接,住宅代理搭配按次计费的轮换模式是社区里更常见的稳定组合。建议每个 IP 在采集一个会话内只承担不超过 200 次请求,会话结束后强制切换 IP。如果单账号长期高频,记录日志里出现大量-1或验证码时,立刻停止,等待一小时以上再换账号继续。多账号可以落到一个配置表里,形如account_id, cookie, 今日请求量, 状态,由调度器统一分配。

5. 用户数据落地:去重、结构化与画像字段的扩展

5.1 三层去重:笔记、用户、内容指纹

采集系统最怕的不是跑得慢,而是同一批数据写进数据库后把统计口径打乱。笔记详情只应该有一条记录,用户主页无论采集多少次都该更新而不是追加,搜索和主页返回的内容需要跨来源去重。所以常规做法是定一个三层去重规则:笔记按note_id、用户按sec_uid、内容指纹按标题+正文摘要的 MD5 值。

import hashlib def dedupe_key(kind: str, *fields) -> str: raw = "|".join(fields) return f"{kind}:{hashlib.md5(raw.encode('utf-8')).hexdigest()}" # 使用示例 note_key = dedupe_key("note", note_id) user_key = dedupe_key("user", sec_uid) fingerprint = dedupe_key( "content", display_title, str(desc.get("desc", ""))[:200] )

dedupe_key把去重逻辑收敛到一个函数里,方便后续换 Redis SET 或 SQLite 唯一索引;内容指纹截取前 200 字可以避免超长字符导致 hash 性能下降。对评论数据需要注意:同一用户名可能在不同笔记下重复出现,所以评论实体的唯一键应该是note_id + 时间戳 + 用户ID,三层去重规则不适用于评论表。

5.2 SQLite 增量归档:采集任务和数据表结构分离

采集任务本身是有状态的:跑了一半挂掉,下次要能接续上游。维护一个轻量 SQLite 数据库做两级落库:任务表和明细表。任务表记录每个sec_uid的采集状态(待处理、采集中、已完成、失败原因),明细表存笔记正文和用户画像字段。

CREATE TABLE IF NOT EXISTS task_queue ( sec_uid TEXT PRIMARY KEY, status TEXT DEFAULT 'pending', last_cursor TEXT, retry_count INT DEFAULT 0, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS user_profile ( sec_uid TEXT PRIMARY KEY, nickname TEXT, avatar TEXT, total_note INT, following INT, follower INT, raw_json TEXT, fetched_at DATETIME ); CREATE TABLE IF NOT EXISTS note_content ( note_id TEXT PRIMARY KEY, sec_uid TEXT, title TEXT, liked_count INT, collected_count INT, comment_count INT, raw_json TEXT, created_at DATETIME );

raw_json字段非常重要。API 返回的结构化字段可能随版本迭代新增,只保存固定列会把新的好字段丢弃。任何时候都保留原始 JSON,之后做字段扩展只需跑一次回刷任务。task_queue.statuslast_cursor的设计让断点续采不再需要重新从第一页跑,直接从上次cursor接上即可。

5.3 从「数据」到「用户侧写」的最小计算

最后一层是把采集到的原始字段加工为可分析的画像数据。这个过程在小红书用户数据挖掘里一般包含三个维度:互动率、发布频率、内容方向。以user_profilenote_content两表为基础,一条 SQL 就能算出用户的互动倾向和活跃度:

SELECT n.sec_uid, COUNT(*) AS total_notes, AVG(liked_count + collected_count + comment_count) AS avg_interaction, MIN(n.created_at) AS first_post_time, MAX(n.created_at) AS latest_post_time FROM note_content n GROUP BY n.sec_uid HAVING total_notes >= 10 ORDER BY avg_interaction DESC;

avg_interaction反映的是基础数据,但「爆款率」更能说明内容质量:比如SUM(CASE WHEN liked_count >= 1000 THEN 1 ELSE 0 END) / COUNT(*)就是千赞率。你可以把这些聚合结果写回user_profile表的新增列里,也可以直接落到数仓做后续建模。这里有个容易踩的坑:liked_count是快照值而非增量值,如果做时间序列分析,必须保留每次采集的历史记录表,不能只存最新字段。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询