报刊历史期号索引平台爬虫实战：年月期号级联、动态日历与历史分页解析

㊗️本期内容已收录至专栏《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 为什么不一开始就用 Scrapy
  - 4.3 整体流程
- 5️⃣ 环境准备与依赖安装
- - 5.1 Python 版本
  - 5.2 创建虚拟环境
  - 5.3 安装依赖
  - 5.4 推荐项目结构
  - 5.5 requirements.txt
- 6️⃣ 核心实现：请求层（Fetcher）
- - 6.1 headers 设置
  - 6.2 timeout
  - 6.3 session/cookie
  - 6.4 失败处理：重试与退避
- 7️⃣ 核心实现：解析层（Parser）
- - 7.1 解析方式选择
  - 7.2 列表页如何拿详情链接
  - 7.3 详情页如何抽字段
  - 7.4 缺失字段怎么办
- 8️⃣ 数据存储与导出（Storage）
- - 8.1 为什么选择 SQLite 起步
  - 8.2 字段映射表
  - 8.3 去重策略
- 9️⃣ 完整可运行代码
- - 9.1 运行结果
- 10. 真实站点改造方法
- - 10.1 关闭样例模式
  - 10.2 修改日历 URL 规则
  - 10.3 修改 CSS 选择器
- 11. 年月期号级联解析
- - 11.1 年份层
  - 11.2 月份层
  - 11.3 日期层
  - 11.4 期号层
  - 11.5 分页层
- 12. 动态日历解析
- - 12.1 什么是动态日历
  - 12.2 JSON 日历解析示例
  - 12.3 只有渲染结果，没有接口怎么办
- 13. 历史分页解析细节
- - 13.1 从第一页顺着下一页走
  - 13.2 根据总页数构造 URL
  - 13.3 空页处理
- 14. 数据清洗策略
- - 14.1 日期清洗
  - 14.2 数字化状态清洗
  - 14.3 URL 补全
- 15. 运行方式与结果展示
- - 15.1 启动命令
  - 15.2 输出位置
  - 15.3 查看 SQLite 数据
  - 15.4 示例结果
- 16. 常见问题与排错
- - 16.1 403 怎么办
  - 16.2 429 怎么办
  - 16.3 HTML 抓到空壳怎么办
  - 16.4 解析报错怎么办
  - 16.5 编码乱码怎么办
  - 16.6 日期解析失败怎么办
  - 16.7 重复数据太多怎么办
  - 16.8 下一页死循环怎么办
- 17. 进阶优化
- - 17.1 并发采集
  - 17.2 asyncio
  - 17.3 Scrapy 改造方向
  - 17.4 断点续跑
  - 17.5 游标记录
  - 17.6 日志与监控
  - 17.7 定时任务
  - 17.8 增量采集
- 18. 工程化拆分建议
- - 18.1 config.py
  - 18.2 crawler/fetcher.py
  - 18.3 crawler/parser.py
  - 18.4 crawler/storage.py
- 19. 字段质量检查
- - 19.1 检查空字段
  - 19.2 检查日期范围
  - 19.3 检查重复主题
  - 19.4 检查状态分布
- 20. MySQL 存储扩展示例
- 21. 选择器稳定性建议
- - 21.1 不建议的写法
  - 21.2 推荐写法
  - 21.3 多选择器兜底
- 22. 请求失败 URL 的补抓
- 23. 面向真实业务的采集边界
- 24. 总结与延伸阅读
- 🌟 文末
- - ✅ 专栏持续更新中｜建议收藏 + 订阅
  - ✅ 互动征集
  - ✅ 免责声明

🌟 开篇语

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

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

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

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

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

0️⃣ 前言（Preface）

这篇文章要做的事情很明确：用 Python 抓取一个“报刊历史期号索引平台”的公开索引数据，字段包括报刊名、期号、出版日期、栏目、主题、数字化状态和详情链接，最后把结果保存到 SQLite 与 CSV 文件中，方便后续检索、分析和归档。

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

一套适合报刊、期刊、公告、档案类网站的通用爬虫设计思路。

一个能够处理“年份 → 月份 → 期号 → 详情页”的级联采集范式。

一份可本地运行、可改造成真实站点爬虫的完整 Python 示例代码。

这不是一篇为了炫技而写的文章。报刊历史数据的爬取并不一定难在请求本身，真正麻烦的是结构：有些站点按年份归档，有些站点按月份提供日历，有些站点的历史分页规则还带着老系统的痕迹。我的经验是，写这种爬虫不能一上来就猛堆并发，先把“路径”和“数据模型”想清楚，后面反而会稳很多。

1️⃣ 摘要（Abstract）

本文围绕“报刊历史期号索引平台”设计并实现一个 Python 爬虫，使用 requests、BeautifulSoup、lxml、SQLite 和 pandas 完成公开索引页采集、详情页解析、字段清洗、去重存储与 CSV 导出。

你将学到：

如何处理年月期号级联结构，比如先抓年份，再抓月份，再进入对应期号列表。

如何解析动态日历和历史分页，尤其是“某月有多少期、每页有哪些栏目文章”的结构。

如何设计一个稳定、可维护、可扩展的爬虫工程，而不是写一个只能跑一次的临时代码。

最终产出是一张结构化数据表，核心字段如下：

字段说明

报刊名	报纸或期刊名称
期号	第几期、总第几期或页面显示的期号
出版日期	该期出版日期
栏目	文章或索引所属栏目
主题	文章标题、主题词或索引主题
数字化状态	已数字化、待数字化、仅目录、缺页等
链接	详情页 URL

2️⃣ 背景与需求（Why）

2.1 为什么要爬报刊历史期号索引

报刊历史索引有很高的信息价值。很多时候，我们并不是为了抓取正文内容，而是为了建立一个“可搜索的目录”。比如：

做资料检索：按时间、栏目、主题快速定位某一期内容。

做数据分析：统计某个主题在不同年份、月份出现的频率。

做信息聚合：把分散在网站历史页里的期号统一整理成结构化数据。

做自动化归档：定期抓取新增期号，减少手工维护表格的工作量。

我个人比较喜欢把这类爬虫称为“索引型爬虫”。它和抓电商商品、新闻列表不太一样。索引型爬虫的重点通常不是“抓得越多越好”，而是“字段要准、层级要清楚、可以复查来源”。

2.2 目标站点的典型结构

本文假设目标站点是一个公开的报刊历史期号索引平台，页面结构大致如下：

首页
├── 年份归档
│ ├── 2021
│ ├── 2022
│ ├── 2023
│ └── 2024
│
├── 月份日历
│ ├── 2024-01
│ ├── 2024-02
│ ├── 2024-03
│ └── …
│
├── 期号列表页
│ ├── 第 1 页
│ ├── 第 2 页
│ └── 第 N 页
│
└── 详情页
├── 报刊名
├── 期号
├── 出版日期
├── 栏目
├── 主题
├── 数字化状态
└── 链接

很多历史报刊站点并不是现代化的前后端分离架构。有的页面是纯 HTML，有的页面用 JavaScript 生成日历，有的则隐藏了一个接口。真实工作中，我一般会先用浏览器开发者工具观察三件事：

切换年份和月份时，页面是否真的刷新。

点击日历日期时，浏览器是否请求了接口。

翻页时 URL 是否有明确参数，比如 page=2、year=2024、month=05。

2.3 目标字段清单

本项目采集字段如下：

字段名英文字段说明

报刊名	newspaper_name	报刊、期刊或平台展示名称
期号	issue_no	如“2024年第05期”“总第238期”
出版日期	publish_date	标准化为 YYYY-MM-DD
栏目	column_name	文章或索引所属栏目
主题	topic	标题、主题或索引名称
数字化状态	digitization_status	已数字化、未数字化、仅目录等
链接	url	详情页完整链接

3️⃣ 合规与注意事项

爬虫不是“看到页面就随便抓”。技术上能做到，不代表场景上就一定合适。尤其是报刊、档案、历史文献类站点，很多属于公共服务、学术机构或资料馆系统，更应该控制请求频率，尊重站点规则。

3.1 robots.txt 基本说明

正式采集前建议查看目标站点根目录下的 robots.txt 文件，例如：

https://example.com/robots.txt

robots.txt 通常会声明哪些路径允许抓取，哪些路径不建议抓取。例如：

User-agent: *
Disallow: /admin/
Disallow: /login/
Allow: /archive/

它不是访问控制系统，但它体现了站点对爬虫访问的基本态度。对于技术分享和个人学习来说，应该尽量遵守这些声明。

在本文示例代码中，我们不会绕过登录，不会采集个人敏感信息，也不会访问后台路径。示例只面向公开索引页。

3.2 频率控制

不要使用攻击式并发。对于索引型站点来说，请求量往往不需要很大。更稳妥的方式是：

每次请求之间增加随机等待。

失败后进行指数退避。

限制最大重试次数。

日志记录失败 URL，后续单独补抓。

示例策略：

正常请求间隔：1.0 ~ 2.5 秒
失败后等待：2 秒、4 秒、8 秒
最大重试：3 次
单机并发：初始建议 1，不建议一开始就开几十个线程

3.3 不采集敏感信息

本文只处理公开页面中的报刊索引信息，不采集手机号、身份证、私人邮箱、账户信息等敏感数据。如果站点需要登录或付费阅读，应使用平台提供的正规访问方式，不建议通过技术手段绕过限制。

3.4 不绕过付费或登录限制

有些报刊站点会公开目录，但正文需要授权。这种情况下，爬虫可以只保存目录索引和公开链接，不应该尝试绕过权限。比较中性的做法是：

目录可公开访问：采集目录字段
正文需要登录：不抓正文，只保留详情页链接
正文需要付费：不抓正文，只记录数字化状态或访问说明

这样既能完成索引整理，也能避免越界。

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

4.1 静态、动态和 API 的判断

报刊历史期号平台常见三种形态。

第一种是静态 HTML。年份、月份、分页、详情都直接写在 HTML 中。这种最适合使用 requests + BeautifulSoup/lxml。

第二种是动态渲染。页面初始 HTML 只有一个空壳，真正的数据由 JavaScript 请求后再渲染到页面。这时可以先找接口，如果接口清晰，仍然用 requests；如果接口复杂或签名难以维护，再考虑 Playwright。

第三种是半动态页面。年月日历是接口返回，详情页是静态 HTML。这类站点很常见，也是本文重点处理的类型。我们会把请求层、解析层和存储层分开，方便以后替换。

本文采用的主方案：

requests + BeautifulSoup + lxml + SQLite + pandas

可选增强：

Playwright：用于处理纯前端渲染页面
Scrapy：用于大规模队列、去重、调度和中间件管理

4.2 为什么不一开始就用 Scrapy

Scrapy 很强，但不是所有项目都需要一上来就用 Scrapy。对于一个字段明确、层级清晰、请求量可控的报刊索引采集任务，先用轻量工程写清楚流程，往往更容易调试。

我的建议是：

小规模验证：requests + bs4
长期采集：requests 工程化 + SQLite
多站点采集：Scrapy
强动态页面：Playwright
大规模生产：Scrapy + Redis + 定时任务 + 监控

4.3 整体流程

本项目采用四段式流程：

采集 → 解析 → 清洗 → 存储

更具体一点：

1. 获取年份入口
2. 获取某年可用月份
3. 进入某月历史日历
4. 解析该月所有期号入口
5. 遍历期号分页
6. 提取列表页详情链接
7. 请求详情页
8. 抽取字段
9. 清洗日期、状态、URL
10. 写入 SQLite
11. 导出 CSV

用文字画成流程图：

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

5.1 Python 版本

建议使用：

Python 3.10+

我平时写爬虫会尽量避免依赖太新的语法，主要是考虑部署环境。Python 3.10 或 3.11 都比较稳。

查看版本：

python –version

5.2 创建虚拟环境

mkdir newspaper_index_crawler
cd newspaper_index_crawler

python -m venv .venv

# Windows
.venv\\Scripts\\activate

# macOS / Linux
source .venv/bin/activate

5.3 安装依赖

pip install requests beautifulsoup4 lxml pandas python-dateutil

可选安装 Playwright：

pip install playwright
playwright install

本文主代码不强制依赖 Playwright，只有遇到纯动态站点时才需要。

5.4 推荐项目结构

newspaper_index_crawler/
├── main.py
├── config.py
├── requirements.txt
├── data/
│ ├── newspaper_issues.db
│ └── newspaper_issues.csv
├── fixtures/
│ ├── calendar_2024_05.html
│ ├── issue_2024_05_01_page_1.html
│ ├── issue_2024_05_01_page_2.html
│ ├── detail_001.html
│ ├── detail_002.html
│ └── detail_003.html
└── crawler/
├── __init__.py
├── fetcher.py
├── parser.py
├── storage.py
└── cleaner.py

为了让文章里的代码更容易复制运行，后文会给出一个“单文件可运行版”。真实项目中再拆成上面的结构即可。

5.5 requirements.txt

requests==2.32.3
beautifulsoup4==4.12.3
lxml==5.2.2
pandas==2.2.2
python-dateutil==2.9.0.post0

版本不一定非得完全一致，但锁定版本有一个好处：别人复现实验时不会因为依赖升级导致行为变化。

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

请求层负责和目标站点打交道。它的职责不是解析页面，也不是存储数据，而是稳定地拿到 HTML 或 JSON。

6.1 headers 设置

常见 headers：

headers = {
"User-Agent": "Mozilla/5.0 …",
"Referer": "https://example.com/archive",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
}

User-Agent 不应该伪装得过度夸张，保持常规浏览器访问即可。Referer 有时是必要的，因为部分站点会检查来源页面。

6.2 timeout

不要写没有 timeout 的请求。

错误写法：

requests.get(url)

推荐写法：

requests.get(url, timeout=(5, 20))

其中 (5, 20) 的含义是：

连接超时：5 秒
读取超时：20 秒

这样可以避免某个请求一直卡住。

6.3 session/cookie

如果目标站点只是公开索引页，通常不需要登录 cookie。但使用 requests.Session() 仍然有好处：

自动复用 TCP 连接。

保持部分站点下发的普通 cookie。

便于统一设置 headers。

示例：

session = requests.Session()
session.headers.update(headers)

6.4 失败处理：重试与退避

常见错误：

403：访问被拒绝
404：页面不存在
429：请求过多
500/502/503：服务端异常
timeout：网络超时

比较稳的策略是：

403：降低频率，检查 headers 和 robots，不建议硬冲
404：记录并跳过
429：明显触发频控，延长等待
5xx：重试几次，不要无限重试
timeout：重试，仍失败则记录

退避策略示例：

第 1 次失败：等待 2 秒
第 2 次失败：等待 4 秒
第 3 次失败：等待 8 秒

我不建议在这类文章里把代理作为默认方案。代理不是万能药，也容易把简单问题复杂化。多数索引采集任务先做好频控和重试就够了。

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

解析层负责从 HTML 中抽取结构化字段。这里有一个原则：不要让解析逻辑散落在请求循环里。请求是请求，解析是解析，两者分开后，调试会舒服很多。

7.1 解析方式选择

本文使用：

BeautifulSoup + CSS Selector

原因是：

入门成本低。

对不规范 HTML 容忍度较好。

CSS 选择器写起来比较直观。

如果页面结构非常稳定，也可以使用 XPath：

from lxml import etree
tree = etree.HTML(html)
links = tree.xpath('//a[contains(@class, "detail-link")]/@href')

7.2 列表页如何拿详情链接

期号列表页通常长这样：

<div class="issue-list">
<div class="issue-item">
<a class="detail-link" href="/detail/001.html">城市建设专题</a>
<span class="column">综合新闻</span>
<span class="status">已数字化</span>
</div>
</div>

解析思路：

1. 找到所有 issue-item
2. 提取 detail-link 的 href
3. 补全成绝对 URL
4. 顺带提取列表页已有字段

有些字段列表页就有，有些字段详情页才有。我的做法是：列表页先保存一份“浅字段”，详情页再用“深字段”覆盖或补齐。

7.3 详情页如何抽字段

详情页可能长这样：

<h1 class="topic">城市建设专题</h1>
<ul class="meta">
<li><span>报刊名：</span><b>示例日报</b></li>
<li><span>期号：</span><b>2024年第05期</b></li>
<li><span>出版日期：</span><b>2024-05-01</b></li>
<li><span>栏目：</span><b>综合新闻</b></li>
<li><span>数字化状态：</span><b>已数字化</b></li>
</ul>

解析时不要只依赖位置，例如第一个 li 是报刊名、第二个是期号。更稳的方式是按照标签文字映射：

看到“报刊名” → newspaper_name
看到“期号” → issue_no
看到“出版日期” → publish_date

这样即使页面多了一个“责任编辑”字段，也不会影响核心字段。

7.4 缺失字段怎么办

真实页面一定会缺字段。不要因为一个字段为空就让整个程序崩掉。

建议策略：

报刊名缺失：使用配置中的默认报刊名
期号缺失：尝试从 URL 或列表页继承
出版日期缺失：尝试从日历日期继承
栏目缺失：填空字符串
主题缺失：使用列表页标题
数字化状态缺失：填“未知”
链接缺失：丢弃该条

字段缺失时应该记录日志，而不是假装一切正常。

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

8.1 为什么选择 SQLite 起步

本文使用 SQLite。原因很简单：

不需要单独安装数据库服务。

单文件存储，便于备份和移动。

支持唯一约束，方便去重。

后续可以很容易迁移到 MySQL 或 PostgreSQL。

对于个人学习、资料归档、小规模索引采集，SQLite 是非常合适的。

8.2 字段映射表

中文字段数据库字段类型示例值

报刊名	newspaper_name	TEXT	示例日报
期号	issue_no	TEXT	2024年第05期
出版日期	publish_date	TEXT	2024-05-01
栏目	column_name	TEXT	综合新闻
主题	topic	TEXT	城市建设专题
数字化状态	digitization_status	TEXT	已数字化
链接	url	TEXT UNIQUE	https://example.com/detail/001.html
内容 hash	content_hash	TEXT	md5 摘要
创建时间	created_at	TEXT	2024-05-20 12:30:00
更新时间	updated_at	TEXT	2024-05-20 12:30:00

8.3 去重策略

去重可以分两层：

第一层：URL 唯一。

url TEXT UNIQUE

第二层：内容 hash。

content_hash = md5(
newspaper_name + issue_no + publish_date + column_name + topic + digitization_status
)

一般来说，详情页 URL 足够作为唯一键。如果站点 URL 不稳定，比如同一篇内容会有多个镜像链接，就可以再加内容 hash。

9️⃣ 完整可运行代码

下面给出一个单文件版本。它会自动生成本地 HTML 样例，然后模拟“日历页 → 历史分页 → 详情页”的采集流程。你复制后可以直接运行。

保存为：

main.py

运行：

python main.py

代码如下：

import csv
import hashlib
import logging
import random
import re
import sqlite3
import time
from dataclasses import dataclass, asdict
from datetime import datetime
from pathlib import Path
from typing import Dict, Iterable, List, Optional
from urllib.parse import urljoin

import pandas as pd
import requests
from bs4 import BeautifulSoup
from dateutil import parser as date_parser

BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = BASE_DIR / "data"
FIXTURE_DIR = BASE_DIR / "fixtures"

DB_PATH = DATA_DIR / "newspaper_issues.db"
CSV_PATH = DATA_DIR / "newspaper_issues.csv"

BASE_URL = "https://example.com"
DEFAULT_NEWSPAPER_NAME = "示例日报"

USE_FIXTURE = True

REQUEST_TIMEOUT = (5, 20)
MIN_DELAY = 0.8
MAX_DELAY = 1.6
MAX_RETRIES = 3

logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(message)s",
)

@dataclass
class IssueRecord:
newspaper_name: str
issue_no: str
publish_date: str
column_name: str
topic: str
digitization_status: str
url: str
content_hash: str = ""

def fill_hash(self) –> None:
raw = "|".join([
self.newspaper_name or "",
self.issue_no or "",
self.publish_date or "",
self.column_name or "",
self.topic or "",
self.digitization_status or "",
self.url or "",
])
self.content_hash = hashlib.md5(raw.encode("utf-8")).hexdigest()

def ensure_dirs() –> None:
DATA_DIR.mkdir(exist_ok=True)
FIXTURE_DIR.mkdir(exist_ok=True)

def write_fixture_files() –> None:
"""
生成本地样例页面。

真实项目中，这部分不需要。
这里只是为了让示例代码不依赖任何外部网站，也能完整跑通：
1. 日历页
2. 期号分页第 1 页
3. 期号分页第 2 页
4. 三个详情页
"""
calendar_html = """
<!doctype html>
<html>
<head><meta charset="utf-8"><title>2024年05月历史日历</title></head>
<body>
<div id="calendar" data-year="2024" data-month="05">
<a class="issue-day" data-date="2024-05-01" href="/archive/2024/05/01/page/1.html">2024-05-01</a>
<a class="issue-day" data-date="2024-05-08" href="/archive/2024/05/08/page/1.html">2024-05-08</a>
</div>
</body>
</html>
"""

issue_page_1 = """
<!doctype html>
<html>
<head><meta charset="utf-8"><title>2024-05-01 第1页</title></head>
<body>
<div class="issue-head">
<span class="newspaper">示例日报</span>
<span class="issue-no">2024年第05期</span>
<span class="publish-date">2024-05-01</span>
</div>

<div class="issue-item">
<a class="detail-link" href="/detail/002.html">老照片里的街区记忆</a>
<span class="column">文化副刊</span>
<span class="status">仅目录</span>
</div>
</div>

<div class="pagination">
<a class="next" href="/archive/2024/05/01/page/2.html">下一页</a>
</div>
</body>
</html>
"""

issue_page_2 = """
<!doctype html>
<html>
<head><meta charset="utf-8"><title>2024-05-01 第2页</title></head>
<body>
<div class="issue-head">
<span class="newspaper">示例日报</span>
<span class="issue-no">2024年第05期</span>
<span class="publish-date">2024-05-01</span>
</div>

<div class="issue-list">
<div class="issue-item">
<a class="detail-link" href="/detail/003.html">历史版面数字化说明</a>
<span class="column">资料索引</span>
<span class="status">待数字化</span>
</div>
</div>

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

detail_001 = """
<!doctype html>
<html>
<head><meta charset="utf-8"><title>城市建设专题</title></head>
<body>
<h1 class="topic">城市建设专题</h1>
<ul class="meta">
<li><span>报刊名：</span><b>示例日报</b></li>
<li><span>期号：</span><b>2024年第05期</b></li>
<li><span>出版日期：</span><b>2024-05-01</b></li>
<li><span>栏目：</span><b>综合新闻</b></li>
<li><span>数字化状态：</span><b>已数字化</b></li>
</ul>
</body>
</html>
"""

detail_002 = """
<!doctype html>
<html>
<head><meta charset="utf-8"><title>老照片里的街区记忆</title></head>
<body>
<h1 class="topic">老照片里的街区记忆</h1>
<ul class="meta">
<li><span>报刊名：</span><b>示例日报</b></li>
<li><span>期号：</span><b>2024年第05期</b></li>
<li><span>出版日期：</span><b>2024年5月1日</b></li>
<li><span>栏目：</span><b>文化副刊</b></li>
<li><span>数字化状态：</span><b>仅目录</b></li>
</ul>
</body>
</html>
"""

detail_003 = """
<!doctype html>
<html>
<head><meta charset="utf-8"><title>历史版面数字化说明</title></head>
<body>
<h1 class="topic">历史版面数字化说明</h1>
<ul class="meta">
<li><span>报刊名：</span><b>示例日报</b></li>
<li><span>期号：</span><b>2024年第05期</b></li>
<li><span>出版日期：</span><b>2024-05-01</b></li>
<li><span>栏目：</span><b>资料索引</b></li>
<li><span>数字化状态：</span><b>待数字化</b></li>
</ul>
</body>
</html>
"""

files = {
"calendar_2024_05.html": calendar_html,
"issue_2024_05_01_page_1.html": issue_page_1,
"issue_2024_05_01_page_2.html": issue_page_2,
"detail_001.html": detail_001,
"detail_002.html": detail_002,
"detail_003.html": detail_003,
}

for filename, html in files.items():
path = FIXTURE_DIR / filename
path.write_text(html.strip(), encoding="utf-8")

class Fetcher:
"""
请求层。

USE_FIXTURE=True 时，从本地 fixtures 目录读取 HTML。
USE_FIXTURE=False 时，使用 requests 请求真实站点。

这样设计的好处：
1. 本地可以稳定复现实验。
2. 改真实站点时，只需要替换 URL 映射和选择器。
"""

def __init__(self) –> None:
self.session = requests.Session()
self.session.headers.update({
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0 Safari/537.36"
),
"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.8",
"Connection": "keep-alive",
})

def fetch(self, url: str, referer: Optional[str] = None) –> str:
if USE_FIXTURE:
return self.fetch_fixture(url)

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

last_error = None

for attempt in range(1, MAX_RETRIES + 1):
try:
self.polite_sleep()
response = self.session.get(
url,
headers=headers,
timeout=REQUEST_TIMEOUT,
)

if response.status_code == 404:
logging.warning("404 not found: %s", url)
return ""

if response.status_code == 429:
wait_seconds = 2 ** attempt * 3
logging.warning("429 too many requests, sleep %s seconds: %s", wait_seconds, url)
time.sleep(wait_seconds)
continue

if response.status_code in (500, 502, 503, 504):
wait_seconds = 2 ** attempt
logging.warning("server error %s, retry after %s seconds: %s",
response.status_code, wait_seconds, url)
time.sleep(wait_seconds)
continue

response.raise_for_status()

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

return response.text

except requests.RequestException as exc:
last_error = exc
wait_seconds = 2 ** attempt
logging.warning("request failed attempt=%s url=%s error=%s", attempt, url, exc)
time.sleep(wait_seconds)

logging.error("request finally failed url=%s error=%s", url, last_error)
return ""

def fetch_fixture(self, url: str) –> str:
"""
将示例 URL 映射到本地 HTML 文件。

示例：
https://example.com/archive/2024/05/calendar.html
-> fixtures/calendar_2024_05.html
"""
mapping = {
"/archive/2024/05/calendar.html": "calendar_2024_05.html",
"/archive/2024/05/01/page/1.html": "issue_2024_05_01_page_1.html",
"/archive/2024/05/01/page/2.html": "issue_2024_05_01_page_2.html",
"/detail/001.html": "detail_001.html",
"/detail/002.html": "detail_002.html",
"/detail/003.html": "detail_003.html",
}

path_part = url.replace(BASE_URL, "")
filename = mapping.get(path_part)

if not filename:
logging.warning("fixture mapping not found: %s", url)
return ""

path = FIXTURE_DIR / filename
if not path.exists():
logging.warning("fixture file not found: %s", path)
return ""

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

@staticmethod
def polite_sleep() –> None:
time.sleep(random.uniform(MIN_DELAY, MAX_DELAY))

class Parser:
"""
解析层。

这里的选择器是针对示例 HTML 写的。
真实项目中，你需要根据目标站点页面结构修改选择器。
"""

@staticmethod
def soup(html: str) –> BeautifulSoup:
return BeautifulSoup(html, "lxml")

def parse_calendar(self, html: str) –> List[str]:
"""
解析某年某月日历页面，返回该月所有期号入口 URL。
"""
if not html:
return []

soup = self.soup(html)
urls = []

for a in soup.select("a.issue-day"):
href = a.get("href", "").strip()
if not href:
continue
urls.append(urljoin(BASE_URL, href))

return list(dict.fromkeys(urls))

def parse_issue_list(self, html: str, page_url: str) –> Dict[str, object]:
"""
解析某一期的列表页。

返回：
{
"page_context": {…},
"items": […],
"next_url": "…"
}
"""
result = {
"page_context": {},
"items": [],
"next_url": "",
}

if not html:
return result

soup = self.soup(html)

page_context = {
"newspaper_name": self.text_or_empty(soup.select_one(".issue-head .newspaper")),
"issue_no": self.text_or_empty(soup.select_one(".issue-head .issue-no")),
"publish_date": self.text_or_empty(soup.select_one(".issue-head .publish-date")),
}
result["page_context"] = page_context

items = []
for node in soup.select(".issue-list .issue-item"):
link_node = node.select_one("a.detail-link")
if not link_node:
continue

href = link_node.get("href", "").strip()
if not href:
continue

item = {
"topic": self.text_or_empty(link_node),
"url": urljoin(BASE_URL, href),
"column_name": self.text_or_empty(node.select_one(".column")),
"digitization_status": self.text_or_empty(node.select_one(".status")),
}
items.append(item)

result["items"] = items

next_node = soup.select_one(".pagination a.next")
if next_node and next_node.get("href"):
result["next_url"] = urljoin(BASE_URL, next_node.get("href").strip())

return result

def parse_detail(self, html: str, detail_url: str) –> Dict[str, str]:
"""
解析详情页字段。
"""
if not html:
return {}

soup = self.soup(html)

data = {
"newspaper_name": "",
"issue_no": "",
"publish_date": "",
"column_name": "",
"topic": self.text_or_empty(soup.select_one("h1.topic")),
"digitization_status": "",
"url": detail_url,
}

label_mapping = {
"报刊名": "newspaper_name",
"期号": "issue_no",
"出版日期": "publish_date",
"栏目": "column_name",
"数字化状态": "digitization_status",
}

for li in soup.select("ul.meta li"):
full_text = self.normalize_space(li.get_text(" ", strip=True))
if not full_text:
continue

key_text = ""
value_text = ""

span = li.select_one("span")
b = li.select_one("b")

if span:
key_text = span.get_text(strip=True).replace("：", "").replace(":", "").strip()

if b:
value_text = b.get_text(strip=True)
else:
value_text = re.sub(r"^[^:：]+[:：]\\s*", "", full_text).strip()

field = label_mapping.get(key_text)
if field:
data[field] = value_text

return data

@staticmethod
def text_or_empty(node) –> str:
if not node:
return ""
return Parser.normalize_space(node.get_text(" ", strip=True))

@staticmethod
def normalize_space(text: str) –> str:
return re.sub(r"\\s+", " ", text or "").strip()

class Cleaner:
"""
清洗层。
"""

@staticmethod
def clean_record(raw: Dict[str, str], fallback: Dict[str, str]) –> Optional[IssueRecord]:
url = raw.get("url") or fallback.get("url") or ""
if not url:
return None

newspaper_name = raw.get("newspaper_name") or fallback.get("newspaper_name") or DEFAULT_NEWSPAPER_NAME
issue_no = raw.get("issue_no") or fallback.get("issue_no") or ""
publish_date = raw.get("publish_date") or fallback.get("publish_date") or ""
column_name = raw.get("column_name") or fallback.get("column_name") or ""
topic = raw.get("topic") or fallback.get("topic") or ""
digitization_status = raw.get("digitization_status") or fallback.get("digitization_status") or "未知"

publish_date = Cleaner.normalize_date(publish_date)
digitization_status = Cleaner.normalize_status(digitization_status)

record = IssueRecord(
newspaper_name=newspaper_name.strip(),
issue_no=issue_no.strip(),
publish_date=publish_date,
column_name=column_name.strip(),
topic=topic.strip(),
digitization_status=digitization_status.strip(),
url=url.strip(),
)
record.fill_hash()
return record

@staticmethod
def normalize_date(value: str) –> str:
value = (value or "").strip()
if not value:
return ""

value = value.replace("年", "-").replace("月", "-").replace("日", "")
value = re.sub(r"[./]", "-", value)
value = re.sub(r"\\s+", "", value)

try:
dt = date_parser.parse(value)
return dt.strftime("%Y-%m-%d")
except Exception:
logging.warning("date parse failed: %s", value)
return value

@staticmethod
def normalize_status(value: str) –> str:
value = (value or "").strip()
if not value:
return "未知"

if "已" in value and "数字" in value:
return "已数字化"
if "待" in value and "数字" in value:
return "待数字化"
if "未" in value and "数字" in value:
return "未数字化"
if "目录" in value:
return "仅目录"
return value

class Storage:
"""
SQLite 存储层。
"""

def __init__(self, db_path: Path) –> None:
self.db_path = db_path
self.conn = sqlite3.connect(str(db_path))
self.conn.row_factory = sqlite3.Row

def init_db(self) –> None:
sql = """
CREATE TABLE IF NOT EXISTS newspaper_issues (
id INTEGER PRIMARY KEY AUTOINCREMENT,
newspaper_name TEXT NOT NULL,
issue_no TEXT,
publish_date TEXT,
column_name TEXT,
topic TEXT,
digitization_status TEXT,
url TEXT NOT NULL UNIQUE,
content_hash TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
"""
self.conn.execute(sql)
self.conn.execute("CREATE INDEX IF NOT EXISTS idx_publish_date ON newspaper_issues(publish_date);")
self.conn.execute("CREATE INDEX IF NOT EXISTS idx_issue_no ON newspaper_issues(issue_no);")
self.conn.execute("CREATE INDEX IF NOT EXISTS idx_column_name ON newspaper_issues(column_name);")
self.conn.commit()

def upsert(self, record: IssueRecord) –> None:
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

sql = """
INSERT INTO newspaper_issues (
newspaper_name,
issue_no,
publish_date,
column_name,
topic,
digitization_status,
url,
content_hash,
created_at,
updated_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(url) DO UPDATE SET
newspaper_name=excluded.newspaper_name,
issue_no=excluded.issue_no,
publish_date=excluded.publish_date,
column_name=excluded.column_name,
topic=excluded.topic,
digitization_status=excluded.digitization_status,
content_hash=excluded.content_hash,
updated_at=excluded.updated_at;
"""

self.conn.execute(sql, (
record.newspaper_name,
record.issue_no,
record.publish_date,
record.column_name,
record.topic,
record.digitization_status,
record.url,
record.content_hash,
now,
now,
))
self.conn.commit()

def export_csv(self, csv_path: Path) –> None:
query = """
SELECT
newspaper_name AS 报刊名,
issue_no AS 期号,
publish_date AS 出版日期,
column_name AS 栏目,
topic AS 主题,
digitization_status AS 数字化状态,
url AS 链接
FROM newspaper_issues
ORDER BY publish_date, issue_no, id;
"""
df = pd.read_sql_query(query, self.conn)
df.to_csv(csv_path, index=False, encoding="utf-8-sig")

def preview(self, limit: int = 5) –> List[sqlite3.Row]:
sql = """
SELECT newspaper_name, issue_no, publish_date, column_name, topic, digitization_status, url
FROM newspaper_issues
ORDER BY id
LIMIT ?;
"""
return list(self.conn.execute(sql, (limit,)))

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

class NewspaperIssueCrawler:
"""
爬虫调度层。
"""

def __init__(self) –> None:
self.fetcher = Fetcher()
self.parser = Parser()
self.cleaner = Cleaner()
self.storage = Storage(DB_PATH)

def run(self) –> None:
self.storage.init_db()

years_months = [
(2024, 5),
]

total_saved = 0

for year, month in years_months:
calendar_url = self.build_calendar_url(year, month)
logging.info("fetch calendar: %s", calendar_url)
calendar_html = self.fetcher.fetch(calendar_url)

issue_entry_urls = self.parser.parse_calendar(calendar_html)
logging.info("calendar parsed issue entries: %s", len(issue_entry_urls))

for issue_entry_url in issue_entry_urls:
saved = self.crawl_issue_pages(issue_entry_url)
total_saved += saved

self.storage.export_csv(CSV_PATH)
logging.info("done, saved=%s, db=%s, csv=%s", total_saved, DB_PATH, CSV_PATH)

print("\\n采集结果预览：")
rows = self.storage.preview(5)
for row in rows:
print(dict(row))

self.storage.close()

def crawl_issue_pages(self, first_page_url: str) –> int:
current_url = first_page_url
visited_pages = set()
saved_count = 0

while current_url:
if current_url in visited_pages:
logging.warning("page already visited, break loop: %s", current_url)
break

visited_pages.add(current_url)

logging.info("fetch issue list page: %s", current_url)
html = self.fetcher.fetch(current_url)
parsed = self.parser.parse_issue_list(html, current_url)

page_context = parsed.get("page_context", {})
items = parsed.get("items", [])
next_url = parsed.get("next_url", "")

logging.info("items found: %s next=%s", len(items), next_url)

for item in items:
detail_url = item.get("url", "")
if not detail_url:
continue

logging.info("fetch detail: %s", detail_url)
detail_html = self.fetcher.fetch(detail_url, referer=current_url)
detail_data = self.parser.parse_detail(detail_html, detail_url)

fallback = {}
fallback.update(page_context)
fallback.update(item)

record = self.cleaner.clean_record(detail_data, fallback)
if not record:
logging.warning("record cleaned empty: %s", detail_url)
continue

self.storage.upsert(record)
saved_count += 1

current_url = next_url

return saved_count

@staticmethod
def build_calendar_url(year: int, month: int) –> str:
return f"{BASE_URL}/archive/{year}/{month:02d}/calendar.html"

def main() –> None:
ensure_dirs()
write_fixture_files()
crawler = NewspaperIssueCrawler()
crawler.run()

if __name__ == "__main__":
main()

9.1 运行结果

运行：

python main.py

你会看到类似输出：

输出文件：

data/newspaper_issues.db
data/newspaper_issues.csv

CSV 示例：

报刊名期号出版日期栏目主题数字化状态链接

示例日报	2024年第05期	2024-05-01	综合新闻	城市建设专题	已数字化	https://example.com/detail/001.html
示例日报	2024年第05期	2024-05-01	文化副刊	老照片里的街区记忆	仅目录	https://example.com/detail/002.html
示例日报	2024年第05期	2024-05-01	资料索引	历史版面数字化说明	待数字化	https://example.com/detail/003.html

10. 真实站点改造方法

上面的代码默认使用本地样例页面。如果要改成真实站点，需要做几处调整。

10.1 关闭样例模式

USE_FIXTURE = False
BASE_URL = "https://目标站点域名"

10.2 修改日历 URL 规则

示例代码：

@staticmethod
def build_calendar_url(year: int, month: int) –> str:
return f"{BASE_URL}/archive/{year}/{month:02d}/calendar.html"

真实站点可能是：

return f"{BASE_URL}/paper/history?year={year}&month={month:02d}"

也可能是：

return f"{BASE_URL}/index.php?m=archive&y={year}&mth={month:02d}"

不要猜。用浏览器打开目标站点，切换年月，看地址栏和 Network 面板。

10.3 修改 CSS 选择器

示例日历选择器：

soup.select("a.issue-day")

真实站点可能是：

soup.select(".calendar a")
soup.select("td.has-paper a")
soup.select("div.date-list a")

列表页详情链接可能是：

soup.select(".issue-list .issue-item a.detail-link")

真实站点可能是：

soup.select("ul.news-list li a")
soup.select("table.archive-table tr td.title a")
soup.select(".list-con h3 a")

详情页字段也要按真实页面改：

soup.select_one("h1.topic")
soup.select("ul.meta li")

重点不是记住某个选择器，而是把字段抽取逻辑写在 Parser 里，后续只改 Parser，不动 Fetcher 和 Storage。

11. 年月期号级联解析

报刊历史平台最容易踩坑的地方就是层级。很多初学者会直接从首页开始找所有链接，然后一层一层爬。这种方式看似简单，实际很容易失控。

更好的方式是显式建立层级。

11.1 年份层

年份层负责确定采集范围。

years = range(2020, 2025)

真实项目中，我会把年份写进配置文件：

CRAWL_YEARS = [2021, 2022, 2023, 2024]

如果站点有年份列表页，可以解析它：

def parse_years(html: str) –> List[int]:
soup = BeautifulSoup(html, "lxml")
years = []
for a in soup.select(".year-list a"):
text = a.get_text(strip=True)
match = re.search(r"(19|20)\\d{2}", text)
if match:
years.append(int(match.group()))
return sorted(set(years))

11.2 月份层

月份通常是 1 到 12，但不是每个月都有内容。可以先构造，再请求：

for year in years:
for month in range(1, 13):
url = build_calendar_url(year, month)

也可以从页面解析可用月份：

def parse_months(html: str) –> List[int]:
soup = BeautifulSoup(html, "lxml")
months = []
for a in soup.select(".month-list a"):
text = a.get_text(strip=True)
match = re.search(r"(\\d{1,2})月", text)
if match:
months.append(int(match.group(1)))
return sorted(set(months))

11.3 日期层

报刊不一定每天出版。日报通常日期多，周报、月刊、专刊则日期少。日历页的价值就在这里：它告诉我们哪些日期有期号。

示例：

解析：

def parse_issue_days(html: str) –> List[dict]:
soup = BeautifulSoup(html, "lxml")
result = []
for a in soup.select("a.issue-day"):
href = a.get("href", "").strip()
date = a.get("data-date", "").strip()
if href:
result.append({
"date": date,
"url": urljoin(BASE_URL, href),
})
return result

11.4 期号层

有些平台的某一天只有一期，有些平台某一天可能有多个版面或增刊。期号层最好不要写死。

可以从页面中提取：

issue_no = soup.select_one(".issue-no").get_text(strip=True)

也可以从标题中提取：

title = soup.select_one("title").get_text(strip=True)
match = re.search(r"(\\d{4}年第\\d+期|总第\\d+期)", title)

11.5 分页层

历史分页一般有几种形式。

第一种：下一页按钮。

第二种：页码列表。

第三种：接口分页。

/api/archive?year=2024&month=05&page=1
/api/archive?year=2024&month=05&page=2

示例代码中采用“下一页按钮”方式。优点是不用提前知道总页数，解析到没有下一页为止。

while current_url:
html = fetcher.fetch(current_url)
parsed = parser.parse_issue_list(html, current_url)
current_url = parsed["next_url"]

为了避免死循环，需要加访问集合：

visited_pages = set()

if current_url in visited_pages:
break

visited_pages.add(current_url)

这点很重要。有些老站点的“下一页”会错误地指向当前页，如果没有防护，程序会一直跑。

12. 动态日历解析

12.1 什么是动态日历

动态日历通常不是直接写在 HTML 中，而是由 JavaScript 请求接口后渲染。例如页面源代码里只有：

浏览器实际看到的日期，是 JS 运行后生成的。

遇到这种情况，不要急着上 Playwright。先打开浏览器开发者工具，切到 Network 面板，看有没有类似接口：

/api/calendar?year=2024&month=05
/api/issues/dates?y=2024&m=05
/archive/calendar.json?year=2024&month=05

如果能找到接口，优先请求接口，因为它更轻、更稳定。

12.2 JSON 日历解析示例

假设接口返回：

{
"year": 2024,
"month": 5,
"days": [
{
"date": "2024-05-01",
"url": "/archive/2024/05/01/page/1.html"
},
{
"date": "2024-05-08",
"url": "/archive/2024/05/08/page/1.html"
}
]
}

可以这样解析：

def parse_calendar_json(data: dict) –> List[str]:
urls = []
for item in data.get("days", []):
url = item.get("url", "").strip()
if url:
urls.append(urljoin(BASE_URL, url))
return urls

请求接口：

def fetch_json(session: requests.Session, url: str) –> dict:
response = session.get(url, timeout=(5, 20))
response.raise_for_status()
return response.json()

12.3 只有渲染结果，没有接口怎么办

如果页面是纯前端渲染，接口又不好拆，可以考虑 Playwright。

示例：

from playwright.sync_api import sync_playwright

def render_html_with_playwright(url: str, wait_selector: str = ".issue-day") –> str:
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page(
user_agent=(
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0 Safari/537.36"
)
)
page.goto(url, wait_until="networkidle", timeout=30000)
page.wait_for_selector(wait_selector, timeout=10000)
html = page.content()
browser.close()
return html

但 Playwright 不应该被滥用。它启动浏览器，资源消耗比 requests 高很多。如果接口能解决问题，尽量不要用浏览器渲染。

13. 历史分页解析细节

13.1 从第一页顺着下一页走

这种方式最稳。

def crawl_by_next(first_url: str):
current_url = first_url
visited = set()

while current_url:
if current_url in visited:
break

visited.add(current_url)
html = fetcher.fetch(current_url)
parsed = parser.parse_issue_list(html, current_url)

for item in parsed["items"]:
yield item

current_url = parsed["next_url"]

优点：

不需要知道总页数
页面结构稍微变化也能适配
不会请求不存在的 page=999

13.2 根据总页数构造 URL

如果页面提供总页数：

可以解析后构造：

def parse_total_pages(html: str) –> int:
soup = BeautifulSoup(html, "lxml")
text = soup.select_one(".total-page").get_text(strip=True)
match = re.search(r"共\\s*(\\d+)\\s*页", text)
return int(match.group(1)) if match else 1

然后：

for page in range(1, total_pages + 1):
url = f"{base_url}?page={page}"

这种方式速度略快，但对分页规则依赖更强。

13.3 空页处理

老站点常见空页。比如总页数显示 10，但第 10 页没有数据。处理方式：

if not items:
logging.warning("empty list page: %s", current_url)

如果连续多个空页，可以停止：

empty_count += 1
if empty_count >= 3:
break

但在本文这种“下一页按钮”模式下，一般不需要连续空页判断。

14. 数据清洗策略

14.1 日期清洗

历史报刊站点的日期格式非常混乱：

2024-05-01
2024/05/01
2024.05.01
2024年5月1日
二〇二四年五月一日

本文示例处理前三类和常见中文年月日：

value = value.replace("年", "-").replace("月", "-").replace("日", "")
dt = date_parser.parse(value)

如果遇到中文数字日期，比如“二〇二四年五月一日”，可以额外写映射函数。但我建议先看目标站点是否真的出现这种格式，不要提前把代码写得过度复杂。

14.2 数字化状态清洗

数字化状态可能有很多写法：

已数字化
已完成数字化
未数字化
待数字化
仅目录
目录已开放
暂无全文
缺页

可以先归一成几个核心状态：

已数字化
未数字化
待数字化
仅目录
未知

示例：

def normalize_status(value: str) –> str:
if "已" in value and "数字" in value:
return "已数字化"
if "待" in value and "数字" in value:
return "待数字化"
if "未" in value and "数字" in value:
return "未数字化"
if "目录" in value:
return "仅目录"
return value or "未知"

后续如果分析需要，再细分状态也不迟。

14.3 URL 补全

详情链接经常是相对路径：

<a href="/detail/001.html">城市建设专题</a>

必须补全：

urljoin(BASE_URL, href)

不要手动字符串拼接：

BASE_URL + href

因为遇到 ../detail/001.html 或 detail/001.html 时，手动拼接很容易出错。

15. 运行方式与结果展示

15.1 启动命令

安装依赖：

pip install requests beautifulsoup4 lxml pandas python-dateutil

运行：

python main.py

15.2 输出位置

SQLite 数据库：
data/newspaper_issues.db

CSV 文件：
data/newspaper_issues.csv

15.3 查看 SQLite 数据

可以用 DB Browser for SQLite 打开，也可以用命令行：

sqlite3 data/newspaper_issues.db

进入后执行：

SELECT newspaper_name, issue_no, publish_date, column_name, topic, digitization_status, url
FROM newspaper_issues
LIMIT 5;

15.4 示例结果

报刊名期号出版日期栏目主题数字化状态
示例日报 2024年第05期 2024-05-01 综合新闻城市建设专题已数字化
示例日报 2024年第05期 2024-05-01 文化副刊老照片里的街区记忆仅目录
示例日报 2024年第05期 2024-05-01 资料索引历史版面数字化说明待数字化

CSV 打开后字段更直观：

报刊名,期号,出版日期,栏目,主题,数字化状态,链接
示例日报,2024年第05期,2024-05-01,综合新闻,城市建设专题,已数字化,https://example.com/detail/001.html
示例日报,2024年第05期,2024-05-01,文化副刊,老照片里的街区记忆,仅目录,https://example.com/detail/002.html
示例日报,2024年第05期,2024-05-01,资料索引,历史版面数字化说明,待数字化,https://example.com/detail/003.html

16. 常见问题与排错

16.1 403 怎么办

403 表示访问被拒绝。原因可能很多：

缺少 User-Agent
缺少 Referer
请求频率过高
访问了不允许抓取的路径
站点有访问策略

处理建议：

先检查 URL 是否能在浏览器中直接打开。

检查 robots.txt 和站点说明。

增加常规 headers。

降低请求频率。

不要尝试绕过登录或付费限制。

代码里可以加日志：

if response.status_code == 403:
logging.warning("403 forbidden, check access policy and headers: %s", url)

16.2 429 怎么办

429 通常表示请求太频繁。处理方式：

降低频率
增加随机 sleep
减少并发
使用指数退避
分批采集

示例：

if response.status_code == 429:
time.sleep(30)
continue

我个人更喜欢保守一些。历史索引不是实时行情，不值得为了快几分钟把站点打出压力。

16.3 HTML 抓到空壳怎么办

如果 requests 抓到的 HTML 里没有数据，浏览器里却能看到内容，通常说明页面由 JavaScript 渲染。

排查步骤：

右键查看网页源代码，看是否有目标字段。

打开开发者工具 Network。

切换年月或翻页。

找 XHR 或 Fetch 请求。

优先请求接口。

找不到接口再考虑 Playwright。

典型空壳：

这种页面用 bs4 直接解析通常拿不到数据。

16.4 解析报错怎么办

常见错误：

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

原因是选择器没匹配到节点。

错误写法：

title = soup.select_one("h1.topic").get_text(strip=True)

稳妥写法：

node = soup.select_one("h1.topic")
title = node.get_text(strip=True) if node else ""

或者封装：

def text_or_empty(node):
return node.get_text(" ", strip=True) if node else ""

16.5 编码乱码怎么办

有些老站点不是 UTF-8，而是 GBK 或 GB2312。表现为：

åŸŽå¸‚å»ºè®¾

处理方式：

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

如果知道站点编码，也可以强制：

response.encoding = "gb18030"

CSV 建议用：

encoding="utf-8-sig"

这样 Excel 打开中文更稳。

16.6 日期解析失败怎么办

遇到奇怪日期：

民国三十六年五月一日
二〇二四年五月一日
2024年第五期

不要硬猜。可以先保留原值，同时记录日志：

logging.warning("date parse failed: %s", value)
return value

如果后续确实要分析，再单独写日期标准化脚本。

16.7 重复数据太多怎么办

检查唯一键是否合理。

如果 URL 唯一：

url TEXT UNIQUE

如果 URL 不唯一，可以建立联合唯一键：

UNIQUE(newspaper_name, issue_no, publish_date, column_name, topic)

但联合唯一键要小心。标题相同、栏目相同、日期相同的情况并不是完全不可能。

16.8 下一页死循环怎么办

有些站点最后一页的“下一页”仍然指向当前页。解决方式：

visited_pages = set()

if current_url in visited_pages:
break

visited_pages.add(current_url)

这行代码看起来普通，但在真实项目里非常救命。

17. 进阶优化

17.1 并发采集

当字段规则稳定后，可以考虑并发。但建议从详情页并发开始，不要一开始就并发所有层级。

例如：

from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch_detail_task(detail_url: str):
html = fetcher.fetch(detail_url)
return parser.parse_detail(html, detail_url)

with ThreadPoolExecutor(max_workers=3) as executor:
futures = [executor.submit(fetch_detail_task, url) for url in detail_urls]

for future in as_completed(futures):
data = future.result()

max_workers=3 已经足够温和。除非你得到站点授权或明确接口限额，否则不建议开太高。

17.2 asyncio

如果请求量较大，可以用 aiohttp。不过异步爬虫的错误处理、限速、重试会更复杂。不要为了“看起来高级”而引入异步。

适合 asyncio 的场景：

详情页数量大
接口响应稳定
有明确限速规则
需要提高吞吐

不适合的场景：

站点结构还没摸清
选择器经常变化
反复出现 403/429
需要大量人工校验

17.3 Scrapy 改造方向

如果项目变成多站点、多任务、长期运行，可以迁移到 Scrapy。

Scrapy 优势：

请求调度
去重队列
中间件
失败重试
日志体系
导出管道
增量采集

Scrapy 中可以把字段定义为 Item：

import scrapy

class NewspaperIssueItem(scrapy.Item):
newspaper_name = scrapy.Field()
issue_no = scrapy.Field()
publish_date = scrapy.Field()
column_name = scrapy.Field()
topic = scrapy.Field()
digitization_status = scrapy.Field()
url = scrapy.Field()

然后写 Pipeline 存 SQLite 或 MySQL。

17.4 断点续跑

断点续跑非常实用。思路有两种。

第一种：依赖数据库唯一键。已经抓过的 URL 再插入时自动更新。

第二种：维护已抓集合。

def load_done_urls(conn) –> set:
rows = conn.execute("SELECT url FROM newspaper_issues").fetchall()
return {row["url"] for row in rows}

请求详情页前判断：

if detail_url in done_urls:
continue

17.5 游标记录

可以单独建一张任务进度表：

CREATE TABLE IF NOT EXISTS crawl_cursor (
task_name TEXT PRIMARY KEY,
year INTEGER,
month INTEGER,
page_url TEXT,
updated_at TEXT
);

每处理完一个月份就更新游标。程序异常退出后，从最近游标继续。

17.6 日志与监控

至少记录这些指标：

请求成功数
请求失败数
解析成功数
解析失败数
入库数量
重复数量
空字段数量
平均响应时间

简单日志示例：

logging.info("items found: %s", len(items))
logging.warning("field missing topic url=%s", detail_url)
logging.error("request finally failed url=%s", url)

长期任务可以把日志写入文件：

logging.basicConfig(
filename="crawler.log",
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(message)s",
)

17.7 定时任务

Linux cron 示例：

0 2 * * * /usr/bin/python3 /opt/newspaper_index_crawler/main.py >> /opt/newspaper_index_crawler/run.log 2>&1

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

如果任务更复杂，可以用 Airflow、Prefect 或其他工作流工具。但对于单站点每日增量采集，cron 已经够用。

17.8 增量采集

报刊索引很适合增量采集。比如只采集最近 3 个月：

years_months = [
(2024, 3),
(2024, 4),
(2024, 5),
]

或者根据当前日期自动生成：

from datetime import date

def recent_year_months(n: int = 3):
today = date.today()
result = []
year = today.year
month = today.month

for _ in range(n):
result.append((year, month))
month -= 1
if month == 0:
month = 12
year -= 1

return result

18. 工程化拆分建议

单文件代码适合演示，但真实项目建议拆分。

18.1 config.py

from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = BASE_DIR / "data"
DB_PATH = DATA_DIR / "newspaper_issues.db"
CSV_PATH = DATA_DIR / "newspaper_issues.csv"

BASE_URL = "https://example.com"
DEFAULT_NEWSPAPER_NAME = "示例日报"

REQUEST_TIMEOUT = (5, 20)
MIN_DELAY = 1.0
MAX_DELAY = 2.5
MAX_RETRIES = 3

CRAWL_YEARS_MONTHS = [
(2024, 5),
]

18.2 crawler/fetcher.py

import logging
import random
import time
from typing import Optional

import requests

from config import REQUEST_TIMEOUT, MIN_DELAY, MAX_DELAY, MAX_RETRIES

class Fetcher:
def __init__(self) –> None:
self.session = requests.Session()
self.session.headers.update({
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0 Safari/537.36"
),
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
})

def get_text(self, url: str, referer: Optional[str] = None) –> str:
headers = {}
if referer:
headers["Referer"] = referer

for attempt in range(1, MAX_RETRIES + 1):
try:
time.sleep(random.uniform(MIN_DELAY, MAX_DELAY))
response = self.session.get(
url,
headers=headers,
timeout=REQUEST_TIMEOUT,
)

if response.status_code == 404:
logging.warning("404: %s", url)
return ""

if response.status_code == 429:
wait = 2 ** attempt * 3
logging.warning("429, sleep %s seconds: %s", wait, url)
time.sleep(wait)
continue

response.raise_for_status()

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

return response.text

except requests.RequestException as exc:
wait = 2 ** attempt
logging.warning("request error attempt=%s url=%s err=%s", attempt, url, exc)
time.sleep(wait)

return ""

18.3 crawler/parser.py

import re
from typing import Dict, List
from urllib.parse import urljoin

from bs4 import BeautifulSoup

from config import BASE_URL

class Parser:
def soup(self, html: str) –> BeautifulSoup:
return BeautifulSoup(html or "", "lxml")

def parse_calendar(self, html: str) –> List[str]:
soup = self.soup(html)
urls = []
for a in soup.select("a.issue-day"):
href = a.get("href", "").strip()
if href:
urls.append(urljoin(BASE_URL, href))
return list(dict.fromkeys(urls))

def parse_issue_list(self, html: str) –> Dict[str, object]:
soup = self.soup(html)

context = {
"newspaper_name": self.text(soup.select_one(".issue-head .newspaper")),
"issue_no": self.text(soup.select_one(".issue-head .issue-no")),
"publish_date": self.text(soup.select_one(".issue-head .publish-date")),
}

items = []
for item in soup.select(".issue-list .issue-item"):
a = item.select_one("a.detail-link")
if not a:
continue

href = a.get("href", "").strip()
if not href:
continue

items.append({
"topic": self.text(a),
"url": urljoin(BASE_URL, href),
"column_name": self.text(item.select_one(".column")),
"digitization_status": self.text(item.select_one(".status")),
})

next_url = ""
next_node = soup.select_one(".pagination a.next")
if next_node and next_node.get("href"):
next_url = urljoin(BASE_URL, next_node.get("href").strip())

return {
"page_context": context,
"items": items,
"next_url": next_url,
}

def parse_detail(self, html: str, url: str) –> Dict[str, str]:
soup = self.soup(html)
data = {
"newspaper_name": "",
"issue_no": "",
"publish_date": "",
"column_name": "",
"topic": self.text(soup.select_one("h1.topic")),
"digitization_status": "",
"url": url,
}

mapping = {
"报刊名": "newspaper_name",
"期号": "issue_no",
"出版日期": "publish_date",
"栏目": "column_name",
"数字化状态": "digitization_status",
}

for li in soup.select("ul.meta li"):
span = li.select_one("span")
b = li.select_one("b")

key = self.text(span).replace("：", "").replace(":", "")
value = self.text(b)

field = mapping.get(key)
if field:
data[field] = value

return data

@staticmethod
def text(node) –> str:
if not node:
return ""
return re.sub(r"\\s+", " ", node.get_text(" ", strip=True)).strip()

18.4 crawler/storage.py

import sqlite3
from datetime import datetime
from pathlib import Path

import pandas as pd

class Storage:
def __init__(self, db_path: Path):
self.conn = sqlite3.connect(str(db_path))

def init_db(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS newspaper_issues (
id INTEGER PRIMARY KEY AUTOINCREMENT,
newspaper_name TEXT NOT NULL,
issue_no TEXT,
publish_date TEXT,
column_name TEXT,
topic TEXT,
digitization_status TEXT,
url TEXT NOT NULL UNIQUE,
content_hash TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
""")
self.conn.commit()

def upsert_dict(self, item: dict):
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.conn.execute("""
INSERT INTO newspaper_issues (
newspaper_name,
issue_no,
publish_date,
column_name,
topic,
digitization_status,
url,
content_hash,
created_at,
updated_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(url) DO UPDATE SET
newspaper_name=excluded.newspaper_name,
issue_no=excluded.issue_no,
publish_date=excluded.publish_date,
column_name=excluded.column_name,
topic=excluded.topic,
digitization_status=excluded.digitization_status,
content_hash=excluded.content_hash,
updated_at=excluded.updated_at;
""", (
item["newspaper_name"],
item["issue_no"],
item["publish_date"],
item["column_name"],
item["topic"],
item["digitization_status"],
item["url"],
item["content_hash"],
now,
now,
))
self.conn.commit()

def export_csv(self, csv_path: Path):
df = pd.read_sql_query("""
SELECT
newspaper_name AS 报刊名,
issue_no AS 期号,
publish_date AS 出版日期,
column_name AS 栏目,
topic AS 主题,
digitization_status AS 数字化状态,
url AS 链接
FROM newspaper_issues
ORDER BY publish_date, issue_no, id
""", self.conn)
df.to_csv(csv_path, index=False, encoding="utf-8-sig")

def close(self):
self.conn.close()

拆分之后，代码更适合团队维护，也更方便写单元测试。

19. 字段质量检查

爬虫不是跑完就结束。数据质量检查非常关键。

19.1 检查空字段

SELECT COUNT(*) FROM newspaper_issues WHERE topic = '';
SELECT COUNT(*) FROM newspaper_issues WHERE publish_date = '';
SELECT COUNT(*) FROM newspaper_issues WHERE digitization_status = '未知';

如果空字段比例很高，说明选择器可能不稳定。

19.2 检查日期范围

SELECT MIN(publish_date), MAX(publish_date)
FROM newspaper_issues;

如果你本来只抓 2024 年，结果出现 1970 或空日期，就要检查清洗逻辑。

19.3 检查重复主题

SELECT publish_date, topic, COUNT(*) AS cnt
FROM newspaper_issues
GROUP BY publish_date, topic
HAVING cnt > 1
ORDER BY cnt DESC;

重复不一定是错，但值得看一眼。

19.4 检查状态分布

SELECT digitization_status, COUNT(*) AS cnt
FROM newspaper_issues
GROUP BY digitization_status
ORDER BY cnt DESC;

如果全部都是“未知”，大概率是状态字段没有解析到。

20. MySQL 存储扩展示例

如果后续数据量变大，SQLite 可以迁移到 MySQL。表结构可以这样设计：

CREATE TABLE newspaper_issues (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
newspaper_name VARCHAR(255) NOT NULL,
issue_no VARCHAR(100),
publish_date DATE,
column_name VARCHAR(255),
topic VARCHAR(500),
digitization_status VARCHAR(100),
url VARCHAR(1000) NOT NULL,
content_hash CHAR(32),
created_at DATETIME NOT NULL,
updated_at DATETIME NOT NULL,
UNIQUE KEY uk_url (url),
KEY idx_publish_date (publish_date),
KEY idx_issue_no (issue_no),
KEY idx_column_name (column_name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

Python 写入可以使用 pymysql：

pip install pymysql

示例：

import pymysql

conn = pymysql.connect(
host="127.0.0.1",
port=3306,
user="root",
password="your_password",
database="crawler_db",
charset="utf8mb4",
)

with conn.cursor() as cursor:
cursor.execute("""
INSERT INTO newspaper_issues (
newspaper_name,
issue_no,
publish_date,
column_name,
topic,
digitization_status,
url,
content_hash,
created_at,
updated_at
)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
ON DUPLICATE KEY UPDATE
newspaper_name=VALUES(newspaper_name),
issue_no=VALUES(issue_no),
publish_date=VALUES(publish_date),
column_name=VALUES(column_name),
topic=VALUES(topic),
digitization_status=VALUES(digitization_status),
content_hash=VALUES(content_hash),
updated_at=NOW()
""", (
item["newspaper_name"],
item["issue_no"],
item["publish_date"],
item["column_name"],
item["topic"],
item["digitization_status"],
item["url"],
item["content_hash"],
))
conn.commit()

不过对于本文这种索引平台入门项目，SQLite 已经足够。先把采集链路打通，比一开始就搭数据库服务更实际。

21. 选择器稳定性建议

真实站点改版是常态。选择器写得太死，后面维护会很痛苦。

21.1 不建议的写法

soup.select_one("body > div:nth-child(3) > div:nth-child(2) > ul > li:nth-child(1) > a")

这种选择器极其脆弱，页面多一个广告位就会失效。

21.2 推荐写法

优先找语义类名：

soup.select(".issue-list .issue-item")
soup.select("a.detail-link")
soup.select(".publish-date")

如果没有类名，可以结合文本：

for li in soup.select("li"):
text = li.get_text(" ", strip=True)
if "出版日期" in text:
...

21.3 多选择器兜底

可以写一个兜底函数：

def first_text(soup, selectors: list[str]) –> str:
for selector in selectors:
node = soup.select_one(selector)
if node:
text = node.get_text(" ", strip=True)
if text:
return text
return ""

使用：

topic = first_text(soup, [
"h1.topic",
".article-title",
".detail-title",
"h1",
])

这种方式在站点小改版时很有用。

22. 请求失败 URL 的补抓

长期采集时，不要让失败 URL 消失在日志里。可以写入文件：

def append_failed_url(url: str, reason: str):
with open("data/failed_urls.csv", "a", encoding="utf-8-sig", newline="") as f:
writer = csv.writer(f)
writer.writerow([
datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
url,
reason,
])

请求失败时：

append_failed_url(url, "timeout")

后续可以单独补抓：

def load_failed_urls(path: str) –> list[str]:
urls = []
with open(path, "r", encoding="utf-8-sig") as f:
reader = csv.reader(f)
for row in reader:
if len(row) >= 2:
urls.append(row[1])
return urls

这比“失败了就算了”靠谱很多。

23. 面向真实业务的采集边界

报刊历史索引爬虫很容易被误解成“我要把所有内容都搬下来”。但在实际项目里，我更建议把它定位为“索引整理工具”。

合理采集：

公开目录
公开日期
公开期号
公开栏目
公开标题
公开详情链接
公开数字化状态

谨慎处理：

正文全文
图片版面
PDF 文件
需要登录后的内容
需要授权的数据库内容

不建议处理：

账号相关数据
个人敏感信息
后台接口
绕过访问限制的内容

这类边界写清楚，项目会更健康，也更容易长期维护。

24. 总结与延伸阅读

本文完成了一个“报刊历史期号索引平台”的爬虫方案设计和代码实现。我们从需求分析开始，明确了目标字段，然后讨论了合规要求、技术选型、请求层、解析层、存储层、运行方式和常见排错。最后给出了一份可以直接运行的 Python 示例代码，模拟了年月期号级联、动态日历入口、历史分页解析和详情页字段抽取。

这套方案的核心不是某个选择器，也不是某个库，而是结构化思维：

先确定层级
再处理请求
再抽取字段
再清洗入库
最后做增量和监控

下一步可以继续做这些优化：

用 Scrapy 重构项目，支持多站点采集。

接入 Playwright，处理强动态渲染页面。

增加断点续跑和失败补抓。

加入字段质量报告，比如空字段率、状态分布、日期覆盖范围。

做一个简单的检索前端，把 SQLite 数据展示成可搜索页面。

将 CSV 导入分析工具，统计不同年份、不同栏目和不同主题的变化趋势。

我一直觉得，爬虫写到后面比的不是谁能更快地请求页面，而是谁能更稳地把数据变成可以长期使用的资产。报刊历史索引这种场景尤其如此。它看起来朴素，但一旦结构化完成，后续检索、分析、归档都会轻松很多。技术最终还是要落到可维护、可复查、可复用上，这也是这类项目最有价值的地方。

🌟 文末

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

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

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

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

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

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

✅ 互动征集

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

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

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

✅ 免责声明

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

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

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