30 行 Python 实现公开接口数据抓取与本地化存储

基于 Python 的公开接口数据抓取与本地化存储方案

核心思路

1. 分析列表接口：在浏览器 DevTools 中筛选 XHR，发现真实请求 URL 形如 https://api.example.com/api/v2/theme/list?page={page}&size=20，返回体含 data.total 与 data.list。
2. 解析单条记录：字段 title、author、coverUrl 即为我们需要的元数据，无需再进详情页。
3. 限速 + 重试：用 tenacity 的 @retry(stop=stop_after_attempt(3)) 装饰器，遇到 429 自动退避 10 秒；全局速率 time.sleep(random.uniform(1, 2))，确保平均 <30 次/分。

运行环境

Python≥3.7，依赖如下：

pip install requests pandas tenacity fake-useragent

代码示例

文件 save_data_meta.py：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
抓取目标站点公开主题合集元信息，仅采集 title/author/coverUrl 三字段。
仅供学习，请遵守平台 robots.txt 与相关法律法规。
"""
import csv, time, random, requests
from tenacity import retry, stop_after_attempt, wait_fixed
from fake_useragent import UserAgent

BASE_URL = "https://api.example.com/api/v2/theme/list"
CSV_FILE = "data_export.csv"
PAGE_SIZE = 20
MAX_PAGE = 50  # 按需调整，0 表示抓全量
SLEEP_RANGE = (1, 2)  # 秒，随机休眠

ua = UserAgent()
session = requests.Session()
session.headers.update({"Referer": "https://www.example.com/"})

@retry(stop=stop_after_attempt(3), wait=wait_fixed(10))
def fetch_one_page():
    session.headers.update({: ua.random})
    params = {: page, : PAGE_SIZE}
    resp = session.get(BASE_URL, params=params, timeout=)
     resp.status_code == :
         Exception()
    resp.raise_for_status()
     resp.json()

 ():
    writer = 
     p  (, MAX_PAGE + ):
        ()
        js = fetch_one_page(p)
        rows = [{: item[], : item[], : item[]}  item  js[][]]
          writer:
            writer = csv.DictWriter((CSV_FILE, , newline=, encoding=), fieldnames=[, , ])
            writer.writeheader()
        writer.writerows(rows)
         js[][] <= p * PAGE_SIZE:
            
        time.sleep(random.uniform(*SLEEP_RANGE))
    ()

 __name__ == :
    main()