企业官网建设流程全解析

简介：面向想学习小红书数据采集与接口算法还原的Python开发者，这份项目围绕毫秒级抓取、x-s与x-common参数解析、请求类型及接口封装展开，适合具备一定爬虫基础、希望理解JavaScript逆向与请求模拟的中级读者参考。压缩包共29个文件，以11个py源码为主，辅以yml配置、txt依赖与说明、md文档及license等，整体约28KB，体量轻便但结构完整，涵盖核心逻辑、异常处理、辅助工具、测试用例与文档配置。已有4451人学习下载，说明其在同类主题中具备一定关注度。读者可从中梳理平台请求封装的组织方式、参数生成与校验思路、测试验证流程和模块分层习惯，并借助现成脚本快速搭建自己的采集实验环境，减少从零摸索接口细节的成本。

1. 从“第二页开始全是空数组”说起：这套 Python 采集器解决的是什么问题

一位做选题分析的朋友找我，说他写的采集脚本重启后能出数据，翻到第二页就只回一个{"data": {"items": []}}，状态码却是 200，日志里一点错都不报。我打开浏览器开发者工具看了十分钟就明白了：问题出在请求头里那几个由前端 JavaScript 拼出来的签名参数，跟 requests 用法没关系。前端算一个哈希值塞进X-s，服务端校验不过，就直接返回空列表，连错误码都不给。

这份采集项目就是围绕这件事做的：把“参数拼装 → JS 签名生成 → 请求发送 → 响应归一化 → 落库”串成一条流水线，签名部分保留原始 JavaScript 并在 Node 里跑，Python 只负责调度、频控和数据。它适合两类开发者：一类是把采集脚本从“只能跑第一页”修好的，另一类是想完整走一遍 JavaScript 逆向补环境流程的。项目本身以学习和技术验证为主，实际使用时记得控制频率，只取公开可见的内容。

2. 项目结构与技术栈：Python 调度层加 Node 签名层的双进程设计

这个项目最容易被误读的地方，是很多人以为它是一套纯 Python 爬虫。实际上它有两个运行时：Python 负责网络、并发、解析和存储，Node 负责跑那段被混淆过的签名 JavaScript。两层之间用标准输入输出传 JSON，边界很清楚。这样的好处是签名算法升级时只换js/vendor/目录下的文件，Python 侧一行不用动；坏处是多一个运行时依赖，部署时要确认目标机器上有 Node，且版本不能太老。

技术栈上没什么花哨的：requests做 HTTP 会话，PyYAML读配置，sqlite3或 MySQL 做落库，pytest跑回归用例，Node 侧只用到vm、fs两个内置模块，不引入任何 npm 包。常见做法是能少一个依赖就少一个，因为一旦签名算法里有环境检测，多引入的库反而会带进一堆全局对象，把补环境的水搅浑。

2.1 目录结构：五层拆分，每层只干一件事

项目根目录的实际布局大概是这样，我按职责重新整理过一遍，方便你对照自己的代码：

note_spider/ ├── config/ │ ├── settings.yaml # 频控、重试、并发、超时 │ └── devices.json # 设备指纹池，X-s-common 用 ├── core/ │ ├── signer.py # Node 子进程管理器，暴露 sign() │ ├── request.py # 会话、重试、退避 │ └── limiter.py # 令牌桶 + 随机抖动 ├── js/ │ ├── entry.js # 签名入口，读 stdin 写 stdout │ ├── env.js # window/navigator/document 最小桩 │ └── vendor/ # 混淆后的原始算法，只读，不改 ├── pipeline/ │ ├── note.py # 笔记列表 / 详情 │ └── user.py # 用户主页 ├── store/ │ ├── schema.sql # 建表语句，note_id 唯一索引 │ └── dao.py # upsert 封装 ├── tests/ │ └── golden_cases.json # 输入参数 → 期望签名头 └── run.py # 命令行入口

core/是唯一需要理解内部实现的一层，pipeline/是业务层，加一个新的采集目标基本就是在这里加一个文件。js/vendor/建议直接设成只读，我见过有人在里面加console.log调试，结果把签名函数内部的执行时序改了，输出全错还查了半天。config/devices.json里放的是多个设备指纹，请求时轮换使用，避免所有请求长得一模一样。

tests/golden_cases.json这个文件看似可有可无，实际上是最有价值的部分：它把“算法还原对了没有”从主观判断变成了可以一键运行的断言。后面第四章会专门讲这个文件怎么攒。

2.2 签名层为什么不翻译成 Python

新手最常见的想法是：既然 JS 能算出结果，那我把它翻译成 Python 不就行了？逻辑上可行，工程上我不建议，原因有三个。

第一是体量。那类签名函数经过混淆后动辄两三千行，里面混着控制流平坦化和大量无用分支，人工翻译的漏检率很高，而且你没法证明翻译后的结果和原版等价，只能靠对拍。第二是迭代成本。前端一周改一次算法是常事，翻译版每次都要重来，而保留 JS 只需要替换vendor/下的文件，再跑一遍黄金用例就能确认。第三是隐藏依赖，算法内部往往会读navigator.userAgent、document.cookie、Date.now()，翻译过程中很容易把这些悄悄换成常量，最后结果是“本地永远对、线上永远错”。

代价当然也有，主要是进程开销。冷启动一个 Node 进程大概 30 到 80 毫秒，如果每条请求都subprocess.run一次，QPS 上不去。我的处理是两级：轻量场景用子进程池，启动时预热四个进程，用完归还；请求量大就改成常驻 Node HTTP 服务，Python 侧走localhost调用，一次调用压到 1 到 2 毫秒。两种模式在core/signer.py里用同一个接口封装，换实现不影响上层。

2.3 环境准备与依赖安装

在动手读代码之前，先把环境跑通，不然后面看代码全是抽象的。Python 建议 3.9 以上，Node 建议 16 以上，太低会对vm模块的某些行为不兼容。

# 1. 建虚拟环境，别装到系统 Python 里 python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate # 2. 装 Python 侧依赖，项目只用了这几个 pip install requests pyyaml pytest # 3. 确认 Node 可用，版本号打不出来说明 PATH 没配对 node --version # 4. 先单独验证签名脚本能不能跑通，这一步很关键 echo '{"note_id":"abc","cursor":"0","page_size":"20","ts":"1700000000000"}' \ | node js/entry.js

第四步如果输出了形如{"X-s":"...","X-t":"...","X-s-common":"..."}的 JSON，说明签名层是通的，后面所有问题都好定位。如果这里就报Cannot find module，八成是js/vendor/目录不完整，或者entry.js里的相对路径写错了。我一般会在run.py启动时先做一次这样的探活，失败就直接退出，别让程序带着坏的签名层往下跑，那样只会得到一堆空数据，排查成本翻倍。

提示：把 Node 版本、Python 版本、vendor/目录的文件哈希写进启动日志。签名突然全错的时候，第一件事就是看这三个值有没有变。

3. 核心链路拆解：一次笔记列表请求从参数拼装到落库

把一次请求拆开看，其实只有四步：拼参数、算签名、发请求、解响应。麻烦的地方在于每一步都有隐式约定，比如参数必须全部转成字符串、签名用的时间戳和请求头里的时间戳必须一致、翻页游标必须来自上一次响应而不是自己加一。这一章按执行顺序把这些约定铺开，代码可以直接对照你的项目改。

3.1 参数拼装：哪些字段进 URL，哪些进签名体

先明确一个概念：签名体和查询串不是一回事。查询串是发给服务端的业务参数，签名体是喂给签名函数的原材料，两者往往高度重合但顺序和类型要求不同。我见过最常见的翻车是把page_size写成整数传进签名函数，JS 里一拼接变成20和"20"的差异，哈希出来完全对不上。

import time from typing import Any def build_payload(note_id: str, cursor: str, page_size: int = 20) -> tuple[dict, int]: """拼装签名体，返回 (payload, ts_ms)。 约定： - 所有值统一转字符串，避免 JS 侧隐式转换带来差异 - ts 同时用于签名体和 X-t 请求头，必须同源，不能各取一次 """ ts_ms = int(time.time() * 1000) payload = { "note_id": str(note_id), "cursor": str(cursor), # 首次传 "0"，之后用响应里的 next_cursor "page_size": str(page_size), "ts": str(ts_ms), } return payload, ts_ms

note_id从列表页或搜索页拿到，cursor首次固定为"0"，后续必须取响应体里的next_cursor字段。这一点是新手最容易想当然的地方，以为游标就是页码乘以每页条数，结果第二页开始就对不上。ts_ms单独返回是为了让请求头复用它，签名体里用的是它，X-t头用的也必须是它，两处各调一次time.time()会导致毫秒级偏差，短链路上看不出来，网络一抖就会偶发校验失败。

page_size建议不要贪大，超过 30 容易被服务端截断或者直接判定异常。我在settings.yaml里把它做成可配的，默认 20，压测时临时调到 10 反而更稳。

3.2 Node 桥接签名：一次调用的完整代码

签名这一层是项目的核心。Python 侧只做两件事：把 payload 序列化成 JSON 写进子进程的标准输入，从标准输出读回签名头。中间不传文件、不拼命令行参数，避免转义和长度限制的坑。

import json import shutil import subprocess from pathlib import Path class Signer: """Node 子进程签名器。 js_entry: entry.js 的绝对路径 timeout: 单次签名超时（秒），超过就 kill，防止混淆代码里的死循环拖死主进程 """ def __init__(self, js_entry: str | Path, timeout: float = 3.0): self.js_entry = str(Path(js_entry).resolve()) self.timeout = timeout self.node = shutil.which("node") if not self.node: raise RuntimeError("PATH 里找不到 node，先确认 Node 装好并加入环境变量") def sign(self, payload: dict) -> dict: proc = subprocess.run( [self.node, self.js_entry], input=json.dumps(payload, ensure_ascii=False), capture_output=True, text=True, timeout=self.timeout, ) # 混淆代码有时会往 stderr 打调试信息，不能因为 stderr 非空就判失败 if proc.returncode != 0: raise RuntimeError(f"签名进程退出码={proc.returncode} stderr={proc.stderr[-500:]}") try: return json.loads(proc.stdout) except json.JSONDecodeError as exc: raise RuntimeError(f"签名输出不是合法 JSON: {proc.stdout[:300]!r}") from exc

对应的js/entry.js只要十几行，重点在加载顺序：先补环境，再加载算法。顺序反了，算法在初始化阶段读到的window就是undefined，后面怎么调都救不回来。

const fs = require('fs'); const path = require('path'); // 1. 先补环境，且必须在 vendor 被 require 之前 require(path.join(__dirname, 'env.js')); // 2. 再加载混淆后的签名函数 const signFn = require(path.join(__dirname, 'vendor', 'entry.js')); // 3. 从标准输入读参数，标准输出写结果，不污染 stdout 的其他内容 const raw = fs.readFileSync(0, 'utf8'); const payload = JSON.parse(raw); const headers = signFn(payload); // { "X-s": "...", "X-t": "...", "X-s-common": "..." } process.stdout.write(JSON.stringify(headers));

几个参数值得单独说一下。timeout=3.0不是拍脑袋定的，正常签名调用在 100 毫秒内完成，超过 3 秒基本可以判定卡死，此时必须kill，否则子进程会一直挂着占内存。ensure_ascii=False是为了让中文参数原样传过去，某些签名函数内部会对参数做encodeURIComponent，如果 Python 侧先转成\uXXXX，结果就变了。

至于为什么不直接用PyExecJS：它能用，但对新版 V8 的支持停滞，多个调用共享同一个上下文容易互相污染，出错时的异常栈也几乎没法看。走子进程虽然多一层，但隔离得干净，出问题一眼能定位到是哪个 payload 崩的。

注意：vendor/entry.js里如果带module.exports之外的副作用（比如立即执行的初始化、定时器），每次起进程都会重跑一遍。跑通之后用console.time打一下首行输出耗时，超过 200 毫秒就该考虑改成常驻服务了。

3.3 响应归一化：从原始 JSON 到能入库的行

拿到响应只是中间态，真正要落库的字段往往是嵌套的，而且数值字段被格式化过。点赞数在小于一万时是"1234"，超过一万变成"1.2万"，超过一亿变成"1.2亿"。这类字段如果直接按字符串存，后面做排序和统计全乱。

import re import sqlite3 _UNIT = {"万": 10_000, "w": 10_000, "W": 10_000, "亿": 100_000_000} def to_int(value) -> int: """把 "1.2万" / "1234" / "10w+" 这类计数统一成整数，解析失败返回 0。""" if isinstance(value, int): return value text = str(value or "").strip().replace("+", "") match = re.match(r"^([\d.]+)\s*([万wW亿]?)$", text) if not match: return 0 number, unit = float(match.group(1)), match.group(2) return int(number * _UNIT.get(unit, 1)) def parse_note(item: dict) -> dict: card = item.get("note_card") or {} interact = card.get("interact_info") or {} return { "note_id": item.get("id"), "title": (card.get("display_title") or "").strip(), "author_id": (card.get("user") or {}).get("user_id"), "liked": to_int(interact.get("liked_count")), "collected": to_int(interact.get("collected_count")), "comment": to_int(interact.get("comment_count")), "next_cursor": item.get("cursor") or "", } def upsert(conn: sqlite3.Connection, row: dict) -> None: conn.execute( """INSERT INTO notes(note_id, title, author_id, liked, collected, comment) VALUES(:note_id, :title, :author_id, :liked, :collected, :comment) ON CONFLICT(note_id) DO UPDATE SET liked=excluded.liked, collected=excluded.collected, comment=excluded.comment""", row, )

to_int里把匹配失败的返回值定成 0 而不是抛异常，是因为采集过程中总能碰到格式奇怪的字段，为了一个计数让整批任务挂掉不划算。upsert依赖note_id上的唯一索引，这个索引在建表时就加上，后面所有的增量更新都靠它，重复跑同一批数据不会产生重复行，还能顺带更新点赞数这类会变的字段。

解析层还有个容易被忽略的细节：不要把整个原始 JSON 塞进一个raw字段就完事。短期看省事，长期看查询时要反复解析 JSON，成本高得多。我一般只保留解析失败的样本到dead_letter表，正常数据全部结构化。

4. 算法还原现场：从混淆代码到可读函数的四步走

这一段是整个项目里最像“手艺活”的部分。工具能帮上忙的地方有限，更多是靠断点、回溯和反复排除。我把它拆成四步，每一步都有一个明确的“完成标志”，达不到就别往下走，不然会在错误的方向上越陷越深。

4.1 定位入口：XHR 断点与调用栈回溯

打开开发者工具，切到 Network，筛选 XHR，清空列表，然后在页面上做一次翻页操作。找到那个返回笔记列表的请求，点进 Initiator 标签看调用栈，通常从下往上数第三四层就能看到业务代码，再往上就是签名函数的调用点。

如果栈太深或者被混淆得看不出层级，换个思路：在 Sources 面板全局搜索请求头的名字，比如搜X-s，命中的位置大概率就是赋值的地方，在那里下一行断点，再触发一次请求，断住之后看调用栈。这个办法比顺着栈爬快得多，我一般都是先搜再断。

完成标志是：你能在断点处拿到签名函数的入参payload和出参X-s，并且两者在多次请求里对应关系稳定。拿到这一对样本，后面补环境才有验证基准。

4.2 补环境：最小桩写到“刚好够用”

混淆代码被扔到 Node 里跑，第一件事通常是ReferenceError: window is not defined。这时候不要急着去找现成的环境模拟库，先从最小桩开始，缺什么补什么，补到不报错为止。

// js/env.js —— 最小桩，按需增补，别一次写全 const root = globalThis; root.window = root; root.self = root; root.navigator = { userAgent: 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ' + '(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', platform: 'Win32', languages: ['zh-CN', 'zh'], hardwareConcurrency: 8, }; root.document = { cookie: '', // 无登录态场景置空，别写假值 createElement: () => ({ getContext: () => null, style: {} }), documentElement: { clientWidth: 1920, clientHeight: 1080 }, }; root.location = { href: 'https://example.com/', protocol: 'https:' }; root.localStorage = { getItem: () => null, setItem: () => {}, removeItem: () => {} }; root.screen = { width: 1920, height: 1080, colorDepth: 24 }; root.performance = root.performance || { now: () => Date.now() }; // 混淆代码常用，Node 里没有的两个 root.atob = (s) => Buffer.from(s, 'base64').toString('binary'); root.btoa = (s) => Buffer.from(s, 'binary').toString('base64');

补环境的判断标准很朴素：不报错不代表对了，要看签名结果和浏览器里抓到的是否一致。所以每补一个对象，都要用 4.1 里拿到的样本对拍一次，一致就留下，不一致就回退。我见过有人一口气把网上找的几百行模拟代码全贴进去，结果某个对象的toString被改写，算法内部的隐式转换悄悄变了，输出永远是错的，还查不出原因。

navigator.userAgent这个值不要随便改，它往往直接参与哈希。document.cookie在只采公开数据时应该是空的，写个假的反而可能触发别的分支。

4.3 固化成黄金用例：把“好像跑通了”变成可回归

样本对拍是手工活，做一次管不了三天。真正的做法是把它写成测试文件，每次改动签名层先跑一遍。

[ { "name": "list_page_first", "payload": {"note_id": "demo001", "cursor": "0", "page_size": "20", "ts": "1700000000000"}, "expect_keys": ["X-s", "X-t", "X-s-common"], "expect_x_t": "1700000000000" }, { "name": "list_page_second", "payload": {"note_id": "demo001", "cursor": "eyJvZmZzZXQiOjIwfQ==", "page_size": "20", "ts": "1700000080000"}, "expect_keys": ["X-s", "X-t", "X-s-common"], "expect_x_t": "1700000080000" } ]

import json import pytest from core.signer import Signer @pytest.fixture(scope="session") def signer(): return Signer("js/entry.js") def load_cases(): with open("tests/golden_cases.json", encoding="utf-8") as fp: return json.load(fp) @pytest.mark.parametrize("case", load_cases(), ids=lambda c: c["name"]) def test_sign_headers(signer, case): headers = signer.sign(case["payload"]) # 键名必须齐，缺一个服务端就会走降级分支返回空列表 for key in case["expect_keys"]: assert key in headers, f"缺少签名头 {key}" # X-t 必须和入参时间戳完全一致，这是最常见的错位点 assert headers["X-t"] == case["expect_x_t"] # X-s 每次都会变（内含随机量），只校验长度和字符集 assert len(headers["X-s"]) > 20

注意这里没有断言X-s的具体值。签名内部通常掺了时间戳和随机串，同一次输入算两次结果不同是正常的，硬写死值只会让测试天天红。能断言的是键名齐全、长度合理、字符集在预期范围内，以及X-t这种确定性字段的严格相等。

黄金用例攒到十来个、覆盖首屏、翻页、详情、空结果这几种形态之后，你对签名层的信心就够了。以后每次vendor/更新，先跑pytest tests/，全绿再上量。

5. 避坑与排查：五类高频故障的现象、成因和解法

下面这五条全是我实际撞过的，按发生频率排。每一条都按“现象 → 原因 → 解决”写，遇到问题时对着症状找，比从头读代码快。

5.1 请求层：签名与频控相关的三条

第一条：首页有数据，第二页开始返回空数组，状态码 200

现象是日志里一条错误都没有，items长度稳定为 0，换个note_id又正常。原因基本是两个：一是cursor没有用响应里的next_cursor，而是自己按 offset 推算；二是签名被复用了，很多实现会把签名结果缓存几十秒，但服务端对同一签名加同一游标的组合做了重放校验。解决方法是每次请求都重新调一次Signer.sign()，游标严格从响应体取，取不到就判定列表结束，别硬翻。

第二条：间歇性 461 或 403，重试一次就好

现象是失败率大概百分之几，重试就过，看起来像网络抖动。原因多半是X-t时间戳和服务端差超过阈值（常见是 300 秒），机器时钟漂移或者容器时区没配对都会触发。解决办法是在启动时做一次时间校准：发一个任意请求，读响应头里的Date，算出本地与服务端的偏移量，之后所有ts_ms都加上这个偏移。这个偏移要定期刷新，长时间运行的进程时钟会继续漂。

from email.utils import parsedate_to_datetime import time class Clock: """维护本地与服务端的时间偏移，避免长时间运行后 X-t 过期。""" def __init__(self): self.offset_ms = 0 def sync(self, date_header: str) -> None: server = parsedate_to_datetime(date_header).timestamp() * 1000 self.offset_ms = int(server - time.time() * 1000) def now_ms(self) -> int: return int(time.time() * 1000) + self.offset_ms

date_header的精度只到秒，够用了；offset_ms会引入正负几百毫秒误差，比起几百秒的漂移可以忽略。这个类挂到Signer上，所有时间戳来源统一。

第三条：Node 调用偶发超时，输出为空字符串

现象是subprocess抛TimeoutExpired，或者stdout是空。原因有三个：混淆代码里有setInterval之类的定时器让进程不退出；并发起太多子进程把 CPU 打满；vendor/里的算法做了执行时长检测，跑太慢直接不返回。解决方法是三件事一起做——在entry.js末尾显式调process.exit(0)，把并发限制在 CPU 核数以内，以及给子进程池设上限。第三点最容易被忽略，机器一忙，签名耗时从 50 毫秒涨到 400 毫秒，算法内部一算就判定为异常环境。

5.2 数据层：解析与存储相关的两条

第四条：点赞数排序完全错乱，“1.2万”排在“9999”前面

现象是导出表格按点赞排序，前几名全是几千的数字。原因是原始字段是格式化字符串，直接按字典序排了。解决办法就是 3.3 里的to_int，在入库前统一转整数，同时保留一个liked_raw字段存原字符串，方便回溯。这个转换要在写入时做，不要等到查询时再算，否则每次分析都得重跑一遍。

第五条：跑一夜磁盘涨几十 G，去重后只剩几万条

现象是数据库文件膨胀得离谱，SELECT COUNT(*)和实际去重数量差一个数量级。原因是分页出现了回环——某些情况下next_cursor会返回已经翻过的位置，脚本无脑跟就会一直循环；另一个原因是note_id没建唯一索引，INSERT直接堆。解决办法：在建表时就给note_id加UNIQUE，写入走ON CONFLICT DO UPDATE；同时在内存里维护一个已见游标的集合，同一个游标出现第二次就熔断退出，并打一条警告日志。

CREATE TABLE IF NOT EXISTS notes ( note_id TEXT PRIMARY KEY, -- 主键即唯一索引，重复写入走 upsert title TEXT NOT NULL DEFAULT '', author_id TEXT, liked INTEGER NOT NULL DEFAULT 0, collected INTEGER NOT NULL DEFAULT 0, comment INTEGER NOT NULL DEFAULT 0, updated_at TEXT NOT NULL DEFAULT (datetime('now')) ); CREATE INDEX IF NOT EXISTS idx_notes_liked ON notes(liked DESC);

注意：采集频率、请求总量、目标页面的公开范围，这三件事在动手前先确认清楚。项目里的限速和退避不是为了绕过什么，是为了不给对方服务器添麻烦，也让自己少踩封禁的坑。

6. 二次开发：把采集器改成你自己的数据管道

跑通之后，多数人下一步想的不是继续采，而是“这套东西能不能改成我要的样子”。可以，而且改动量比想象中小，因为项目的分层把变化点圈得很清楚。下面是我按实际改造经验整理的几个扩展方向，以及各自要动的地方。

最关键的一步是把run.py从“脚本”改成“可被调用的入口”。现在的写法大概是if __name__ == "__main__": main()，改造时把main()拆成一个接收参数的函数，返回结构化结果而不是直接打印。这样它既能被命令行调用，也能被定时任务或者消息消费者调用，不用复制代码。

from dataclasses import dataclass @dataclass class CollectTask: target: str # note / user / search key: str # note_id 或关键词 max_pages: int = 50 since_ts: int = 0 # 只保留更新时间大于该值的数据，做增量 def run_task(task: CollectTask, signer, clock, dao) -> dict: """执行一次采集，返回统计信息而不是直接打印，方便上层编排。""" stats = {"pages": 0, "inserted": 0, "skipped": 0, "stop_reason": ""} cursor, seen = "0", set() while stats["pages"] < task.max_pages: if cursor in seen: # 游标回环保护 stats["stop_reason"] = "cursor_loop" break seen.add(cursor) payload, ts_ms = build_payload(task.key, cursor) headers = signer.sign(payload) resp = request_with_retry(task.target, payload, headers, clock) items = (resp.get("data") or {}).get("items") or [] if not items: stats["stop_reason"] = "empty_page" break for item in items: row = parse_note(item) if task.since_ts and row.get("updated_at", 0) < task.since_ts: stats["skipped"] += 1 continue dao.upsert(row) stats["inserted"] += 1 cursor = items[-1].get("cursor") or "" stats["pages"] += 1 if not cursor: stats["stop_reason"] = "no_cursor" break return stats

这个函数里有两处值得单独说。seen集合是游标回环的最后一道防线，前面提到的磁盘膨胀问题，靠它兜底最省事。since_ts这个参数是增量更新的关键，第一次全量跑完把当前时间记下来，之后每次只处理更新的数据，采集量能从每天几十万条压到几千条。返回值改成字典而不是打印，是为了让上层能做重试决策——stop_reason是empty_page就正常结束，是cursor_loop就该告警。

验证改造是否成功，我的习惯是准备一套离线回放：把线上抓到的二十条原始响应存成 JSON 文件，写个脚本喂给解析层，比对输出的行数和字段值。这套回放跑得快、不依赖网络，改解析逻辑时特别有用。签名层则继续用第四章的黄金用例兜着，两层各管各的，出问题能立刻分清是算法错了还是解析错了。

从那以后，我每次动签名相关的代码，都会先把黄金用例跑一遍再上量，哪怕只改了一行环境桩。这个习惯救过我好几次——有一次只是把navigator.platform从Win32改成了MacIntel，看着无害，结果签名长度直接变了，服务端全量返回空。希望帮到你。

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