㊗️本期内容已收录至专栏《Python爬虫实战》，持续完善知识体系与项目实战，建议先订阅收藏，后续查阅更方便～
㊙️本期爬虫难度指数：⭐⭐⭐⭐⭐（专家级）
🉐福利： 一次订阅后，专栏内的所有文章可永久免费看，持续更新中，保底1000+(篇)硬核实战内容。

🌟 开篇语

哈喽，各位小伙伴们你们好呀～我是【喵手】。
运营社区： C站 / 掘金 / 腾讯云 / 阿里云 / 华为云 / 51CTO
欢迎大家常来逛逛，一起学习，一起进步～🌟

  我长期专注 Python 爬虫工程化实战，主理专栏👉 《Python爬虫实战》：从采集策略到反爬对抗，从数据清洗到分布式调度，持续输出可复用的方法论与可落地案例。内容主打一个“能跑、能用、能扩展”，让数据价值真正做到——抓得到、洗得净、用得上。

  📌 专栏食用指南（建议收藏）

• ✅ 入门基础：环境搭建 / 请求与解析 / 数据落库
• ✅ 进阶提升：登录鉴权 / 动态渲染 / 反爬对抗
• ✅ 工程实战：异步并发 / 分布式调度 / 监控与容错
• ✅ 项目落地：数据治理 / 可视化分析 / 场景化应用

📣 专栏推广时间：如果你想系统学爬虫，而不是碎片化东拼西凑，欢迎订阅专栏👉《Python爬虫实战》👈，一次订阅后，专栏内的所有文章可永久免费阅读，持续更新中。
  
💕订阅后更新会优先推送，按目录学习更高效💯～

0️⃣ 前言（Preface）

这次要做的事情很具体：使用 Python 对国家科技资源共享服务平台公开资源目录进行低频、可恢复式采集，递归遍历可能不断向下展开的分类树，从公开资源详情中整理 resource_id、标题、负责人、领域、数据量和 DOI 等字段，最后生成一个本地 SQLite/CSV 元数据索引。

这类任务表面看起来像普通“列表页 → 详情页”的爬虫，但真正麻烦的地方其实不在详情页，而在目录。

很多科研数据门户的分类不是：

一级分类
    二级分类
        数据

这么规整。

更常见的是：

科学数据
├── 地球科学
│   ├── 地理学
│   │   ├── 自然地理
│   │   │   ├── 气候
│   │   │   │   ├── ...
│   │   │   │   └── 数据集
│   │   │   └── 水文
│   │   └── 人文地理
│   └── 地球物理
├── 生物学
│   ├── ...
└── ...

层级深度不固定，有些节点本身还有数据，有些节点只有子节点，有些节点第一次展开时才向服务器请求下一级。

所以我更愿意把这类爬虫理解为：

先遍历一棵远程树，再抓取树叶与树枝关联的数据资源。

读完本文，你至少可以获得三样东西：

1. 一套不依赖“固定三层目录”的通用树遍历策略，可以应付任意深度分类；
2. 一套 Playwright 与 requests 混合使用的动态站点采集框架；
3. 一套带重试、限速、SQLite 去重、断点续跑和字段容错的完整 Python 项目骨架。

本文只讨论公开页面与公开元数据索引，不涉及账户权限内容，也不尝试绕过网站的访问控制。

---

1️⃣ 摘要（Abstract）

本文使用 Python 构建一个面向公开科学数据资源目录的元数据采集程序，以浏览器自动化发现动态分类请求，以递归/队列算法遍历不确定深度的树状目录，再通过 requests 获取公开详情页面，解析并标准化 resource_id、title、pi_name、domain、data_volume、doi 六类核心字段，最终写入 SQLite 并导出 CSV。

国家科技资源共享门户当前公开页面包含“资源目录”“国家科学数据中心”“国家资源库”等入口，资源详情也可以公开展示主题分类、学科分类等元数据信息。部分科学数据详情还可以出现 DOI、数据量等字段。

本文的关键收获不是某几个 XPath，而是完整的数据获取思路：

• 目录未知深度时，不写死层级，而是维护待访问节点队列；
• 动态站点不盲猜 API，而是让 Playwright 监听浏览器真实网络请求；
• 页面字段名称不统一时，通过字段别名映射统一成自己的 Schema；
• 爬虫随时可能中断，因此状态必须落盘，而不是只放在 Python set 中。

2️⃣ 背景与需求（Why）

2.1 为什么要采集公开科研资源元数据

如果只是偶尔寻找一个数据集，直接在网站搜索当然最简单。

但当需求变成：

• 分析不同领域开放数据资源数量；
• 建立 DOI 与数据集标题对应关系；
• 统计科研数据大致体量；
• 构建本地可全文检索的资源索引；
• 做数据资源知识图谱；
• 对元数据质量进行统计；
• 定期检查新增公开资源；

手工检索就不现实了。

举一个非常直观的例子。

假设本地数据库最终得到：

resource_id | title          | pi_name | domain       | data_volume | doi
------------+----------------+---------+--------------+-------------+------------------
A001        | 某气候数据集     | 张某     | 地球科学       | 3.08 MB     | 10.xxxx/xxx
A002        | 某生态观测数据集   | 李某     | 生态学         | 1.2 GB      | 10.xxxx/xxx
...

那么后续可以直接：

SELECT domain, COUNT(*)
FROM resources
GROUP BY domain
ORDER BY COUNT(*) DESC;

甚至可以做 DOI 完整率：

SELECT
    COUNT(*) AS total,
    SUM(CASE WHEN doi <> '' THEN 1 ELSE 0 END) AS has_doi
FROM resources;

这时候爬虫就不再是“下载网页的小脚本”，而是一个数据采集基础设施。

---

2.2 目标站点

本文实际以国家科技资源共享相关公开门户为研究对象。

当前公开入口中，“资源目录”对应元数据目录页面，网站自身也表明页面依赖 JavaScript 环境才能正常运行。

这一点非常重要。

因为它意味着：

requests.get(url).text

拿到的 HTML 未必等于用户浏览器最后看到的 DOM。

这也是为什么本文不会只使用 requests。

---

2.3 目标字段

我们建立如下内部 Schema：

resource_id
title
pi_name
domain
data_volume
doi

具体含义：

字段	含义
resource_id	资源唯一标识，优先从详情页参数、CSTR/id 等稳定标识获得
title	数据集或资源标题
pi_name	项目负责人、负责人、数据负责人或其他等价公开字段
domain	学科分类/领域/主题分类
data_volume	数据量，如 MB、GB、TB
doi	DOI 标识

这里需要强调一个真实工程问题：

网页展示字段不一定正好叫 pi_name。

不同资源中心可能出现：

负责人
项目负责人
联系人
数据贡献者
作者
责任者
首席科学家

所以我们的数据库字段叫：

pi_name

但解析器不能只搜索：

项目负责人

而应该维护别名集合。

同样：

domain

可以来自：

学科分类
领域
所属学科
主题分类

这一步叫做字段归一化。

---

3️⃣ 合规与注意事项

做公开数据爬虫，技术只是其中一部分。

一个能跑的爬虫不意味着应该无限制运行。

3.1 先查看 robots.txt 与网站公开规则

运行前建议首先检查：

https://目标域名/robots.txt

robots.txt 的主要意义是向自动化程序表达哪些路径允许或不建议自动访问。

需要注意：

robots.txt 不是“看到 Allow 就可以无限请求”的许可证。

还应该同时查看：

• 网站使用说明；
• 数据使用声明；
• 下载许可；
• API 使用规则；
• 页面公开范围；
• 数据引用要求。

部分科学数据详情页面会明确给出数据使用声明和引用方式，因此抓取“公开元数据”和下载“实际数据文件”应该视为两个不同层级的问题。

本文只构建公开元数据索引。

---

3.2 控制请求频率

不要这样：

for url in urls:
    requests.get(url)

几十个线程一起跑。

一个研究型元数据采集器通常根本不需要这样的速度。

本文默认：

REQUEST_INTERVAL_MIN = 1.0
REQUEST_INTERVAL_MAX = 2.5

也就是说请求间随机等待 1～2.5 秒。

如果网站返回：

429 Too Many Requests

第一反应应该是：

降低频率。

而不是立即增加代理数量。

---

3.3 不做攻击式并发

我个人处理公共科研站点时，通常遵循一个很简单的原则：

如果单线程一晚上可以完成，就没必要为了十几分钟的速度提升把请求并发堆到几十甚至几百。

后续确实要加并发，也建议从：

2

或：

3

开始。

而不是：

50
100
200

---

3.4 不采集与任务无关的信息

本文目标只有公开元数据：

resource_id
title
pi_name
domain
data_volume
doi

不需要：

• 用户账号；
• 手机号码；
• 私人邮箱；
• 登录 Cookie；
• 身份信息；
• 后台页面数据。

即使某些页面包含额外信息，也应该坚持最小化采集。

---

3.5 不处理登录和付费限制

如果某项数据提示：

请登录
申请后获取
仅注册用户可下载
授权后查看

那么采集器应该停留在公开元数据层面。

本文不会设计：

• 验证码绕过；
• 登录限制绕过；
• 付费限制绕过；
• 权限检查绕过。

---

4️⃣ 技术选型与整体流程（What / How）

4.1 这是静态站、动态站还是 API？

从实际网页表现看，这是一个典型的：

前端动态渲染 + 后端接口提供数据

类型的网站。

公开页面会提示：

必须在浏览器中启用 Javascript

才能正常工作。

所以：

requests only

不是最稳妥的第一步。

但是：

Playwright everything

也不是最好的长期方案。

因为浏览器非常重。

最终采用：

Playwright
    ↓
发现真实网络请求 / 展开动态目录
    ↓
requests.Session
    ↓
大量低频 HTTP 请求
    ↓
lxml / BeautifulSoup
    ↓
SQLite

---

4.2 整体流程

完整链路：

        ┌─────────────────────┐
        │    启动 Playwright   │
        └──────────┬──────────┘
                   ↓
        ┌─────────────────────┐
        │ 打开公开资源目录页面   │
        └──────────┬──────────┘
                   ↓
        ┌─────────────────────┐
        │ 监听 XHR / fetch JSON │
        └──────────┬──────────┘
                   ↓
        ┌─────────────────────┐
        │ 识别分类树请求结构     │
        └──────────┬──────────┘
                   ↓
        ┌─────────────────────┐
        │ BFS/DFS 遍历分类节点   │
        └──────────┬──────────┘
                   ↓
        ┌─────────────────────┐
        │ 获取分类下资源列表     │
        └──────────┬──────────┘
                   ↓
        ┌─────────────────────┐
        │ 收集详情 URL / ID     │
        └──────────┬──────────┘
                   ↓
        ┌─────────────────────┐
        │ requests 获取详情页   │
        └──────────┬──────────┘
                   ↓
        ┌─────────────────────┐
        │ lxml/BS4 解析字段     │
        └──────────┬──────────┘
                   ↓
        ┌─────────────────────┐
        │ 字段归一化 + 数据清洗  │
        └──────────┬──────────┘
                   ↓
        ┌─────────────────────┐
        │ SQLite UPSERT       │
        └──────────┬──────────┘
                   ↓
        ┌─────────────────────┐
        │ 导出 CSV             │
        └─────────────────────┘

概括起来就是：

采集 → 解析 → 清洗 → 存储。

---

4.3 为什么不是纯 Scrapy

Scrapy 很强。

如果最终规模到了几十万甚至百万 URL，我会优先考虑 Scrapy。

但这篇文章的核心难点是：

未知深度动态树

以及：

不知道树节点接口最开始如何触发

所以 Playwright 的“真实浏览器 + 网络监听”非常有价值。

第一次分析时：

Playwright

比单纯阅读压缩后的 JavaScript 快得多。

等接口结构明确后，再逐步把浏览器替换成 requests 或 Scrapy。

---

5️⃣ 环境准备与依赖安装

5.1 Python 版本

推荐：

Python 3.10+

本文以：

Python 3.11

为基准。

查看版本：

python --version

---

5.2 创建虚拟环境

Windows：

python -m venv .venv
.venv\Scripts\activate

macOS / Linux：

python3 -m venv .venv
source .venv/bin/activate

---

5.3 安装依赖

pip install requests beautifulsoup4 lxml playwright tenacity pandas

安装 Chromium：

playwright install chromium

如果是 Linux 服务器：

playwright install --with-deps chromium

---

5.4 推荐项目结构

nstr_metadata_crawler/
│
├── crawler/
│   ├── __init__.py
│   ├── config.py
│   ├── fetcher.py
│   ├── parser.py
│   ├── storage.py
│   ├── models.py
│   └── tree.py
│
├── tools/
│   └── discover_api.py
│
├── data/
│   ├── metadata.db
│   └── resources.csv
│
├── logs/
│   └── crawler.log
│
├── main.py
└── requirements.txt

创建：

mkdir -p crawler tools data logs

---

6️⃣ 核心实现：请求层（Fetcher）

我习惯先把请求层独立出来。

不要在业务代码里到处写：

requests.get()

否则一旦需要：

• 改 UA；
• 加 timeout；
• 加 retry；
• 控制频率；
• 统一 Referer；

整个项目都会很难维护。

---

6.1 config.py

# crawler/config.py

from pathlib import Path

BASE_URL = "https://www.escience.org.cn"

CATALOG_URL = f"{BASE_URL}/metadata"

DATA_DIR = Path("data")
LOG_DIR = Path("logs")

DB_PATH = DATA_DIR / "metadata.db"
CSV_PATH = DATA_DIR / "resources.csv"

REQUEST_TIMEOUT = 20

REQUEST_INTERVAL_MIN = 1.0
REQUEST_INTERVAL_MAX = 2.5

MAX_RETRIES = 4

USER_AGENT = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
    "AppleWebKit/537.36 (KHTML, like Gecko) "
    "Chrome/126.0 Safari/537.36"
)

目录 URL 使用当前公开的：

https://www.escience.org.cn/metadata

官方公开导航中的“资源目录”目前确实指向该路径。

---

6.2 fetcher.py

# crawler/fetcher.py

import logging
import random
import time

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

from crawler.config import (
    BASE_URL,
    REQUEST_TIMEOUT,
    REQUEST_INTERVAL_MIN,
    REQUEST_INTERVAL_MAX,
    USER_AGENT,
)


logger = logging.getLogger(__name__)


class Fetcher:
    def __init__(self):
        self.session = requests.Session()

        retry = Retry(
            total=4,
            connect=4,
            read=4,
            status=4,
            backoff_factor=1.2,
            status_forcelist=(429, 500, 502, 503, 504),
            allowed_methods=frozenset(["GET"]),
            respect_retry_after_header=True,
            raise_on_status=False,
        )

        adapter = HTTPAdapter(
            max_retries=retry,
            pool_connections=4,
            pool_maxsize=4,
        )

        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)

        self.session.headers.update(
            {
                "User-Agent": USER_AGENT,
                "Accept": (
                    "text/html,application/xhtml+xml,"
                    "application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8"
                ),
                "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.7",
                "Referer": BASE_URL + "/",
                "Connection": "keep-alive",
            }
        )

    @staticmethod
    def _sleep():
        delay = random.uniform(
            REQUEST_INTERVAL_MIN,
            REQUEST_INTERVAL_MAX,
        )

        time.sleep(delay)

    def get(self, url, **kwargs):
        self._sleep()

        timeout = kwargs.pop(
            "timeout",
            REQUEST_TIMEOUT,
        )

        logger.info("GET %s", url)

        response = self.session.get(
            url,
            timeout=timeout,
            **kwargs,
        )

        if response.status_code == 429:
            logger.warning(
                "服务器返回 429，当前访问频率可能过高：%s",
                url,
            )

        response.raise_for_status()

        return response

---

6.3 Headers 为什么需要 UA

最基本的是：

User-Agent

它让服务器知道客户端类型。

不建议把 Header 伪装得极其复杂。

本文只保留常规浏览器字段。

---

6.4 Referer

设置：

"Referer": "https://www.escience.org.cn/"

原因不是为了“绕过检测”，而是很多前端接口本身就会依赖正常页面访问上下文。

---

6.5 Timeout

绝对不要：

requests.get(url)

然后无限等待。

建议：

timeout=20

大型公共平台偶尔响应变慢很正常。

---

6.6 Session

使用：

requests.Session()

可以复用：

TCP 连接
HTTP Keep-Alive
Cookie
默认 Headers

即使不登录，也比每次重新建立连接更合理。

---

6.7 重试与指数退避

我们的配置：

backoff_factor=1.2

并针对：

429
500
502
503
504

进行重试。

大概表现为：

第一次失败
↓
稍等
↓
第二次
↓
等待更久
↓
第三次

这就是退避。

千万不要写：

while True:
    requests.get(url)

这种无限重试反而可能把暂时的服务异常变成额外压力。

---

7️⃣ 核心实现：解析层（Parser）

这一部分是本文重点。

先处理最难的问题：

无限向下的树到底怎么爬？

---

7.1 错误方案：把目录层级写死

很多人第一次会这样：

for level1 in level1_nodes:

    for level2 in get_children(level1):

        for level3 in get_children(level2):

            crawl(level3)

这种程序只适用于：

固定三级

如果第四级突然出现：

level4

整个逻辑就失效。

因此不要让代码知道：

一级
二级
三级

只让代码知道：

节点

---

7.2 把分类目录抽象成树

任何目录节点都表示成：

{
    "id": "node_id",
    "name": "节点名称",
    "parent_id": "parent_id"
}

它只需要支持一个动作：

get_children(node_id)

然后递归：

def walk(node):
    children = get_children(node)

    if not children:
        crawl_resources(node)
        return

    for child in children:
        walk(child)

理论上这样已经可以无限向下。

但生产环境我通常不会优先用真正的 Python 递归。

为什么？

因为树可能非常深。

Python 默认递归深度有限。

所以更稳的是：

显式栈 DFS

或者：

队列 BFS。

---

7.3 DFS 版本

def walk_tree_dfs(root):
    stack = [root]

    while stack:

        node = stack.pop()

        print("访问：", node)

        children = get_children(node)

        for child in children:
            stack.append(child)

---

7.4 BFS 版本

from collections import deque


def walk_tree_bfs(root):
    queue = deque([root])

    while queue:

        node = queue.popleft()

        print("访问：", node)

        children = get_children(node)

        queue.extend(children)

我个人更喜欢 BFS。

因为：

先扫浅层
再逐渐下钻

调试时比较容易观察。

---

7.5 一个很容易被忽略的问题：树可能不是树

前端看上去是树，不代表后台数据结构一定是严格树。

比如：

A
└── C

B
└── C

节点 C 可能同时属于两个分类。

甚至错误数据可能出现：

A → B
B → C
C → A

形成环。

所以必须增加：

visited_nodes = set()

代码：

from collections import deque


def walk_tree(root):
    queue = deque([root])

    visited_nodes = set()

    while queue:

        node = queue.popleft()

        node_id = node["id"]

        if node_id in visited_nodes:
            continue

        visited_nodes.add(node_id)

        children = get_children(node_id)

        queue.extend(children)

这一步非常关键。

否则如果后台分类出现循环关系，程序会永远运行。

---

7.6 不猜 API：先发现真实 XHR

动态站我一直不推荐做一件事：

看一眼页面，然后自己猜：

/api/category/list
/api/tree
/getCategory

看起来很合理。

但很可能完全不存在。

正确方式是：

浏览器真实打开页面
↓
监听 response
↓
记录 JSON 请求
↓
观察哪一个接口返回分类数据

Playwright 非常适合做这件事。

---

7.7 tools/discover_api.py

# tools/discover_api.py

import asyncio
import json

from playwright.async_api import async_playwright

from crawler.config import CATALOG_URL


async def main():

    async with async_playwright() as p:

        browser = await p.chromium.launch(
            headless=False
        )

        context = await browser.new_context(
            locale="zh-CN"
        )

        page = await context.new_page()

        async def handle_response(response):

            content_type = response.headers.get(
                "content-type",
                "",
            )

            if "application/json" not in content_type:
                return

            url = response.url

            try:
                payload = await response.json()
            except Exception:
                return

            print("\n" + "=" * 80)
            print("JSON URL:")
            print(url)

            try:
                text = json.dumps(
                    payload,
                    ensure_ascii=False,
                )
                print(text[:1500])
            except Exception:
                pass

        page.on(
            "response",
            handle_response,
        )

        await page.goto(
            CATALOG_URL,
            wait_until="domcontentloaded",
            timeout=60000,
        )

        print(
            "\n浏览器已经打开。"
            "请正常点击或展开左侧公开分类目录，"
            "终端会打印返回 JSON 的请求。\n"
        )

        await page.wait_for_timeout(
            120_000
        )

        await browser.close()


if __name__ == "__main__":
    asyncio.run(main())

运行：

python tools/discover_api.py

然后浏览器会打开。

此时正常：

展开一级分类
展开二级分类
翻页
点击某个资源

终端就会不断出现：

JSON URL:
https://...

{"code": ..., "data": ...}

我们需要寻找返回内容类似：

{
  "data": [
    {
      "id": "...",
      "name": "地球科学",
      "children": [...]
    }
  ]
}

或者：

{
  "records": [...],
  "total": 123
}

这种请求。

这是整个项目最重要的侦察阶段。

---

7.8 为什么 Playwright 只用于“发现”

浏览器开一个页面，会加载：

HTML
CSS
JS
字体
图片
埋点
接口

如果有 10 万个资源详情，全部用浏览器打开显然成本非常高。

所以最合理的结构通常是：

Playwright
   ↓
发现接口
   ↓
requests

浏览器是显微镜。

requests 才是流水线。

---

7.9 通用树节点模型

创建：

# crawler/models.py

from dataclasses import dataclass


@dataclass
class CategoryNode:
    node_id: str
    name: str
    parent_id: str = ""
    depth: int = 0


@dataclass
class ResourceItem:
    resource_id: str = ""
    title: str = ""
    pi_name: str = ""
    domain: str = ""
    data_volume: str = ""
    doi: str = ""
    detail_url: str = ""

---

7.10 通用树遍历器

# crawler/tree.py

import logging
from collections import deque

from crawler.models import CategoryNode


logger = logging.getLogger(__name__)


class TreeWalker:

    def __init__(
        self,
        get_children_func,
        on_node_func=None,
    ):
        self.get_children = get_children_func
        self.on_node = on_node_func

    def walk(self, roots):

        queue = deque(roots)

        visited = set()

        while queue:

            node = queue.popleft()

            if node.node_id in visited:
                continue

            visited.add(node.node_id)

            logger.info(
                "访问分类 depth=%s id=%s name=%s",
                node.depth,
                node.node_id,
                node.name,
            )

            if self.on_node:
                self.on_node(node)

            try:
                children = self.get_children(
                    node
                )
            except Exception:
                logger.exception(
                    "分类节点获取失败：%s",
                    node.node_id,
                )
                continue

            for child in children:

                if not child.parent_id:
                    child.parent_id = node.node_id

                child.depth = node.depth + 1

                if child.node_id not in visited:
                    queue.append(child)

        return visited

注意这里完全没有：

一级
二级
三级
四级

概念。

因此理论上：

3 层
30 层
300 层

代码都一样。

---

7.11 如果接口一次直接返回整棵树

有些站点会返回：

[
  {
    "id": "1",
    "name": "科学数据",
    "children": [
      {
        "id": "11",
        "name": "地球科学",
        "children": [
          ...
        ]
      }
    ]
  }
]

这种情况甚至不需要反复请求。

可以直接 flatten：

def flatten_tree(nodes):

    stack = [
        (node, 0, "")
        for node in reversed(nodes)
    ]

    result = []

    while stack:

        node, depth, parent_id = stack.pop()

        node_id = str(
            node.get("id")
            or node.get("value")
            or node.get("code")
            or ""
        )

        name = (
            node.get("name")
            or node.get("label")
            or node.get("title")
            or ""
        )

        result.append(
            {
                "node_id": node_id,
                "name": name,
                "parent_id": parent_id,
                "depth": depth,
            }
        )

        children = (
            node.get("children")
            or []
        )

        for child in reversed(children):
            stack.append(
                (
                    child,
                    depth + 1,
                    node_id,
                )
            )

    return result

这个函数的好处是完全不依赖深度。

---

7.12 延迟加载树的情况

还有一种更常见：

第一次：

GET children?id=ROOT

返回：

[
    {"id": "A"},
    {"id": "B"}
]

然后展开 A：

GET children?id=A

再返回：

[
    {"id": "A1"},
    {"id": "A2"}
]

这种就是典型：

lazy tree。

实现：

def get_children(node):

    response = fetcher.get(
        CATEGORY_ENDPOINT,
        params={
            "parentId": node.node_id
        },
    )

    payload = response.json()

    rows = extract_rows(payload)

    children = []

    for row in rows:

        children.append(
            CategoryNode(
                node_id=str(
                    row.get("id")
                    or row.get("value")
                    or row.get("code")
                ),
                name=(
                    row.get("name")
                    or row.get("label")
                    or ""
                ),
                parent_id=node.node_id,
            )
        )

    return children

这里故意没有把：

CATEGORY_ENDPOINT

写成某个“貌似真实”的地址。

原因很简单：

应该填写你通过浏览器实际发现的公开接口，而不是从教程里复制一个可能已经失效甚至从未存在的地址。

---

7.13 详情页 URL 收集

另一条路径是直接从 HTML 中收集：

/metadata/detail

链接。

实现：

from urllib.parse import urljoin

from bs4 import BeautifulSoup

from crawler.config import BASE_URL


def extract_detail_urls(html):

    soup = BeautifulSoup(
        html,
        "lxml",
    )

    result = set()

    for a in soup.select("a[href]"):

        href = a.get("href", "").strip()

        if "/metadata/detail" not in href:
            continue

        url = urljoin(
            BASE_URL,
            href,
        )

        result.add(url)

    return sorted(result)

---

7.14 resource_id 如何获取

公开详情地址中可能出现：

?cstrId=...

或者：

?id=...

所以最简单的 resource_id 来源是 URL。

from urllib.parse import urlparse, parse_qs


def extract_resource_id(url):

    parsed = urlparse(url)

    params = parse_qs(
        parsed.query
    )

    for key in (
        "id",
        "cstrId",
        "resourceId",
        "resource_id",
    ):

        values = params.get(key)

        if values and values[0]:
            return values[0].strip()

    return ""

这样比：

url.split("=")[1]

健壮很多。

因为 URL 可能有两个参数。

---

7.15 详情页字段抽取

详情页解析我不建议只依赖一个 CSS Selector。

比如写成：

soup.select_one(
    ".detail-info > div:nth-child(4)"
)

这种选择器非常脆。

前端稍微增加一个元素：

nth-child(4)

就错了。

更稳妥的办法是：

基于字段标签语义抽取。

比如页面文本：

学科分类
地球科学

DOI标识
10.xxxx/xxx

我们可以先转成：

{
    "学科分类": "地球科学",
    "DOI标识": "10.xxxx/xxx"
}

再映射。

---

7.16 parser.py

# crawler/parser.py

import re
from urllib.parse import parse_qs, urlparse

from bs4 import BeautifulSoup

from crawler.models import ResourceItem


FIELD_ALIASES = {

    "title": [
        "资源名称（中文）",
        "资源名称",
        "数据名称",
        "数据集名称",
        "标题",
    ],

    "pi_name": [
        "项目负责人",
        "负责人",
        "责任者",
        "数据贡献者",
        "联系人",
        "作者",
    ],

    "domain": [
        "学科分类",
        "学科类别",
        "所属领域",
        "领域",
        "主题分类",
    ],

    "data_volume": [
        "数据量",
        "数据大小",
        "数据体量",
        "文件大小",
    ],

    "doi": [
        "DOI标识",
        "DOI",
        "doi",
    ],
}


DOI_PATTERN = re.compile(
    r"\b10\.\d{4,9}/[-._;()/:A-Z0-9]+\b",
    re.IGNORECASE,
)


def normalize_text(text):

    if text is None:
        return ""

    text = str(text)

    text = text.replace("\xa0", " ")

    text = re.sub(
        r"\s+",
        " ",
        text,
    )

    return text.strip()


def extract_resource_id(url):

    parsed = urlparse(url)

    params = parse_qs(
        parsed.query
    )

    priority = (
        "id",
        "cstrId",
        "resourceId",
        "resource_id",
    )

    for key in priority:

        values = params.get(key)

        if values:
            value = values[0].strip()

            if value:
                return value

    return ""


def find_by_alias(
    field_map,
    aliases,
):

    for alias in aliases:

        value = field_map.get(alias)

        if value:
            return normalize_text(value)

    return ""


def html_to_field_map(html):

    soup = BeautifulSoup(
        html,
        "lxml",
    )

    field_map = {}

    # 方案一：表格
    for row in soup.select("tr"):

        cells = row.find_all(
            ["th", "td"],
            recursive=False,
        )

        if len(cells) < 2:
            continue

        key = normalize_text(
            cells[0].get_text(
                " ",
                strip=True,
            )
        )

        value = normalize_text(
            cells[1].get_text(
                " ",
                strip=True,
            )
        )

        if key and value:
            field_map.setdefault(
                key.rstrip("：:"),
                value,
            )

    # 方案二：dt / dd
    for dt in soup.select("dt"):

        dd = dt.find_next_sibling(
            "dd"
        )

        if not dd:
            continue

        key = normalize_text(
            dt.get_text(
                " ",
                strip=True,
            )
        )

        value = normalize_text(
            dd.get_text(
                " ",
                strip=True,
            )
        )

        if key and value:
            field_map.setdefault(
                key.rstrip("：:"),
                value,
            )

    # 方案三：简单 label/value DOM
    candidates = soup.find_all(
        ["div", "span", "p"]
    )

    alias_set = {
        alias
        for aliases in FIELD_ALIASES.values()
        for alias in aliases
    }

    for node in candidates:

        text = normalize_text(
            node.get_text(
                " ",
                strip=True,
            )
        ).rstrip("：:")

        if text not in alias_set:
            continue

        sibling = node.find_next_sibling()

        if sibling:

            value = normalize_text(
                sibling.get_text(
                    " ",
                    strip=True,
                )
            )

            if value:
                field_map.setdefault(
                    text,
                    value,
                )

    return field_map


def extract_title_from_page(soup):

    # h1 优先
    h1 = soup.find("h1")

    if h1:
        text = normalize_text(
            h1.get_text(
                " ",
                strip=True,
            )
        )

        if text:
            return text

    # 再尝试常见标题节点
    selectors = [
        ".title",
        ".resource-title",
        ".detail-title",
        "h2",
    ]

    for selector in selectors:

        node = soup.select_one(
            selector
        )

        if node:

            text = normalize_text(
                node.get_text(
                    " ",
                    strip=True,
                )
            )

            if text:
                return text

    # 最后 fallback 到 title 标签
    if soup.title:

        title = normalize_text(
            soup.title.get_text(
                " ",
                strip=True,
            )
        )

        title = re.sub(
            r"\s*[-|_]\s*资源详情.*$",
            "",
            title,
        )

        return title

    return ""


def extract_doi_fallback(text):

    match = DOI_PATTERN.search(
        text
    )

    if not match:
        return ""

    return match.group(0).rstrip(
        ".,;，；"
    )


def parse_detail(
    html,
    detail_url,
):

    soup = BeautifulSoup(
        html,
        "lxml",
    )

    fields = html_to_field_map(
        html
    )

    title = find_by_alias(
        fields,
        FIELD_ALIASES["title"],
    )

    if not title:
        title = extract_title_from_page(
            soup
        )

    pi_name = find_by_alias(
        fields,
        FIELD_ALIASES["pi_name"],
    )

    domain = find_by_alias(
        fields,
        FIELD_ALIASES["domain"],
    )

    data_volume = find_by_alias(
        fields,
        FIELD_ALIASES["data_volume"],
    )

    doi = find_by_alias(
        fields,
        FIELD_ALIASES["doi"],
    )

    if not doi:

        page_text = normalize_text(
            soup.get_text(
                " ",
                strip=True,
            )
        )

        doi = extract_doi_fallback(
            page_text
        )

    return ResourceItem(
        resource_id=extract_resource_id(
            detail_url
        ),
        title=title,
        pi_name=pi_name,
        domain=domain,
        data_volume=data_volume,
        doi=doi,
        detail_url=detail_url,
    )

---

7.17 为什么一定要做缺失字段容错

科研资源门户特别容易出现这种情况：

A 数据：

负责人：张三
DOI：10.xxx/aaa

B 数据：

联系人：李四
DOI：

C 数据：

数据贡献者：王五

如果你写：

doi = soup.select_one(".doi").text

遇到没有 DOI 的记录直接：

AttributeError

整个任务就挂了。

更合理：

doi = ""

缺失并不是异常。

缺失是数据本身的一种状态。

---

7.18 DOI 二次正则兜底

一些网页不是：

DOI标识 | 10.xxxx/xxx

而是正文里：

doi=10.xxxx/xxx

因此准备一个 DOI regex 很有价值：

DOI_PATTERN = re.compile(
    r"\b10\.\d{4,9}/[-._;()/:A-Z0-9]+\b",
    re.I,
)

如果结构化字段没找到，就在页面文本中寻找。

---

8️⃣ 数据存储与导出（Storage）

第一版我推荐 SQLite。

理由：

不需要安装数据库服务器
支持 SQL
支持唯一约束
支持事务
支持 UPSERT
单文件方便备份

比 CSV 更适合作为爬虫主存储。

CSV 应该是：

导出格式

而不是：

任务状态数据库。

---

8.1 建表

# crawler/storage.py

import csv
import sqlite3
from pathlib import Path

from crawler.config import (
    DB_PATH,
    CSV_PATH,
)


class Storage:

    def __init__(
        self,
        db_path=DB_PATH,
    ):

        Path(db_path).parent.mkdir(
            parents=True,
            exist_ok=True,
        )

        self.conn = sqlite3.connect(
            db_path
        )

        self.conn.execute(
            """
            PRAGMA journal_mode=WAL
            """
        )

        self.create_tables()

    def create_tables(self):

        self.conn.execute(
            """
            CREATE TABLE IF NOT EXISTS resources (
                resource_id TEXT PRIMARY KEY,
                title TEXT NOT NULL DEFAULT '',
                pi_name TEXT NOT NULL DEFAULT '',
                domain TEXT NOT NULL DEFAULT '',
                data_volume TEXT NOT NULL DEFAULT '',
                doi TEXT NOT NULL DEFAULT '',
                detail_url TEXT NOT NULL DEFAULT '',
                fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP
            )
            """
        )

        self.conn.execute(
            """
            CREATE UNIQUE INDEX IF NOT EXISTS
            idx_resources_detail_url
            ON resources(detail_url)
            """
        )

        self.conn.execute(
            """
            CREATE TABLE IF NOT EXISTS visited_categories (
                node_id TEXT PRIMARY KEY,
                node_name TEXT NOT NULL DEFAULT '',
                parent_id TEXT NOT NULL DEFAULT '',
                depth INTEGER NOT NULL DEFAULT 0,
                fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP
            )
            """
        )

        self.conn.execute(
            """
            CREATE TABLE IF NOT EXISTS crawl_urls (
                url TEXT PRIMARY KEY,
                status TEXT NOT NULL DEFAULT 'pending',
                retry_count INTEGER NOT NULL DEFAULT 0,
                last_error TEXT NOT NULL DEFAULT '',
                updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
            )
            """
        )

        self.conn.commit()

---

8.2 保存资源

    def save_resource(
        self,
        item,
    ):

        self.conn.execute(
            """
            INSERT INTO resources (
                resource_id,
                title,
                pi_name,
                domain,
                data_volume,
                doi,
                detail_url
            )
            VALUES (?, ?, ?, ?, ?, ?, ?)

            ON CONFLICT(resource_id)
            DO UPDATE SET

                title = excluded.title,
                pi_name = excluded.pi_name,
                domain = excluded.domain,
                data_volume = excluded.data_volume,
                doi = excluded.doi,
                detail_url = excluded.detail_url,
                fetched_at = CURRENT_TIMESTAMP
            """,
            (
                item.resource_id,
                item.title,
                item.pi_name,
                item.domain,
                item.data_volume,
                item.doi,
                item.detail_url,
            ),
        )

        self.conn.commit()

---

8.3 resource_id 为空怎么办

有些 URL 可能没有稳定 ID。

这时可以 fallback 到 URL hash。

import hashlib


def make_resource_id(
    resource_id,
    detail_url,
):

    if resource_id:
        return resource_id

    return hashlib.sha256(
        detail_url.encode("utf-8")
    ).hexdigest()

更稳。

---

8.4 保存待抓 URL

    def add_url(
        self,
        url,
    ):

        self.conn.execute(
            """
            INSERT OR IGNORE INTO
            crawl_urls(url)
            VALUES (?)
            """,
            (url,),
        )

        self.conn.commit()

---

8.5 修改状态

    def mark_url_done(
        self,
        url,
    ):

        self.conn.execute(
            """
            UPDATE crawl_urls
            SET
                status='done',
                updated_at=CURRENT_TIMESTAMP
            WHERE url=?
            """,
            (url,),
        )

        self.conn.commit()


    def mark_url_failed(
        self,
        url,
        error,
    ):

        self.conn.execute(
            """
            UPDATE crawl_urls
            SET
                status='failed',
                retry_count=retry_count+1,
                last_error=?,
                updated_at=CURRENT_TIMESTAMP
            WHERE url=?
            """,
            (
                str(error)[:1000],
                url,
            ),
        )

        self.conn.commit()

---

8.6 获取待处理 URL

    def pending_urls(
        self,
        limit=100,
    ):

        cursor = self.conn.execute(
            """
            SELECT url
            FROM crawl_urls
            WHERE
                status='pending'
                OR (
                    status='failed'
                    AND retry_count < 3
                )
            LIMIT ?
            """,
            (limit,),
        )

        return [
            row[0]
            for row in cursor.fetchall()
        ]

---

8.7 字段映射表

最终数据库：

字段	SQLite 类型	示例
resource_id	TEXT	20146.11.2018.05.13.V1
title	TEXT	中国某区域气候数据集
pi_name	TEXT	张某
domain	TEXT	地球科学
data_volume	TEXT	3.08 MB
doi	TEXT	10.xxxx/example.v1
detail_url	TEXT	/metadata/detail?...

为什么 data_volume 不直接用 INTEGER？

因为网页可能显示：

1.2 GB
850 MB
3.08 MB
约500TB
--

直接保存原始字符串更安全。

后续如果需要统计，再增加：

data_volume_bytes

字段。

---

8.8 数据量标准化

可以增加：

import re


UNIT_MAP = {
    "B": 1,
    "KB": 1024,
    "MB": 1024 ** 2,
    "GB": 1024 ** 3,
    "TB": 1024 ** 4,
    "PB": 1024 ** 5,
}


def parse_data_volume(value):

    if not value:
        return None

    value = (
        value.upper()
        .replace(" ", "")
        .replace("约", "")
    )

    match = re.search(
        r"([\d.]+)(PB|TB|GB|MB|KB|B)",
        value,
    )

    if not match:
        return None

    number = float(
        match.group(1)
    )

    unit = match.group(2)

    return int(
        number * UNIT_MAP[unit]
    )

例如：

print(
    parse_data_volume(
        "3.08 MB"
    )
)

得到：

3229614

以后统计：

SUM(data_volume_bytes)

就很方便。

---

8.9 去重策略

推荐两层。

第一层：

resource_id UNIQUE

第二层：

detail_url UNIQUE

如果两者都不存在，再考虑：

内容 hash

例如：

import hashlib


def content_hash(
    title,
    pi_name,
    doi,
):

    raw = "|".join(
        [
            title.strip(),
            pi_name.strip(),
            doi.strip().lower(),
        ]
    )

    return hashlib.sha256(
        raw.encode("utf-8")
    ).hexdigest()

不建议只使用：

title

去重。

因为不同版本资源可能同名。

---

8.10 导出 CSV

    def export_csv(
        self,
        path=CSV_PATH,
    ):

        Path(path).parent.mkdir(
            parents=True,
            exist_ok=True,
        )

        cursor = self.conn.execute(
            """
            SELECT
                resource_id,
                title,
                pi_name,
                domain,
                data_volume,
                doi,
                detail_url
            FROM resources
            ORDER BY resource_id
            """
        )

        with open(
            path,
            "w",
            encoding="utf-8-sig",
            newline="",
        ) as f:

            writer = csv.writer(f)

            writer.writerow(
                [
                    "resource_id",
                    "title",
                    "pi_name",
                    "domain",
                    "data_volume",
                    "doi",
                    "detail_url",
                ]
            )

            writer.writerows(
                cursor.fetchall()
            )

这里使用：

utf-8-sig

是为了让 Windows Excel 直接打开中文 CSV 时更少遇到乱码。

---

9️⃣ 运行方式与结果展示

现在把整个过程串起来。

---

9.1 main.py

import logging

from crawler.fetcher import Fetcher
from crawler.parser import (
    parse_detail,
)
from crawler.storage import Storage


def setup_logging():

    logging.basicConfig(
        level=logging.INFO,
        format=(
            "%(asctime)s "
            "[%(levelname)s] "
            "%(name)s - "
            "%(message)s"
        ),
        handlers=[
            logging.FileHandler(
                "logs/crawler.log",
                encoding="utf-8",
            ),
            logging.StreamHandler(),
        ],
    )


def crawl_pending():

    fetcher = Fetcher()

    storage = Storage()

    while True:

        urls = storage.pending_urls(
            limit=50
        )

        if not urls:
            break

        for url in urls:

            try:

                response = fetcher.get(
                    url
                )

                item = parse_detail(
                    response.text,
                    url,
                )

                if not item.resource_id:

                    import hashlib

                    item.resource_id = (
                        hashlib.sha256(
                            url.encode(
                                "utf-8"
                            )
                        ).hexdigest()
                    )

                storage.save_resource(
                    item
                )

                storage.mark_url_done(
                    url
                )

                logging.info(
                    "完成：%s | %s",
                    item.resource_id,
                    item.title,
                )

            except Exception as exc:

                logging.exception(
                    "抓取失败：%s",
                    url,
                )

                storage.mark_url_failed(
                    url,
                    exc,
                )

    storage.export_csv()


def main():

    setup_logging()

    crawl_pending()


if __name__ == "__main__":
    main()

---

9.2 第一次运行前如何准备 URL

通过前文：

python tools/discover_api.py

确认真实分类/搜索接口。

然后：

分类树
↓
分类节点
↓
资源列表
↓
详情链接

将详情 URL 写入：

crawl_urls

表。

例如：

from crawler.storage import Storage


storage = Storage()


urls = [
    "https://www.escience.org.cn/metadata/detail?cstrId=...",
]


for url in urls:
    storage.add_url(url)

正式工程中，则由分类遍历模块自动调用：

storage.add_url(url)

---

9.3 启动

python main.py

日志：

2026-08-10 10:21:13 [INFO] crawler.fetcher - GET https://...
2026-08-10 10:21:15 [INFO] root - 完成：xxxx | 某科学数据集
2026-08-10 10:21:17 [INFO] crawler.fetcher - GET https://...

---

9.4 输出位置

SQLite：

data/metadata.db

CSV：

data/resources.csv

日志：

logs/crawler.log

---

9.5 查询结果

打开 SQLite：

sqlite3 data/metadata.db

查询：

SELECT
    resource_id,
    title,
    pi_name,
    domain,
    data_volume,
    doi
FROM resources
LIMIT 5;

结构示意：

resource_id                 title                   pi_name   domain       data_volume   doi
--------------------------  ----------------------  --------  -----------  ------------  -----------------------
20146.xxxx                  某气候区划数据集           张某       地球科学       3.08 MB       10.xxxx/example01
17099.xxxx                  某逐日环境数据集           李某       环境科学       1.26 GB       10.xxxx/example02
17099.xxxx                  某生态观测数据集           王某       生态学         842 MB        10.xxxx/example03
xxxxxx                      某地理空间数据集           陈某       地理学         5.4 GB        10.xxxx/example04
xxxxxx                      某基础科学数据集           赵某       基础科学                      10.xxxx/example05

这里的内容是数据格式示意，实际采集结果应以程序运行时公开页面返回内容为准，不应把示例行当成真实项目记录。

---

🔟 常见问题与排错

10.1 403 Forbidden 怎么办

403 不意味着：

赶紧换代理

先检查：

第一，URL 对不对

浏览器能打开吗？

第二，Referer 是否合理

headers = {
    "Referer": "https://www.escience.org.cn/"
}

第三，请求是不是太频繁

如果刚才：

20 threads

直接降为：

1 thread

第四，是不是这个接口本来就只供浏览器上下文使用

如果依赖动态 token，那么优先重新分析公开页面流程，而不是尝试绕过。

---

10.2 429 Too Many Requests

429 的含义非常直白：

请求过多

正确处理方式：

if response.status_code == 429:
    sleep(...)

如果服务器返回：

Retry-After

应该尊重它。

urllib3：

respect_retry_after_header=True

正是干这个的。

---

10.3 HTML 抓到了，但只有空壳

例如：

html = requests.get(url).text

print(len(html))

看起来很多内容。

但你搜索：

DOI

根本没有。

通常意味着：

数据通过 JavaScript 后加载

解决方式：

打开：

F12
→ Network
→ Fetch/XHR

或者运行：

python tools/discover_api.py

找到真正的数据来源。

---

10.4 Playwright 为什么看到数据，requests 看不到？

因为：

浏览器执行了 JavaScript
requests 不执行 JavaScript

浏览器最终 DOM：

HTML 初始模板
+
JS
+
接口数据

requests 只有：

HTML 初始响应

二者不是一回事。

---

10.5 XPath 突然失效

如果你写：

//*[@id="app"]/div[2]/div[3]/div[1]

它基本注定会失效。

尽量根据语义：

DOI标识
学科分类
数据量

寻找相邻值。

---

10.6 数据量抓错成访问量

这是科研平台解析里非常常见的坑。

页面可能同时显示：

数据量
10259 数据访问量
583 数据下载量

如果简单使用：

re.search(
    r"数据量.*?(\d+)",
    text
)

有可能把：

10259

当成数据量。

所以应优先根据 DOM 标签和字段对应关系解析，而不是整页正则。

---

10.7 DOI 后面多出文字

例如：

10.1234/abc.v1数据联系人

说明正则范围过宽。

推荐：

DOI_PATTERN = re.compile(
    r"\b10\.\d{4,9}/[-._;()/:A-Z0-9]+\b",
    re.I,
)

并：

rstrip(".,;，；")

---

10.8 中文乱码

requests 一般：

response.encoding

会自动判断。

如果明显错误，可以：

response.encoding = (
    response.apparent_encoding
)

但是不要无脑覆盖。

CSV：

encoding="utf-8-sig"

通常对 Excel 兼容比较友好。

---

10.9 分类一直重复

说明树不是严格树，或者后台返回重复节点。

加入：

visited = set()

：

if node_id in visited:
    continue

---

10.10 程序运行几个小时后突然挂了

这正是为什么不能只使用：

visited = set()

保存在内存。

应该保存：

visited_categories
crawl_urls

到 SQLite。

下次启动：

done

跳过，

pending

继续，

failed

在限定次数内重试。

这才是真正的断点续跑。

---

1️⃣1️⃣ 进阶优化

11.1 完整断点续跑模型

爬虫状态可以设计成：

pending
processing
done
failed

开始：

UPDATE crawl_urls
SET status='processing'
WHERE url=?;

成功：

done

失败：

failed

如果程序异常退出，一些 URL 会一直是：

processing

启动时恢复：

UPDATE crawl_urls
SET status='pending'
WHERE status='processing';

这招非常实用。

---

11.2 批量提交事务

前面的示例每条：

commit()

简单但不是最快。

数据量大后可以：

buffer = []

for item in items:

    buffer.append(item)

    if len(buffer) >= 100:

        save_batch(buffer)

        buffer.clear()

每 100 条 commit。

---

11.3 轻量并发

如果确认网站允许，并且单线程过慢，可以：

from concurrent.futures import (
    ThreadPoolExecutor,
    as_completed,
)

例如：

MAX_WORKERS = 3

而不是：

MAX_WORKERS = 100

示例：

def crawl_one(url):

    response = fetcher.get(
        url
    )

    return parse_detail(
        response.text,
        url,
    )


with ThreadPoolExecutor(
    max_workers=3
) as executor:

    futures = {
        executor.submit(
            crawl_one,
            url,
        ): url
        for url in urls
    }

    for future in as_completed(
        futures
    ):

        url = futures[future]

        try:
            item = future.result()

            storage.save_resource(
                item
            )

        except Exception:
            logging.exception(
                "失败：%s",
                url,
            )

不过注意：

当前 Fetcher：

session

是共享的。

严格工程环境下，可以使用：

thread-local Session

减少共享状态问题。

---

11.4 asyncio

如果后期发现接口非常标准，而且服务器允许一定并发，可以迁移：

aiohttp
httpx.AsyncClient

结构：

async def fetch(url):
    ...


async def worker(queue):
    ...

但不要为了“看起来高级”强行 async。

如果总共只有：

3000

个资源，一秒一条也没有多大的工程压力。

---

11.5 日志

至少记录：

访问 URL
分类节点 ID
资源 ID
HTTP 状态
解析成功
解析失败
重试次数
耗时

推荐：

logger.info(
    "resource_id=%s title=%s",
    item.resource_id,
    item.title,
)

不要：

print("ok")

因为出了问题以后：

ok
ok
ok
error
ok

基本无法排查。

---

11.6 成功率监控

增加表：

CREATE TABLE crawl_stats (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    success_count INTEGER,
    failed_count INTEGER,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);

或者运行完直接统计：

SELECT status, COUNT(*)
FROM crawl_urls
GROUP BY status;

得到：

done       9652
failed       17
pending       0

成功率：

9652 / (9652 + 17)

就能快速知道任务质量。

---

11.7 字段完整率

甚至可以做：

SELECT
    COUNT(*) AS total,

    SUM(
        CASE
            WHEN doi <> ''
            THEN 1
            ELSE 0
        END
    ) AS doi_count,

    SUM(
        CASE
            WHEN pi_name <> ''
            THEN 1
            ELSE 0
        END
    ) AS pi_count

FROM resources;

这个结果甚至比“爬了多少条”更有意义。

因为：

爬到 10000 条

但：

9000 条 title 都是空的

显然不能叫成功。

---

11.8 分类路径保存

现在：

domain

只保存最终领域。

实际上最好额外保存：

category_path

例如：

科学数据
>
地球科学
>
地理学
>
自然地理
>
气候

实现时给 CategoryNode 增加：

path: tuple[str, ...]

：

child.path = (
    *node.path,
    child.name,
)

最终：

category_path = " > ".join(
    child.path
)

这个字段后续做分类统计特别好用。

---

11.9 保存原始 HTML

对于重要采集任务，我喜欢保留一部分原始响应。

目录：

data/raw/

按 URL hash：

hashlib.sha256(
    url.encode()
).hexdigest()

保存：

xxxxxxxx.html

好处是：

半年以后解析器出问题，你还有原始数据可以重新解析。

而不是重新访问网站。

---

11.10 内容 Hash

网页发生更新怎么办？

加入：

content_hash

：

def sha256_text(text):

    import hashlib

    return hashlib.sha256(
        text.encode(
            "utf-8",
            errors="ignore",
        )
    ).hexdigest()

数据库：

content_hash

下一次抓取：

hash 相同
→ 不重新解析

hash 不同
→ 更新

这样可以构建增量更新器。

---

11.11 增量采集

第一轮：

全量

之后：

每天 / 每周

只检查：

新增资源
变更资源

理想流程：

抓目录
↓
获取 resource_id
↓
数据库 EXISTS？
├─ no → 新资源
└─ yes
    ↓
检查更新时间/hash
    ↓
变化？
├─ no → skip
└─ yes → 更新

---

11.12 定时任务

Linux cron：

crontab -e

每周日凌晨 3 点：

0 3 * * 0 cd /opt/nstr_metadata_crawler && /opt/nstr_metadata_crawler/.venv/bin/python main.py >> logs/cron.log 2>&1

如果任务规模继续扩大，可以考虑：

Airflow
Prefect
Dagster

但对于一个单站点元数据任务：

cron + SQLite

往往已经足够稳定。

---

11.13 从 SQLite 升级 MySQL/PostgreSQL

当需求变成：

多个采集节点
多人访问
Web 查询
百万级以上数据

再考虑：

PostgreSQL
MySQL

数据库表设计基本不用大改。

---

11.14 Scrapy 化

如果后期已经确认：

分类 API
列表 API
详情 API

全部可稳定通过 HTTP 调用，那么 Playwright 可以退居辅助地位。

Scrapy Spider 大概：

import scrapy


class MetadataSpider(
    scrapy.Spider
):

    name = "metadata"

    custom_settings = {
        "CONCURRENT_REQUESTS": 4,
        "DOWNLOAD_DELAY": 1.5,
        "RANDOMIZE_DOWNLOAD_DELAY": True,
        "RETRY_TIMES": 3,
    }

    def start_requests(self):

        yield scrapy.Request(
            self.start_url,
            callback=self.parse_tree,
        )

    def parse_tree(
        self,
        response,
    ):

        ...

    def parse_list(
        self,
        response,
    ):

        ...

    def parse_detail(
        self,
        response,
    ):

        ...

Scrapy 的优势在：

Scheduler
Downloader
Retry
Pipeline
DupeFilter
AutoThrottle
日志
并发

这些基础设施已经替你做好。

---

11.15 Playwright 作为兜底，而不是默认

实际项目中我通常按这个优先级：

公开 API
↓
requests HTML
↓
Playwright

而不是：

Playwright
↓
什么都 Playwright

原因只有一个：

越简单的技术，长期维护成本通常越低。

---

11.16 无限树爬虫真正的核心公式

如果只让我留下本文一个代码片段，我会留这个：

from collections import deque


queue = deque(
    root_nodes
)

visited = set()


while queue:

    node = queue.popleft()

    if node.id in visited:
        continue

    visited.add(
        node.id
    )

    save_node(
        node
    )

    crawl_resources(
        node
    )

    children = get_children(
        node
    )

    for child in children:

        if child.id not in visited:
            queue.append(
                child
            )

所有复杂的分类结构，最后都可以压缩成三个动作：

访问节点
抓节点资源
加入子节点

只要不假设深度：

无限层级

就不再是特殊问题。

---

11.17 为什么我更推荐 BFS 而不是递归 DFS

传统递归：

def crawl(node):

    for child in children(node):
        crawl(child)

确实漂亮。

但生产环境容易有三个问题。

第一：

递归深度

第二：

中断状态不好落盘

第三：

队列优先级不好控制

BFS：

queue

可以直接序列化到数据库。

例如：

node_id
status
priority
depth

于是爬虫重启之后仍然知道：

下一步该处理谁。

这才是工程化和算法题之间最大的区别。

---

11.18 一个更完整的任务队列表

如果数据量真的很大，可以创建：

CREATE TABLE category_queue (

    node_id TEXT PRIMARY KEY,

    node_name TEXT,

    parent_id TEXT,

    depth INTEGER,

    status TEXT DEFAULT 'pending',

    retry_count INTEGER DEFAULT 0,

    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,

    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
);

取一个：

SELECT *
FROM category_queue
WHERE status='pending'
ORDER BY depth ASC
LIMIT 1;

完成：

UPDATE category_queue
SET status='done'
WHERE node_id=?;

子节点：

INSERT OR IGNORE INTO category_queue (
    node_id,
    node_name,
    parent_id,
    depth
)
VALUES (?, ?, ?, ?);

到这里，所谓：

无限目录递归

其实已经变成：

数据库驱动的图遍历任务。

---

11.19 防止错误页面污染数据

有时候返回：

HTTP 200

但实际上是：

系统繁忙
访问异常
验证码
维护页面

所以不能：

if status_code == 200:
    save()

还应该检查字段。

def validate_item(
    item
):

    if not item.title:
        return False

    if len(item.title) > 500:
        return False

    return True

：

item = parse_detail(
    html,
    url,
)

if not validate_item(item):

    raise ValueError(
        "页面解析结果异常"
    )

---

11.20 建立异常样本目录

解析失败时保存：

data/errors/

：

def save_error_html(
    url,
    html,
):

    import hashlib
    from pathlib import Path

    folder = Path(
        "data/errors"
    )

    folder.mkdir(
        parents=True,
        exist_ok=True,
    )

    name = hashlib.md5(
        url.encode()
    ).hexdigest()

    path = folder / f"{name}.html"

    path.write_text(
        html,
        encoding="utf-8",
    )

一天后打开失败样本，比盯日志猜：

为什么 selector 不对

快得多。

---

11.21 数据质量检查

采集完建议至少跑四项检查。

空标题

SELECT COUNT(*)
FROM resources
WHERE title='';

DOI 空值率

SELECT
    ROUND(
        100.0 *
        SUM(CASE WHEN doi='' THEN 1 ELSE 0 END)
        / COUNT(*),
        2
    )
FROM resources;

resource_id 重复

理论上主键已经杜绝：

SELECT resource_id, COUNT(*)
FROM resources
GROUP BY resource_id
HAVING COUNT(*) > 1;

异常 DOI

SELECT doi
FROM resources
WHERE
    doi <> ''
    AND doi NOT LIKE '10.%';

这些 SQL 很朴素，但实际特别有用。

---

11.22 数据量排行

如果增加：

data_volume_bytes

：

SELECT
    title,
    data_volume,
    data_volume_bytes
FROM resources
WHERE data_volume_bytes IS NOT NULL
ORDER BY data_volume_bytes DESC
LIMIT 20;

就可以分析大体量开放科学资源。

---

11.23 DOI 覆盖情况

按领域：

SELECT
    domain,

    COUNT(*) AS total,

    SUM(
        CASE
            WHEN doi <> ''
            THEN 1
            ELSE 0
        END
    ) AS doi_count

FROM resources

GROUP BY domain

ORDER BY total DESC;

这样一个原本只是：

爬虫

的项目，就开始转向：

开放科学元数据分析

了。

---

1️⃣2️⃣ 总结与延伸阅读

这次我们并没有写一个：

requests.get()
BeautifulSoup()
for

就结束的小脚本。

完整流程实际包含：

公开目录入口
        ↓
判断页面动态性
        ↓
Playwright 监听真实请求
        ↓
识别分类树接口
        ↓
建立 CategoryNode
        ↓
BFS 遍历未知深度目录
        ↓
visited 防重复、防环
        ↓
采集分类中的资源列表
        ↓
生成 detail_url
        ↓
requests.Session 低频请求
        ↓
Retry + Backoff
        ↓
HTML / JSON 解析
        ↓
字段别名归一化
        ↓
缺失字段容错
        ↓
resource_id / URL 去重
        ↓
SQLite UPSERT
        ↓
断点续跑
        ↓
CSV 导出
        ↓
数据质量检查

最终我们的目标表只有六个核心字段：

resource_id
title
pi_name
domain
data_volume
doi

但真正有价值的代码并不是：

doi = xxx
title = xxx

而是前面的整个采集框架。

尤其是树状分类。

如果写死：

一级 → 二级 → 三级

今天能跑。

明天网站增加第四级，代码就得重写。

如果抽象成：

node → children

再使用：

queue + visited

网站目录变成五级、十级甚至更深，算法仍然成立。

这是本文最值得记住的一点：

不要爬“第几层目录”，要爬“节点关系”。

同样，动态网站也不要执着于：

HTML 上有没有数据。

应该先问：

浏览器到底从哪里得到这些数据？

Network/XHR 往往比反复试 XPath 更快。

而在存储层，也不要把：

CSV

当成全部。

真正长期运行的采集任务一定需要：

任务状态
失败记录
已访问节点
唯一约束
更新时间

这些持久化信息。

因此，即使最终只是为了得到一份：

resources.csv

背后最好仍然使用：

SQLite

作为任务核心。

后续这个项目可以继续向三个方向扩展。

第一条路线：

requests
↓
Scrapy

适用于资源量继续扩大，需要更加成熟的调度、去重、并发和 Pipeline。

第二条路线：

Playwright
↓
浏览器自动发现
↓
接口自动识别

适合应对越来越多的前端动态资源门户。

第三条路线：

元数据采集
↓
清洗
↓
DOI 规范化
↓
领域统计
↓
全文搜索
↓
知识图谱

这会让项目从“一个爬虫”变成真正的科研数据索引系统。

我一直觉得，爬虫写到一定阶段以后，最重要的能力已经不是：

会多少 XPath
会不会异步

而是能不能把网页抽象成稳定的数据模型。

页面会改。

CSS 类名会改。

前端框架会改。

接口参数甚至也会改。

但是：

分类节点
资源
负责人
领域
DOI
数据量

这些业务实体相对稳定。

所以最值得投入时间的地方，是把：

页面结构

和：

自己的数据结构

彻底解耦。

做到这一点以后，即使某一天网站更新前端，通常也只是：

修改 Fetcher
修改 Parser

而：

Storage
TreeWalker
任务状态
数据分析 SQL

都可以原样留下。

这才是一个可以维护、可以复跑、也适合继续扩展的 Python 数据采集项目。

🌟 文末

好啦～以上就是本期的全部内容啦！如果你在实践过程中遇到任何疑问，欢迎在评论区留言交流，我看到都会尽量回复～咱们下期见！

小伙伴们在批阅的过程中，如果觉得文章不错，欢迎点赞、收藏、关注哦～
三连就是对我写作道路上最好的鼓励与支持！ ❤️🔥

✅ 专栏持续更新中｜建议收藏 + 订阅

墙裂推荐订阅专栏 👉 《Python爬虫实战》，本专栏秉承着以“入门 → 进阶 → 工程化 → 项目落地”的路线持续更新，争取让每一期内容都做到：

✅ 讲得清楚（原理）｜✅ 跑得起来（代码）｜✅ 用得上（场景）｜✅ 扛得住（工程化）

📣 想系统提升的小伙伴：强烈建议先订阅专栏 《Python爬虫实战》，再按目录大纲顺序学习，效率十倍上升～

✅ 互动征集

想让我把【某站点/某反爬/某验证码/某分布式方案】等写成某期实战？

评论区留言告诉我你的需求，我会优先安排实现(更新)哒~

---

⭐️ 若喜欢我，就请关注我叭～（更新不迷路）
⭐️ 若对你有用，就请点赞支持一下叭～（给我一点点动力）
⭐️ 若有疑问，就请评论留言告诉我叭～（我会补坑 & 更新迭代）

---

✅ 免责声明

本文爬虫思路、相关技术和代码仅用于学习参考，对阅读本文后的进行爬虫行为的用户本作者不承担任何法律责任。

使用或者参考本项目即表示您已阅读并同意以下条款：

• 合法使用： 不得将本项目用于任何违法、违规或侵犯他人权益的行为，包括但不限于网络攻击、诈骗、绕过身份验证、未经授权的数据抓取等。
• 风险自负： 任何因使用本项目而产生的法律责任、技术风险或经济损失，由使用者自行承担，项目作者不承担任何形式的责任。
• 禁止滥用： 不得将本项目用于违法牟利、黑产活动或其他不当商业用途。
• 使用或者参考本项目即视为同意上述条款,即 “谁使用，谁负责” 。如不同意，请立即停止使用并删除本项目。！！！