小红书数据采集工程化实战：从笔记、评论到图片的爬虫与落库

简介这份资源面向小红书运营者、数据采集开发者与自动化脚本学习者围绕平台数据采集、爬虫工程、账号养号、内容发布、评论互动与数据分析等场景提供一套可参考的实践代码与前端工程结构。包内共277个文件以107个ts脚本、77个vue组件、18个scss样式为主辅以js、json配置、png/svg图标及rs、toml、yml等工程文件压缩包约709KB整体偏向中小型前端项目与自动化模块的源码组织。已有74人学习下载。读者可从中了解登录模拟、动态渲染解析、频率控制、Session管理等采集思路以及笔记发布、评论响应、竞品监控、数据漏斗分析等运营自动化的实现框架适合作为搭建自有工具链、排查接口与理解目录结构的参考素材。1. 小红书数据采集从笔记、评论到图片一套能跑通的工程化路径小红书的页面结构这两年改得越来越勤很多人第一次写小红书爬虫打开开发者工具能拿到数据换成代码请求就返回空或者跑两天账号就被限流。问题不在 Python 写得对不对而在于小红书的数据采集本质上是一套「请求构造 会话维持 频率控制 数据落库」的工程活不是几十行 requests 就能收工的。这篇笔记面向的是想正经做小红书运营、小红书数据分析、小红书自动化的从业者你需要把笔记、评论、图片这些数据稳定拿下来存进数据库再喂给后面的分析或发布流程。我会按「先搞清楚数据从哪来 → 再搭最小可跑采集 → 再处理图片和评论 → 最后讲风控和避坑」的顺序把每一步的命令、参数和翻车点都摊开讲。新手能照着复现熟手能直接看到边界在哪。2. 小红书数据到底藏在哪接口、签名与页面结构2.1 先分清三类数据源别一上来就写爬虫做小红书采集之前先要判断你要的数据属于哪一类因为不同类别的获取成本差一个数量级。第一类是笔记列表和笔记详情包括标题、正文、点赞收藏数、发布时间、作者信息。这类数据在小红书的 Web 端和 App 端都有对应的接口返回的是 JSON字段结构相对稳定是采集的主力目标。第二类是评论数据包括一级评论和二级回复。评论接口通常带分页游标而且和笔记详情不在同一个请求里需要单独构造。评论的翻页逻辑比笔记列表更容易触发风控因为请求频率天然更高。第三类是图片和视频资源笔记里的图片是 CDN 直链拿到 URL 之后可以直接下载但要注意图片有防盗链和尺寸参数直接存 URL 和存文件是两回事。常见做法是先用浏览器抓一次真实请求把 URL、请求头、Cookie、签名参数完整复制下来再在 Python 里复现。不要凭猜测拼参数小红书的接口对缺失字段的容忍度很低。2.2 签名参数是第一个黑匣子先观察再复现小红书 Web 接口里通常带几个看起来像乱码的参数比如x-s、x-t这类签名字段。它们由前端 JS 根据请求路径、时间戳、Cookie 里的某些值计算出来。你直接改 URL 里的关键词签名就对不上服务端会返回错误码或者空数据。处理思路有两种。一种是用浏览器自动化工具比如 Playwright驱动真实页面让页面自己发请求你只负责拦截响应。这种方式不用逆向签名稳定性高代价是速度慢、资源占用大。另一种是复现签名算法把前端 JS 里的计算逻辑用 Python 重写速度快但前端一改算法你就得跟着改。我一般会这样选如果只是做小规模的小红书数据分析数据量在几千条以内直接用 Playwright 拦截响应最省心如果要长期、大批量采集才值得投入精力去复现签名并且要做好算法随时失效的准备。from playwright.sync_api import sync_playwright import json captured [] def handle_response(response): # 只拦截笔记详情接口URL 关键词按实际抓包结果替换 if /api/sns/web/v1/feed in response.url: try: data response.json() captured.append(data) except Exception: pass with sync_playwright() as p: browser p.chromium.launch(headlessFalse) context browser.new_context() page context.new_page() page.on(response, handle_response) # 打开目标笔记页触发真实请求 page.goto(https://www.xiaohongshu.com/explore/笔记ID) page.wait_for_timeout(5000) browser.close() print(json.dumps(captured, ensure_asciiFalse)[:500])这段代码的逻辑是启动一个带界面的 Chromium注册响应监听凡是 URL 里包含笔记接口特征的响应都尝试解析成 JSON 并收集。headlessFalse是为了方便你观察页面是否正常加载正式跑可以改成True。wait_for_timeout给页面留出加载和发请求的时间具体数值按网络情况调整太短会漏掉响应。参数说明/api/sns/web/v1/feed只是示例路径你必须用自己抓包得到的真实路径替换否则拦截不到任何东西。提示第一次跑一定要开headlessFalse看着页面动起来确认接口真的被触发了再去调后面的解析逻辑。2.3 请求头里哪几个字段不能省即使用 Playwright 拦截你也会发现有些接口在无头模式下返回的数据和真实浏览器不一样。原因通常在请求头。小红书的接口对User-Agent、Referer、Cookie这几个字段比较敏感。User-Agent要和你实际使用的浏览器版本一致不要用一个几年前的 UA 字符串。Referer一般要指向笔记所在页面缺失时可能被判定为异常来源。Cookie里包含登录态和部分风控标识是维持会话的关键采集过程中要保证 Cookie 有效过期了就要重新获取。如果你用 requests 直接请求建议把抓包得到的完整请求头原样复制只替换掉会变化的时间戳和签名参数。不要自作聪明删掉看起来没用的字段很多字段就是用来做风控校验的。3. 搭一套最小可跑的小红书采集从请求到 SQLAlchemy 落库3.1 用 requests 复现单条笔记请求在正式写采集循环之前先用一条笔记把请求跑通。这一步的目标不是拿多少数据而是确认你的请求头、Cookie、签名参数能换来正常响应。import requests headers { User-Agent: 你的浏览器UA, Referer: https://www.xiaohongshu.com/, Cookie: 你抓包得到的完整Cookie, # 签名相关字段按抓包结果补齐 x-s: 抓包得到的值, x-t: 抓包得到的时间戳, } url https://www.xiaohongshu.com/api/sns/web/v1/feed payload { source_note_id: 笔记ID, image_formats: [jpg, webp, avif], extra: {need_body_topic: 1}, } resp requests.post(url, headersheaders, jsonpayload, timeout10) print(resp.status_code) print(resp.text[:800])逻辑说明用 POST 提交 JSON body这是小红书笔记详情接口的常见形式。source_note_id是笔记的唯一标识从分享链接里解析得到。image_formats决定返回的图片格式按需保留。参数说明timeout一定要设否则网络卡住时脚本会一直挂着resp.text[:800]是为了先看返回结构确认字段名之后再写解析。如果返回的是空数据或者错误码优先检查三件事Cookie 是否过期、签名参数是否和当前请求匹配、请求体字段名是否写错。这三样占失败原因的八成以上。3.2 用 SQLAlchemy 定义笔记和评论表数据拿到之后要有地方存。用 SQLAlchemy 定义模型比裸写 SQL 更好维护也方便后面接数据分析。from sqlalchemy import create_engine, Column, String, Integer, Text, DateTime from sqlalchemy.orm import declarative_base, sessionmaker from datetime import datetime Base declarative_base() class Note(Base): __tablename__ xhs_note id Column(Integer, primary_keyTrue, autoincrementTrue) note_id Column(String(64), uniqueTrue, indexTrue) title Column(String(255)) content Column(Text) author_id Column(String(64), indexTrue) liked_count Column(Integer, default0) collected_count Column(Integer, default0) comment_count Column(Integer, default0) publish_time Column(DateTime, nullableTrue) created_at Column(DateTime, defaultdatetime.now) class Comment(Base): __tablename__ xhs_comment id Column(Integer, primary_keyTrue, autoincrementTrue) comment_id Column(String(64), uniqueTrue, indexTrue) note_id Column(String(64), indexTrue) user_id Column(String(64), indexTrue) content Column(Text) liked_count Column(Integer, default0) created_at Column(DateTime, defaultdatetime.now) engine create_engine(sqlite:///xhs.db, echoFalse) Base.metadata.create_all(engine) Session sessionmaker(bindengine)逻辑说明note_id和comment_id都加了uniqueTrue配合后面的去重写入避免重复采集产生脏数据。indexTrue是为了后面按作者、按笔记查询时速度不至于太慢。参数说明数据库先用 SQLite 跑通数据量上到几十万条再换 MySQL 或 PostgreSQLcreate_engine的连接串换掉即可模型代码不用动。3.3 写入时怎么做去重和更新采集最怕的是重复写入。常见做法是「先查后写」但数据量大时这样效率低。更稳的方式是用数据库的唯一约束配合 upsert 逻辑。from sqlalchemy.dialects.sqlite import insert def save_note(session, note_data): stmt insert(Note).values(**note_data) # 冲突时更新互动数据保留原始发布时间 stmt stmt.on_conflict_do_update( index_elements[note_id], set_{ liked_count: stmt.excluded.liked_count, collected_count: stmt.excluded.collected_count, comment_count: stmt.excluded.comment_count, }, ) session.execute(stmt) session.commit()逻辑说明on_conflict_do_update在 note_id 冲突时只更新互动数据不覆盖标题和正文避免因为接口返回字段缺失把已有内容冲掉。参数说明index_elements必须和模型里的唯一约束对应如果你用的是 MySQL这段要换成on_duplicate_key_update语法不同但思路一致。注意互动数据是随时间变化的重复采集同一条笔记时更新计数是合理的但正文和发布时间一般不变不要无脑全字段覆盖。4. 评论翻页、图片下载与小红书图片提取的实操细节4.1 评论分页的游标怎么接评论接口一般返回一个cursor或者end_id下一页请求要把它带上。很多人第一次写评论采集只拿到第一页就停了就是因为没处理游标。def fetch_comments(note_id, session, max_pages20): cursor for page in range(max_pages): payload { note_id: note_id, cursor: cursor, top_comment_id: , image_formats: [jpg, webp, avif], } resp requests.post(COMMENT_URL, headersheaders, jsonpayload, timeout10) data resp.json() comments data.get(data, {}).get(comments, []) if not comments: break for c in comments: save_comment(session, parse_comment(c, note_id)) cursor data.get(data, {}).get(cursor, ) if not cursor: break time.sleep(random.uniform(1.5, 3.5))逻辑说明循环里每次用上一页返回的 cursor 请求下一页拿不到评论或拿不到 cursor 就退出。max_pages是保护性上限防止接口异常时无限循环。参数说明time.sleep的随机区间是控制频率的关键固定间隔比随机间隔更容易被识别建议在 1.5 到 3.5 秒之间浮动具体按你的采集规模和账号情况调整。4.2 小红书图片下载URL 处理和防盗链笔记里的图片 URL 通常带尺寸参数和签名直接下载可能返回 403。处理方式是保留完整 URL请求时带上 Referer。import os def download_image(img_url, save_dir, filename): os.makedirs(save_dir, exist_okTrue) headers_img { User-Agent: headers[User-Agent], Referer: https://www.xiaohongshu.com/, } r requests.get(img_url, headersheaders_img, timeout15, streamTrue) if r.status_code 200: path os.path.join(save_dir, filename) with open(path, wb) as f: for chunk in r.iter_content(8192): f.write(chunk) return path return None逻辑说明streamTrue配合分块写入避免大图一次性读进内存。Referer是防盗链校验的关键缺失时大概率被拒。参数说明filename建议用「笔记ID_序号.jpg」的格式方便后面按笔记归类chunk大小 8192 字节是通用值网络差可以调小。4.3 图片提取后怎么和笔记关联图片下载完只是第一步真正做小红书数据分析时你需要知道哪张图属于哪条笔记。常见做法是在文件名里带 note_id或者在数据库里单独建一张图片表记录 note_id、图片序号、本地路径、原始 URL。如果后面要做图片内容分析建议保留原始 URL因为本地文件可能被清理而 URL 可以重新下载。图片表的结构不用复杂关键是 note_id 和序号这两个字段它们决定了你能不能把图片和笔记正文对应起来。5. 小红书风控与采集避坑账号、频率、数据一致性5.1 账号被限流的三个典型现象和原因现象一请求返回正常状态码但数据字段为空。原因通常是签名参数过期或 Cookie 失效服务端识别出请求不是来自真实会话返回了空壳数据。解决方式是重新抓包更新 Cookie 和签名参数不要试图用旧参数硬撑。现象二连续请求后突然全部返回错误码。原因是请求频率超过了风控阈值。解决方式是立刻停止采集降低频率增加随机等待必要时更换会话。硬顶着继续请求只会让限制时间变长。现象三账号能正常浏览但采集脚本拿不到数据。原因是脚本使用的请求特征和真实浏览器差异太大比如 UA 不匹配、缺少关键请求头。解决方式是把抓包得到的完整请求头逐字段比对补齐缺失项。5.2 频率控制不是加个 sleep 就完事很多人以为在循环里加time.sleep(1)就安全了实际上风控看的不只是间隔还有请求的规律性。固定间隔、固定顺序、固定数量的请求模式很容易被识别。我一般会这样做间隔用随机数范围根据采集量调整每采集 N 条插入一次较长的等待不要按笔记 ID 顺序连续采集打乱顺序能降低模式特征。这些手段不能保证绝对安全但能明显降低被限流的概率。5.3 数据一致性别让重复采集把库写乱采集脚本跑久了同一条笔记会被多次采集。如果没有去重逻辑数据库里会出现重复记录后面做数据分析时统计结果全是错的。除了前面说的唯一约束和 upsert还要注意时间字段的处理。互动数据用最新值覆盖是合理的但发布时间、作者 ID 这类不变字段不要覆盖。如果接口某次返回缺少某个字段upsert 时不要用空值覆盖已有值否则数据会被写坏。5.4 采集和运营的边界要清楚做小红书运营的人容易把采集和自动化发布混在一起。采集是读发布是写两者的风控策略完全不同。发布类操作的频率要更低内容要更接近真实用户行为否则账号风险比采集高得多。如果你的目标是小红书养号或自动化发布采集只能作为数据支撑不能把采集的高频策略直接套到发布上。6. 把采集数据用起来从原始表到可分析的数据看板采集跑通之后数据躺在 SQLite 里只是原料。真正体现价值的是把它变成能看、能查、能支撑决策的东西。我一般会先做一层轻量聚合再考虑接可视化。第一步是建一张按天聚合的统计表把笔记的点赞、收藏、评论按发布时间归到天这样能快速看出哪类内容在涨。from sqlalchemy import func, extract def daily_stats(session): rows ( session.query( func.date(Note.publish_time).label(day), func.count(Note.id).label(note_cnt), func.sum(Note.liked_count).label(total_liked), func.sum(Note.comment_count).label(total_comment), ) .filter(Note.publish_time.isnot(None)) .group_by(func.date(Note.publish_time)) .all() ) return rows逻辑说明按发布时间的天粒度分组统计每天的笔记数和互动总量。参数说明filter排除发布时间为空的记录避免脏数据影响统计如果数据库是 MySQLfunc.date可以直接用PostgreSQL 需要换成func.date_trunc。第二步是把聚合结果导出成 CSV喂给 Python 数据分析与可视化的常规流程。pandas 读进来之后按天画折线、按作者画柱状都是几行代码的事。这一步不用追求花哨能把趋势看清楚就够了。第三步是验证采集质量。随机抽几条笔记拿数据库里的互动数据和页面上显示的数字对一遍差得太多说明采集字段解析有问题。这个习惯能帮你早发现接口字段变更而不是等分析结果明显不对了才回头查。我自己踩过最深的一个坑是早期采集时没做去重跑了半个月发现同一条爆款笔记被写进去几十次导致那段时间的统计曲线完全失真后面花了一整天清洗数据。从那以后任何采集脚本上线前我都会先用小批量数据验证唯一约束和 upsert 逻辑确认重复采集不会污染库再放开跑。希望帮到你。本文还有配套的精品资源点击获取