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

🌟 开篇语

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

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

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

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

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

0️⃣ 前言（Preface）

这篇文章要爬取的是 Python 官方 PEP 索引与每个 PEP 详情页，使用 requests + BeautifulSoup + lxml + SQLite/CSV 完成一个“索引页采集链接、详情页抽取字段”的二段式爬虫，最终产出一份结构化的 PEP 数据表。

读完这篇文章，你可以获得：

1. 掌握技术文档目录页到详情页的二段式抓取方法。
2. 学会为爬虫补齐请求层、解析层、存储层、容错、日志、去重等工程化细节。
3. 获得一套可以直接运行、可维护、可扩展的 Python PEP 索引采集项目代码。

我个人一直觉得，爬虫学习不能只停在“请求一个网页，然后打印标题”这个层面。真正写起来顺手的爬虫，往往不是代码量特别夸张，而是边界处理足够扎实：超时怎么办、页面结构变了怎么办、重复链接怎么去掉、失败任务怎么重跑、字段缺失怎么落库。PEP 索引这个案例刚好适合练这些东西，它不像电商站那样充满风控，也不像 SPA 应用那样一上来就动态渲染，它更像一份规整但真实的技术文档目录，非常适合用来练“二段式抓取”。

本文不会做攻击式并发，不会绕过登录限制，也不会采集任何敏感信息。所有内容只围绕公开技术文档的结构化整理展开。

1️⃣ 摘要（Abstract）

本文将以 Python 官方 PEP 索引为目标站点，使用 requests 获取页面、BeautifulSoup/lxml 解析 HTML、SQLite + CSV 存储导出，完成一个从索引页到详情页的二段式采集程序，输出字段包括 PEP 编号、标题、状态、作者、创建日期和详情页 URL。

读完本文，你将能够：

1. 理解“目录页抓链接、详情页抓字段”的通用爬虫模型。
2. 写出一个带重试、超时、User-Agent、频率控制、去重、日志和导出的完整项目。
3. 知道遇到 403、429、空 HTML、编码异常、选择器失效时如何排查。

这不是一个花哨的案例，但它非常接近我平时处理技术文档聚合任务时的写法。先把流程打通，再考虑性能；先把字段抓准，再考虑并发；先把失败留痕，再考虑自动化。爬虫不是越快越好，能稳定、可解释、可复跑，才是长期可用。

2️⃣ 背景与需求（Why）

2.1 为什么要爬 PEP 索引

PEP，全称是 Python Enhancement Proposal，也就是 Python 增强提案。它记录了 Python 语言、标准库、流程治理、发布节奏等方面的大量设计讨论和历史决策。对 Python 使用者来说，PEP 不只是“文档”，更像 Python 语言演化的档案库。

爬取 PEP 索引有几个实际价值：

第一，方便做数据分析。

例如我们可以统计不同状态的 PEP 数量：

• Final 有多少？
• Rejected 有多少？
• Active 有多少？
• Draft 有多少？
• 哪些年份创建的 PEP 最多？
• 哪些作者参与最多？

第二，方便做信息聚合。

PEP 页面本身是面向阅读的，但如果我们想把它做成一个检索表、知识库入口、内部技术导航页，就需要先将页面内容结构化。

第三，方便做自动化更新。

PEP 是长期维护的技术文档。手动整理一次还可以接受，但如果要每周、每月更新，就应该让脚本自动完成采集、解析、存储和导出。

第四，适合作为二段式爬虫练习。

PEP 索引页中包含很多详情页链接，而每个详情页中都有相似的元信息区域，比如 Author、Status、Type、Created 等。这个结构非常适合练习：

索引页 -> 提取详情页 URL -> 请求详情页 -> 抽取字段 -> 清洗 -> 存储

这正是很多技术文档站、标准文档站、规范索引站常见的数据结构。

2.2 目标站点

本文目标站点：

https://peps.python.org/

目标详情页示例：

https://peps.python.org/pep-0001/
https://peps.python.org/pep-0008/
https://peps.python.org/pep-0257/

2.3 目标字段

本次采集字段如下：

字段	说明	示例
pep_number	PEP 编号	8
title	PEP 标题	Style Guide for Python Code
status	当前状态	Active
authors	作者	Guido van Rossum, Barry Warsaw, Alyssa Coghlan
created	创建日期	05-Jul-2001
url	详情页 URL	https://peps.python.org/pep-0008/
source	数据来源	html
crawled_at	抓取时间	2026-06-08T10:12:00

虽然需求中只要求“PEP 编号、标题、状态、作者、创建日期”，但我会额外保留 url、source、crawled_at。这三个字段在实战里很重要，尤其是后续排查数据来源、对比更新、断点续跑时非常有用。

---

3️⃣ 合规与注意事项

爬虫文章必须讲合规。技术不是为了制造压力，更不是为了绕过网站规则。尤其是技术文档站点，公开内容本身通常是为了让开发者阅读、学习和引用，我们更应该以克制的方式访问。

3.1 robots.txt 基本说明

robots.txt 是网站放在根路径下的爬虫访问建议文件，常见位置类似：

https://example.com/robots.txt

爬虫在启动前，建议先检查目标站点的 robots 规则，判断自己的 User-Agent 是否允许访问目标路径。需要注意，robots.txt 不是权限系统，它更像是一种约定和声明。工程实践里，即使 robots 没有限制，也应该保持低频访问。

在本文代码中，我们会提供一个简单的 robots 检查函数：

from urllib.robotparser import RobotFileParser

def can_fetch(robots_url: str, user_agent: str, target_url: str) -> bool:
    rp = RobotFileParser()
    rp.set_url(robots_url)
    rp.read()
    return rp.can_fetch(user_agent, target_url)

实际运行时，可以这样检查：

robots_url = "https://peps.python.org/robots.txt"
target_url = "https://peps.python.org/"
ua = "Mozilla/5.0 (compatible; PepIndexCrawler/1.0; +https://example.com/bot-info)"
print(can_fetch(robots_url, ua, target_url))

如果读取失败，不建议直接高并发开爬。我的习惯是：读取失败时按保守策略处理，降低频率，减少请求量，必要时先手动确认站点规则。

3.2 频率控制

本文示例默认采用串行抓取，并在请求之间加入短暂休眠：

REQUEST_INTERVAL = 0.5

如果抓取全部 PEP 页面，数量并不算特别大，但依然不建议无间隔访问。比较合理的方式是：

• 每次请求设置 timeout。
• 每次请求之间 sleep。
• 失败时指数退避。
• 不要使用攻击式并发。
• 不要反复刷新同一个失败页面。

3.3 不采集敏感信息

本文只采集公开 PEP 技术文档中的元信息，不采集：

• 账号信息。
• 登录后内容。
• 个人隐私数据。
• 非公开接口数据。
• 付费墙内容。
• 需要授权才能访问的内容。

作者字段中有时可能包含公开显示的作者名，这属于技术文档公开元信息。示例中只做文档索引整理，不做画像分析，也不做任何不必要的个人信息扩展。

3.4 不绕过限制

本文不讨论绕过登录、不讨论破解验证码、不讨论绕过付费限制、不讨论隐藏真实访问目的。遇到 403 或 429 时，我们优先从合规角度处理：

• 降低频率。
• 检查请求头。
• 检查 robots。
• 确认是否访问了不该访问的路径。
• 等待一段时间后再试。
• 必要时停止任务。

这是我写爬虫时很坚持的一点：能慢一点，就不要硬冲。能用公开 API，就不要硬扒 HTML。能缓存结果，就不要重复请求。

---

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

4.1 静态、动态还是 API？

目标站点属于典型的静态文档站。PEP 页面不依赖复杂的前端渲染，页面内容直接存在于 HTML 中。因此，本文选择：

requests + BeautifulSoup + lxml

而不是直接使用 Playwright 或 Selenium。

当然，PEP 站点也提供了官方 JSON 元数据接口。生产环境中，如果需求只是获取 PEP 元信息，API 会更稳定、更省请求。但本篇的主题是“技术文档目录二段式抓取”，所以正文仍然以 HTML 抓取为主。API 可以作为校验方案或备用方案。

简单对比一下：

方案	适用场景	本文是否采用
静态 HTML 抓取	页面内容直接在 HTML 中	采用
动态渲染抓取	内容由 JavaScript 渲染	不需要
官方 API	站点提供稳定 JSON 数据	作为补充
Scrapy	大规模、多任务、工程化爬虫	可进阶
Playwright	需要浏览器执行 JS	不需要

4.2 整体流程

本项目流程如下：

采集
  |
  |-- 请求 PEP 索引页 https://peps.python.org/
  |
解析
  |
  |-- 从索引页提取所有 /pep-xxxx/ 详情页链接
  |-- 对链接去重并按编号排序
  |
采集
  |
  |-- 逐个请求 PEP 详情页
  |
解析
  |
  |-- 从详情页元信息区域抽取 Author / Status / Created
  |-- 从标题中抽取 PEP 编号和标题
  |
清洗
  |
  |-- 去除多余空白
  |-- 标准化日期字符串
  |-- 处理缺失字段
  |
存储
  |
  |-- 写入 SQLite
  |-- 导出 CSV
  |-- 保存失败任务日志

也可以画成更紧凑的流程：

Index Page
   ↓
Extract Detail Links
   ↓
Fetch Detail Pages
   ↓
Parse Metadata
   ↓
Normalize Records
   ↓
Deduplicate
   ↓
Save SQLite / Export CSV

4.3 为什么选 requests

requests 足够轻量，适合静态页面抓取。它的优势是：

• API 简洁。
• 容易设置 headers。
• 容易设置 timeout。
• 可以使用 Session 复用连接。
• 可以和 urllib3 Retry 结合做重试。

4.4 为什么选 BeautifulSoup + lxml

BeautifulSoup 的容错性很好，适合解析技术文档这种结构比较稳定但细节可能有变化的 HTML。lxml 作为解析器，速度更快，兼容性也不错。

本文解析思路不会依赖过于脆弱的选择器。比如我们不会写死“第几个 div、第几个 p”，而是尽量通过语义信息提取：

• 从所有链接中筛选 /pep-0001/ 这类 URL。
• 从页面标题中提取 PEP 1 – xxx。
• 从元信息区域中寻找 Author:、Status:、Created:。

这样的解析方式更耐用。

4.5 为什么起步用 SQLite 和 CSV

CSV 适合查看和分享，SQLite 适合去重和复跑。实战里我喜欢两个都保留：

• SQLite：作为主存储，便于唯一约束、更新、查询。
• CSV：作为交付结果，便于打开、分析、导入 Excel 或 Pandas。

MySQL 当然也可以，但这个案例数据量不大，用 SQLite 更清爽。

---

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

5.1 Python 版本

建议使用：

Python 3.10+

我推荐 Python 3.11 或 3.12。低版本不是不能跑，但新版本在类型注解、标准库体验和性能上都更舒服。

查看版本：

python --version

或者：

python3 --version

5.2 创建虚拟环境

Linux / macOS：

mkdir pep_index_crawler
cd pep_index_crawler

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

Windows PowerShell：

mkdir pep_index_crawler
cd pep_index_crawler

python -m venv .venv
.\.venv\Scripts\Activate.ps1

5.3 安装依赖

pip install requests beautifulsoup4 lxml tqdm

如果你想导出更复杂的数据格式，也可以安装 pandas：

pip install pandas

本文主代码不强依赖 pandas，避免项目变重。

5.4 requirements.txt

新建 requirements.txt：

requests>=2.31.0
beautifulsoup4>=4.12.0
lxml>=5.0.0
tqdm>=4.66.0

安装：

pip install -r requirements.txt

5.5 推荐项目结构

本文采用下面这个结构：

pep_index_crawler/
├── README.md
├── requirements.txt
├── run.py
├── data/
│   ├── pep_index.sqlite3
│   ├── pep_index.csv
│   └── failed_urls.txt
├── logs/
│   └── crawler.log
└── pep_crawler/
    ├── __init__.py
    ├── config.py
    ├── models.py
    ├── fetcher.py
    ├── parser.py
    ├── storage.py
    ├── utils.py
    └── crawler.py

这套结构看起来比单文件麻烦一点，但好处很明显：

• 请求逻辑在 fetcher.py。
• 解析逻辑在 parser.py。
• 存储逻辑在 storage.py。
• 调度逻辑在 crawler.py。
• 配置集中在 config.py。
• 入口只有 run.py。

后期想换 Scrapy、换 MySQL、加并发，都不会把一个文件改成一团。

---

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

请求层负责“如何安全、稳定、克制地拿到 HTML”。

这一层至少要考虑：

• headers。
• User-Agent。
• Referer。
• timeout。
• Session。
• Cookie 是否需要。
• 重试。
• 退避。
• 编码。
• 日志。

PEP 页面是公开静态文档，不需要登录，也不需要 Cookie。我们会使用 requests.Session 复用连接，并设置自定义 User-Agent。

6.1 config.py

创建文件：

# pep_crawler/config.py

from pathlib import Path

BASE_URL = "https://peps.python.org/"
INDEX_URL = "https://peps.python.org/"

USER_AGENT = (
    "Mozilla/5.0 (compatible; PepIndexCrawler/1.0; "
    "+https://example.com/bot-info)"
)

DEFAULT_HEADERS = {
    "User-Agent": USER_AGENT,
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "en-US,en;q=0.9",
    "Referer": BASE_URL,
    "Connection": "keep-alive",
}

REQUEST_TIMEOUT = 15
REQUEST_INTERVAL = 0.5

MAX_RETRIES = 3
BACKOFF_FACTOR = 1.5

PROJECT_ROOT = Path(__file__).resolve().parent.parent
DATA_DIR = PROJECT_ROOT / "data"
LOG_DIR = PROJECT_ROOT / "logs"

SQLITE_PATH = DATA_DIR / "pep_index.sqlite3"
CSV_PATH = DATA_DIR / "pep_index.csv"
FAILED_URLS_PATH = DATA_DIR / "failed_urls.txt"
LOG_PATH = LOG_DIR / "crawler.log"

这里有几个点：

第一，User-Agent 不要空着。虽然我们写的是学习案例，也应该尽量让请求看起来规范。更严格的生产环境里，User-Agent 最好能包含项目说明页或联系邮箱。

第二，Referer 可以写站点首页。对这个案例来说不是必须，但技术文档站通常不会因为没有 Referer 就拒绝请求。这里加上主要是演示请求头设计。

第三，REQUEST_INTERVAL 设置为 0.5 秒。这个案例页面数量不算巨大，串行抓取也能接受，不需要上来就并发。

第四，目录路径集中配置。很多脚本最烦的是路径散落各处，过几天自己都找不到输出文件在哪里。

6.2 utils.py

创建文件：

# pep_crawler/utils.py

import logging
import re
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable


def ensure_dirs(paths: Iterable[Path]) -> None:
    for path in paths:
        path.mkdir(parents=True, exist_ok=True)


def setup_logger(log_path: Path) -> logging.Logger:
    logger = logging.getLogger("pep_crawler")
    logger.setLevel(logging.INFO)

    if logger.handlers:
        return logger

    formatter = logging.Formatter(
        fmt="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )

    stream_handler = logging.StreamHandler(sys.stdout)
    stream_handler.setFormatter(formatter)
    stream_handler.setLevel(logging.INFO)

    file_handler = logging.FileHandler(log_path, encoding="utf-8")
    file_handler.setFormatter(formatter)
    file_handler.setLevel(logging.INFO)

    logger.addHandler(stream_handler)
    logger.addHandler(file_handler)

    return logger


def normalize_space(text: str | None) -> str:
    if not text:
        return ""
    return re.sub(r"\s+", " ", text).strip()


def utc_now_iso() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()


def write_failed_url(path: Path, url: str, reason: str) -> None:
    line = f"{utc_now_iso()}\t{url}\t{reason}\n"
    with path.open("a", encoding="utf-8") as f:
        f.write(line)

日志、目录创建、空白清洗、失败 URL 记录都放在工具层。很多新手项目一开始不写日志，等出问题时只能靠 print 猜。我的建议是：哪怕是小爬虫，也尽量写日志，后期能省很多时间。

6.3 fetcher.py

创建文件：

# pep_crawler/fetcher.py

from __future__ import annotations

import time
from dataclasses import dataclass

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

from pep_crawler.config import (
    DEFAULT_HEADERS,
    MAX_RETRIES,
    BACKOFF_FACTOR,
    REQUEST_TIMEOUT,
)


@dataclass
class FetchResult:
    url: str
    ok: bool
    status_code: int | None
    text: str
    error: str = ""


class Fetcher:
    def __init__(
        self,
        headers: dict[str, str] | None = None,
        timeout: int = REQUEST_TIMEOUT,
        max_retries: int = MAX_RETRIES,
        backoff_factor: float = BACKOFF_FACTOR,
    ) -> None:
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update(headers or DEFAULT_HEADERS)

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

        adapter = HTTPAdapter(max_retries=retry)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)

    def get(self, url: str) -> FetchResult:
        try:
            response = self.session.get(url, timeout=self.timeout)
            self._fix_encoding(response)

            if 200 <= response.status_code < 300:
                return FetchResult(
                    url=url,
                    ok=True,
                    status_code=response.status_code,
                    text=response.text,
                )

            return FetchResult(
                url=url,
                ok=False,
                status_code=response.status_code,
                text=response.text or "",
                error=f"unexpected status code: {response.status_code}",
            )

        except requests.Timeout as exc:
            return FetchResult(
                url=url,
                ok=False,
                status_code=None,
                text="",
                error=f"timeout: {exc}",
            )
        except requests.RequestException as exc:
            return FetchResult(
                url=url,
                ok=False,
                status_code=None,
                text="",
                error=f"request error: {exc}",
            )

    @staticmethod
    def _fix_encoding(response: Response) -> None:
        if not response.encoding:
            response.encoding = response.apparent_encoding or "utf-8"

    @staticmethod
    def polite_sleep(seconds: float) -> None:
        if seconds > 0:
            time.sleep(seconds)

6.4 headers 设计说明

本文请求头主要包含：

DEFAULT_HEADERS = {
    "User-Agent": "...",
    "Accept": "...",
    "Accept-Language": "...",
    "Referer": BASE_URL,
    "Connection": "keep-alive",
}

说明如下：

Header	作用
User-Agent	标识客户端来源
Accept	告诉服务器希望接收 HTML/XML 等内容
Accept-Language	语言偏好
Referer	来源页面
Connection	尝试复用连接

这里不伪装成具体浏览器版本，不使用随机 UA 池，也不加入代理池。因为这个案例是公开技术文档采集，不需要做对抗，也不应该做对抗。保持透明、低频、可控即可。

6.5 timeout 说明

REQUEST_TIMEOUT = 15

timeout 是必须写的。很多爬虫卡死不是因为代码错，而是因为某个请求一直等待。设置 timeout 后，请求层才能把失败交给上层处理。

6.6 session/cookie 说明

PEP 站点不需要登录，因此不需要 Cookie。使用 Session 的主要目的是：

• 复用 TCP 连接。
• 统一 headers。
• 统一 retry 配置。
• 方便后续扩展。

如果目标站点需要登录，那就不是本文范围了。本文不讨论登录后采集，也不讨论绕过访问限制。

6.7 失败处理：重试和退避

这里使用 urllib3.util.retry.Retry，对以下状态码进行重试：

status_forcelist=(429, 500, 502, 503, 504)

这些状态一般代表：

状态码	含义	处理建议
429	请求过多	降频、等待
500	服务端错误	重试少量次数
502	网关错误	重试少量次数
503	服务不可用	等待后重试
504	网关超时	重试少量次数

退避因子：

BACKOFF_FACTOR = 1.5

失败时不会马上狂刷，而是逐渐等待。爬虫要像正常用户一样克制，这是基本功。

---

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

解析层负责“从 HTML 里拿出结构化数据”。

本案例分两步：

1. 从索引页提取详情页链接。
2. 从详情页提取 PEP 编号、标题、状态、作者、创建日期。

7.1 models.py

先定义数据模型。

# pep_crawler/models.py

from __future__ import annotations

from dataclasses import dataclass, asdict


@dataclass
class PepLink:
    pep_number: int
    url: str


@dataclass
class PepRecord:
    pep_number: int
    title: str
    status: str
    authors: str
    created: str
    url: str
    source: str
    crawled_at: str

    def to_dict(self) -> dict:
        return asdict(self)

为什么使用 dataclass？

因为它足够轻量，又比裸 dict 更清晰。字段多起来之后，裸 dict 很容易写错 key，比如 created 写成 create_at，排查起来不舒服。

7.2 parser.py：提取详情页链接

创建文件：

# pep_crawler/parser.py

from __future__ import annotations

import re
from urllib.parse import urljoin

from bs4 import BeautifulSoup

from pep_crawler.config import BASE_URL
from pep_crawler.models import PepLink, PepRecord
from pep_crawler.utils import normalize_space, utc_now_iso


PEP_URL_PATTERN = re.compile(r"/pep-(\d{4})/?$")
TITLE_PATTERN = re.compile(r"^PEP\s+(\d+)\s+[–-]\s+(.+)$")


class PepParser:
    def parse_index_links(self, html: str, base_url: str = BASE_URL) -> list[PepLink]:
        soup = BeautifulSoup(html, "lxml")

        links: dict[int, str] = {}

        for a in soup.find_all("a", href=True):
            href = a.get("href", "")
            absolute_url = urljoin(base_url, href)

            match = PEP_URL_PATTERN.search(absolute_url)
            if not match:
                continue

            pep_number = int(match.group(1))
            links[pep_number] = absolute_url

        return [
            PepLink(pep_number=number, url=url)
            for number, url in sorted(links.items(), key=lambda item: item[0])
        ]

这里没有写死某个 table 的 class，也没有依赖某个栏目标题，而是从所有链接里筛选符合 /pep-0001/ 这种模式的 URL。这样做有两个好处：

第一，页面结构小幅调整时，链接模式一般不会变。

第二，PEP 索引页里可能同一个 PEP 链接出现多次，用 dict 可以自然按编号去重。

这里的正则：

PEP_URL_PATTERN = re.compile(r"/pep-(\d{4})/?$")

含义是匹配：

/pep-0001/
/pep-0008/
/pep-0257/

并提取四位数字。

7.3 parser.py：抽取详情页字段

继续补充 PepParser：

# pep_crawler/parser.py

from __future__ import annotations

import re
from urllib.parse import urljoin

from bs4 import BeautifulSoup

from pep_crawler.config import BASE_URL
from pep_crawler.models import PepLink, PepRecord
from pep_crawler.utils import normalize_space, utc_now_iso


PEP_URL_PATTERN = re.compile(r"/pep-(\d{4})/?$")
TITLE_PATTERN = re.compile(r"^PEP\s+(\d+)\s+[–-]\s+(.+)$")


class PepParser:
    def parse_index_links(self, html: str, base_url: str = BASE_URL) -> list[PepLink]:
        soup = BeautifulSoup(html, "lxml")

        links: dict[int, str] = {}

        for a in soup.find_all("a", href=True):
            href = a.get("href", "")
            absolute_url = urljoin(base_url, href)

            match = PEP_URL_PATTERN.search(absolute_url)
            if not match:
                continue

            pep_number = int(match.group(1))
            links[pep_number] = absolute_url

        return [
            PepLink(pep_number=number, url=url)
            for number, url in sorted(links.items(), key=lambda item: item[0])
        ]

    def parse_detail(self, html: str, url: str) -> PepRecord:
        soup = BeautifulSoup(html, "lxml")

        pep_number, title = self._parse_title(soup, url)
        meta = self._parse_metadata(soup)

        return PepRecord(
            pep_number=pep_number,
            title=title,
            status=meta.get("Status", ""),
            authors=meta.get("Author", ""),
            created=meta.get("Created", ""),
            url=url,
            source="html",
            crawled_at=utc_now_iso(),
        )

    def _parse_title(self, soup: BeautifulSoup, url: str) -> tuple[int, str]:
        h1 = soup.find("h1")
        raw_title = normalize_space(h1.get_text(" ", strip=True) if h1 else "")

        match = TITLE_PATTERN.match(raw_title)
        if match:
            return int(match.group(1)), normalize_space(match.group(2))

        url_match = PEP_URL_PATTERN.search(url)
        if url_match:
            pep_number = int(url_match.group(1))
        else:
            pep_number = -1

        fallback_title = raw_title.replace("Python Enhancement Proposals", "").strip()
        return pep_number, fallback_title

    def _parse_metadata(self, soup: BeautifulSoup) -> dict[str, str]:
        meta: dict[str, str] = {}

        # PEP 页面顶部通常有类似：
        # <dt>Author:</dt><dd>...</dd>
        # <dt>Status:</dt><dd>...</dd>
        # <dt>Created:</dt><dd>...</dd>
        for dt in soup.find_all("dt"):
            key = normalize_space(dt.get_text(" ", strip=True)).rstrip(":")
            if not key:
                continue

            dd = dt.find_next_sibling("dd")
            if not dd:
                continue

            value = normalize_space(dd.get_text(" ", strip=True))
            if value:
                meta[key] = value

        # 容错：有些页面或转换版本可能不是 dt/dd 结构，
        # 则尝试从纯文本行里做弱解析。
        if not meta:
            meta = self._parse_metadata_from_text(soup)

        return meta

    def _parse_metadata_from_text(self, soup: BeautifulSoup) -> dict[str, str]:
        wanted_keys = {"Author", "Status", "Created"}
        meta: dict[str, str] = {}

        text = soup.get_text("\n", strip=True)
        lines = [normalize_space(line) for line in text.splitlines() if normalize_space(line)]

        for idx, line in enumerate(lines):
            key = line.rstrip(":")
            if key not in wanted_keys:
                continue

            if idx + 1 < len(lines):
                meta[key] = normalize_space(lines[idx + 1])

        return meta

7.4 解析方式说明

本文使用的是：

BeautifulSoup + lxml + 正则

具体策略：

目标	方法
详情页链接	扫描全部 <a href>，用正则匹配 /pep-xxxx/
PEP 编号	优先从 h1 标题解析，失败时从 URL 解析
标题	从 h1 中解析 PEP 8 – Style Guide for Python Code
作者	从元信息 Author 抽取
状态	从元信息 Status 抽取
创建日期	从元信息 Created 抽取

7.5 列表页如何拿详情链接

索引页上会出现大量 PEP 链接。我们不需要关心它们位于哪个表格、哪个分类、哪个章节，只要链接满足下面模式即可：

https://peps.python.org/pep-0001/
https://peps.python.org/pep-0008/
https://peps.python.org/pep-0257/

因此代码写成：

for a in soup.find_all("a", href=True):
    href = a.get("href", "")
    absolute_url = urljoin(base_url, href)

    match = PEP_URL_PATTERN.search(absolute_url)
    if not match:
        continue

    pep_number = int(match.group(1))
    links[pep_number] = absolute_url

这样比写：

soup.select("table.docutils tbody tr td a")

更稳一些。后者一旦页面表格结构或 class 名变化，可能就失效。

7.6 详情页如何抽字段

详情页顶部通常有元信息，例如：

Author:
    ...
Status:
    ...
Type:
    ...
Created:
    ...

HTML 中这类信息通常会被渲染为定义列表结构，也就是 dt/dd。解析代码：

for dt in soup.find_all("dt"):
    key = normalize_space(dt.get_text(" ", strip=True)).rstrip(":")
    dd = dt.find_next_sibling("dd")
    value = normalize_space(dd.get_text(" ", strip=True))

解析结果类似：

{
    "Author": "Guido van Rossum, Barry Warsaw, Alyssa Coghlan",
    "Status": "Active",
    "Type": "Process",
    "Created": "05-Jul-2001"
}

7.7 缺失字段怎么办

字段缺失在实战中很常见。本文处理策略：

情况	策略
标题缺失	从 URL 中提取编号，标题置空或使用 fallback
作者缺失	写空字符串
状态缺失	写空字符串
创建日期缺失	写空字符串
页面请求失败	写入 failed_urls.txt
解析失败	捕获异常并记录日志

不要因为一个页面字段缺失就让整个程序中断。结构化采集最重要的是“整体任务可完成，异常数据可追踪”。

---

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

本文使用 SQLite 作为主存储，并导出 CSV。

8.1 字段映射表

字段名	类型	示例值	说明
pep_number	INTEGER	8	PEP 编号
title	TEXT	Style Guide for Python Code	标题
status	TEXT	Active	状态
authors	TEXT	Guido van Rossum, Barry Warsaw	作者
created	TEXT	05-Jul-2001	创建日期
url	TEXT	https://peps.python.org/pep-0008/	详情页
source	TEXT	html	数据来源
crawled_at	TEXT	2026-06-08T10:12:00+00:00	抓取时间

8.2 去重策略

这里使用两层去重。

第一层，索引页链接去重：

links[pep_number] = absolute_url

同一个编号只保留一个 URL。

第二层，数据库唯一约束：

pep_number INTEGER PRIMARY KEY

同一个 PEP 编号重复写入时，使用 upsert 更新：

ON CONFLICT(pep_number) DO UPDATE SET ...

为什么不用 URL 唯一？

因为 PEP 编号更稳定。URL 可能因为域名、尾部斜杠、镜像站等变化产生差异，但 PEP 编号是业务主键。

8.3 storage.py

创建文件：

# pep_crawler/storage.py

from __future__ import annotations

import csv
import sqlite3
from pathlib import Path

from pep_crawler.models import PepRecord


class PepStorage:
    def __init__(self, sqlite_path: Path) -> None:
        self.sqlite_path = sqlite_path
        self.conn = sqlite3.connect(self.sqlite_path)
        self.conn.row_factory = sqlite3.Row
        self.init_db()

    def init_db(self) -> None:
        sql = """
        CREATE TABLE IF NOT EXISTS pep_records (
            pep_number INTEGER PRIMARY KEY,
            title TEXT NOT NULL DEFAULT '',
            status TEXT NOT NULL DEFAULT '',
            authors TEXT NOT NULL DEFAULT '',
            created TEXT NOT NULL DEFAULT '',
            url TEXT NOT NULL DEFAULT '',
            source TEXT NOT NULL DEFAULT '',
            crawled_at TEXT NOT NULL DEFAULT ''
        );
        """
        self.conn.execute(sql)
        self.conn.commit()

    def upsert(self, record: PepRecord) -> None:
        sql = """
        INSERT INTO pep_records (
            pep_number,
            title,
            status,
            authors,
            created,
            url,
            source,
            crawled_at
        )
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(pep_number) DO UPDATE SET
            title = excluded.title,
            status = excluded.status,
            authors = excluded.authors,
            created = excluded.created,
            url = excluded.url,
            source = excluded.source,
            crawled_at = excluded.crawled_at;
        """
        self.conn.execute(
            sql,
            (
                record.pep_number,
                record.title,
                record.status,
                record.authors,
                record.created,
                record.url,
                record.source,
                record.crawled_at,
            ),
        )
        self.conn.commit()

    def bulk_upsert(self, records: list[PepRecord]) -> None:
        for record in records:
            self.upsert(record)

    def count(self) -> int:
        row = self.conn.execute("SELECT COUNT(*) AS total FROM pep_records").fetchone()
        return int(row["total"])

    def fetch_all(self) -> list[dict]:
        rows = self.conn.execute(
            """
            SELECT
                pep_number,
                title,
                status,
                authors,
                created,
                url,
                source,
                crawled_at
            FROM pep_records
            ORDER BY pep_number ASC;
            """
        ).fetchall()

        return [dict(row) for row in rows]

    def export_csv(self, csv_path: Path) -> None:
        rows = self.fetch_all()

        fieldnames = [
            "pep_number",
            "title",
            "status",
            "authors",
            "created",
            "url",
            "source",
            "crawled_at",
        ]

        with csv_path.open("w", newline="", encoding="utf-8-sig") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(rows)

    def close(self) -> None:
        self.conn.close()

8.4 为什么 CSV 使用 utf-8-sig

encoding="utf-8-sig"

主要是为了兼容 Excel。普通 utf-8 在一些本地 Excel 环境中可能显示乱码，utf-8-sig 会多一个 BOM 标记，Excel 通常能正确识别。

如果你只用 Python、Pandas、数据库，不一定需要 utf-8-sig。但面向交付时，我一般会这样写，省得别人打开 CSV 后第一句话就是“乱码了”。

---

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

9.1 crawler.py

创建调度器：

# pep_crawler/crawler.py

from __future__ import annotations

from tqdm import tqdm

from pep_crawler.config import (
    CSV_PATH,
    DATA_DIR,
    FAILED_URLS_PATH,
    INDEX_URL,
    LOG_DIR,
    LOG_PATH,
    REQUEST_INTERVAL,
    SQLITE_PATH,
)
from pep_crawler.fetcher import Fetcher
from pep_crawler.parser import PepParser
from pep_crawler.storage import PepStorage
from pep_crawler.utils import ensure_dirs, setup_logger, write_failed_url


class PepCrawler:
    def __init__(self) -> None:
        ensure_dirs([DATA_DIR, LOG_DIR])

        self.logger = setup_logger(LOG_PATH)
        self.fetcher = Fetcher()
        self.parser = PepParser()
        self.storage = PepStorage(SQLITE_PATH)

    def run(self, limit: int | None = None) -> None:
        self.logger.info("crawler started")
        self.logger.info("fetching index page: %s", INDEX_URL)

        index_result = self.fetcher.get(INDEX_URL)
        if not index_result.ok:
            self.logger.error("failed to fetch index page: %s", index_result.error)
            write_failed_url(FAILED_URLS_PATH, INDEX_URL, index_result.error)
            return

        links = self.parser.parse_index_links(index_result.text, INDEX_URL)
        self.logger.info("found %s pep detail links", len(links))

        if limit is not None and limit > 0:
            links = links[:limit]
            self.logger.info("limit enabled, only crawling first %s links", len(links))

        success_count = 0
        failed_count = 0

        for link in tqdm(links, desc="Crawling PEP detail pages"):
            result = self.fetcher.get(link.url)

            if not result.ok:
                failed_count += 1
                reason = result.error or f"status={result.status_code}"
                self.logger.warning("failed to fetch detail page: %s | %s", link.url, reason)
                write_failed_url(FAILED_URLS_PATH, link.url, reason)
                self.fetcher.polite_sleep(REQUEST_INTERVAL)
                continue

            try:
                record = self.parser.parse_detail(result.text, link.url)
                self.storage.upsert(record)
                success_count += 1
            except Exception as exc:
                failed_count += 1
                reason = f"parse error: {exc}"
                self.logger.exception("failed to parse detail page: %s", link.url)
                write_failed_url(FAILED_URLS_PATH, link.url, reason)

            self.fetcher.polite_sleep(REQUEST_INTERVAL)

        self.storage.export_csv(CSV_PATH)

        total = self.storage.count()
        self.logger.info(
            "crawler finished, success=%s, failed=%s, db_total=%s, csv=%s",
            success_count,
            failed_count,
            total,
            CSV_PATH,
        )

    def close(self) -> None:
        self.storage.close()

9.2 run.py

创建入口文件：

# run.py

from __future__ import annotations

import argparse

from pep_crawler.crawler import PepCrawler


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Crawl Python PEP index and export structured metadata."
    )
    parser.add_argument(
        "--limit",
        type=int,
        default=None,
        help="Only crawl first N PEP detail pages. Useful for testing.",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()

    crawler = PepCrawler()
    try:
        crawler.run(limit=args.limit)
    finally:
        crawler.close()


if __name__ == "__main__":
    main()

9.3 如何启动

先测试抓取前 5 条：

python run.py --limit 5

确认正常后，抓取全部：

python run.py

9.4 输出位置

输出文件：

data/pep_index.sqlite3
data/pep_index.csv
data/failed_urls.txt
logs/crawler.log

其中：

• pep_index.sqlite3 是主数据库。
• pep_index.csv 是导出的结果文件。
• failed_urls.txt 记录失败 URL。
• crawler.log 记录运行日志。

9.5 示例日志

运行时可能看到类似日志：

2026-06-08 10:12:01 | INFO | pep_crawler | crawler started
2026-06-08 10:12:01 | INFO | pep_crawler | fetching index page: https://peps.python.org/
2026-06-08 10:12:02 | INFO | pep_crawler | found 600 pep detail links
Crawling PEP detail pages: 100%|██████████| 5/5 [00:04<00:00,  1.20it/s]
2026-06-08 10:12:07 | INFO | pep_crawler | crawler finished, success=5, failed=0, db_total=5, csv=data/pep_index.csv

实际链接数量会随 PEP 文档更新而变化，以运行时为准。

9.6 展示 3–5 行示例结果

CSV 示例结果大致如下：

pep_number	title	status	authors	created	url
1	PEP Purpose and Guidelines	Active	Barry Warsaw, Jeremy Hylton, David Goodger, Alyssa Coghlan	13-Jun-2000	https://peps.python.org/pep-0001/
2	Procedure for Adding New Modules	Active	Brett Cannon, Martijn Faassen	07-Jul-2001	https://peps.python.org/pep-0002/
3	Guidelines for Handling Bug Reports	Withdrawn	Jeremy Hylton	25-Sep-2000	https://peps.python.org/pep-0003/
4	Deprecation of Standard Modules	Active	Brett Cannon, Martin von Löwis	01-Oct-2000	https://peps.python.org/pep-0004/
8	Style Guide for Python Code	Active	Guido van Rossum, Barry Warsaw, Alyssa Coghlan	05-Jul-2001	https://peps.python.org/pep-0008/

注意：示例结果用于说明字段形态，真实结果以你运行时页面内容为准。

9.7 用 SQLite 查看结果

可以使用 sqlite3 命令行：

sqlite3 data/pep_index.sqlite3

进入后执行：

.headers on
.mode column

SELECT pep_number, title, status, created
FROM pep_records
ORDER BY pep_number
LIMIT 10;

统计不同状态数量：

SELECT status, COUNT(*) AS total
FROM pep_records
GROUP BY status
ORDER BY total DESC;

统计每年创建数量：

SELECT substr(created, -4) AS year, COUNT(*) AS total
FROM pep_records
WHERE created != ''
GROUP BY year
ORDER BY year;

SQLite 对这类小规模数据分析非常够用。后面如果要接 BI、Web 后台或定时任务，再迁移到 MySQL/PostgreSQL 也不迟。

---

🔟 常见问题与排错

10.1 403 怎么办

403 表示服务器拒绝访问。遇到 403，不建议第一反应就是“换代理”。更合理的排查顺序是：

第一，检查是否访问了错误路径。

有些页面可以正常浏览，但某些资源路径不允许爬虫访问。确认目标 URL 是否真的是公开页面。

第二，检查 robots。

如果 robots 不允许访问，就不要继续抓。

第三，检查 User-Agent。

空 UA 或非常异常的 UA 可能被一些服务器拒绝。可以设置一个清晰、克制的 UA。

第四，降低访问频率。

短时间请求过快，也可能触发限制。

第五，停止任务并观察。

如果持续 403，说明站点不希望你这样访问。尊重限制，比硬绕过去更重要。

10.2 429 怎么办

429 表示请求过多。处理建议：

• 增大 REQUEST_INTERVAL。
• 降低并发或改为串行。
• 增加指数退避。
• 减少重复请求。
• 使用本地缓存。
• 等待一段时间后再运行。

本文请求层已经对 429 加入重试：

status_forcelist=(429, 500, 502, 503, 504)

但重试不是万能药。429 的根本解决办法通常是降频。

可以把配置改成：

REQUEST_INTERVAL = 2.0
MAX_RETRIES = 2
BACKOFF_FACTOR = 3.0

这样会更温和。

10.3 HTML 抓到空壳怎么办

如果你打印 HTML 后发现只有一个根节点、几个 script、没有正文内容，通常说明页面是动态渲染的。

处理思路：

第一，打开浏览器开发者工具，查看 Network 面板。

看看页面是否请求了 JSON 接口。

第二，优先抓接口。

如果有公开接口，而且接口返回数据结构清晰，优先使用接口。

第三，必要时再考虑 Playwright。

如果数据必须浏览器执行 JS 后才出现，可以使用 Playwright。

但 PEP 页面属于静态文档，不需要 Playwright。用浏览器自动化反而更重。

10.4 解析报错怎么办

常见报错：

AttributeError: 'NoneType' object has no attribute 'get_text'

通常是选择器没选到元素。处理方式：

第一，不要直接链式调用。

不推荐：

title = soup.find("h1").get_text(strip=True)

推荐：

h1 = soup.find("h1")
title = h1.get_text(strip=True) if h1 else ""

第二，保留失败页面。

解析失败时，可以把 HTML 保存下来：

debug_path = DATA_DIR / "debug_failed.html"
debug_path.write_text(result.text, encoding="utf-8")

然后手动打开分析结构。

第三，选择器不要太脆弱。

不要依赖“第几个 div”。尽量依赖稳定语义，比如标题、链接模式、字段名。

10.5 编码或乱码如何处理

本文请求层中有：

if not response.encoding:
    response.encoding = response.apparent_encoding or "utf-8"

CSV 导出使用：

encoding="utf-8-sig"

如果你看到乱码，可以检查：

• response.encoding。
• HTML 中的 <meta charset>。
• CSV 打开软件是否识别 UTF-8。
• 终端编码。
• 数据库客户端编码。

PEP 页面主要是英文，但作者名中可能包含非 ASCII 字符，例如带重音符号的名字，所以不要用 gbk 之类的编码保存。

10.6 链接数量不对怎么办

如果提取出来的详情页链接明显太少，检查正则：

PEP_URL_PATTERN = re.compile(r"/pep-(\d{4})/?$")

如果站点 URL 形式变化，比如多了锚点：

/pep-0008/#introduction

那么当前正则不会匹配。可以改成：

PEP_URL_PATTERN = re.compile(r"/pep-(\d{4})/?(?:#.*)?$")

不过对本文目标站点，详情页链接一般不需要锚点。

10.7 日期要不要转成标准格式

PEP 页面日期常见格式类似：

05-Jul-2001
13-Jun-2000

为了保持原始信息，本文先按字符串存储。如果你后续要做时间分析，可以额外转换成 ISO 日期。

示例函数：

from datetime import datetime

def parse_pep_date(value: str) -> str:
    if not value:
        return ""

    try:
        dt = datetime.strptime(value, "%d-%b-%Y")
        return dt.date().isoformat()
    except ValueError:
        return value

然后把 05-Jul-2001 转成：

2001-07-05

不过要注意，有些字段可能包含多个日期或格式不一致，转换失败时不要硬报错。

---

1️⃣1️⃣ 进阶优化

11.1 使用官方 API 做校验

虽然本文主线是 HTML 二段式抓取，但 PEP 站点提供 JSON 元数据接口。我们可以写一个简单校验器，把 HTML 抓取结果和 API 返回结果对比。

创建 pep_crawler/api_checker.py：

# pep_crawler/api_checker.py

from __future__ import annotations

import requests

PEPS_API_URL = "https://peps.python.org/api/peps.json"


def fetch_api_records() -> dict[int, dict]:
    response = requests.get(PEPS_API_URL, timeout=15)
    response.raise_for_status()
    data = response.json()

    return {int(number): item for number, item in data.items()}


def compare_one(html_record: dict, api_record: dict) -> dict[str, tuple[str, str]]:
    diff: dict[str, tuple[str, str]] = {}

    mapping = {
        "title": "title",
        "status": "status",
        "authors": "authors",
        "created": "created",
    }

    for html_key, api_key in mapping.items():
        html_value = str(html_record.get(html_key, "")).strip()
        api_value = str(api_record.get(api_key, "")).strip()

        if html_value != api_value:
            diff[html_key] = (html_value, api_value)

    return diff

这个校验器的意义不是替代爬虫，而是告诉我们：当目标站点提供权威结构化数据时，应该善用它。HTML 抓取适合练解析能力，也适合没有 API 的站点；但在生产中，官方 API 往往更稳定。

11.2 加本地缓存

如果你经常调试解析器，不应该每次都重新请求网站。可以加入简单缓存：

# pep_crawler/cache.py

from __future__ import annotations

import hashlib
from pathlib import Path


class HtmlCache:
    def __init__(self, cache_dir: Path) -> None:
        self.cache_dir = cache_dir
        self.cache_dir.mkdir(parents=True, exist_ok=True)

    def _key(self, url: str) -> str:
        return hashlib.sha256(url.encode("utf-8")).hexdigest()

    def get_path(self, url: str) -> Path:
        return self.cache_dir / f"{self._key(url)}.html"

    def exists(self, url: str) -> bool:
        return self.get_path(url).exists()

    def read(self, url: str) -> str:
        return self.get_path(url).read_text(encoding="utf-8")

    def write(self, url: str, html: str) -> None:
        self.get_path(url).write_text(html, encoding="utf-8")

然后在请求前先查缓存：

if cache.exists(url):
    html = cache.read(url)
else:
    result = fetcher.get(url)
    html = result.text
    cache.write(url, html)

这对调试非常有帮助，也能减少重复请求。

11.3 并发优化

这个案例默认串行，已经足够。但如果你确实需要并发，可以使用线程池。注意，并发不是越大越好。对技术文档站，我通常会设置很低的并发，比如 3 或 5。

示例：

from concurrent.futures import ThreadPoolExecutor, as_completed

def crawl_one(link):
    result = fetcher.get(link.url)
    if not result.ok:
        return None
    return parser.parse_detail(result.text, link.url)

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(crawl_one, link) for link in links]

    for future in as_completed(futures):
        record = future.result()
        if record:
            storage.upsert(record)

但要注意：并发后，频率控制会更复杂。如果每个线程都没有 sleep，请求会瞬间变密。更稳妥的办法是使用队列和限速器，或者直接用 Scrapy 的下载延迟配置。

11.4 asyncio 版本思路

如果换成异步，可以使用：

aiohttp + asyncio + async_timeout

适合大量 IO 请求。但对 PEP 这种规模，异步不是必须。异步代码调试成本更高，新手容易把错误处理写散。

我的建议：

• 页面数量小：requests 串行。
• 页面数量中等：requests + 小线程池。
• 页面数量大：Scrapy。
• 页面动态渲染：Playwright。
• 页面有官方 API：优先 API。

11.5 断点续跑

断点续跑有两种简单做法。

第一，数据库查重。

如果 pep_number 已存在，就跳过：

def exists(self, pep_number: int) -> bool:
    row = self.conn.execute(
        "SELECT 1 FROM pep_records WHERE pep_number = ? LIMIT 1",
        (pep_number,),
    ).fetchone()
    return row is not None

调度时：

if storage.exists(link.pep_number):
    logger.info("skip existing pep: %s", link.pep_number)
    continue

第二，失败任务重跑。

failed_urls.txt 中记录失败 URL，后续可以只读取失败 URL 再跑一次。

示例：

from pathlib import Path

def load_failed_urls(path: Path) -> list[str]:
    if not path.exists():
        return []

    urls = []
    for line in path.read_text(encoding="utf-8").splitlines():
        parts = line.split("\t")
        if len(parts) >= 2:
            urls.append(parts[1])

    return sorted(set(urls))

11.6 日志与监控

小爬虫至少记录：

• 启动时间。
• 目标 URL。
• 详情页总数。
• 成功数量。
• 失败数量。
• 输出路径。
• 异常堆栈。

如果是生产任务，还可以记录：

• 平均响应时间。
• 不同状态码数量。
• 每分钟请求数。
• 解析成功率。
• 字段缺失率。
• 重试次数。

这些指标能帮你判断问题出在哪里。比如失败率突然升高，可能是网络波动，也可能是页面结构变了。

11.7 定时任务

Linux 可以使用 cron：

crontab -e

每周一凌晨 2 点运行：

0 2 * * 1 cd /path/to/pep_index_crawler && /path/to/pep_index_crawler/.venv/bin/python run.py >> logs/cron.log 2>&1

更复杂的场景可以用 Airflow、Prefect、Dagster。这个案例没必要一开始就上调度平台，cron 已经够用。

11.8 用 Pandas 做简单分析

采集完成后，可以用 Pandas 读 CSV：

import pandas as pd

df = pd.read_csv("data/pep_index.csv")

print(df.head())
print(df["status"].value_counts())

按创建年份统计：

df["year"] = df["created"].str.extract(r"(\d{4})$")
print(df["year"].value_counts().sort_index())

按作者粗略统计：

from collections import Counter

counter = Counter()

for authors in df["authors"].dropna():
    for author in str(authors).split(","):
        name = author.strip()
        if name:
            counter[name] += 1

print(counter.most_common(20))

注意作者字段有时包含复杂格式，简单逗号切分不一定完美，但做初步分析够用。更严谨的作者拆分要结合官方 API 的 author_names 字段。

---

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

本文完成了一个完整的 Python PEP 索引二段式爬虫：

索引页采集 -> 详情页链接提取 -> 详情页请求 -> 元信息解析 -> 数据清洗 -> SQLite 存储 -> CSV 导出

我们不仅写了能跑的代码，还补齐了实际项目中经常被忽略的部分：

• 请求头。
• timeout。
• Session。
• 重试。
• 退避。
• 频率控制。
• 字段缺失容错。
• 日志。
• 失败 URL 记录。
• SQLite 去重。
• CSV 导出。
• 断点续跑思路。
• API 校验思路。

这个案例不复杂，但它很“正”。很多爬虫问题并不是选择器有多难，而是工程细节做得不够：没有超时，没有日志，没有去重，没有失败记录，没有缓存，最后脚本看似能跑，稍微出点问题就要从头猜。把这些基础补齐，后面再去写更复杂的站点，会轻松很多。

下一步你可以继续做几件事：

第一，把 HTML 抓取结果和官方 API 结果做字段对比。

这样可以发现自己的解析器是否遗漏字段，也可以理解 API 与页面之间的差异。

第二，把存储层换成 MySQL 或 PostgreSQL。

适合多人使用、长期定时更新、接入后台系统的场景。

第三，用 Scrapy 重写。

Scrapy 的请求调度、下载延迟、失败重试、管道存储更完整。这个案例非常适合作为 Scrapy 入门练习。

第四，加 Playwright 版本。

虽然 PEP 页面不需要动态渲染，但你可以用同样的架构去处理一个动态技术文档站，对比 requests 和浏览器自动化的区别。

第五，把数据做成一个小型检索页面。

例如用 FastAPI + SQLite 做一个 PEP 查询接口：

GET /peps?status=Final
GET /peps/8
GET /stats/status

这样，一个简单爬虫就变成了一个小型数据产品。

我写完这个案例最大的感受是：技术文档类爬虫不应该追求刺激，应该追求稳定。慢一点、干净一点、留痕多一点，反而更接近真实工作里的可维护代码。爬虫不是只要“抓到”就结束，能反复抓、能解释、能修、能交付，才算真正完成。

---

附录：完整代码汇总

为了方便复制，下面把完整代码按文件再汇总一次。

requirements.txt

requests>=2.31.0
beautifulsoup4>=4.12.0
lxml>=5.0.0
tqdm>=4.66.0

pep_crawler/config.py

from pathlib import Path

BASE_URL = "https://peps.python.org/"
INDEX_URL = "https://peps.python.org/"

USER_AGENT = (
    "Mozilla/5.0 (compatible; PepIndexCrawler/1.0; "
    "+https://example.com/bot-info)"
)

DEFAULT_HEADERS = {
    "User-Agent": USER_AGENT,
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "en-US,en;q=0.9",
    "Referer": BASE_URL,
    "Connection": "keep-alive",
}

REQUEST_TIMEOUT = 15
REQUEST_INTERVAL = 0.5

MAX_RETRIES = 3
BACKOFF_FACTOR = 1.5

PROJECT_ROOT = Path(__file__).resolve().parent.parent
DATA_DIR = PROJECT_ROOT / "data"
LOG_DIR = PROJECT_ROOT / "logs"

SQLITE_PATH = DATA_DIR / "pep_index.sqlite3"
CSV_PATH = DATA_DIR / "pep_index.csv"
FAILED_URLS_PATH = DATA_DIR / "failed_urls.txt"
LOG_PATH = LOG_DIR / "crawler.log"

pep_crawler/models.py

from __future__ import annotations

from dataclasses import dataclass, asdict


@dataclass
class PepLink:
    pep_number: int
    url: str


@dataclass
class PepRecord:
    pep_number: int
    title: str
    status: str
    authors: str
    created: str
    url: str
    source: str
    crawled_at: str

    def to_dict(self) -> dict:
        return asdict(self)

pep_crawler/utils.py

import logging
import re
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable


def ensure_dirs(paths: Iterable[Path]) -> None:
    for path in paths:
        path.mkdir(parents=True, exist_ok=True)


def setup_logger(log_path: Path) -> logging.Logger:
    logger = logging.getLogger("pep_crawler")
    logger.setLevel(logging.INFO)

    if logger.handlers:
        return logger

    formatter = logging.Formatter(
        fmt="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )

    stream_handler = logging.StreamHandler(sys.stdout)
    stream_handler.setFormatter(formatter)
    stream_handler.setLevel(logging.INFO)

    file_handler = logging.FileHandler(log_path, encoding="utf-8")
    file_handler.setFormatter(formatter)
    file_handler.setLevel(logging.INFO)

    logger.addHandler(stream_handler)
    logger.addHandler(file_handler)

    return logger


def normalize_space(text: str | None) -> str:
    if not text:
        return ""
    return re.sub(r"\s+", " ", text).strip()


def utc_now_iso() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()


def write_failed_url(path: Path, url: str, reason: str) -> None:
    line = f"{utc_now_iso()}\t{url}\t{reason}\n"
    with path.open("a", encoding="utf-8") as f:
        f.write(line)

pep_crawler/fetcher.py

from __future__ import annotations

import time
from dataclasses import dataclass

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

from pep_crawler.config import (
    DEFAULT_HEADERS,
    MAX_RETRIES,
    BACKOFF_FACTOR,
    REQUEST_TIMEOUT,
)


@dataclass
class FetchResult:
    url: str
    ok: bool
    status_code: int | None
    text: str
    error: str = ""


class Fetcher:
    def __init__(
        self,
        headers: dict[str, str] | None = None,
        timeout: int = REQUEST_TIMEOUT,
        max_retries: int = MAX_RETRIES,
        backoff_factor: float = BACKOFF_FACTOR,
    ) -> None:
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update(headers or DEFAULT_HEADERS)

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

        adapter = HTTPAdapter(max_retries=retry)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)

    def get(self, url: str) -> FetchResult:
        try:
            response = self.session.get(url, timeout=self.timeout)
            self._fix_encoding(response)

            if 200 <= response.status_code < 300:
                return FetchResult(
                    url=url,
                    ok=True,
                    status_code=response.status_code,
                    text=response.text,
                )

            return FetchResult(
                url=url,
                ok=False,
                status_code=response.status_code,
                text=response.text or "",
                error=f"unexpected status code: {response.status_code}",
            )

        except requests.Timeout as exc:
            return FetchResult(
                url=url,
                ok=False,
                status_code=None,
                text="",
                error=f"timeout: {exc}",
            )
        except requests.RequestException as exc:
            return FetchResult(
                url=url,
                ok=False,
                status_code=None,
                text="",
                error=f"request error: {exc}",
            )

    @staticmethod
    def _fix_encoding(response: Response) -> None:
        if not response.encoding:
            response.encoding = response.apparent_encoding or "utf-8"

    @staticmethod
    def polite_sleep(seconds: float) -> None:
        if seconds > 0:
            time.sleep(seconds)

pep_crawler/parser.py

from __future__ import annotations

import re
from urllib.parse import urljoin

from bs4 import BeautifulSoup

from pep_crawler.config import BASE_URL
from pep_crawler.models import PepLink, PepRecord
from pep_crawler.utils import normalize_space, utc_now_iso


PEP_URL_PATTERN = re.compile(r"/pep-(\d{4})/?$")
TITLE_PATTERN = re.compile(r"^PEP\s+(\d+)\s+[–-]\s+(.+)$")


class PepParser:
    def parse_index_links(self, html: str, base_url: str = BASE_URL) -> list[PepLink]:
        soup = BeautifulSoup(html, "lxml")

        links: dict[int, str] = {}

        for a in soup.find_all("a", href=True):
            href = a.get("href", "")
            absolute_url = urljoin(base_url, href)

            match = PEP_URL_PATTERN.search(absolute_url)
            if not match:
                continue

            pep_number = int(match.group(1))
            links[pep_number] = absolute_url

        return [
            PepLink(pep_number=number, url=url)
            for number, url in sorted(links.items(), key=lambda item: item[0])
        ]

    def parse_detail(self, html: str, url: str) -> PepRecord:
        soup = BeautifulSoup(html, "lxml")

        pep_number, title = self._parse_title(soup, url)
        meta = self._parse_metadata(soup)

        return PepRecord(
            pep_number=pep_number,
            title=title,
            status=meta.get("Status", ""),
            authors=meta.get("Author", ""),
            created=meta.get("Created", ""),
            url=url,
            source="html",
            crawled_at=utc_now_iso(),
        )

    def _parse_title(self, soup: BeautifulSoup, url: str) -> tuple[int, str]:
        h1 = soup.find("h1")
        raw_title = normalize_space(h1.get_text(" ", strip=True) if h1 else "")

        match = TITLE_PATTERN.match(raw_title)
        if match:
            return int(match.group(1)), normalize_space(match.group(2))

        url_match = PEP_URL_PATTERN.search(url)
        if url_match:
            pep_number = int(url_match.group(1))
        else:
            pep_number = -1

        fallback_title = raw_title.replace("Python Enhancement Proposals", "").strip()
        return pep_number, fallback_title

    def _parse_metadata(self, soup: BeautifulSoup) -> dict[str, str]:
        meta: dict[str, str] = {}

        for dt in soup.find_all("dt"):
            key = normalize_space(dt.get_text(" ", strip=True)).rstrip(":")
            if not key:
                continue

            dd = dt.find_next_sibling("dd")
            if not dd:
                continue

            value = normalize_space(dd.get_text(" ", strip=True))
            if value:
                meta[key] = value

        if not meta:
            meta = self._parse_metadata_from_text(soup)

        return meta

    def _parse_metadata_from_text(self, soup: BeautifulSoup) -> dict[str, str]:
        wanted_keys = {"Author", "Status", "Created"}
        meta: dict[str, str] = {}

        text = soup.get_text("\n", strip=True)
        lines = [normalize_space(line) for line in text.splitlines() if normalize_space(line)]

        for idx, line in enumerate(lines):
            key = line.rstrip(":")
            if key not in wanted_keys:
                continue

            if idx + 1 < len(lines):
                meta[key] = normalize_space(lines[idx + 1])

        return meta

pep_crawler/storage.py

from __future__ import annotations

import csv
import sqlite3
from pathlib import Path

from pep_crawler.models import PepRecord


class PepStorage:
    def __init__(self, sqlite_path: Path) -> None:
        self.sqlite_path = sqlite_path
        self.conn = sqlite3.connect(self.sqlite_path)
        self.conn.row_factory = sqlite3.Row
        self.init_db()

    def init_db(self) -> None:
        sql = """
        CREATE TABLE IF NOT EXISTS pep_records (
            pep_number INTEGER PRIMARY KEY,
            title TEXT NOT NULL DEFAULT '',
            status TEXT NOT NULL DEFAULT '',
            authors TEXT NOT NULL DEFAULT '',
            created TEXT NOT NULL DEFAULT '',
            url TEXT NOT NULL DEFAULT '',
            source TEXT NOT NULL DEFAULT '',
            crawled_at TEXT NOT NULL DEFAULT ''
        );
        """
        self.conn.execute(sql)
        self.conn.commit()

    def upsert(self, record: PepRecord) -> None:
        sql = """
        INSERT INTO pep_records (
            pep_number,
            title,
            status,
            authors,
            created,
            url,
            source,
            crawled_at
        )
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(pep_number) DO UPDATE SET
            title = excluded.title,
            status = excluded.status,
            authors = excluded.authors,
            created = excluded.created,
            url = excluded.url,
            source = excluded.source,
            crawled_at = excluded.crawled_at;
        """
        self.conn.execute(
            sql,
            (
                record.pep_number,
                record.title,
                record.status,
                record.authors,
                record.created,
                record.url,
                record.source,
                record.crawled_at,
            ),
        )
        self.conn.commit()

    def count(self) -> int:
        row = self.conn.execute("SELECT COUNT(*) AS total FROM pep_records").fetchone()
        return int(row["total"])

    def fetch_all(self) -> list[dict]:
        rows = self.conn.execute(
            """
            SELECT
                pep_number,
                title,
                status,
                authors,
                created,
                url,
                source,
                crawled_at
            FROM pep_records
            ORDER BY pep_number ASC;
            """
        ).fetchall()

        return [dict(row) for row in rows]

    def export_csv(self, csv_path: Path) -> None:
        rows = self.fetch_all()

        fieldnames = [
            "pep_number",
            "title",
            "status",
            "authors",
            "created",
            "url",
            "source",
            "crawled_at",
        ]

        with csv_path.open("w", newline="", encoding="utf-8-sig") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(rows)

    def close(self) -> None:
        self.conn.close()

pep_crawler/crawler.py

from __future__ import annotations

from tqdm import tqdm

from pep_crawler.config import (
    CSV_PATH,
    DATA_DIR,
    FAILED_URLS_PATH,
    INDEX_URL,
    LOG_DIR,
    LOG_PATH,
    REQUEST_INTERVAL,
    SQLITE_PATH,
)
from pep_crawler.fetcher import Fetcher
from pep_crawler.parser import PepParser
from pep_crawler.storage import PepStorage
from pep_crawler.utils import ensure_dirs, setup_logger, write_failed_url


class PepCrawler:
    def __init__(self) -> None:
        ensure_dirs([DATA_DIR, LOG_DIR])

        self.logger = setup_logger(LOG_PATH)
        self.fetcher = Fetcher()
        self.parser = PepParser()
        self.storage = PepStorage(SQLITE_PATH)

    def run(self, limit: int | None = None) -> None:
        self.logger.info("crawler started")
        self.logger.info("fetching index page: %s", INDEX_URL)

        index_result = self.fetcher.get(INDEX_URL)
        if not index_result.ok:
            self.logger.error("failed to fetch index page: %s", index_result.error)
            write_failed_url(FAILED_URLS_PATH, INDEX_URL, index_result.error)
            return

        links = self.parser.parse_index_links(index_result.text, INDEX_URL)
        self.logger.info("found %s pep detail links", len(links))

        if limit is not None and limit > 0:
            links = links[:limit]
            self.logger.info("limit enabled, only crawling first %s links", len(links))

        success_count = 0
        failed_count = 0

        for link in tqdm(links, desc="Crawling PEP detail pages"):
            result = self.fetcher.get(link.url)

            if not result.ok:
                failed_count += 1
                reason = result.error or f"status={result.status_code}"
                self.logger.warning("failed to fetch detail page: %s | %s", link.url, reason)
                write_failed_url(FAILED_URLS_PATH, link.url, reason)
                self.fetcher.polite_sleep(REQUEST_INTERVAL)
                continue

            try:
                record = self.parser.parse_detail(result.text, link.url)
                self.storage.upsert(record)
                success_count += 1
            except Exception as exc:
                failed_count += 1
                reason = f"parse error: {exc}"
                self.logger.exception("failed to parse detail page: %s", link.url)
                write_failed_url(FAILED_URLS_PATH, link.url, reason)

            self.fetcher.polite_sleep(REQUEST_INTERVAL)

        self.storage.export_csv(CSV_PATH)

        total = self.storage.count()
        self.logger.info(
            "crawler finished, success=%s, failed=%s, db_total=%s, csv=%s",
            success_count,
            failed_count,
            total,
            CSV_PATH,
        )

    def close(self) -> None:
        self.storage.close()

run.py

from __future__ import annotations

import argparse

from pep_crawler.crawler import PepCrawler


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Crawl Python PEP index and export structured metadata."
    )
    parser.add_argument(
        "--limit",
        type=int,
        default=None,
        help="Only crawl first N PEP detail pages. Useful for testing.",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()

    crawler = PepCrawler()
    try:
        crawler.run(limit=args.limit)
    finally:
        crawler.close()


if __name__ == "__main__":
    main()

运行命令

python run.py --limit 5

确认没问题后：

python run.py

最终输出：

data/pep_index.sqlite3
data/pep_index.csv
data/failed_urls.txt
logs/crawler.log

🌟 文末

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

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

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

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

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

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

✅ 互动征集

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

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

---

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

---

✅ 免责声明

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

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

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