应急预案目录分页爬虫实战：用 Python 完成目录页与详情页二段式采集

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

全文目录：

• • 🌟 开篇语
  • 0️⃣ 前言（Preface）
  • 1️⃣ 摘要（Abstract）
  • 2️⃣ 背景与需求（Why）
  • • 2.1 为什么要爬应急预案目录
    • 2.2 目标站点类型
    • 2.3 目标字段清单
  • 3️⃣ 合规与注意事项（必写）
  • • 3.1 robots.txt 基本说明
    • 3.2 频率控制，不要攻击式并发
    • 3.3 不采集敏感信息，不绕过限制
    • 3.4 记录来源，保留链接
  • 4️⃣ 技术选型与整体流程（What / How）
  • • 4.1 静态、动态、API：这篇属于哪种
    • 4.2 整体流程
    • 4.3 为什么选 requests + bs4 + lxml
  • 5️⃣ 环境准备与依赖安装（可复现）
  • • 5.1 Python 版本
    • 5.2 创建虚拟环境
    • 5.3 安装依赖
    • 5.4 推荐项目结构
  • 6️⃣ 核心实现：请求层（Fetcher）
  • • 6.1 headers
    • 6.2 timeout
    • 6.3 session / cookie
    • 6.4 失败处理：重试与退避
    • 6.5 请求层代码
  • 7️⃣ 核心实现：解析层（Parser）
  • • 7.1 解析方式选择
    • 7.2 列表页如何拿详情链接
    • 7.3 详情页如何抽字段
    • 7.4 缺失字段怎么办
    • 7.5 工具函数
    • 7.6 数据模型
    • 7.7 Parser 完整代码
  • 8️⃣ 数据存储与导出（Storage）
  • • 8.1 起步选择 CSV
    • 8.2 字段映射表
    • 8.3 去重策略
    • 8.4 存储层代码
  • 9️⃣ 运行方式与结果展示（必写）
  • • 9.1 配置文件
    • 9.2 本地 fixture 页面
    • 9.3 入口文件
    • 9.4 启动命令
    • 9.5 输出路径
    • 9.6 示例结果
  • 🔟 常见问题与排错
  • • 10.1 403 怎么办
    • 10.2 429 怎么办
    • 10.3 HTML 抓到空壳怎么办
    • 10.4 解析报错怎么办
    • 10.5 编码或乱码如何处理
    • 10.6 日期格式不统一怎么办
    • 10.7 下一页抓不到怎么办
  • 1️⃣1️⃣ 进阶优化（可选但加分）
  • • 11.1 并发优化
    • 11.2 断点续跑
    • 11.3 SQLite 存储版本
    • 11.4 日志与监控
    • 11.5 定时任务
    • 11.6 失败 URL 单独落盘
    • 11.7 内容变更检测
  • 1️⃣2️⃣ 总结与延伸阅读
  • 🌟 文末
  • • ✅ 专栏持续更新中｜建议收藏 + 订阅
    • ✅ 互动征集
    • ✅ 免责声明

🌟 开篇语

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

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

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

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

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

0️⃣ 前言（Preface）

这篇文章要做的事情很明确：使用 Python、requests、BeautifulSoup 与 lxml，抓取一个“应急预案公开目录”的分页列表，再进入详情页提取“预案名、适用领域、发布单位、发布日期、状态、链接”等字段，最终导出为 CSV 文件。

读完这篇文章，你至少能获得三件东西：

我个人比较喜欢把这类爬虫称为“信息整理型爬虫”。它不追求什么炫技，也不需要高并发压测，更不应该把目标站点当成压力测试对象。它的重点是稳定、克制、可维护：能把公开目录里的信息干净地采回来，能记录失败，能重复执行，能在页面结构有小变化时尽量不崩。

1️⃣ 摘要（Abstract）

本文围绕“应急预案目录分页”这一公开信息聚合场景，使用 Python 的 requests、BeautifulSoup、lxml、csv、sqlite 可选方案，实现从目录页分页采集到详情页字段抽取，再到 CSV 导出的完整流程。

读完后你能掌握：

本文示例采用“静态 HTML 页面优先”的技术路线。如果目标站点返回的是完整 HTML，用 requests + bs4 就可以完成；如果列表页只有空壳，数据由 JavaScript 渲染，则需要进一步检查接口或使用 Playwright。BeautifulSoup 官方文档支持使用 select() 按 CSS 选择器查找元素，requests 文档也说明可以通过 headers 参数设置请求头；同时，requests 高级文档提醒，如果不显式设置 timeout，请求可能长时间挂起，所以本文会把 timeout 当成请求层的必备项。([Crummy][1])

2️⃣ 背景与需求（Why）

2.1 为什么要爬应急预案目录

应急预案类公开目录通常具有几个共同特点：

第一，信息分散。 很多预案信息散落在不同栏目、不同年份、不同分页中。人工复制不仅慢，而且容易漏掉。

第二，字段相对固定。 这类公开目录通常会包含标题、发布时间、发布单位、分类、状态、详情链接等字段。字段规整，适合结构化处理。

第三，适合后续分析。 拿到结构化数据之后，可以做很多事情，例如：

• 按年份统计预案发布数量。
• 按适用领域聚合，例如自然灾害、公共卫生、生产安全、交通运输等。
• 检查哪些预案状态为现行有效、废止、修订中。
• 建立内部知识库索引，方便搜索和归档。
• 定期巡检公开页面是否新增内容。

这里要强调一点：本文只讨论公开网页上的普通信息整理，不讨论绕过登录、不讨论付费墙、不讨论敏感信息采集，也不鼓励对任何站点进行高频访问。

2.2 目标站点类型

本文假设目标页面是一个公开目录页面，结构大致如下：

目录页第 1 页
├── 预案 A 标题 + 日期 + 详情链接
├── 预案 B 标题 + 日期 + 详情链接
├── 预案 C 标题 + 日期 + 详情链接
└── 下一页链接

目录页第 2 页
├── 预案 D 标题 + 日期 + 详情链接
├── 预案 E 标题 + 日期 + 详情链接
└── 下一页链接

详情页
├── 预案名
├── 适用领域
├── 发布单位
├── 发布日期
├── 状态
└── 正文内容

2.3 目标字段清单

本项目采集字段如下：

我建议字段不要一开始就设计得过多。先把核心字段跑通，再逐步增加正文、附件链接、来源栏目、抓取时间、内容 hash 等辅助字段。字段过多会让第一版爬虫变得脆弱，尤其在目标站点页面结构不统一的时候。

3️⃣ 合规与注意事项（必写）

爬虫不是“能抓就抓”。做公开数据采集时，我习惯先看三件事：robots、访问频率、采集边界。

3.1 robots.txt 基本说明

robots.txt 是站点给自动化访问程序提供的访问规则文件。Python 标准库提供了 urllib.robotparser.RobotFileParser，可以用来判断某个 user-agent 是否允许抓取指定 URL。Python 官方文档说明，RobotFileParser 可以回答某个 user agent 是否可以抓取网站发布的 robots.txt 所约束的 URL。([Python documentation][2])

需要注意的是，robots 不是唯一判断依据。即使 robots 没有限制，也应该保持低频、克制和必要性原则。反过来，如果 robots 明确不允许访问某些路径，就应该停止采集对应路径。

本文代码会提供 robots 检查能力。为了保证本地演示可运行，默认 fixture 模式不检查 robots；当切换到真实 HTTP 模式时，会加载目标站点 robots。

3.2 频率控制，不要攻击式并发

应急预案目录这类页面通常更新频率并不高，没有必要高并发抓取。更稳妥的做法是：

• 每次请求之间 sleep 0.8 到 2 秒。
• 失败后做指数退避，不要立刻重试。
• 只抓必要页面，不做全站扫描。
• 控制最大页数，避免分页规则错误导致无限抓取。
• 不在短时间内重复抓同一个详情页。

3.3 不采集敏感信息，不绕过限制

本文只处理公开目录中的一般字段，不采集个人隐私、账号信息、联系方式明细、内部文件，也不绕过登录、验证码、付费、权限控制等限制。

如果目标页面要求登录才能访问，说明它不是普通公开目录，应该停止采集，或通过站点提供的正式接口、授权渠道获取数据。

3.4 记录来源，保留链接

每一条数据都保留详情页 URL。这样做有两个好处：

第一，方便回溯。 当字段抽取结果有疑问时，可以回到原页面核对。

第二，方便去重。 多数目录场景下，详情页 URL 是天然唯一键。后续写入 CSV、SQLite、MySQL 时，都可以用 URL 做主键或唯一索引。

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

4.1 静态、动态、API：这篇属于哪种

网页采集通常先判断三种情况：

第一种，静态 HTML。 浏览器右键查看网页源代码，能看到列表项、标题、详情链接。此时 requests + BeautifulSoup 通常足够。

第二种，动态渲染。 网页源代码里没有数据，打开开发者工具 Elements 才能看到列表。这说明数据可能由 JavaScript 后加载。此时可以进一步分析 Network 请求，寻找 JSON 接口；如果接口复杂，再考虑 Playwright。

第三种，公开 API。 页面数据来自某个 JSON 接口，且接口不需要特殊授权。此时直接请求接口比解析 HTML 更稳定。

本文默认目标是第一种：静态 HTML 目录页。解析层用 BeautifulSoup 的 CSS 选择器，必要时可以换成 lxml XPath。lxml 文档说明它支持 XPath 1.0；如果页面结构比较规整，XPath 在定位表格、兄弟节点、父子关系时会很有优势。([lxml][3])

4.2 整体流程

流程可以概括为四个词：

采集 → 解析 → 清洗 → 存储

展开之后是：

读取配置
↓
检查 robots
↓
请求目录页
↓
解析列表项和详情链接
↓
请求详情页
↓
抽取预案名、适用领域、发布单位、发布日期、状态
↓
清洗字段
↓
URL 去重
↓
写入 CSV
↓
输出日志和失败记录

4.3 为什么选 requests + bs4 + lxml

我没有一上来就用 Scrapy 或 Playwright，原因很简单：这个任务不需要那么重。

requests 负责请求，代码直观，适合中小型采集任务。BeautifulSoup 负责容错解析 HTML，面对不太规范的页面时比较省心。lxml 作为解析器，速度和 XPath 支持都不错。Requests 官方文档展示了如何传入自定义 headers；BeautifulSoup 文档展示了如何使用 select() 执行 CSS 选择器查询，这正好适合列表页和详情页字段抽取。([Requests][4])

Scrapy 适合规模更大的项目，比如多栏目、多站点、任务队列、失败重试、下载中间件、去重调度都要系统化的时候。

Playwright 适合页面强依赖浏览器渲染的情况。但它更重，运行成本更高。能不用浏览器就不用浏览器，这是我写爬虫时一直坚持的原则。

5️⃣ 环境准备与依赖安装（可复现）

5.1 Python 版本

建议使用：

Python 3.10+

我个人更推荐 Python 3.11 或 3.12。原因不是为了追新，而是标准库和类型提示体验更好，运行性能也更稳。

查看版本：

python –version

5.2 创建虚拟环境

mkdir emergency_plan_spider
cd emergency_plan_spider

python -m venv .venv

Windows 激活：

.venv\\Scripts\\activate

macOS / Linux 激活：

source .venv/bin/activate

5.3 安装依赖

创建 requirements.txt：

requests==2.32.3
beautifulsoup4==4.12.3
lxml==5.3.0
PyYAML==6.0.2
python-dateutil==2.9.0.post0

安装：

pip install -r requirements.txt

5.4 推荐项目结构

emergency_plan_spider/
├── crawler/
│ ├── __init__.py
│ ├── config.py
│ ├── fetcher.py
│ ├── parser.py
│ ├── storage.py
│ └── utils.py
├── fixtures/
│ ├── list_page_1.html
│ ├── list_page_2.html
│ ├── detail_001.html
│ ├── detail_002.html
│ ├── detail_003.html
│ └── detail_004.html
├── data/
│ └── .gitkeep
├── config.yaml
├── main.py
└── requirements.txt

这套结构看起来比单文件复杂一点，但后期维护轻松很多。请求、解析、存储分开之后，哪怕目标站点改版，也通常只需要改 parser.py 或 config.yaml，不用把整段逻辑推倒重来。

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

请求层的职责是“安全、稳定、可控地拿到 HTML”。不要在请求层里写字段解析，也不要在解析层里直接发请求。职责混在一起，后面排错会非常难受。

6.1 headers

常见 headers 包括：

• User-Agent
• Accept
• Accept-Language
• Referer

这里的目的不是伪装攻击，而是让请求更接近普通浏览器访问，并且减少因为缺少基础头而被误判的情况。requests 支持传入 headers 字典。([Requests][4])

6.2 timeout

timeout 必须写。requests 文档明确提醒，默认情况下请求不会超时；如果不设置 timeout，代码可能挂很久。([Requests][5])

本文使用：

timeout=(5, 20)

含义是：

• 连接超时 5 秒。
• 读取超时 20 秒。

6.3 session / cookie

如果只是公开静态页面，不需要主动设置 cookie。但使用 requests.Session() 仍然有价值，因为它可以复用连接，并统一维护 headers。

如果目标站点需要登录，本文不建议继续写爬虫绕过，而是应使用授权接口或停止采集。

6.4 失败处理：重试与退避

失败处理思路：

• 403：不要暴力重试，先检查 robots、headers、访问频率。
• 404：记录失败，通常不重试。
• 429：降低频率，指数退避。
• 500 / 502 / 503 / 504：可以重试。
• 网络错误：可以重试，但次数有限。

退避示例：

第 1 次失败：sleep 1 秒
第 2 次失败：sleep 2 秒
第 3 次失败：sleep 4 秒
再失败：记录并跳过

6.5 请求层代码

crawler/fetcher.py：

from __future__ import annotations

import random
import time
from pathlib import Path
from typing import Optional
from urllib.parse import urlparse, urljoin
from urllib.robotparser import RobotFileParser

import requests

class FetchError(Exception):
"""请求失败时抛出的自定义异常。"""

class BaseFetcher:
"""请求器基类。"""

def get_text(self, url: str, referer: Optional[str] = None) –> str:
raise NotImplementedError

class FixtureFetcher(BaseFetcher):
"""
本地演示请求器。

这个类不访问互联网，而是从 fixtures 目录读取 HTML。
这样做的目的有两个：
1. 让读者不依赖任何真实站点，也能完整跑通项目。
2. 方便测试 Parser 的稳定性。
"""

def __init__(self, fixture_dir: str = "fixtures") –> None:
self.fixture_dir = Path(fixture_dir)

def get_text(self, url: str, referer: Optional[str] = None) –> str:
"""
fixture 模式下，url 实际上是文件名，例如：
fixture://list_page_1.html
fixture://detail_001.html
"""
if not url.startswith("fixture://"):
raise FetchError(f"FixtureFetcher only accepts fixture:// URL, got: {url}")

filename = url.replace("fixture://", "").strip("/")
path = self.fixture_dir / filename

if not path.exists():
raise FetchError(f"Fixture file not found: {path}")

return path.read_text(encoding="utf-8")

class HTTPFetcher(BaseFetcher):
"""
真实 HTTP 请求器。

注意：
– 只用于公开页面。
– 默认启用 robots 检查。
– 默认低频访问。
"""

def __init__(
self,
user_agent: str,
timeout_connect: float = 5,
timeout_read: float = 20,
max_retries: int = 3,
backoff_base: float = 1.0,
min_interval: float = 1.0,
max_interval: float = 2.5,
respect_robots: bool = True,
) –> None:
self.user_agent = user_agent
self.timeout = (timeout_connect, timeout_read)
self.max_retries = max_retries
self.backoff_base = backoff_base
self.min_interval = min_interval
self.max_interval = max_interval
self.respect_robots = respect_robots

self.session = requests.Session()
self.session.headers.update(
{
"User-Agent": self.user_agent,
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.6",
"Connection": "keep-alive",
}
)

self._robots_cache: dict[str, RobotFileParser] = {}

def _sleep_between_requests(self) –> None:
delay = random.uniform(self.min_interval, self.max_interval)
time.sleep(delay)

def _get_origin(self, url: str) –> str:
parsed = urlparse(url)
return f"{parsed.scheme}://{parsed.netloc}"

def _get_robot_parser(self, url: str) –> RobotFileParser:
origin = self._get_origin(url)

if origin in self._robots_cache:
return self._robots_cache[origin]

robots_url = urljoin(origin, "/robots.txt")
parser = RobotFileParser()
parser.set_url(robots_url)

try:
parser.read()
except Exception:
# robots 读取失败时，不代表可以放开抓。
# 这里保守处理：仍然缓存 parser，但后续由 can_fetch 结果决定。
pass

self._robots_cache[origin] = parser
return parser

def _allowed_by_robots(self, url: str) –> bool:
if not self.respect_robots:
return True

parser = self._get_robot_parser(url)

try:
return parser.can_fetch(self.user_agent, url)
except Exception:
# 无法判断时，保守返回 False。
return False

def get_text(self, url: str, referer: Optional[str] = None) –> str:
if not self._allowed_by_robots(url):
raise FetchError(f"Blocked by robots.txt or unable to verify robots permission: {url}")

headers = {}
if referer:
headers["Referer"] = referer

last_error: Optional[Exception] = None

for attempt in range(self.max_retries + 1):
if attempt > 0:
backoff = self.backoff_base * (2 ** (attempt – 1))
jitter = random.uniform(0, 0.5)
time.sleep(backoff + jitter)

self._sleep_between_requests()

try:
response = self.session.get(url, headers=headers, timeout=self.timeout)

if response.status_code == 200:
# requests 会根据响应头推断编码。
# 某些老站点编码不规范时，可使用 apparent_encoding 做兜底。
if not response.encoding or response.encoding.lower() == "iso-8859-1":
response.encoding = response.apparent_encoding
return response.text

if response.status_code in {429, 500, 502, 503, 504}:
last_error = FetchError(f"Retryable status {response.status_code}: {url}")
continue

raise FetchError(f"Non-retryable status {response.status_code}: {url}")

except requests.RequestException as exc:
last_error = exc
continue

raise FetchError(f"Failed after retries: {url}, reason: {last_error}")

这段代码做了几件关键的事：

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

解析层的职责是“把 HTML 变成结构化字段”。

7.1 解析方式选择

本文使用 BeautifulSoup CSS 选择器作为主解析方式。原因是目录页通常结构比较简单，比如：

<ul class="plan-list">
<li>
<a href="/detail/001.html">某应急预案</a>
<span class="date">2024-01-10</span>
</li>
</ul>

CSS 选择器写起来很直观：

soup.select("ul.plan-list li")
item.select_one("a")
item.select_one(".date")

如果详情页是表格结构，也可以使用 BeautifulSoup 遍历 tr、td。如果页面结构复杂，XPath 会更适合，例如“找包含某个文本的标签后面的兄弟节点”。

7.2 列表页如何拿详情链接

目录页一般要抽两个东西：

• 详情页链接
• 列表页上已有的字段，如标题、日期

其中链接要用 urljoin 转为绝对 URL，因为页面里常见的是相对路径：

<a href="./detail_001.html">预案标题</a>

如果当前页是：

https://example.com/yjya/index.html

那么详情页应解析为：

https://example.com/yjya/detail_001.html

7.3 详情页如何抽字段

详情页字段可能以多种形式出现：

表格型：

<tr>
<th>发布单位</th>
<td>某某单位</td>
</tr>

段落型：

<p>发布单位：某某单位</p>
<p>发布日期：2024-01-10</p>

列表型：

<li>适用领域：生产安全</li>
<li>状态：现行有效</li>

所以 Parser 不应该只写死一种结构。更稳妥的方式是：

7.4 缺失字段怎么办

我的建议是：缺失字段不要让程序崩。 字段为空是数据质量问题，不应该变成运行错误。

比如状态没有抽到，就给：

未知

发布日期没有抽到，就留空，并在日志里提示。 链接没有抽到，则该列表项不能进入详情页采集，记录 warning 后跳过。

7.5 工具函数

crawler/utils.py：

from __future__ import annotations

import hashlib
import re
from datetime import datetime
from typing import Iterable, Optional
from urllib.parse import urljoin

from dateutil import parser as date_parser

SPACE_RE = re.compile(r"\\s+")
DATE_RE = re.compile(
r"(?P<year>20\\d{2}|19\\d{2})[年\\-/.](?P<month>\\d{1,2})[月\\-/.](?P<day>\\d{1,2})日?"
)

def clean_text(value: Optional[str]) –> str:
"""清洗文本：去掉多余空白、换行、制表符。"""
if value is None:
return ""
value = value.replace("\\u3000", " ")
value = SPACE_RE.sub(" ", value)
return value.strip()

def normalize_url(base_url: str, href: str) –> str:
"""将相对链接转为绝对链接。"""
href = clean_text(href)
if not href:
return ""
if base_url.startswith("fixture://"):
# fixture 模式下不走真实 URL 规则，直接返回 fixture 协议路径。
if href.startswith("fixture://"):
return href
return "fixture://" + href.lstrip("./")
return urljoin(base_url, href)

def normalize_date(value: str) –> str:
"""
尽量把日期统一成 YYYY-MM-DD。
支持：
– 2024-01-10
– 2024/01/10
– 2024.01.10
– 2024年1月10日
"""
value = clean_text(value)
if not value:
return ""

match = DATE_RE.search(value)
if match:
year = int(match.group("year"))
month = int(match.group("month"))
day = int(match.group("day"))
return f"{year:04d}–{month:02d}–{day:02d}"

try:
dt = date_parser.parse(value, fuzzy=True)
return dt.strftime("%Y-%m-%d")
except Exception:
return value

def first_non_empty(values: Iterable[str], default: str = "") –> str:
for value in values:
value = clean_text(value)
if value:
return value
return default

def make_hash(*parts: str) –> str:
raw = "|".join(clean_text(p) for p in parts)
return hashlib.sha256(raw.encode("utf-8")).hexdigest()

7.6 数据模型

crawler/config.py：

from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

import yaml

@dataclass
class ListSelectors:
item: str
title: str
link: str
date: str = ""
next_page: str = ""

@dataclass
class DetailSelectors:
title: str = "h1"
applicable_area: str = ""
publisher: str = ""
publish_date: str = ""
status: str = ""
content: str = "body"

@dataclass
class SpiderConfig:
mode: str
start_urls: list[str]
output_csv: str
user_agent: str
max_pages: int
max_detail_pages: int
respect_robots: bool
list_selectors: ListSelectors
detail_selectors: DetailSelectors

def load_config(path: str = "config.yaml") –> SpiderConfig:
data: dict[str, Any] = yaml.safe_load(Path(path).read_text(encoding="utf-8"))

list_selectors = ListSelectors(**data["selectors"]["list"])
detail_selectors = DetailSelectors(**data["selectors"]["detail"])

return SpiderConfig(
mode=data.get("mode", "fixture"),
start_urls=data["start_urls"],
output_csv=data.get("output_csv", "data/emergency_plans.csv"),
user_agent=data.get("user_agent", "EmergencyPlanResearchBot/1.0"),
max_pages=int(data.get("max_pages", 10)),
max_detail_pages=int(data.get("max_detail_pages", 200)),
respect_robots=bool(data.get("respect_robots", True)),
list_selectors=list_selectors,
detail_selectors=detail_selectors,
)

7.7 Parser 完整代码

crawler/parser.py：

from __future__ import annotations

import re
from dataclasses import dataclass
from typing import Optional

from bs4 import BeautifulSoup, Tag

from crawler.config import DetailSelectors, ListSelectors
from crawler.utils import clean_text, first_non_empty, normalize_date, normalize_url

@dataclass
class ListItem:
"""列表页解析出的临时对象。"""

title: str
detail_url: str
list_publish_date: str = ""

@dataclass
class PlanRecord:
"""最终导出的记录。"""

plan_name: str
applicable_area: str
publisher: str
publish_date: str
status: str
url: str

class EmergencyPlanParser:
"""
应急预案页面解析器。

解析层尽量不假设页面一定规整。
它先按配置选择器取值，再扫描常见键值对，最后用正则兜底。
"""

LABEL_MAP = {
"plan_name": ["预案名称", "预案名", "名称", "标题"],
"applicable_area": ["适用领域", "适用范围", "领域", "分类", "类别"],
"publisher": ["发布单位", "制定单位", "印发单位", "发布机构", "来源"],
"publish_date": ["发布日期", "发布时间", "印发日期", "公开日期", "发文日期"],
"status": ["状态", "有效性", "当前状态", "文件状态"],
}

def __init__(self, list_selectors: ListSelectors, detail_selectors: DetailSelectors) –> None:
self.list_selectors = list_selectors
self.detail_selectors = detail_selectors

def _soup(self, html: str) –> BeautifulSoup:
return BeautifulSoup(html, "lxml")

def _select_text(self, root: BeautifulSoup | Tag, selector: str) –> str:
if not selector:
return ""
node = root.select_one(selector)
if not node:
return ""
return clean_text(node.get_text(" ", strip=True))

def _select_attr(self, root: BeautifulSoup | Tag, selector: str, attr: str) –> str:
if not selector:
return ""
node = root.select_one(selector)
if not node:
return ""
return clean_text(node.get(attr, ""))

def parse_list_page(self, html: str, page_url: str) –> tuple[list[ListItem], str]:
soup = self._soup(html)
items: list[ListItem] = []

for node in soup.select(self.list_selectors.item):
title_node = node.select_one(self.list_selectors.title)
link_node = node.select_one(self.list_selectors.link)

if not title_node and not link_node:
continue

title = ""
if title_node:
title = clean_text(title_node.get_text(" ", strip=True))
if not title and link_node:
title = clean_text(link_node.get_text(" ", strip=True))

href = ""
if link_node:
href = clean_text(link_node.get("href", ""))

detail_url = normalize_url(page_url, href)

date_text = ""
if self.list_selectors.date:
date_node = node.select_one(self.list_selectors.date)
if date_node:
date_text = normalize_date(date_node.get_text(" ", strip=True))

if not title or not detail_url:
continue

items.append(
ListItem(
title=title,
detail_url=detail_url,
list_publish_date=date_text,
)
)

next_url = ""
if self.list_selectors.next_page:
next_node = soup.select_one(self.list_selectors.next_page)
if next_node:
href = clean_text(next_node.get("href", ""))
if href:
next_url = normalize_url(page_url, href)

return items, next_url

def _extract_table_pairs(self, soup: BeautifulSoup) –> dict[str, str]:
"""
从表格中提取键值对。

支持类似：
<tr><th>发布单位</th><td>某单位</td></tr>
<tr><td>发布日期</td><td>2024-01-10</td></tr>
"""
pairs: dict[str, str] = {}

for tr in soup.select("tr"):
cells = tr.find_all(["th", "td"])
if len(cells) < 2:
continue

key = clean_text(cells[0].get_text(" ", strip=True)).rstrip("：:")
value = clean_text(cells[1].get_text(" ", strip=True))

if key and value:
pairs[key] = value

return pairs

def _extract_list_pairs(self, soup: BeautifulSoup) –> dict[str, str]:
"""
从 li、p、div 中提取“键：值”结构。
"""
pairs: dict[str, str] = {}

candidates = soup.select("li, p, div")
for node in candidates:
text = clean_text(node.get_text(" ", strip=True))
if not text:
continue

# 避免正文长段落被误判。
if len(text) > 120:
continue

if "：" in text:
key, value = text.split("：", 1)
elif ":" in text:
key, value = text.split(":", 1)
else:
continue

key = clean_text(key).rstrip("：:")
value = clean_text(value)

if key and value:
pairs[key] = value

return pairs

def _match_field_from_pairs(self, pairs: dict[str, str], field_name: str) –> str:
aliases = self.LABEL_MAP.get(field_name, [])

for key, value in pairs.items():
clean_key = clean_text(key)
for alias in aliases:
if alias in clean_key:
return clean_text(value)

return ""

def _regex_extract(self, text: str, labels: list[str]) –> str:
"""
从正文中按标签兜底提取。
"""
for label in labels:
pattern = re.compile(
rf"{re.escape(label)}\\s*[：:]\\s*(?P<value>[^\\n\\r；;，,。]+)"
)
match = pattern.search(text)
if match:
return clean_text(match.group("value"))
return ""

def parse_detail_page(
self,
html: str,
detail_url: str,
fallback_title: str = "",
fallback_date: str = "",
) –> PlanRecord:
soup = self._soup(html)

selector_title = self._select_text(soup, self.detail_selectors.title)
selector_area = self._select_text(soup, self.detail_selectors.applicable_area)
selector_publisher = self._select_text(soup, self.detail_selectors.publisher)
selector_date = self._select_text(soup, self.detail_selectors.publish_date)
selector_status = self._select_text(soup, self.detail_selectors.status)

table_pairs = self._extract_table_pairs(soup)
list_pairs = self._extract_list_pairs(soup)
pairs = {**table_pairs, **list_pairs}

body_text = self._select_text(soup, self.detail_selectors.content) or clean_text(
soup.get_text(" ", strip=True)
)

plan_name = first_non_empty(
[
selector_title,
self._match_field_from_pairs(pairs, "plan_name"),
self._regex_extract(body_text, self.LABEL_MAP["plan_name"]),
fallback_title,
],
default="",
)

applicable_area = first_non_empty(
[
selector_area,
self._match_field_from_pairs(pairs, "applicable_area"),
self._regex_extract(body_text, self.LABEL_MAP["applicable_area"]),
],
default="",
)

publisher = first_non_empty(
[
selector_publisher,
self._match_field_from_pairs(pairs, "publisher"),
self._regex_extract(body_text, self.LABEL_MAP["publisher"]),
],
default="",
)

publish_date = first_non_empty(
[
selector_date,
self._match_field_from_pairs(pairs, "publish_date"),
self._regex_extract(body_text, self.LABEL_MAP["publish_date"]),
fallback_date,
],
default="",
)
publish_date = normalize_date(publish_date)

status = first_non_empty(
[
selector_status,
self._match_field_from_pairs(pairs, "status"),
self._regex_extract(body_text, self.LABEL_MAP["status"]),
],
default="未知",
)

return PlanRecord(
plan_name=plan_name,
applicable_area=applicable_area,
publisher=publisher,
publish_date=publish_date,
status=status,
url=detail_url,
)

这一层的写法不是最短的，但比较耐用。真实网页经常不按一种结构写，有的详情页是表格，有的是段落，有的是混合布局。Parser 多做几种兜底，后面返工会少很多。

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

8.1 起步选择 CSV

本文起步选 CSV，原因是简单、透明、好检查。对于第一版爬虫，我不建议一开始就上 MySQL。先让数据稳定落地，再考虑数据库。

CSV 的优势：

• Excel 可以直接打开。
• diff 工具可以对比版本。
• 方便给后续脚本读取。
• 不依赖数据库服务。

如果后续要做增量采集和查询，再升级到 SQLite 或 MySQL。

8.2 字段映射表

8.3 去重策略

最简单的去重策略是 URL 唯一。 因为同一条预案通常对应同一个详情页 URL。

但真实网站偶尔会出现这些情况：

• 同一内容有多个 URL。
• URL 后面带追踪参数。
• 同一预案迁移到了新栏目。
• 标题相同但发布单位不同。

所以可以增加内容 hash：

hash = sha256(plan_name + publisher + publish_date + url)

本文示例主要用 URL 去重，同时保留 hash 的扩展空间。

8.4 存储层代码

crawler/storage.py：

from __future__ import annotations

import csv
from pathlib import Path

from crawler.parser import PlanRecord
from crawler.utils import make_hash

class CsvStorage:
"""CSV 存储器。"""

FIELDNAMES = [
"plan_name",
"applicable_area",
"publisher",
"publish_date",
"status",
"url",
"content_hash",
]

def __init__(self, output_path: str) –> None:
self.output_path = Path(output_path)
self.output_path.parent.mkdir(parents=True, exist_ok=True)
self.seen_urls: set[str] = set()

def load_seen_urls(self) –> None:
"""
如果 CSV 已存在，先加载历史 URL，用于断点续跑时去重。
"""
if not self.output_path.exists():
return

with self.output_path.open("r", encoding="utf-8-sig", newline="") as f:
reader = csv.DictReader(f)
for row in reader:
url = row.get("url", "").strip()
if url:
self.seen_urls.add(url)

def exists(self, url: str) –> bool:
return url in self.seen_urls

def append_many(self, records: list[PlanRecord]) –> int:
if not records:
return 0

file_exists = self.output_path.exists()

written = 0
with self.output_path.open("a", encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=self.FIELDNAMES)

if not file_exists:
writer.writeheader()

for record in records:
if record.url in self.seen_urls:
continue

content_hash = make_hash(
record.plan_name,
record.applicable_area,
record.publisher,
record.publish_date,
record.status,
record.url,
)

writer.writerow(
{
"plan_name": record.plan_name,
"applicable_area": record.applicable_area,
"publisher": record.publisher,
"publish_date": record.publish_date,
"status": record.status,
"url": record.url,
"content_hash": content_hash,
}
)

self.seen_urls.add(record.url)
written += 1

return written

我这里使用 utf-8-sig，主要是照顾 Excel。普通 utf-8 本身没问题，但某些环境下 Excel 打开中文 CSV 会乱码，加 BOM 后兼容性更好。

9️⃣ 运行方式与结果展示（必写）

9.1 配置文件

config.yaml：

mode: fixture

start_urls:
– "fixture://list_page_1.html"

output_csv: "data/emergency_plans.csv"

user_agent: "EmergencyPlanResearchBot/1.0 (+for-public-directory-research)"

max_pages: 5
max_detail_pages: 50
respect_robots: true

selectors:
list:
item: "ul.plan-list li"
title: "a.title"
link: "a.title"
date: ".date"
next_page: "a.next"

detail:
title: "h1"
applicable_area: ""
publisher: ""
publish_date: ""
status: ""
content: "body"

在真实站点中，你主要需要改这几个地方：

mode: http

start_urls:
– "https://example.com/yjya/index.html"

selectors:
list:
item: "ul.xxx-list li"
title: "a"
link: "a"
date: ".time"
next_page: "a.next"

如果目标站点的详情页字段有稳定选择器，也可以写：

detail:
title: "h1"
applicable_area: ".meta .area"
publisher: ".meta .publisher"
publish_date: ".meta .date"
status: ".meta .status"
content: ".article"

9.2 本地 fixture 页面

fixtures/list_page_1.html：

<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>应急预案目录 – 第1页</title>
</head>
<body>
<h1>应急预案目录</h1>

<ul class="plan-list">
<li>
<a class="title" href="detail_001.html">城市防汛应急预案</a>
<span class="date">2024-05-12</span>
</li>
<li>
<a class="title" href="detail_002.html">生产安全事故综合应急预案</a>
<span class="date">2024年04月20日</span>
</li>
</ul>

<div class="pager">
<a class="next" href="list_page_2.html">下一页</a>
</div>
</body>
</html>

fixtures/list_page_2.html：

<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>应急预案目录 – 第2页</title>
</head>
<body>
<h1>应急预案目录</h1>

<ul class="plan-list">
<li>
<a class="title" href="detail_003.html">突发公共卫生事件应急预案</a>
<span class="date">2023/12/01</span>
</li>
<li>
<a class="title" href="detail_004.html">道路交通突发事件应急预案</a>
<span class="date">2023.09.18</span>
</li>
</ul>

<div class="pager">
</div>
</body>
</html>

fixtures/detail_001.html：

<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>城市防汛应急预案</title>
</head>
<body>
<article class="article">
<h1>城市防汛应急预案</h1>

<table class="meta-table">
<tr>
<th>适用领域</th>
<td>自然灾害</td>
</tr>
<tr>
<th>发布单位</th>
<td>示例市应急管理部门</td>
</tr>
<tr>
<th>发布日期</th>
<td>2024-05-12</td>
</tr>
<tr>
<th>状态</th>
<td>现行有效</td>
</tr>
</table>

<p>本预案用于规范城市防汛相关应急响应流程。</p>
</article>
</body>
</html>

fixtures/detail_002.html：

<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>生产安全事故综合应急预案</title>
</head>
<body>
<article class="article">
<h1>生产安全事故综合应急预案</h1>

<p>适用领域：生产安全</p>
<p>发布单位：示例区安全生产管理部门</p>
<p>发布日期：2024年04月20日</p>
<p>状态：现行有效</p>

<p>本预案用于生产安全事故的信息报告、响应分级和协同处置。</p>
</article>
</body>
</html>

fixtures/detail_003.html：

<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>突发公共卫生事件应急预案</title>
</head>
<body>
<main>
<h1>突发公共卫生事件应急预案</h1>

<ul class="doc-info">
<li>分类：公共卫生</li>
<li>发布机构：示例卫生健康部门</li>
<li>公开日期：2023/12/01</li>
<li>有效性：现行有效</li>
</ul>

<section>
<p>本预案用于突发公共卫生事件的监测、报告和应急响应。</p>
</section>
</main>
</body>
</html>

fixtures/detail_004.html：

<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>道路交通突发事件应急预案</title>
</head>
<body>
<div class="content">
<h1>道路交通突发事件应急预案</h1>

<div>适用范围：交通运输</div>
<div>发布单位：示例交通运输管理部门</div>
<div>发文日期：2023.09.18</div>
<div>文件状态：修订中</div>

<p>本预案用于道路交通突发事件的协调处置和信息发布。</p>
</div>
</body>
</html>

9.3 入口文件

main.py：

from __future__ import annotations

import argparse
import logging
from typing import Iterable

from crawler.config import SpiderConfig, load_config
from crawler.fetcher import BaseFetcher, FetchError, FixtureFetcher, HTTPFetcher
from crawler.parser import EmergencyPlanParser, ListItem, PlanRecord
from crawler.storage import CsvStorage

def setup_logging() –> None:
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(message)s",
)

def build_fetcher(config: SpiderConfig) –> BaseFetcher:
if config.mode == "fixture":
return FixtureFetcher("fixtures")

if config.mode == "http":
return HTTPFetcher(
user_agent=config.user_agent,
respect_robots=config.respect_robots,
timeout_connect=5,
timeout_read=20,
max_retries=3,
backoff_base=1.0,
min_interval=1.0,
max_interval=2.5,
)

raise ValueError(f"Unsupported mode: {config.mode}")

def crawl_list_pages(
fetcher: BaseFetcher,
parser: EmergencyPlanParser,
start_urls: Iterable[str],
max_pages: int,
) –> list[ListItem]:
all_items: list[ListItem] = []
visited_pages: set[str] = set()

queue = list(start_urls)

while queue and len(visited_pages) < max_pages:
page_url = queue.pop(0)

if page_url in visited_pages:
continue

logging.info("Fetching list page: %s", page_url)

try:
html = fetcher.get_text(page_url)
except FetchError as exc:
logging.warning("List page fetch failed: %s", exc)
visited_pages.add(page_url)
continue

items, next_url = parser.parse_list_page(html, page_url)

logging.info("Parsed %d list items from %s", len(items), page_url)

all_items.extend(items)
visited_pages.add(page_url)

if next_url and next_url not in visited_pages:
queue.append(next_url)

return all_items

def crawl_detail_pages(
fetcher: BaseFetcher,
parser: EmergencyPlanParser,
storage: CsvStorage,
list_items: list[ListItem],
max_detail_pages: int,
) –> list[PlanRecord]:
records: list[PlanRecord] = []
count = 0

for item in list_items:
if count >= max_detail_pages:
logging.info("Reached max_detail_pages=%d", max_detail_pages)
break

if storage.exists(item.detail_url):
logging.info("Skip duplicated URL: %s", item.detail_url)
continue

logging.info("Fetching detail page: %s", item.detail_url)

try:
html = fetcher.get_text(item.detail_url)
except FetchError as exc:
logging.warning("Detail page fetch failed: %s", exc)
continue

try:
record = parser.parse_detail_page(
html=html,
detail_url=item.detail_url,
fallback_title=item.title,
fallback_date=item.list_publish_date,
)
except Exception as exc:
logging.exception("Detail page parse failed: %s, reason=%s", item.detail_url, exc)
continue

records.append(record)
count += 1

return records

def run(config_path: str) –> None:
config = load_config(config_path)
fetcher = build_fetcher(config)
parser = EmergencyPlanParser(config.list_selectors, config.detail_selectors)
storage = CsvStorage(config.output_csv)
storage.load_seen_urls()

list_items = crawl_list_pages(
fetcher=fetcher,
parser=parser,
start_urls=config.start_urls,
max_pages=config.max_pages,
)

logging.info("Total list items collected: %d", len(list_items))

records = crawl_detail_pages(
fetcher=fetcher,
parser=parser,
storage=storage,
list_items=list_items,
max_detail_pages=config.max_detail_pages,
)

written = storage.append_many(records)

logging.info("Parsed detail records: %d", len(records))
logging.info("Written records: %d", written)
logging.info("Output CSV: %s", config.output_csv)

def parse_args() –> argparse.Namespace:
parser = argparse.ArgumentParser(description="Emergency plan directory spider")
parser.add_argument(
"–config",
default="config.yaml",
help="Path to config yaml file",
)
return parser.parse_args()

if __name__ == "__main__":
setup_logging()
args = parse_args()
run(args.config)

9.4 启动命令

python main.py –config config.yaml

运行日志类似：

2026-05-22 10:30:01 | INFO | Fetching list page: fixture://list_page_1.html
2026-05-22 10:30:01 | INFO | Parsed 2 list items from fixture://list_page_1.html
2026-05-22 10:30:01 | INFO | Fetching list page: fixture://list_page_2.html
2026-05-22 10:30:01 | INFO | Parsed 2 list items from fixture://list_page_2.html
2026-05-22 10:30:01 | INFO | Total list items collected: 4
2026-05-22 10:30:01 | INFO | Fetching detail page: fixture://detail_001.html
2026-05-22 10:30:01 | INFO | Fetching detail page: fixture://detail_002.html
2026-05-22 10:30:01 | INFO | Fetching detail page: fixture://detail_003.html
2026-05-22 10:30:01 | INFO | Fetching detail page: fixture://detail_004.html
2026-05-22 10:30:01 | INFO | Parsed detail records: 4
2026-05-22 10:30:01 | INFO | Written records: 4
2026-05-22 10:30:01 | INFO | Output CSV: data/emergency_plans.csv

9.5 输出路径

data/emergency_plans.csv

9.6 示例结果

CSV 实际还会多一个 content_hash 字段，用于辅助判断内容是否变化。

🔟 常见问题与排错

10.1 403 怎么办

403 表示服务器拒绝访问。常见原因包括：

• 请求头太少。
• 访问频率过高。
• robots 不允许。
• 目标路径不允许自动访问。
• 来源页 referer 校验。
• 站点本身有访问限制。

处理建议：

第一，先降低频率。 不要一看到 403 就上代理、上并发，这样只会让情况更糟。

第二，检查 robots。 如果目标路径不允许访问，就停止。

第三，补充基础 headers。 包括 User-Agent、Accept、Accept-Language、Referer。

第四，使用浏览器人工访问确认页面是否公开。 如果浏览器也打不开，就不是爬虫代码问题。

10.2 429 怎么办

429 通常表示访问过快。处理方式：

• 增加请求间隔。
• 减少并发。
• 延长退避时间。
• 缩小抓取范围。
• 做增量抓取，不要全量反复跑。

示例配置可以改成：

min_interval=3.0
max_interval=8.0
backoff_base=5.0

我不建议为了绕过 429 去堆代理。对于这种公开目录采集，正确做法是少抓、慢抓、必要时停止。

10.3 HTML 抓到空壳怎么办

如果 requests 抓到的 HTML 里只有框架，没有列表数据，通常说明页面是动态渲染的。

排查步骤：

如果能找到 JSON 接口，优先请求 JSON 接口。 如果接口参数复杂，或者数据必须通过浏览器执行 JS 才能得到，再考虑 Playwright。

Playwright 不是第一选择。它能解决动态渲染问题，但成本更高，速度更慢，也更容易被站点识别为自动化浏览器。

10.4 解析报错怎么办

解析报错一般有四类：

第一，选择器失效。 页面结构改了，原来的 .plan-list li 找不到数据。

第二，字段位置不统一。 有的详情页用表格，有的详情页用段落。

第三，缺失字段没处理。 select_one() 返回 None 后直接 .text，就会报错。

第四，HTML 不规范。 老站点经常有未闭合标签、嵌套混乱等问题。

解决方式：

node = soup.select_one(".date")
date = node.get_text(strip=True) if node else ""

不要写：

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

后一种写法在页面稍微有点变化时就会炸。

10.5 编码或乱码如何处理

如果抓到的中文乱码，可以检查：

print(response.encoding)
print(response.apparent_encoding)

然后手动设置：

response.encoding = response.apparent_encoding

本文请求层里已经做了一个兜底：

if not response.encoding or response.encoding.lower() == "iso-8859-1":
response.encoding = response.apparent_encoding

导出 CSV 时使用：

encoding="utf-8-sig"

这能减少 Excel 打开中文 CSV 时出现乱码的概率。

10.6 日期格式不统一怎么办

真实页面里日期格式可能是：

2024-01-10
2024/01/10
2024.01.10
2024年1月10日
发布日期：2024年01月10日

所以代码里做了 normalize_date()。 如果你的目标页面还有“2024-1-10 发布”这种格式，可以继续扩展正则。

10.7 下一页抓不到怎么办

分页结构通常有几种：

<a class="next" href="index_2.html">下一页</a>

<a href="index_2.html">下页</a>

<a href="?page=2">下一页</a>

<script>goPage(2)</script>

前两种 CSS 选择器能解决。 第三种需要 urljoin 处理。 第四种可能需要分析 JS 或构造分页 URL。

如果分页规则很稳定，例如：

index.html
index_2.html
index_3.html

也可以不用解析“下一页”，直接生成 URL：

for page in range(1, 11):
if page == 1:
url = "https://example.com/yjya/index.html"
else:
url = f"https://example.com/yjya/index_{page}.html"

但我更喜欢优先解析页面上的下一页链接。因为它能自然停止，不容易抓过界。

1️⃣1️⃣ 进阶优化（可选但加分）

11.1 并发优化

目录类爬虫一般不需要高并发。如果数据量大，可以考虑低并发线程池，例如 2 到 4 个 worker。

示例：

from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch_one_detail(item):
html = fetcher.get_text(item.detail_url)
return parser.parse_detail_page(
html=html,
detail_url=item.detail_url,
fallback_title=item.title,
fallback_date=item.list_publish_date,
)

with ThreadPoolExecutor(max_workers=3) as executor:
future_map = {
executor.submit(fetch_one_detail, item): item
for item in list_items
}

for future in as_completed(future_map):
item = future_map[future]
try:
record = future.result()
except Exception as exc:
logging.warning("Failed: %s, reason=%s", item.detail_url, exc)

注意，并发不是越高越好。对于公开目录页，稳定和礼貌比速度重要。

11.2 断点续跑

本文 CSV 存储层已经实现了基础断点续跑：

storage.load_seen_urls()

如果 CSV 里已经有某个 URL，下次运行会跳过。

更完整的断点续跑可以记录：

• 已抓列表页。
• 已抓详情页。
• 失败 URL。
• 失败原因。
• 最后成功时间。
• 内容 hash。

可以创建一个 SQLite 表：

CREATE TABLE crawl_state (
url TEXT PRIMARY KEY,
url_type TEXT,
status TEXT,
retry_count INTEGER DEFAULT 0,
last_error TEXT,
updated_at TEXT
);

状态可以设计为：

pending
success
failed
skipped

这样即使程序中断，也能从失败点继续。

11.3 SQLite 存储版本

如果 CSV 不够用，可以加 SQLite。

示例 crawler/sqlite_storage.py：

from __future__ import annotations

import sqlite3
from pathlib import Path

from crawler.parser import PlanRecord
from crawler.utils import make_hash

class SQLiteStorage:
def __init__(self, db_path: str = "data/emergency_plans.db") –> None:
self.db_path = Path(db_path)
self.db_path.parent.mkdir(parents=True, exist_ok=True)
self.conn = sqlite3.connect(str(self.db_path))
self._init_table()

def _init_table(self) –> None:
sql = """
CREATE TABLE IF NOT EXISTS emergency_plans (
id INTEGER PRIMARY KEY AUTOINCREMENT,
plan_name TEXT,
applicable_area TEXT,
publisher TEXT,
publish_date TEXT,
status TEXT,
url TEXT UNIQUE,
content_hash TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
"""
self.conn.execute(sql)
self.conn.commit()

def exists(self, url: str) –> bool:
sql = "SELECT 1 FROM emergency_plans WHERE url = ? LIMIT 1"
cur = self.conn.execute(sql, (url,))
return cur.fetchone() is not None

def upsert(self, record: PlanRecord) –> None:
content_hash = make_hash(
record.plan_name,
record.applicable_area,
record.publisher,
record.publish_date,
record.status,
record.url,
)

sql = """
INSERT INTO emergency_plans (
plan_name,
applicable_area,
publisher,
publish_date,
status,
url,
content_hash
)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(url) DO UPDATE SET
plan_name = excluded.plan_name,
applicable_area = excluded.applicable_area,
publisher = excluded.publisher,
publish_date = excluded.publish_date,
status = excluded.status,
content_hash = excluded.content_hash;
"""

self.conn.execute(
sql,
(
record.plan_name,
record.applicable_area,
record.publisher,
record.publish_date,
record.status,
record.url,
content_hash,
),
)
self.conn.commit()

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

SQLite 的优势是可以直接查询：

SELECT applicable_area, COUNT(*)
FROM emergency_plans
GROUP BY applicable_area
ORDER BY COUNT(*) DESC;

11.4 日志与监控

建议记录这些指标：

• 列表页请求成功数。
• 列表页请求失败数。
• 详情页请求成功数。
• 详情页请求失败数。
• 解析成功数。
• 缺字段数量。
• 去重数量。
• 总耗时。

可以在主程序里加一个简单统计器：

from dataclasses import dataclass

@dataclass
class CrawlStats:
list_success: int = 0
list_failed: int = 0
detail_success: int = 0
detail_failed: int = 0
parsed_records: int = 0
written_records: int = 0
skipped_duplicates: int = 0

爬虫不只是“跑完就完”。日志和统计会在维护阶段救命。尤其当某天突然只抓到 0 条，日志能告诉你是请求失败、解析失败，还是页面结构变了。

11.5 定时任务

如果要每天定时执行，可以用 cron。

Linux / macOS：

crontab -e

添加：

0 2 * * * cd /path/to/emergency_plan_spider && /path/to/.venv/bin/python main.py –config config.yaml >> logs/cron.log 2>&1

这表示每天凌晨 2 点执行一次。

Windows 可以用任务计划程序。 更复杂的场景可以用 Airflow、Prefect 或 APScheduler。

11.6 失败 URL 单独落盘

可以把失败详情页写到 data/failed_urls.csv：

import csv
from pathlib import Path

def save_failed_url(url: str, reason: str, path: str = "data/failed_urls.csv") –> None:
output = Path(path)
output.parent.mkdir(parents=True, exist_ok=True)

file_exists = output.exists()

with output.open("a", encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=["url", "reason"])

if not file_exists:
writer.writeheader()

writer.writerow(
{
"url": url,
"reason": reason,
}
)

这样你可以第二天单独补抓失败 URL，而不是重新跑全量。

11.7 内容变更检测

如果同一个 URL 的内容可能更新，可以保存 content_hash。下次抓取时比较 hash：

old_hash = get_old_hash(url)
new_hash = make_hash(plan_name, publisher, publish_date, status, url)

if old_hash != new_hash:
print("内容可能发生变化，需要更新记录")

注意，这里的 hash 是字段级 hash，不是全文 hash。字段级 hash 更轻量，但如果你关心正文变更，可以把正文也加入 hash。

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

这篇文章完成了一个完整的“应急预案目录分页”爬虫工程。

我们没有把代码写成一个几百行的混乱脚本，而是拆成了几个清晰模块：

• fetcher.py：负责请求、headers、timeout、session、robots、重试退避。
• parser.py：负责列表页和详情页解析。
• storage.py：负责 CSV 导出和 URL 去重。
• utils.py：负责文本清洗、日期标准化、URL 规范化、hash。
• main.py：负责调度整个流程。
• config.yaml：负责把站点差异配置化。

从工程角度看，这类项目最重要的不是“能不能抓到一次”，而是“下周页面小改还能不能修得动”。所以本文把请求、解析、存储拆开，也把 fixture 演示模式保留下来。fixture 看起来像多余的东西，但在真实维护中非常有用：你可以不访问目标站点，也能快速测试解析逻辑。

下一步可以继续做这些扩展：

最后，我想补一句比较个人的经验： 爬虫写到最后，拼的不是“请求有多猛”，而是“边界感、耐心和维护习惯”。尤其是公开目录类数据，慢一点没关系，字段干净、日志清楚、能稳定复跑，比一次性抓得飞快更有价值。

🌟 文末

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

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

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

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

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

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

✅ 互动征集

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

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

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

✅ 免责声明

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

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

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