接口抓取：JSON API、分页与参数签名

页面里看到的列表，绝大多数是前端调接口渲染出来的，能拿到那个接口就没必要去啃 HTML。本章按「找接口、判分页、看签名、写代码、做容错」的顺序走一遍，顺便把哪些接口可以抓、哪些碰不得说清楚。

先找官方接口，再考虑抓页面

动手顺序永远是：先查站点有没有开放平台与接口文档，能申请 API Key 就用官方接口，配额、字段、稳定性都有保证；没有官方接口，再去看页面自己在调什么接口。抓之前确认 robots.txt 与服务条款允许该路径，控制请求间隔与并发上限，不采集个人隐私与付费内容，不绕过登录、付费墙、验证码等访问控制。

在 DevTools 里找接口

1. Network 面板勾选 Fetch/XHR，刷新一次页面
2. 按 Size 或 Time 排序，找返回 JSON 的请求
3. 看 Request URL、Query String 与 Payload，区分分页参数与筛选参数
4. 把 URL 复制到 Python 里用 requests 复现一次，能复现才写正式代码

判断标准很直接：接口返回结构化 JSON、分页参数清楚、不依赖一次性 token，就走接口；接口参数里带着看不懂的加密串、或者数据只能靠前端渲染出来，就退回 HTML 解析，别硬啃。

分页的三种形态

形态	典型参数	终止条件	注意
页码式	`page`、`pageSize`、`offset`/`limit`	返回空列表、条数少于 `pageSize`、`total` 已翻完	深翻页常被限制，翻页期间数据变化会错位
游标式	`cursor`、`next_token`、`after`	`next_token` 为空、`has_more=False`	最稳，适合边翻边被写入的数据
增量式	`since`、`updated_after`、时间戳	返回 0 条、时间回退到上次水位线	增量同步首选，必须把水位线存下来

无论哪种形态，代码里都要加一道 MAX_PAGES 或最大条数限制，防止分页条件写错时无限翻页。

参数签名：只讲原理

不少接口会带上 sign 参数，常见构造是「时间戳 + 随机串（nonce）+ 业务参数按字典序拼接 + 密钥，再做 MD5 或 HMAC」，服务端用同样的规则重算一遍比对，用途是防篡改与防重放。理解这个原理是为了看懂请求、排查自己调不通的原因。

本文不教逆向他人签名算法。签名是站点有意设置的技术保护措施，绕过它既违反服务条款，也可能触碰《数据安全法》《个人信息保护法》的边界。需要数据就走下面这些路：

需求	合规做法
要稳定拿数据	查官方开放平台与接口文档，按文档对接
配额不够	申请 API Key 或升级开发者账号
做分析或建模	使用开放数据集、政府公开数据
商用或高频批量	购买授权数据，或联系站点商务合作签署数据使用协议

请求体的两种提交方式

方式	写法	`Content-Type`	服务端按什么解析
JSON	`requests.post(url, json=payload)`	`application/json`	解析请求体 JSON
表单	`requests.post(url, data=payload)`	`application/x-www-form-urlencoded`	解析表单字段

# JSON 提交：requests 自动带上 application/json
r = requests.post(url, json={"page": 1, "keyword": "爬虫"}, headers=HEADERS, timeout=10)
# 表单提交：requests 自动带上 application/x-www-form-urlencoded
r = requests.post(url, data={"page": 1, "keyword": "爬虫"}, headers=HEADERS, timeout=10)

完整可运行示例

下面是一个翻页抓取函数，包含重试、限速、分页终止判断、字段容错与 JSONL 追加落盘：

"""翻页抓取 + 容错 + JSONL 落盘：python api_demo.py"""
import json, time
import requests

API = "https://example.com/api/articles"
HEADERS = {"User-Agent": "RuilinCrawler/1.0 (contact: ops@example.com)"}
PAGE_SIZE, MAX_PAGES = 20, 50          # 保险丝：分页条件写错也不会无限翻

def fetch_page(session, page, retries=3):
    """抓一页；网络错误退避重试，限流或非 JSON 直接返回 None。"""
    for i in range(1, retries + 1):
        try:
            r = session.get(API, params={"page": page, "pageSize": PAGE_SIZE},
                            headers=HEADERS, timeout=10)
        except requests.RequestException as e:
            print(f"第 {i} 次网络失败：{e}")
            time.sleep(i * 1.0)            # 退避，控制对站点的压力
            continue
        if r.status_code == 429:           # 站点明确提示限流，别再硬打
            print("被限流，休眠 60 秒后结束本轮")
            time.sleep(60)
            return None
        if r.status_code >= 400:
            print(f"HTTP {r.status_code}，放弃第 {page} 页")
            return None
        try:
            return r.json()
        except ValueError:
            print(f"第 {page} 页不是 JSON：{r.text[:120]}")
            return None
    return None

def parse_items(payload):
    """字段一律用 get 兜底，缺必填字段的记录不落盘。"""
    rows = []
    for item in (payload.get("data") or {}).get("list") or []:
        row = {"id": item.get("id"), "title": (item.get("title") or "").strip(),
               "url": item.get("url") or "", "published": item.get("published_at") or ""}
        if row["id"] and row["title"]:
            rows.append(row)
    return rows

def crawl(out_path="articles.jsonl"):
    session, total = requests.Session(), 0
    with open(out_path, "a", encoding="utf-8") as fh:
        for page in range(1, MAX_PAGES + 1):
            payload = fetch_page(session, page)
            if payload is None:
                break                      # 重试耗尽或限流，停下来人工确认
            rows = parse_items(payload)
            for row in rows:
                fh.write(json.dumps(row, ensure_ascii=False) + "\n")   # 一行一条追加写
            total += len(rows)
            print(f"第 {page} 页 {len(rows)} 条，累计 {total} 条")
            if not rows:                   # 空列表＝已经翻到底
                break
            if not (payload.get("data") or {}).get("has_more", True):
                break                      # 服务端明确说没有下一页
            time.sleep(1.0)
    return total

if __name__ == "__main__":
    print(crawl())

选 JSONL 而不是一个大 JSON 数组，是因为它可以边抓边追加：进程中途崩了，已抓到的行依然完整可读，用 pandas.read_json(path, lines=True) 能直接读回来。

常见坑与容错要点

情况	表现	处理
字段时有时无	`KeyError` 直接崩	一律 `item.get("title") or ""`
空列表与 `None` 混淆	分不清「没数据」和「没取到」	`payload.get("data") or {}` 再取 `list or []`
限流提示	`code=429` 或 `message` 含「频繁」「稍后重试」	识别后 `time.sleep` 退避，连续命中就退出，别换代理硬刷
返回不是 JSON	`response.json()` 抛 `ValueError`	先看 `response.text[:200]`，确认是登录页还是错误页
记录缺必填字段	落库后出现空 id、空标题	落盘前校验，丢弃并计数
分页条件失效	一直返回同一页，程序跑不完	记录每页首条 id 做去重，并靠 `MAX_PAGES` 兜底

调试顺序建议固定为：先用 curl 或 requests 单独打一页，把 status_code、Content-Type、前 200 个字符打出来；确认结构后再接分页；分页跑通后再接落盘与去重。跳过第一步直接写全量抓取，最容易把「参数错了」误判成「反爬了」。

小结：先找官方接口再考虑抓页面，用 DevTools 的 Fetch/XHR 面板定位接口并按分页形态确定终止条件，签名只理解原理不逆向，请求体分清 json= 与 data=，代码里做重试、限速、字段容错与 JSONL 落盘，采集范围守好 robots.txt、服务条款与三法边界。

笔记加载中…

标签：#Python3 #爬虫

返回知识库