在线教育平台公开目录树采集实战：动态课程章节、试看状态、时长、讲师与标签解析

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

全文目录：

• • 🌟 开篇语
  • 0️⃣ 前言（Preface）
  • 1️⃣ 摘要（Abstract）
  • 2️⃣ 背景与需求（Why）
  • • 为什么要采集课程目录？
    • 目标字段
  • 3️⃣ 合规与注意事项
  • • 3.1 robots.txt 基本说明
    • 3.2 频率控制
    • 3.3 不采集敏感信息
    • 3.4 不绕过限制
  • 4️⃣ 技术选型与整体流程（What/How）
  • • 4.1 静态、动态与 API 的区别
    • 4.2 整体流程
    • 4.3 为什么选择 requests
    • 4.4 什么时候用 Playwright
  • 5️⃣ 环境准备与依赖安装
  • • 5.1 Python 版本
    • 5.2 安装依赖
    • 5.3 推荐项目结构
  • 6️⃣ 核心实现：请求层（Fetcher）
  • • 6.1 配置文件
    • 6.2 示例 JSON
    • 6.3 配置读取
    • 6.4 工具函数
    • 6.5 Fetcher 实现
  • 7️⃣ 核心实现：解析层（Parser）
  • • 7.1 原始 JSON 结构分析
    • 7.2 数据模型
    • 7.3 解析方式说明
    • 7.4 列表页如何拿详情链接
    • 7.5 详情页如何抽字段
    • 7.6 缺失字段怎么办
  • 8️⃣ 数据存储与导出（Storage）
  • • 8.1 字段映射表
    • 8.2 Storage 实现
    • 8.3 去重策略
  • 9️⃣ 运行方式与结果展示
  • • 9.1 主入口文件
    • 9.2 启动命令
    • 9.3 输出位置
    • 9.4 示例结果
  • 🔟 常见问题与排错
  • • 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 日志与监控
    • 11.4 定时任务
    • 11.5 缓存原始响应
    • 11.6 数据质量检查
    • 11.7 扩展到 Scrapy
    • 11.8 扩展到 Playwright
  • 1️⃣2️⃣ 总结与延伸阅读
  • 🌟 文末
  • • ✅ 专栏持续更新中｜建议收藏 + 订阅
    • ✅ 互动征集
    • ✅ 免责声明

🌟 开篇语

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

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

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

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

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

0️⃣ 前言（Preface）

这篇文章要做的事情很明确：采集一个在线教育平台公开课程目录树，用 requests + JSON 递归解析 + CSV/JSON/SQLite 完成从课程、章节、试看、时长、讲师到标签字段的结构化导出。

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

这篇内容只讨论公开目录数据的技术采集，不涉及登录绕过、付费内容获取、账号接口滥用，也不讨论任何非技术议题。我的态度比较简单：爬虫是一把工具，写得稳、用得克制，才有长期价值。

1️⃣ 摘要（Abstract）

本文以在线教育平台公开课程目录树为采集对象，使用 Python 的 requests 请求公开 JSON 数据，递归解析嵌套 children 节点，并最终将课程、章节、试看、时长、讲师、标签等字段导出到 CSV、JSON 和 SQLite。

读完之后，你将能够：

2️⃣ 背景与需求（Why）

在线教育平台的课程详情页通常会展示一个目录树。例如：

• 第一章：Python 基础

  • 1.1 环境安装
  • 1.2 变量与数据类型

• 第二章：网络请求

  • 2.1 HTTP 基础
  • 2.2 requests 入门
  • 2.3 反爬与频控说明

• 第三章：数据解析

  • 3.1 JSON 解析
  • 3.2 XPath 解析
  • 3.3 递归处理 children

这种目录结构看起来像普通列表，实际很多平台会通过前端框架动态渲染。你在浏览器里看到的是完整课程目录，但直接用 requests.get(url).text 拿回来的 HTML 里可能没有章节数据，只有一个空的 <div id="app"></div>。

这类页面的真实数据经常来自公开接口，例如：

/course/detail?id=1001
/api/course/catalog?course_id=1001
/api/course/tree/1001

前端拿到 JSON 后再渲染成目录树。爬虫的核心任务不是“模拟点击页面”，而是找到公开数据来源，并对返回结构进行稳定解析。

为什么要采集课程目录？

常见目的有三类。

第一类是数据分析。比如分析不同课程的章节数量、试看比例、平均课时、讲师覆盖范围、标签分布等。

第二类是信息聚合。比如把多个公开课程目录汇总到一个内部知识库，便于学习路线规划。

第三类是自动化。比如每天检查公开目录是否新增章节，或者课程标题、讲师、试看状态是否变化。

目标字段

本文目标字段如下：

这里特别需要注意的是 parent_id、level 和 sort_path。目录树不是一张扁平表，原始数据通常是嵌套结构。为了后续分析方便，我们会把树形结构展开成扁平记录，同时保留层级关系。

3️⃣ 合规与注意事项

正式写代码前，必须先把边界说清楚。爬虫不是“能抓就抓”，而是应该先判断数据是否公开、使用方式是否合理、频率是否克制。

3.1 robots.txt 基本说明

很多网站会提供 robots.txt 文件，用来说明搜索引擎或自动化程序哪些路径可以访问，哪些路径不希望被访问。它一般位于网站根路径：

https://example.com/robots.txt

在正式采集前，建议先检查目标站点的 robots 规则，尤其关注：

User-agent: *
Disallow: /private/
Disallow: /user/
Disallow: /api/order/
Allow: /course/

如果规则明确不允许访问某些路径，就不要采集这些路径。即便某些接口能直接访问，也不代表适合被批量抓取。

3.2 频率控制

技术上可以开几十个线程、几百个协程，但这并不意味着应该这么做。对公开课程目录这种低频变化的数据，完全没必要高并发。

建议策略：

• 单站点请求间隔设置为 1–3 秒。
• 对失败请求使用指数退避。
• 对 429、503 等状态码主动降低频率。
• 不做攻击式并发。
• 不在短时间内重复请求同一 URL。

本文示例项目默认是串行请求。后面进阶部分会讲并发，但也会强调速率限制。

3.3 不采集敏感信息

本文只采集公开课程目录字段，例如课程名称、章节标题、试看状态、时长、讲师、标签。不要采集：

• 用户个人信息
• 学员评论中的敏感内容
• 订单信息
• 购买记录
• 登录后才可见的数据
• 付费视频真实播放地址
• 需要绕过权限才能获取的内容

对于需要登录、付费或授权的数据，应使用官方 API、开放平台或经过授权的数据导出方式。

3.4 不绕过限制

本文不会演示绕过登录、破解签名、规避风控、批量账号请求、访问付费资源等做法。即使某些数据在浏览器网络面板中能看到，也要判断它是否属于公开目录信息。技术分享应该保持边界感，这一点比代码本身更重要。

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

4.1 静态、动态与 API 的区别

爬取网页前，我习惯先把目标页面分成三类。

第一类是静态 HTML。页面源代码里已经有完整数据，直接使用 requests + BeautifulSoup/lxml 就能解析。

第二类是动态渲染。页面源代码里没有目标数据，数据由 JavaScript 加载后渲染。课程目录树经常属于这一类。

第三类是接口驱动。动态渲染背后通常有 JSON 接口。与其用浏览器自动化去点页面，不如直接请求公开接口，然后解析 JSON。

本文属于第三类：动态页面背后的公开 JSON 目录树解析。

4.2 整体流程

流程可以概括为：

采集 → 解析 → 清洗 → 存储

更具体一点：

#mermaid-svg-7o1JWeouvoDh4JBV{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-7o1JWeouvoDh4JBV .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-7o1JWeouvoDh4JBV .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-7o1JWeouvoDh4JBV .error-icon{fill:#552222;}#mermaid-svg-7o1JWeouvoDh4JBV .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-7o1JWeouvoDh4JBV .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-7o1JWeouvoDh4JBV .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-7o1JWeouvoDh4JBV .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-7o1JWeouvoDh4JBV .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-7o1JWeouvoDh4JBV .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-7o1JWeouvoDh4JBV .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-7o1JWeouvoDh4JBV .marker{fill:#333333;stroke:#333333;}#mermaid-svg-7o1JWeouvoDh4JBV .marker.cross{stroke:#333333;}#mermaid-svg-7o1JWeouvoDh4JBV svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-7o1JWeouvoDh4JBV p{margin:0;}#mermaid-svg-7o1JWeouvoDh4JBV .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-7o1JWeouvoDh4JBV .cluster-label text{fill:#333;}#mermaid-svg-7o1JWeouvoDh4JBV .cluster-label span{color:#333;}#mermaid-svg-7o1JWeouvoDh4JBV .cluster-label span p{background-color:transparent;}#mermaid-svg-7o1JWeouvoDh4JBV .label text,#mermaid-svg-7o1JWeouvoDh4JBV span{fill:#333;color:#333;}#mermaid-svg-7o1JWeouvoDh4JBV .node rect,#mermaid-svg-7o1JWeouvoDh4JBV .node circle,#mermaid-svg-7o1JWeouvoDh4JBV .node ellipse,#mermaid-svg-7o1JWeouvoDh4JBV .node polygon,#mermaid-svg-7o1JWeouvoDh4JBV .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-7o1JWeouvoDh4JBV .rough-node .label text,#mermaid-svg-7o1JWeouvoDh4JBV .node .label text,#mermaid-svg-7o1JWeouvoDh4JBV .image-shape .label,#mermaid-svg-7o1JWeouvoDh4JBV .icon-shape .label{text-anchor:middle;}#mermaid-svg-7o1JWeouvoDh4JBV .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-7o1JWeouvoDh4JBV .rough-node .label,#mermaid-svg-7o1JWeouvoDh4JBV .node .label,#mermaid-svg-7o1JWeouvoDh4JBV .image-shape .label,#mermaid-svg-7o1JWeouvoDh4JBV .icon-shape .label{text-align:center;}#mermaid-svg-7o1JWeouvoDh4JBV .node.clickable{cursor:pointer;}#mermaid-svg-7o1JWeouvoDh4JBV .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-7o1JWeouvoDh4JBV .arrowheadPath{fill:#333333;}#mermaid-svg-7o1JWeouvoDh4JBV .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-7o1JWeouvoDh4JBV .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-7o1JWeouvoDh4JBV .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-7o1JWeouvoDh4JBV .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-7o1JWeouvoDh4JBV .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-7o1JWeouvoDh4JBV .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-7o1JWeouvoDh4JBV .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-7o1JWeouvoDh4JBV .cluster text{fill:#333;}#mermaid-svg-7o1JWeouvoDh4JBV .cluster span{color:#333;}#mermaid-svg-7o1JWeouvoDh4JBV div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-7o1JWeouvoDh4JBV .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-7o1JWeouvoDh4JBV rect.text{fill:none;stroke-width:0;}#mermaid-svg-7o1JWeouvoDh4JBV .icon-shape,#mermaid-svg-7o1JWeouvoDh4JBV .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-7o1JWeouvoDh4JBV .icon-shape p,#mermaid-svg-7o1JWeouvoDh4JBV .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-7o1JWeouvoDh4JBV .icon-shape .label rect,#mermaid-svg-7o1JWeouvoDh4JBV .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-7o1JWeouvoDh4JBV .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-7o1JWeouvoDh4JBV .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-7o1JWeouvoDh4JBV :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

否

是

读取配置

构造请求

Fetcher 请求公开 JSON

请求是否成功

重试与退避

Parser 解析课程信息

递归解析 children 目录树

字段清洗与规范化

去重

导出 CSV

导出 JSON

写入 SQLite

4.3 为什么选择 requests

本文选择 requests 的原因很直接：

4.4 什么时候用 Playwright

如果你找不到公开 JSON 接口，或者页面必须执行 JavaScript 才能拿到数据，可以考虑 Playwright。但 Playwright 不应该成为第一选择，因为它更重，运行成本更高。

合理做法是：

只有在数据确实依赖浏览器运行状态时，再使用 Playwright。

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

5.1 Python 版本

建议使用：

Python 3.10+

本文代码在 Python 3.10、3.11、3.12 语法范围内都可以运行。

5.2 安装依赖

创建项目目录：

mkdir course_tree_spider
cd course_tree_spider
python -m venv .venv

激活虚拟环境。

macOS / Linux：

source .venv/bin/activate

Windows PowerShell：

.venv\\Scripts\\Activate.ps1

安装依赖：

pip install requests pyyaml tenacity pydantic rich

生成 requirements.txt：

requests>=2.31.0
PyYAML>=6.0.1
tenacity>=8.2.3
pydantic>=2.6.0
rich>=13.7.0

如果你希望后续扩展到 HTML 解析，可以额外安装：

pip install beautifulsoup4 lxml

本文主线使用 JSON 解析，所以 bs4/lxml 不是必须依赖。

5.3 推荐项目结构

course_tree_spider/
├── configs/
│ └── settings.yaml
├── crawler/
│ ├── __init__.py
│ ├── config.py
│ ├── fetcher.py
│ ├── parser.py
│ ├── storage.py
│ └── utils.py
├── examples/
│ └── course_tree_sample.json
├── data/
│ ├── output.csv
│ ├── output.json
│ └── courses.db
├── logs/
│ └── spider.log
├── tests/
│ └── test_parser.py
├── main.py
└── requirements.txt

为了让代码可以真实运行，本文会提供一个本地示例 JSON。你可以先用本地文件跑通流程，再把配置切换成公开接口 URL。

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

Fetcher 负责请求数据。一个成熟的请求层至少应该考虑：

• headers
• timeout
• session
• cookie
• retry
• backoff
• 状态码处理
• JSON 解析异常
• 本地示例文件读取

6.1 配置文件

先写 configs/settings.yaml：

spider:
mode: "file" # file or api
api_url: "https://example.com/api/course/catalog?course_id=python-1001"
sample_file: "examples/course_tree_sample.json"
source_url: "https://example.com/course/python-1001"
request_interval: 1.5
timeout: 10
max_retries: 3

headers:
User-Agent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/124.0 Safari/537.36"
Referer: "https://example.com/course/python-1001"
Accept: "application/json, text/plain, */*"
Accept-Language: "zh-CN,zh;q=0.9,en;q=0.8"

storage:
csv_path: "data/output.csv"
json_path: "data/output.json"
sqlite_path: "data/courses.db"
sqlite_table: "course_chapters"

这里的 mode 有两个值：

• file：读取本地示例 JSON，适合先验证解析逻辑。
• api：请求公开 API，适合替换成真实公开目录接口。

6.2 示例 JSON

创建 examples/course_tree_sample.json：

{
"code": 0,
"message": "ok",
"data": {
"course": {
"id": "python-1001",
"title": "Python 网络爬虫实战入门",
"teacher": {
"id": "t-7788",
"name": "林川"
},
"tags": ["Python", "爬虫", "数据采集", "实战"]
},
"catalog": [
{
"id": "c1",
"title": "第 1 章 课程介绍与环境准备",
"duration": 1800,
"preview": true,
"children": [
{
"id": "c1-1",
"title": "1.1 课程适合谁学习",
"duration": 420,
"preview": true,
"children": []
},
{
"id": "c1-2",
"title": "1.2 Python 环境安装",
"duration": 780,
"preview": true,
"children": []
},
{
"id": "c1-3",
"title": "1.3 项目目录规划",
"duration": 600,
"preview": false,
"children": []
}
]
},
{
"id": "c2",
"title": "第 2 章 HTTP 请求基础",
"duration": 3600,
"preview": false,
"children": [
{
"id": "c2-1",
"title": "2.1 GET 与 POST 的区别",
"duration": 900,
"preview": false,
"children": []
},
{
"id": "c2-2",
"title": "2.2 headers 与 referer",
"duration": 960,
"preview": false,
"children": []
},
{
"id": "c2-3",
"title": "2.3 timeout 与重试",
"duration": 1020,
"preview": true,
"children": []
},
{
"id": "c2-4",
"title": "2.4 频率控制与合规采集",
"duration": 720,
"preview": false,
"children": []
}
]
},
{
"id": "c3",
"title": "第 3 章 动态目录树解析",
"duration": 5400,
"preview": false,
"children": [
{
"id": "c3-1",
"title": "3.1 识别动态渲染页面",
"duration": 1200,
"preview": false,
"children": []
},
{
"id": "c3-2",
"title": "3.2 Network 面板定位公开 JSON",
"duration": 1500,
"preview": false,
"children": []
},
{
"id": "c3-3",
"title": "3.3 递归解析 children",
"duration": 1800,
"preview": true,
"children": [
{
"id": "c3-3-1",
"title": "3.3.1 递归函数设计",
"duration": 600,
"preview": true,
"children": []
},
{
"id": "c3-3-2",
"title": "3.3.2 层级与 sort_path 维护",
"duration": 720,
"preview": false,
"children": []
}
]
},
{
"id": "c3-4",
"title": "3.4 解析异常与容错",
"duration": 900,
"preview": false,
"children": []
}
]
}
]
}
}

6.3 配置读取

创建 crawler/config.py：

from __future__ import annotations

from pathlib import Path
from typing import Any

import yaml
from pydantic import BaseModel, Field

class SpiderSettings(BaseModel):
mode: str = Field(default="file")
api_url: str = Field(default="")
sample_file: str = Field(default="examples/course_tree_sample.json")
source_url: str = Field(default="")
request_interval: float = Field(default=1.0)
timeout: int = Field(default=10)
max_retries: int = Field(default=3)

class StorageSettings(BaseModel):
csv_path: str = Field(default="data/output.csv")
json_path: str = Field(default="data/output.json")
sqlite_path: str = Field(default="data/courses.db")
sqlite_table: str = Field(default="course_chapters")

class AppSettings(BaseModel):
spider: SpiderSettings
headers: dict[str, str] = Field(default_factory=dict)
storage: StorageSettings

def load_settings(path: str = "configs/settings.yaml") –> AppSettings:
config_path = Path(path)
if not config_path.exists():
raise FileNotFoundError(f"配置文件不存在: {config_path}")

with config_path.open("r", encoding="utf-8") as f:
raw: dict[str, Any] = yaml.safe_load(f)

return AppSettings(**raw)

6.4 工具函数

创建 crawler/utils.py：

from __future__ import annotations

import hashlib
import json
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

def ensure_parent_dir(path: str) –> None:
target = Path(path)
target.parent.mkdir(parents=True, exist_ok=True)

def now_iso() –> str:
return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")

def content_hash(value: dict[str, Any]) –> str:
raw = json.dumps(value, ensure_ascii=False, sort_keys=True)
return hashlib.sha256(raw.encode("utf-8")).hexdigest()

def format_duration(seconds: int | None) –> str:
if seconds is None:
return ""

try:
total = int(seconds)
except (TypeError, ValueError):
return ""

if total < 0:
return ""

hours = total // 3600
minutes = (total % 3600) // 60
secs = total % 60

if hours:
return f"{hours:02d}:{minutes:02d}:{secs:02d}"

return f"{minutes:02d}:{secs:02d}"

def setup_logger(log_path: str = "logs/spider.log") –> logging.Logger:
Path(log_path).parent.mkdir(parents=True, exist_ok=True)

logger = logging.getLogger("course_tree_spider")
logger.setLevel(logging.INFO)

if logger.handlers:
return logger

formatter = logging.Formatter(
fmt="%(asctime)s | %(levelname)s | %(name)s | %(message)s"
)

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

console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)

logger.addHandler(file_handler)
logger.addHandler(console_handler)

return logger

6.5 Fetcher 实现

创建 crawler/fetcher.py：

from __future__ import annotations

import json
import time
from pathlib import Path
from typing import Any

import requests
from requests import Response, Session
from tenacity import (
retry,
retry_if_exception_type,
stop_after_attempt,
wait_exponential,
)

from crawler.config import SpiderSettings
from crawler.utils import setup_logger

logger = setup_logger()

class FetchError(RuntimeError):
pass

class CourseCatalogFetcher:
def __init__(self, settings: SpiderSettings, headers: dict[str, str]):
self.settings = settings
self.headers = headers
self.session: Session = requests.Session()
self.session.headers.update(headers)

def fetch(self) –> dict[str, Any]:
mode = self.settings.mode.lower().strip()

if mode == "file":
return self.fetch_from_file(self.settings.sample_file)

if mode == "api":
return self.fetch_from_api(self.settings.api_url)

raise ValueError(f"未知 mode: {self.settings.mode}，只支持 file 或 api")

def fetch_from_file(self, path: str) –> dict[str, Any]:
file_path = Path(path)
if not file_path.exists():
raise FileNotFoundError(f"示例 JSON 文件不存在: {file_path}")

logger.info("读取本地示例数据: %s", file_path)

with file_path.open("r", encoding="utf-8") as f:
return json.load(f)

@retry(
retry=retry_if_exception_type((requests.RequestException, FetchError)),
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10),
reraise=True,
)
def fetch_from_api(self, url: str) –> dict[str, Any]:
if not url:
raise ValueError("api_url 不能为空")

logger.info("请求公开课程目录接口: %s", url)

time.sleep(max(self.settings.request_interval, 0))

response = self.session.get(
url,
timeout=self.settings.timeout,
)

self._check_response(response)

try:
return response.json()
except ValueError as exc:
raise FetchError(f"响应不是合法 JSON: {exc}") from exc

@staticmethod
def _check_response(response: Response) –> None:
status = response.status_code

if status == 200:
return

if status in {403, 401}:
raise FetchError(
f"请求被拒绝，状态码 {status}。请确认数据是否公开，"
"不要尝试绕过登录、权限或付费限制。"
)

if status == 429:
raise FetchError(
"请求过于频繁，状态码 429。请降低频率，增加请求间隔。"
)

if 500 <= status < 600:
raise FetchError(f"服务端错误，状态码 {status}，稍后可重试。")

raise FetchError(f"请求失败，状态码 {status}，响应片段: {response.text[:200]}")

这里的请求层有几个细节。

第一，使用 requests.Session()。它可以复用连接，也方便统一设置 headers。

第二，设置 timeout。没有 timeout 的请求在网络不稳定时可能一直卡住，这在定时任务里很麻烦。

第三，加入重试和指数退避。网络抖动、短暂 5xx 错误都可能出现，直接失败并不优雅。

第四，对 401、403、429 做单独提示。403 不应该第一反应就是“怎么绕过”，更应该先确认数据是否公开、请求是否过快、headers 是否缺失、目标站点是否允许自动化访问。

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

解析层是本文重点。课程目录树最常见的问题是：只解析到第一层章节，漏掉 children 里的子章节。

7.1 原始 JSON 结构分析

示例数据的核心结构如下：

{
"data": {
"course": {
"id": "python-1001",
"title": "Python 网络爬虫实战入门",
"teacher": {
"name": "林川"
},
"tags": ["Python", "爬虫", "数据采集", "实战"]
},
"catalog": [
{
"id": "c1",
"title": "第 1 章 课程介绍与环境准备",
"duration": 1800,
"preview": true,
"children": []
}
]
}
}

目录节点可能有很多变体，比如：

{
"chapterId": "xxx",
"chapterName": "xxx",
"trial": 1,
"length": "12:30",
"childList": []
}

所以解析器不能写得太死。我的习惯是写几个小工具函数，兼容不同字段名：

• pick_str
• pick_bool
• pick_int
• normalize_tags
• parse_duration_to_seconds

7.2 数据模型

创建 crawler/parser.py：

from __future__ import annotations

from dataclasses import asdict, dataclass
from typing import Any, Iterable

from crawler.utils import content_hash, format_duration, now_iso

@dataclass
class ChapterRecord:
course_id: str
course_title: str
teacher: str
tags: str
chapter_id: str
chapter_title: str
parent_id: str
level: int
is_preview: bool
duration_seconds: int | None
duration_text: str
sort_path: str
source_url: str
crawled_at: str
row_hash: str

def to_dict(self) –> dict[str, Any]:
return asdict(self)

def pick_value(data: dict[str, Any], keys: Iterable[str], default: Any = None) –> Any:
for key in keys:
if key in data and data[key] not in (None, ""):
return data[key]
return default

def pick_str(data: dict[str, Any], keys: Iterable[str], default: str = "") –> str:
value = pick_value(data, keys, default)
if value is None:
return default
return str(value).strip()

def pick_bool(data: dict[str, Any], keys: Iterable[str], default: bool = False) –> bool:
value = pick_value(data, keys, default)

if isinstance(value, bool):
return value

if isinstance(value, int):
return value == 1

if isinstance(value, str):
normalized = value.strip().lower()
return normalized in {"1", "true", "yes", "y", "试看", "free", "preview"}

return default

def parse_duration_to_seconds(value: Any) –> int | None:
if value is None or value == "":
return None

if isinstance(value, int):
return value if value >= 0 else None

if isinstance(value, float):
return int(value) if value >= 0 else None

if isinstance(value, str):
text = value.strip()
if not text:
return None

if text.isdigit():
return int(text)

parts = text.split(":")
if all(part.isdigit() for part in parts):
if len(parts) == 2:
minutes, seconds = parts
return int(minutes) * 60 + int(seconds)

if len(parts) == 3:
hours, minutes, seconds = parts
return int(hours) * 3600 + int(minutes) * 60 + int(seconds)

return None

def normalize_tags(value: Any) –> str:
if value is None:
return ""

if isinstance(value, list):
return ",".join(str(item).strip() for item in value if str(item).strip())

if isinstance(value, str):
return value.strip()

return str(value).strip()

def find_first_list(data: dict[str, Any], keys: Iterable[str]) –> list[dict[str, Any]]:
for key in keys:
value = data.get(key)
if isinstance(value, list):
return [item for item in value if isinstance(item, dict)]
return []

class CourseCatalogParser:
def __init__(self, source_url: str = ""):
self.source_url = source_url

def parse(self, payload: dict[str, Any]) –> list[ChapterRecord]:
data = payload.get("data", payload)
if not isinstance(data, dict):
raise ValueError("payload.data 不是对象结构，无法解析")

course = self._extract_course(data)
catalog = self._extract_catalog(data)

if not catalog:
return []

records: list[ChapterRecord] = []
self._walk_nodes(
nodes=catalog,
course=course,
records=records,
parent_id="",
level=1,
prefix=[],
)
return records

def _extract_course(self, data: dict[str, Any]) –> dict[str, Any]:
course = data.get("course") or data.get("courseInfo") or data.get("detail") or data

if not isinstance(course, dict):
course = {}

teacher_value = pick_value(course, ["teacher", "lecturer", "instructor"], "")

if isinstance(teacher_value, dict):
teacher = pick_str(teacher_value, ["name", "nickname", "title"], "")
else:
teacher = str(teacher_value).strip() if teacher_value else ""

return {
"course_id": pick_str(course, ["id", "course_id", "courseId"], ""),
"course_title": pick_str(course, ["title", "name", "course_title", "courseName"], ""),
"teacher": teacher,
"tags": normalize_tags(pick_value(course, ["tags", "tagList", "labels"], [])),
}

def _extract_catalog(self, data: dict[str, Any]) –> list[dict[str, Any]]:
catalog = find_first_list(
data,
[
"catalog",
"chapters",
"chapterList",
"lessons",
"items",
"tree",
"children",
],
)
return catalog

def _walk_nodes(
self,
nodes: list[dict[str, Any]],
course: dict[str, Any],
records: list[ChapterRecord],
parent_id: str,
level: int,
prefix: list[int],
) –> None:
for index, node in enumerate(nodes, start=1):
if not isinstance(node, dict):
continue

sort_path_parts = prefix + [index]
sort_path = ".".join(str(part) for part in sort_path_parts)

chapter_id = pick_str(
node,
["id", "chapter_id", "chapterId", "lesson_id", "lessonId", "nodeId"],
default=f"auto-{sort_path}",
)

title = pick_str(
node,
["title", "name", "chapter_title", "chapterName", "lessonName"],
default="",
)

duration_seconds = parse_duration_to_seconds(
pick_value(
node,
["duration", "duration_seconds", "durationSeconds", "length", "time"],
None,
)
)

is_preview = pick_bool(
node,
["preview", "is_preview", "isPreview", "trial", "free"],
default=False,
)

base = {
"course_id": course["course_id"],
"course_title": course["course_title"],
"teacher": course["teacher"],
"tags": course["tags"],
"chapter_id": chapter_id,
"chapter_title": title,
"parent_id": parent_id,
"level": level,
"is_preview": is_preview,
"duration_seconds": duration_seconds,
"duration_text": format_duration(duration_seconds),
"sort_path": sort_path,
"source_url": self.source_url,
}

row_hash = content_hash(base)

record = ChapterRecord(
**base,
crawled_at=now_iso(),
row_hash=row_hash,
)
records.append(record)

children = find_first_list(
node,
["children", "childList", "subChapters", "sections", "lessons"],
)

if children:
self._walk_nodes(
nodes=children,
course=course,
records=records,
parent_id=chapter_id,
level=level + 1,
prefix=sort_path_parts,
)

7.3 解析方式说明

本文主线使用 JSON 解析。原因是目标数据本身就是动态页面背后的 JSON 数据，不需要先把它渲染成 HTML 再解析。

如果目标站点是静态 HTML，可以使用：

• BeautifulSoup
• XPath
• CSS Selector
• lxml

如果目标站点是动态页面，建议先定位接口。实在无法定位时，再用 Playwright。

7.4 列表页如何拿详情链接

有些平台不是直接给一个课程接口，而是先有课程列表页。典型流程如下：

课程列表接口 → 获取 course_id / detail_url → 逐个请求课程目录接口

示例结构可能是：

{
"data": {
"list": [
{
"id": "python-1001",
"title": "Python 网络爬虫实战入门",
"detailUrl": "https://example.com/course/python-1001"
},
{
"id": "data-2001",
"title": "数据分析基础",
"detailUrl": "https://example.com/course/data-2001"
}
]
}
}

这种情况下可以加一个列表解析器：

def parse_course_list(payload: dict) –> list[dict]:
data = payload.get("data", payload)
items = data.get("list") or data.get("items") or data.get("courses") or []

result = []
for item in items:
if not isinstance(item, dict):
continue

course_id = item.get("id") or item.get("courseId") or item.get("course_id")
detail_url = item.get("detailUrl") or item.get("url") or item.get("link")
title = item.get("title") or item.get("name") or ""

if not course_id and not detail_url:
continue

result.append(
{
"course_id": str(course_id or "").strip(),
"title": str(title).strip(),
"detail_url": str(detail_url or "").strip(),
}
)

return result

如果目录接口需要 course_id，就把它拼进 URL：

def build_catalog_url(template: str, course_id: str) –> str:
return template.format(course_id=course_id)

例如：

url = build_catalog_url(
"https://example.com/api/course/catalog?course_id={course_id}",
"python-1001",
)

7.5 详情页如何抽字段

详情页字段通常分两部分。

第一部分是课程级字段：

• 课程 ID
• 课程名称
• 讲师
• 标签

第二部分是章节级字段：

• 章节 ID
• 章节标题
• 是否试看
• 时长
• 父章节
• 层级

课程级字段只解析一次，章节字段递归解析多次。每个章节记录都会带上课程字段，这样导出 CSV 后更容易做统计。

7.6 缺失字段怎么办

实际项目里，缺字段非常常见：

• 某些章节没有时长。
• 某些课程没有标签。
• 某些父节点只是分类，没有试看状态。
• 某些节点没有 ID。

本文处理方式如下：

这种策略的好处是不中断采集。后续如果需要更严格的数据质量校验，可以在清洗层增加规则。

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

本文同时实现 CSV、JSON、SQLite 三种存储方式。起步阶段 CSV 最方便，调试阶段 JSON 最清晰，长期查询 SQLite 更稳。

8.1 字段映射表

8.2 Storage 实现

创建 crawler/storage.py：

from __future__ import annotations

import csv
import json
import sqlite3
from pathlib import Path
from typing import Any

from crawler.parser import ChapterRecord
from crawler.utils import ensure_parent_dir, setup_logger

logger = setup_logger()

FIELDNAMES = [
"course_id",
"course_title",
"teacher",
"tags",
"chapter_id",
"chapter_title",
"parent_id",
"level",
"is_preview",
"duration_seconds",
"duration_text",
"sort_path",
"source_url",
"crawled_at",
"row_hash",
]

def deduplicate_records(records: list[ChapterRecord]) –> list[ChapterRecord]:
seen: set[tuple[str, str, str]] = set()
result: list[ChapterRecord] = []

for record in records:
key = (record.course_id, record.chapter_id, record.sort_path)
if key in seen:
continue
seen.add(key)
result.append(record)

return result

class CourseStorage:
def __init__(self, csv_path: str, json_path: str, sqlite_path: str, table: str):
self.csv_path = csv_path
self.json_path = json_path
self.sqlite_path = sqlite_path
self.table = table

def save_all(self, records: list[ChapterRecord]) –> None:
clean_records = deduplicate_records(records)

logger.info("解析记录数: %s，去重后记录数: %s", len(records), len(clean_records))

self.save_csv(clean_records)
self.save_json(clean_records)
self.save_sqlite(clean_records)

def save_csv(self, records: list[ChapterRecord]) –> None:
ensure_parent_dir(self.csv_path)

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

for record in records:
writer.writerow(record.to_dict())

logger.info("CSV 已导出: %s", self.csv_path)

def save_json(self, records: list[ChapterRecord]) –> None:
ensure_parent_dir(self.json_path)

data = [record.to_dict() for record in records]

with open(self.json_path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)

logger.info("JSON 已导出: %s", self.json_path)

def save_sqlite(self, records: list[ChapterRecord]) –> None:
ensure_parent_dir(self.sqlite_path)

conn = sqlite3.connect(self.sqlite_path)

try:
self._create_table(conn)

sql = f"""
INSERT OR REPLACE INTO {self.table} (
course_id,
course_title,
teacher,
tags,
chapter_id,
chapter_title,
parent_id,
level,
is_preview,
duration_seconds,
duration_text,
sort_path,
source_url,
crawled_at,
row_hash
) VALUES (
:course_id,
:course_title,
:teacher,
:tags,
:chapter_id,
:chapter_title,
:parent_id,
:level,
:is_preview,
:duration_seconds,
:duration_text,
:sort_path,
:source_url,
:crawled_at,
:row_hash
)
"""

payload: list[dict[str, Any]] = []
for record in records:
item = record.to_dict()
item["is_preview"] = int(item["is_preview"])
payload.append(item)

conn.executemany(sql, payload)
conn.commit()

logger.info("SQLite 已写入: %s，表: %s", self.sqlite_path, self.table)
finally:
conn.close()

def _create_table(self, conn: sqlite3.Connection) –> None:
sql = f"""
CREATE TABLE IF NOT EXISTS {self.table} (
course_id TEXT NOT NULL,
course_title TEXT,
teacher TEXT,
tags TEXT,
chapter_id TEXT NOT NULL,
chapter_title TEXT,
parent_id TEXT,
level INTEGER,
is_preview INTEGER,
duration_seconds INTEGER,
duration_text TEXT,
sort_path TEXT NOT NULL,
source_url TEXT,
crawled_at TEXT,
row_hash TEXT,
PRIMARY KEY (course_id, chapter_id, sort_path)
);
"""

conn.execute(sql)

index_sql = f"""
CREATE INDEX IF NOT EXISTS idx_{self.table}_course
ON {self.table} (course_id, level, sort_path);
"""

conn.execute(index_sql)
conn.commit()

8.3 去重策略

本文采用组合键去重：

course_id + chapter_id + sort_path

为什么不只用 chapter_id？

因为不同课程里可能存在相同的章节 ID，尤其是平台内部如果用短 ID 或迁移数据时，单独用 chapter_id 不够稳。

为什么加 sort_path？

因为有些节点没有真实 ID，我们会生成 auto-1.2.3。加入 sort_path 可以避免同名节点互相覆盖。

如果你要做内容变化监控，可以使用 row_hash 判断记录是否变化。比如同一个章节标题或试看状态变了，hash 会发生变化。

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

9.1 主入口文件

创建 main.py：

from __future__ import annotations

from rich.console import Console
from rich.table import Table

from crawler.config import load_settings
from crawler.fetcher import CourseCatalogFetcher
from crawler.parser import CourseCatalogParser
from crawler.storage import CourseStorage
from crawler.utils import setup_logger

console = Console()
logger = setup_logger()

def preview_records(records, limit: int = 5) –> None:
table = Table(title=f"Course Catalog Preview Top {limit}")

table.add_column("course_title", overflow="fold")
table.add_column("chapter_title", overflow="fold")
table.add_column("level")
table.add_column("preview")
table.add_column("duration")
table.add_column("teacher")

for record in records[:limit]:
table.add_row(
record.course_title,
record.chapter_title,
str(record.level),
"yes" if record.is_preview else "no",
record.duration_text,
record.teacher,
)

console.print(table)

def main() –> None:
settings = load_settings()

fetcher = CourseCatalogFetcher(
settings=settings.spider,
headers=settings.headers,
)

payload = fetcher.fetch()

parser = CourseCatalogParser(
source_url=settings.spider.source_url or settings.spider.api_url,
)

records = parser.parse(payload)

if not records:
logger.warning("没有解析到任何课程章节记录")
return

storage = CourseStorage(
csv_path=settings.storage.csv_path,
json_path=settings.storage.json_path,
sqlite_path=settings.storage.sqlite_path,
table=settings.storage.sqlite_table,
)

storage.save_all(records)

preview_records(records, limit=5)

if __name__ == "__main__":
main()

9.2 启动命令

先确保目录完整：

mkdir -p configs crawler examples data logs tests
touch crawler/__init__.py

运行：

python main.py

默认读取本地示例文件：

spider:
mode: "file"

如果要请求公开接口，把配置改成：

spider:
mode: "api"
api_url: "https://example.com/api/course/catalog?course_id=python-1001"

然后再次运行：

python main.py

9.3 输出位置

运行后会生成：

data/output.csv
data/output.json
data/courses.db

SQLite 查询示例：

sqlite3 data/courses.db

进入 SQLite 后执行：

SELECT
course_title,
chapter_title,
level,
is_preview,
duration_text,
teacher
FROM course_chapters
ORDER BY course_id, sort_path
LIMIT 5;

9.4 示例结果

示例输出类似：

你会发现，父章节和子章节都被展开到了同一张表里。后续要统计每门课有多少小节、试看小节比例、总课时，都很方便。

🔟 常见问题与排错

10.1 遇到 403 怎么办？

403 表示请求被拒绝。常见原因有：

• 目标数据并非公开数据。
• 请求缺少合理 headers。
• Referer 不符合站点预期。
• 请求频率过高。
• 接口需要登录或授权。

建议处理方式：

本文 Fetcher 对 403 的处理是直接提示并停止，不会继续高频重试。

10.2 遇到 429 怎么办？

429 通常表示请求过多。解决方式不是继续加代理猛打，而是降低频率。

建议：

• 增加 request_interval。
• 开启指数退避。
• 减少并发。
• 做本地缓存。
• 不重复请求同一课程。
• 对稳定目录数据设置较长更新周期。

课程目录不是秒级变化的数据，很多场景每天采一次已经足够。

10.3 HTML 抓到空壳怎么办？

如果你用：

html = requests.get(url).text
print(html)

发现只有：

<div id="app"></div>
<script src="/static/js/app.js"></script>

说明页面大概率是前端动态渲染。处理步骤：

不要急着上 Playwright。很多时候真正需要的只是一个公开 JSON 接口。

10.4 解析报错怎么办？

解析报错常见原因：

• 字段名变化。
• children 改成了 childList。
• duration 从整数变成了 "12:30"。
• teacher 从字符串变成了对象。
• 某些节点缺少 ID。
• 接口返回错误信息而不是课程数据。

本文解析器通过 pick_value、find_first_list 和 parse_duration_to_seconds 做了一层兼容。但真实项目中仍然建议把异常样本保存下来。

可以在 main.py 中加一段调试输出：

import json

with open("data/debug_payload.json", "w", encoding="utf-8") as f:
json.dump(payload, f, ensure_ascii=False, indent=2)

当解析失败时，先看原始 JSON，不要只盯着代码猜。

10.5 编码乱码怎么办？

CSV 乱码通常是 Excel 打开方式导致的。本文写 CSV 时使用：

encoding="utf-8-sig"

这样在 Windows Excel 中打开中文更稳。

JSON 建议使用：

json.dump(data, f, ensure_ascii=False, indent=2)

这样中文不会被转义成 \\u4e2d\\u6587。

10.6 数据重复怎么办？

先确认重复类型。

如果是完全重复，多半是翻页或任务调度重复请求导致的。

如果是同名章节重复，可能是课程确实有多个同名小节。

如果是父章节和子章节看起来重复，要检查 level 和 parent_id。

本文使用：

course_id + chapter_id + sort_path

作为去重主键，适合课程目录树场景。

10.7 试看字段不准怎么办？

试看字段在不同平台可能叫：

preview
isPreview
trial
free
canTry

有的平台还会用数字：

1 表示试看
0 表示不可试看

也可能用字符串：

"free"
"trial"
"试看"

所以解析时不要只写：

is_preview = node["preview"]

而应该写兼容逻辑。本文 pick_bool 就是为这个场景准备的。

1️⃣1️⃣ 进阶优化

11.1 并发采集

如果你要采集很多课程，可以引入并发。但课程目录接口不适合盲目高并发。建议先做低并发，例如 3–5 个 worker。

线程池示例：

from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Callable

def run_with_threads(
course_ids: list[str],
worker: Callable[[str], list[dict]],
max_workers: int = 3,
) –> list[dict]:
results: list[dict] = []

with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_map = {
executor.submit(worker, course_id): course_id
for course_id in course_ids
}

for future in as_completed(future_map):
course_id = future_map[future]

try:
rows = future.result()
results.extend(rows)
except Exception as exc:
print(f"course_id={course_id} 采集失败: {exc}")

return results

这个例子只是说明结构，不建议直接把并发开大。并发不是越高越专业，稳定、克制、可恢复才更重要。

11.2 断点续跑

课程采集任务可能中途失败，比如网络中断、接口临时错误、机器重启。断点续跑可以避免从头开始。

简单做法是维护一个已完成集合：

import json
from pathlib import Path

class Checkpoint:
def __init__(self, path: str = "data/checkpoint.json"):
self.path = Path(path)
self.done: set[str] = set()
self.load()

def load(self) –> None:
if not self.path.exists():
self.done = set()
return

with self.path.open("r", encoding="utf-8") as f:
data = json.load(f)

self.done = set(data.get("done", []))

def save(self) –> None:
self.path.parent.mkdir(parents=True, exist_ok=True)

with self.path.open("w", encoding="utf-8") as f:
json.dump(
{"done": sorted(self.done)},
f,
ensure_ascii=False,
indent=2,
)

def is_done(self, course_id: str) –> bool:
return course_id in self.done

def mark_done(self, course_id: str) –> None:
self.done.add(course_id)
self.save()

使用方式：

checkpoint = Checkpoint()

for course_id in course_ids:
if checkpoint.is_done(course_id):
continue

# fetch and parse course
# save records

checkpoint.mark_done(course_id)

11.3 日志与监控

正式任务不建议只用 print。日志至少要记录：

• 请求 URL
• 状态码
• 解析记录数
• 成功课程数
• 失败课程数
• 失败原因
• 导出路径

后续可以统计成功率：

class SpiderStats:
def __init__(self):
self.total = 0
self.success = 0
self.failed = 0
self.records = 0

def mark_success(self, count: int) –> None:
self.total += 1
self.success += 1
self.records += count

def mark_failed(self) –> None:
self.total += 1
self.failed += 1

@property
def success_rate(self) –> float:
if self.total == 0:
return 0.0
return self.success / self.total

def summary(self) –> dict:
return {
"total": self.total,
"success": self.success,
"failed": self.failed,
"records": self.records,
"success_rate": round(self.success_rate, 4),
}

11.4 定时任务

Linux 上可以用 cron。比如每天凌晨 2 点执行：

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

如果任务复杂，可以用 Airflow、Prefect 或其他调度系统。小项目用 cron 就够了，大项目再谈工作流平台。

11.5 缓存原始响应

原始响应很有价值。解析规则变了以后，可以直接用历史 JSON 重新解析，而不用重新请求目标站点。

示例：

import json
from pathlib import Path
from datetime import datetime

def save_raw_payload(course_id: str, payload: dict, root: str = "data/raw") –> None:
today = datetime.now().strftime("%Y%m%d")
directory = Path(root) / today
directory.mkdir(parents=True, exist_ok=True)

path = directory / f"{course_id}.json"

with path.open("w", encoding="utf-8") as f:
json.dump(payload, f, ensure_ascii=False, indent=2)

这种缓存对排错非常有帮助。很多时候，线上失败不是代码突然坏了，而是接口结构变了。保留原始 JSON，排查会轻松很多。

11.6 数据质量检查

可以加一个简单的数据质量函数：

from crawler.parser import ChapterRecord

def validate_records(records: list[ChapterRecord]) –> list[str]:
errors: list[str] = []

for index, record in enumerate(records, start=1):
if not record.course_id:
errors.append(f"第 {index} 行缺少 course_id")

if not record.chapter_title:
errors.append(f"第 {index} 行缺少 chapter_title")

if record.level <= 0:
errors.append(f"第 {index} 行 level 非法: {record.level}")

if record.duration_seconds is not None and record.duration_seconds < 0:
errors.append(f"第 {index} 行 duration_seconds 非法")

return errors

在 main.py 中使用：

errors = validate_records(records)
if errors:
for error in errors[:20]:
logger.warning("数据质量问题: %s", error)

数据质量检查不一定要阻断流程，但至少要让你知道哪里不对。

11.7 扩展到 Scrapy

当目标变成大量课程、多入口、多列表、多详情页时，可以考虑 Scrapy。Scrapy 的优势是：

• 调度器成熟
• 下载中间件完善
• 去重机制内置
• 日志清晰
• item pipeline 适合存储
• 更适合大规模任务

但对于本文这个单接口课程目录树项目，直接上 Scrapy 会略显重。技术选型没有绝对高级，只有适不适合。

11.8 扩展到 Playwright

如果接口需要页面运行后才能得到，Playwright 可以用于：

• 等待页面渲染
• 监听接口响应
• 提取页面中的初始化 JSON
• 处理必要的前端状态

一个中性、安全的 Playwright 思路如下：

from playwright.sync_api import sync_playwright

def capture_public_json(page_url: str, keyword: str = "catalog") –> list[dict]:
matched: list[dict] = []

with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()

def handle_response(response):
url = response.url
if keyword not in url:
return

content_type = response.headers.get("content-type", "")
if "application/json" not in content_type:
return

try:
matched.append(
{
"url": url,
"json": response.json(),
}
)
except Exception:
pass

page.on("response", handle_response)
page.goto(page_url, wait_until="networkidle", timeout=30000)

browser.close()

return matched

这段代码只用于定位公开 JSON 响应，不用于绕过登录、权限或付费限制。

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

本文完成了一个在线教育平台公开目录树采集项目，从配置、请求、解析、清洗、存储到运行展示都走了一遍。

我们做了几件关键事情：

我个人写这类采集任务时，最看重的不是“跑得多快”，而是“下个月还能不能看懂，出了问题能不能定位，目标站点能不能承受”。课程目录这类数据本身不需要高频抓取，稳定性、可维护性和合规边界要比并发数更重要。

下一步你可以继续扩展：

• 接入课程列表页，批量采集多个课程。
• 增加课程变更监控，比较今天和昨天的目录差异。
• 把 SQLite 换成 MySQL 或 PostgreSQL。
• 用 Scrapy 管理大规模课程任务。
• 用 Playwright 辅助定位动态页面公开接口。
• 增加可视化分析，比如试看比例、总课时、章节层级分布。

最后再强调一次：爬虫技术本身没有问题，关键在使用方式。只采集公开数据，控制频率，尊重规则，不触碰权限边界，这样写出来的程序才值得长期维护。

🌟 文末

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

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

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

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

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

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

✅ 互动征集

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

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

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

✅ 免责声明

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

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

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