做数据分析和内容运营的朋友应该都遇到过这种需求想批量获取小红书笔记的详情包括标题、点赞数、评论数、视频播放地址、封面图这些结构化信息。市面上各种插件和在线解析工具不少但要么不稳定要么字段不全要么用着用着就被平台风控拦住了。我最近在给内部的内容监控系统做数据源接入时封装了一个名为item_get_video_pro的统一接口专门用来获取小红书笔记详情。这篇文章不打算讲什么高深理论就把示例代码逐段拆开讲清楚每个参数的设计意图、每次请求背后的交互逻辑以及我在实际调试中踩过的坑。如果你正在写类似的数据采集服务或者想给自己爬虫工具加一个稳定的详情获取入口这篇文章可以参考。1. 项目拆分item_get_video_pro 到底是个什么角色先把这个接口的名字拆开看item_get_video_pro其实是三部分语义的组合item表示它处理的是笔记对象本身get_video点明它重点解决视频类笔记的解析pro后缀则表示这是增强版接口不只是拿基础信息还会返回视频直链、各清晰度地址、封面图、互动数据等完整详情。这个定位非常明确。市面上的通用笔记解析接口往往只返回标题和图片遇到视频笔记就抓瞎或者只给一个含水印的播放页地址没法直接用于内容分析。item_get_video_pro的设计目标是调用方传一个笔记 ID 或分享链接接口内部完成解析、鉴权、数据拼装最后返回一份标准化 JSON。调用方不需要关心小红书页面的 DOM 结构是怎样的、接口签名怎么算的也不需要维护登录态只要拿到 JSON 往里塞业务逻辑就行。从系统架构来看它应该处于你整个数据管道的中游位置。上游是笔记 ID 的采集来源比如你自己运营的账号列表、同行公开的爆款笔记清单、或者用户主动提交的链接下游是数据存储和分析模块比如入库 MySQL、同步到 Elasticsearch、推送企业微信告警。item_get_video_pro只负责把一个笔记 ID 变成完整的结构化数据这件事职责单一便于单独维护和升级。接口的基本交互模型可以画成这样的链路调用方传入 item_id - 服务端校验参数 - 检查本地缓存 - 未命中则请求上游数据源 - 解析原始响应 - 字段清洗与类型转换 - 返回标准 JSON 给调用方实际项目中我还给这个接口加了raw_origin这个调试字段方便出问题时快速判断数据是来自 Redis 缓存、本地静态文件还是在线实时拉取。这个字段在生产排障时非常有用后面细说。2. 前置条件不能跳过登录态、授权与合规边界开始写代码之前必须先解决一个容易被忽视的问题怎么合法地拿到笔记数据。小红书的界面和数据接口都有风控直接裸奔请求结果大概率是验证码、滑块甚至 IP 封禁。我这里说的不是一个可以公开协议的开放 API而是基于平台现有页面的数据解析服务所以在正式使用前有几道前置关卡必须处理。第一关是登录授权。常规思路是维护一个企业自有的小红书账号通过扫码登录拿到 Cookie 和对应的x-s签名参数。登录成功后服务端会颁发一个web_session的 Cookie 凭证后续请求都需要带上这个凭证才能访问笔记详情接口。注意这个凭证有有效期通常几个小时后就会失效所以项目里一定要设计一个登录态保活机制比如用定时任务模拟访问首页来刷新 Cookie或者接打码平台过验证码重新登录。第二关是合规边界。说句实在话批量采集平台内容这件事在对方用户协议里一般是不被允许的。所以我的做法是只采集自家账号产生的笔记、用户明确授权的内容以及用于学术研究和舆论分析的公开数据绝对不碰非公开笔记不做用户身份信息的批量收集不把抓到的视频图片重新打包分发。代码里我也加了一层白名单校验item_id不在授权列表内直接拒绝服务而不是把锅全甩给下游。第三关是频率控制。登录态即使有效每秒请求几十次一样会触发风控。正确做法是给整个采集过程设置一个全局的令牌桶比如每分钟最多 30 个请求单 IP 超过阈值直接熔断。这一步我放在服务端的网关层而不是在每个业务代码里做因为网关层可以统一控制多个接口的流量。提供一个小建议调试阶段不要用真实账号去频繁点页面先用公开的笔记 ID 和低频率做验证。等代码稳定了再上真实数据不然账号很容易被限制后面所有工作都得停下来。3. 核心代码拆解从请求构造到响应解析现在进入正题逐段解析示例代码。先看完整的核心部分然后我会把每一段的关键逻辑都讲清楚。import requests import json import time import hashlib import random from typing import Optional, Dict, Any from urllib.parse import urljoin, quote class XiaoHongShuDetailCrawler: 小红书笔记详情解析器 通过 web 页面数据接口获取笔记完整信息 def __init__(self, cookie: str, user_agent: str None): self.session requests.Session() self.session.headers.update({ Cookie: cookie, User-Agent: user_agent or Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Referer: https://www.xiaohongshu.com/, Origin: https://www.xiaohongshu.com, }) self.base_url https://www.xiaohongshu.com self.token_bucket {capacity: 30, tokens: 30, last_refill: time.time()} def _check_rate_limit(self): 简单令牌桶限速 now time.time() self.token_bucket[tokens] min( self.token_bucket[capacity], self.token_bucket[tokens] (now - self.token_bucket[last_refill]) * 1.0 ) self.token_bucket[last_refill] now if self.token_bucket[tokens] 1: raise RuntimeError(请求频率超限请稍后重试) self.token_bucket[tokens] - 1 def resolve_short_url(self, short_url: str) - str: 解析小红书分享短链得到真实笔记 ID 短链格式一般像 https://xhslink.com/xxxxx resp self.session.get(short_url, allow_redirectsFalse) location resp.headers.get(Location, ) # 从跳转地址中提取 /explore/ 或 /discovery/item/ 后面的 ID if /explore/ in location: return location.split(/explore/)[-1].split(?)[0] elif /discovery/item/ in location: return location.split(/discovery/item/)[-1].split(?)[0] else: # 部分短链需要跟随重定向再提取 for prefix in (/explore/, /discovery/item/): if prefix in resp.url: return resp.url.split(prefix)[-1].split(?)[0] raise ValueError(f无法从短链解析笔记ID, location{location}) def get_note_detail(self, note_id: str) - Dict[str, Any]: 获取笔记详情返回标准化结构 self._check_rate_limit() api_url urljoin(self.base_url, f/api/sns/web/v1/feed) # 实际请求体中需要传入笔记 ID 列表 payload { source_note_id: note_id, note_id: note_id, } headers { Content-Type: application/json;charsetUTF-8, } resp self.session.post(api_url, jsonpayload, headersheaders, timeout10) if resp.status_code ! 200: raise ConnectionError(fHTTP状态码异常: {resp.status_code}) data resp.json() # 业务状态码判断 if data.get(code) ! 0: raise RuntimeError(f接口返回错误: code{data.get(code)}, msg{data.get(msg)}) items data.get(data, {}).get(items, []) if not items: raise ValueError(响应中未找到笔记数据) note items[0] return self._normalize_note(note)这段代码表面上不长但每一行都有讲究。逐步拆开看。_check_rate_limit是一个最简单的令牌桶实现。容量设了 30也就是每轮最多放行 30 个请求令牌按 1 个/秒的速率补充。这个数字在实际使用中可以根据账号权重调整新号建议 10 个/分钟起测跑半小时没问题再慢慢往上加。我见过有人一上来就把并发拉到 50结果 10 分钟后 Cookie 直接失效所有请求都开始跳验证码数据全抓空了。resolve_short_url处理的是用户经常直接扔过来一个xhslink.com短链的情况。短链背后的真实笔记 ID 必须通过一次不跟随重定向的 GET 请求拿到Location头因为小红书分享短链设计成先跳到一个中间页再 302 到详情页。注意这里用了allow_redirectsFalse就是要手动截获第一次跳转地址避免 requests 自动把最后的页面内容也下载下来白白增加流量和风控风险。如果Location为空程序会退回检查resp.url因为有一些旧格式短链不会发Location头而是直接在响应 URL 里体现。get_note_detail是核心方法。接口路径用的是/api/sns/web/v1/feed这是 web 端获取笔记流数据的标准入口请求体是 JSON 格式里面放笔记 ID。这个方法原名是 feed 流接口但对单篇笔记详情同样适用。响应结构里业务状态码为 0 代表成功非 0 时msg会给出具体错误信息比如笔记不存在或笔记已删除。从data.items里取第一个元素就是目标笔记的原始数据。这里有一个关键点POST 请求要带正确的Content-Type并且 body 必须序列化为 JSON 字符串数组相关的字段如果传错格式平台会返回参数校验错误。我踩过一次很蠢的坑payload 里直接放字符串而不是对象结果接口返回缺少参数排查了半天才发现问题。再往下是数据规范化方法_normalize_note这是整个项目里最繁琐的部分def _normalize_note(self, raw_note: Dict[str, Any]) - Dict[str, Any]: 原始笔记数据规范化统一字段命名和类型 note_info raw_note.get(note_card) or raw_note.get(note) or raw_note video_info note_info.get(video, {}) or {} # 视频直链可能有多清晰度取最大的那个 video_url video_cover video_duration 0 video_media video_info.get(media, {}) if video_media: stream_types video_media.get(stream_types, []) if stream_types: # stream_types 是一个列表按清晰度从高到低排 master stream_types[0] video_url master.get(master_url) or master.get(backup_urls, [])[0] video_cover master.get(cover, ) video_duration master.get(duration, 0) # 图片数组解析 image_list [] image_entity note_info.get(image_list) or note_info.get(images) or [] for img in image_entity: info img.get(info_list) or img.get(url) or [] if isinstance(info, list): # 取最大尺寸 sorted_info sorted(info, keylambda x: x.get(width, 0) * x.get(height, 0), reverseTrue) if sorted_info: image_list.append(sorted_info[0].get(url, )) elif isinstance(info, str): image_list.append(info) interactive note_info.get(interact_info) or {} normalized { note_id: note_info.get(note_id) or raw_note.get(note_id), title: note_info.get(title) or note_info.get(display_title, ), desc: note_info.get(desc) or note_info.get(description, ), type: note_info.get(type, normal), # normal / video video_url: video_url, video_cover: video_cover, video_duration: video_duration, images: image_list, liked_count: interactive.get(liked_count, 0), collected_count: interactive.get(collected_count, 0), comment_count: interactive.get(comment_count, 0), share_count: interactive.get(share_count, 0), author: note_info.get(user, {}).get(nickname, ), author_id: note_info.get(user, {}).get(user_id, ), create_time: note_info.get(create_time, ), last_update_time: time.time(), } return normalized规范化这段看似是脏活累活其实决定了下游业务代码好不好写。原始返回里字段名比较混乱有的字段在note_card下有的在note下不同版本的接口字段还不一样。所以这里我的思路是用多重 fallback 取字段比如视频信息先看note_card.video没有就找note.video都没有就用空字典兜底。这样即使上游返回结构有轻微调整代码也不会直接崩溃。视频直链的解析是item_get_video_pro这个 pro 版本的核心卖点。视频信息里media.stream_types是一个按清晰度排列的列表第一个通常是 1080p 或 720p 的 master 流。每一路流里有master_url这是 CDN 直链可以直接用来拉流分析和存储。注意有些情况下master_url是空的需要用backup_urls列表里的第一个备选地址。我在实际使用中发现视频地址的 CDN 域名偶尔会变所以拿到 URL 后最好做一次 HEAD 请求验证是否可访问再决定入不入库。图片数组的处理相对简单但有个隐蔽的坑image_list里的每个元素不是直接给 URL而是info_list这个包含多个不同尺寸版本的结构。如果不做尺寸排序而直接取url字段拿到的往往是模糊的缩略图。我在代码里按宽高乘积分从大到小排序取最大的一张才能保证图片分析时不会因为分辨率太低导致 OCR 识别失败。interact_info里的互动数据全是字符串格式不是整数。这个设计可能是为了兼容超大数值。我在规范化字段里保持字符串类型同时在使用时由下游自行转换这样最大限度保留原始精度也不会因为某个字段偶尔返回 None 字符串导致 int() 转换报错。4. 数据解析的完整链路拿到 JSON 之后还差几步接口能正确返回 JSON 只是第一步真正让数据可用还需要经历一个完整的解析链路。我总结为校验 - 抽取 - 清洗 - 装载四步。校验阶段需要判断返回的笔记当前是否可访问。小红书不少笔记会因为违规或作者主动删除变成幽灵笔记接口可能返回 200 但items是空数组也可能 code 非 0。我的做法是在get_note_detail里对三种常见错误做区分笔记不存在、笔记已删除、请求过于频繁分别对应不同的日志级别。前两种属于数据问题记录下来即可最后一种属于策略问题需要触发全局限速。抽取阶段就是_normalize_note里干的事把扁平的原始字典转成结构清晰的normalized对象。这里我会刻意把视频信息、图片信息、互动信息分成独立的三块方便下游按需取用。比如内容分析服务可能只需要标题和正文运营报表服务需要互动数据视频去重服务只需要video_url加上视频长度。字段分得清楚下游代码就不用每次对着原始结构猜。清洗阶段主要处理脏数据。比如标题里存在大量不可见字符、全角半角混合、emoji 和表情符号夹杂这些对文本分光和搜索非常不友好。我在代码中增加了_clean_text工具函数统一去除控制字符、折叠连续空白、把全角逗号句号转半角。再比如图片 URL 里可能带动态签名参数x-oss-process后缀需要裁掉才能拿到干净的原图地址。这类细节不处理数据入库后做二次清洗的成本比你想象的大得多。装载阶段就是组装最终的结果。我习惯在返回的字典里额外写入一个fetch_source字段标明数据是online实时抓的还是cache缓存读的。这样一旦线上反馈数据不对可以快速判断是不是上游接口改版导致缓存数据过期。还有一个经验不要在解析函数里直接打印整个原始 JSON。一是日志会被刷爆二是部分字段可能包含用户敏感信息。我遇到过有人把完整的响应体打到日志里结果被审计点名麻烦得很。正确做法是只打印note_id、fetch_source、耗时ms、HTTP状态码这几个精简字段。5. 工程化落地缓存、重试、超时与并发控制示例代码能跑通 Demo 距离生产可用还很远真正上线前必须补上三个基础能力缓存、重试和并发控制。缓存的意义不用多说。同一篇笔记经常被不同下游任务重复请求比如运营看一次、算法分析看一次、导出报表又看一次。每个请求都打到线上数据源既浪费配额又增加风控风险。我的方案是引入 Redis 做两级缓存一级缓存过期时间 15 分钟应对热点笔记的短时反复查询二级缓存过期时间 24 小时防止冷门笔记被低频任务反复请求。缓存 key 直接用明文xh_note_detail:{note_id}value 用 JSON 序列化后的normalized结果。读取缓存时加一个from_cache标记方便日志排查。import redis class CacheLayer: def __init__(self, redis_url: str): self.r redis.Redis.from_url(redis_url) def get_note(self, note_id: str) - Optional[Dict[str, Any]]: raw self.r.get(fxh_note_detail:{note_id}) if raw: return json.loads(raw) return None def set_note(self, note_id: str, data: Dict[str, Any], ttl: int 900): self.r.setex(fxh_note_detail:{note_id}, ttl, json.dumps(data, ensure_asciiFalse))重试机制要讲究策略。小红书这种平台接口偶尔超时或者返回 5xx 是常态不能一失败就放弃。但也不能无脑重试不然会加剧对端压力。我的做法是最多重试 2 次采用指数退避第一次等 1 秒第二次等 3 秒。对于业务层面的错误码比如笔记不存在不进行重试因为那是不可恢复错误。这个判断放在except块里需要格外细致只捕获网络异常和 5xx 状态千万不要捕获所有异常就重试否则数据校验错误会被无限重放。并发控制方面除了前面提到的令牌桶我还用threading.Semaphore限制了整体线程数。采集服务里一般开 4 到 8 个 worker 就足够了再多容易触发风控而且也不会有明显的速度提升。真正让采集速度上去的不是无限开线程而是你有效管理 Cookie 和代理池。一个合格的 Cookie 池至少准备 5 个账号轮换使用当某个账号出现风控提示时自动剔除并切换到下一个。代理池同理至少要 10 个不同 C 段 IP 备用不要用共享数据中心 IP容易被关联封禁。from threading import Semaphore import concurrent.futures semaphore Semaphore(4) def safe_fetch(note_id: str): with semaphore: try: detail crawler.get_note_detail(note_id) cache.set_note(note_id, detail) return detail except Exception as e: log.error(fnote_id{note_id} 抓取失败: {e}) return None超时设置也值得单独说。requests.post里的timeout10指的是连接和读取的总超时不是单阶段。如果你只传 10 一个数字requests 会同时用作 connect 和 read 超时如果网络抖动厉害建议拆开写(3.05, 10)连接超时 3 秒读取超时 10 秒。我遇到过有些 CDN 视频地址连接很慢但读取很快的情况拆开设置后整体排队等待时间大幅下降。6. 踩坑实录风控、防盗链与字段陷阱最后这部分是我最想分享的因为这些坑在网上很少被系统性地总结只有自己动手做一遍才会遇到。第一个坑是防盗链问题。接口返回的视频直链和图片直链直接用浏览器打开没问题但用requests下载时会返回 403。原因是 CDN 校验Referer头必须带上小红书域名才给放行。我之前没有把Referer设置好导致视频下载全部失败排查半天才意识到是这一个头的缺失。解决办法是在下载时单独构造一个请求头Referer设为https://www.xiaohongshu.com/User-Agent设置为真实浏览器 UA其他头保持默认。第二个坑是短链解析的 302 链路过长。小红书短链有时候会先跳到xhslink.com/a/xxxx再跳到xiaohongshu.com如果只用一层allow_redirectsFalse拿到的Location还是一个中间链。我的处理是循环重定向最多 3 次每次都取Location直到发现/explore/或/discovery/item/为止。每次跳转后加一个极短的time.sleep(0.1)防止被安全网关识别为自动化。第三个坑是视频stream_types列表排序不稳定。文档上没有明确说第一个就是最高清实际验证发现有些笔记第一个 stream 反而是 360p 流畅版。所以代码里不能只取[0]要遍历stream_types找到width最大的那一路。这个改动很小但直接影响最终拿到的视频质量。类似地图片info_list的排序也会出现width和height缺失的情况排序前必须先做默认值填充。第四个坑是互动数据的水分。有些笔记的点赞数会包含平台自己加的推广计数和作者页看到的数字并不完全一致。如果你的业务是做精确的达人效果分析不能只看接口里的原始值建议在笔记数据表里同时记录source来源和fetch_time抓取时间方便后续做数据对账。我们内部就遇到过某篇笔记接口返回 3 万赞但作者后台实际只有 1.2 万赞的偏差当时就是靠fetch_time字段回溯定位到是推广计数问题。第五个坑是User-Agent不能长期固定同一个版本。Chrome 每几周就发新版本如果 UA 里的版本号和当前主流浏览器差距过大接口开始可能是正常的但随着时间推移会逐步增加风控概率。我的做法是在配置里维护一个 UA 池每次请求随机取一个并且每月更新一批版本号。这个细节很小但对稳定性的提升非常明显。我做了一个简易错误排查表可以直接对应处理现象可能原因处理方式返回 code 非 0提示请求过频频率过高或 Cookie 异常降低并发等待 5-10 分钟检查 Cookie 有效期HTTP 200 但 items 为空数组笔记已删除/违规下架/无权限记录 note_id 到黑名单停止重试视频链接总是 403Referer 头或 UA 被 CDN 拒绝补全 Referer 和小红书 UA短链解析拿不到 ID跳转逻辑变化或 Location 为空升级为重定向循环解析打印中间 URL某篇笔记字段大量缺失接口版本字段差异启用 fallback 字段逻辑兼容 note_card/note 两种结构还有一个容易被忽略的问题接口返回的create_time单位是毫秒还是秒不同版本不一致。我在_normalize_note里统一转成毫秒字符串返回下游不管是从 MySQL 还是从 ClickHouse 读都按毫秒处理避免因为单位问题导致时间统计全部偏差。这个转换极其简单但如果不做后面写时间维度报表时一定会炸一次。7. 从示例代码到稳定服务的最后一步如果只看上面那些代码你可能觉得项目已经完整了。但实际上还差最后一公里运维和可观测性。一个没有日志、没有监控的数据采集服务即使代码写得再漂亮上线两周后一定会让你焦头烂额。我在这个项目里养成的习惯是每一笔抓取都打印一条结构化日志字段固定为note_id、http_code、biz_code、cost_ms、sourceonline/cache、worker_id。所有日志通过标准输出流收集到日志平台再配两个必看的告警一是成功率低于 90% 就列入 P2 告警二是缓存命中率异常升高说明源站可能出问题导致大量回源失败就列入 P3 告警。关于缓存命中率这里有个反直觉的点缓存命中率太高不一定是好事。如果你很久没抓到新数据缓存命中率长期接近 100%可能说明你的抓取任务全在打老数据而没有把新增笔记纳入队列。所以我在格面板里同时展示抓取总量变化趋势如果总量一周都只有十几条那大概率是上游采集断了而不是服务变好了。最后给一个建议先小规模跑一周看数据质量再规模化。我见过太多人第一天就把 10 万条笔记塞进队列跑结果第二天账号被限制队列全变成失败重试日志刷疯了。正确节奏是第一天只跑 100 条把数据字段和真实笔记逐一比对确认没有解析偏差第二天扩到 1000 条观察耗时和成功率确认稳定后再放开到全量。这个节奏听起来保守但实际算下来反而比一次性踩坑重跑更快。这个项目做完之后我的体会是item_get_video_pro这类接口的难点从来不在能不能调到数据而在于怎么把一个临时能跑的脚本变成一个经得起长期运行的稳定服务。代码只是最表面的一层背后的限速策略、缓存设计、字段兼容、日志体系每一项都需要在真实数据上磨过一遍才靠谱。希望这篇文章的代码拆解和踩坑记录能帮你少走几步弯路。