欢迎光临
我们一直在努力

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 为什么不用 Scrapy 起步
        • 4.4 为什么不用 Playwright 起步
      • 5️⃣ 环境准备与依赖安装(可复现)
        • 5.1 Python 版本
        • 5.2 创建项目目录
        • 5.3 安装依赖
        • 5.4 项目运行方式预览
      • 6️⃣ 核心实现:请求层(Fetcher)
        • 6.1 headers 设置
        • 6.2 timeout
        • 6.3 session/cookie
        • 6.4 失败处理:重试与退避
      • 7️⃣ 核心实现:解析层(Parser)
        • 7.1 列表页如何拿详情链接
        • 7.2 详情页如何抽字段
        • 7.3 缺失字段怎么办
      • 8️⃣ 数据存储与导出(Storage)
        • 8.1 字段映射表
        • 8.2 去重策略
      • 9️⃣ 完整代码:可运行版本
      • 9️⃣ 运行方式与结果展示(必写)
        • 9.1 使用内置示例运行
        • 9.2 指定真实课程目录页运行
        • 9.3 输出位置
        • 9.4 示例结果
        • 9.5 使用 SQLite 查询
      • 🔟 常见问题与排错(强烈建议写)
        • 10.1 遇到 403 怎么办
        • 10.2 遇到 429 怎么办
        • 10.3 HTML 抓到空壳怎么办
        • 10.4 解析报错怎么办
        • 10.5 编码或乱码怎么办
        • 10.6 详情页链接拼接错误怎么办
        • 10.7 字段缺失怎么办
      • 1️⃣1️⃣ 进阶优化(可选但加分)
        • 11.1 并发优化
        • 11.2 asyncio 版本思路
        • 11.3 断点续跑
        • 11.4 日志与监控
        • 11.5 定时任务
        • 11.6 配置化多个站点
        • 11.7 数据质量校验
      • 1️⃣2️⃣ 总结与延伸阅读
      • 🌟 文末
        • ✅ 专栏持续更新中|建议收藏 + 订阅
        • ✅ 互动征集
        • ✅ 免责声明

🌟 开篇语

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

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

  📌 专栏食用指南(建议收藏)

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

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

0️⃣ 前言(Preface)

这篇文章要做一件很具体的事:用 Python 爬取“老年大学课程目录页”,采集课程名、课程类别、开课时间、校区/地点、课程链接,并最终导出为 CSV 与 SQLite 数据库。

我选择的技术栈是 requests + BeautifulSoup + lxml + SQLite。这不是为了炫技,而是因为课程目录页通常结构比较规整,字段清晰,很多页面并不需要一上来就用 Scrapy 或 Playwright。先用轻量方案跑通,再根据页面复杂度升级,这是我个人比较认可的爬虫开发节奏。

读完这篇文章,你可以获得:

  • 一个能直接运行的老年大学课程目录采集脚本;
  • 一套从请求、解析、清洗、存储到排错的完整爬虫项目思路;
  • 一些面对真实网页结构变化时的容错写法与实战经验。
  • 这类项目看起来不复杂,但很适合练基本功。课程目录型页面没有太多商业化干扰,字段稳定,数据规模适中,很适合做信息聚合、课程检索、数据分析或自动化整理。


    1️⃣ 摘要(Abstract)

    本文围绕“老年大学课程目录页”展开,使用 Python 的 requests 请求页面,用 BeautifulSoup/lxml 解析列表页和详情页,提取课程名、课程类别、开课时间、校区/地点、链接等字段,最后将数据保存为 CSV 和 SQLite。

    读完本文,你将掌握:

  • 如何设计一个面向课程目录页的轻量爬虫;
  • 如何处理 headers、timeout、重试、退避、字段缺失、去重等实战问题;
  • 如何将采集结果结构化保存,并展示可分析的样例数据。
  • 文章中的代码以可复现为原则。由于不同老年大学官网的 HTML 结构不完全一致,代码默认提供一组本地示例页面,保证复制后可以直接运行;如果你要换成真实站点,只需要把列表页 URL 和解析选择器按实际页面做少量调整即可。


    2️⃣ 背景与需求(Why)

    2.1 为什么要爬老年大学课程目录页

    老年大学的课程信息通常分散在不同校区、不同栏目或不同通知页面里。有的学校按“课程表”发布,有的按“招生简章”发布,也有的在官网中维护了课程目录页。对于学习者、家属、社区服务人员,或者做信息聚合的小工具来说,手工打开网页逐条复制并不高效。

    爬取课程目录页的价值主要体现在三个方面。

    第一,方便信息聚合。 例如把多个校区的课程统一整理到一个表格里,按课程类别、上课地点、开课时间筛选,就比在网页里一页页翻找舒服很多。

    第二,方便数据分析。 采集到结构化数据后,可以统计不同类别课程的数量,比如书画类、声乐类、舞蹈类、健康养生类、手机摄影类等。也可以分析哪些校区开课更多,哪些时间段课程更集中。

    第三,方便自动化更新。 如果网站课程信息定期更新,可以设置定时任务,每周或每天抓取一次,自动比对新增课程和下架课程。这类自动化不一定复杂,但对日常信息维护很有帮助。

    我一直觉得,爬虫最适合从这种“页面干净、字段明确、场景真实”的项目开始。不要一上来就抓特别复杂的网站,否则很容易把时间耗在登录、验证码、反爬和动态渲染上,反而忽略了数据工程最基础的东西:字段定义、异常处理、数据清洗和可复现。

    2.2 目标站点类型

    本文面向的是公开可访问的老年大学课程目录页面,常见页面形式包括:

  • 学校官网中的课程目录列表;
  • 招生简章中的课程表;
  • 某个校区的课程通知页;
  • 社区教育平台中的银龄课程栏目;
  • 静态 HTML 列表页,列表项链接到详情页。
  • 为了让代码稳定可运行,本文默认使用本地示例 HTML。真实项目中,你可以将 –list-url 替换为目标课程目录页地址。

    2.3 目标字段清单

    本项目采集字段如下:

    字段含义示例
    course_name 课程名 智能手机摄影基础班
    category 课程类别 摄影 / 数字生活
    start_time 开课时间 2026-03-10 09:00
    campus 校区/地点 东城校区 302 教室
    url 课程详情链接 https://example.org/course/1001.html

    在实际项目里,还可以继续扩展字段,例如授课老师、学费、课时数、招生人数、报名状态等。但这篇文章聚焦目录型项目,不把字段做得太重。


    3️⃣ 合规与注意事项(必写)

    爬虫技术本身是中性的,关键在于怎么使用。做公开网页采集时,我一般会先确认三件事:是否允许访问、访问频率是否温和、采集范围是否合理。

    3.1 robots.txt 基本说明

    robots.txt 是网站放在根目录下的爬虫访问规则文件。它通常位于:

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

    这个文件会说明哪些路径允许或不允许自动化程序访问。虽然它不是严格意义上的权限控制系统,但它代表了站点管理方对爬虫访问边界的公开说明。

    在写采集脚本前,建议先查看:

    https://example.org/robots.txt

    需要重点关注:

  • 是否禁止访问课程目录所在路径;
  • 是否设置了 Crawl-delay;
  • 是否禁止特定 User-Agent;
  • 是否对搜索页、接口页、附件页有额外限制。
  • 如果规则明确禁止访问某些目录,就不要采集这些路径。技术上能访问,并不代表应该访问。

    3.2 频率控制

    课程目录页的数据变化通常不高,没有必要高并发请求。本文代码默认设置了请求间隔 delay,建议真实站点使用:

    1 秒到 3 秒请求一次

    如果目标站点响应较慢,可以继续加大间隔。对于这类公共信息页面,我个人不建议一上来就开几十个线程。数据量通常也不大,温和采集反而更稳定。

    3.3 不要攻击式并发

    所谓攻击式并发,就是短时间内对网站发起大量请求,导致服务器压力升高,甚至影响正常用户访问。课程目录页采集没有任何理由这么做。

    比较合适的做法是:

  • 控制访问频率;
  • 设置合理 timeout;
  • 失败后退避重试;
  • 限制最大页数;
  • 缓存已抓过的详情页;
  • 不重复请求同一个 URL。
  • 3.4 不采集敏感信息

    本文只采集公开课程信息,不涉及个人身份信息、联系方式、账号数据、报名名单、内部系统数据等内容。

    如果真实站点需要登录才能访问,或者某些信息只面向特定用户展示,不建议绕过登录、验证码、权限校验或付费限制。本文也不会讨论绕过限制的做法。

    技术分享应该有边界。尤其是教育类网站,很多服务器并不是为高频访问设计的,写爬虫时要尽量克制。


    4️⃣ 技术选型与整体流程(What/How)

    4.1 静态、动态还是 API

    网页采集通常可以分成三类。

    第一类是静态 HTML。 页面源码里直接包含课程名、时间、地点和详情链接。这种最适合用 requests + BeautifulSoup。

    第二类是动态渲染。 浏览器打开页面能看到课程列表,但 requests.get() 拿到的是空壳 HTML,真正的数据由 JavaScript 后续加载。这时可以分析接口,或者使用 Playwright 渲染页面。

    第三类是接口数据。 页面前端通过 JSON API 获取课程列表。这种一般可以直接请求接口,解析 JSON,比解析 HTML 更稳定。

    本文把老年大学课程目录页按“静态或轻动态页面”处理,优先使用:

    requests + BeautifulSoup + lxml

    如果你发现 HTML 里没有课程数据,再考虑接口分析或 Playwright。

    4.2 整体流程

    本项目流程可以概括为:

    采集 → 解析 → 清洗 → 存储

    更细一点是:

    读取列表页
    ↓
    解析课程卡片/课程行
    ↓
    提取详情页链接
    ↓
    请求详情页
    ↓
    抽取课程字段
    ↓
    字段清洗与补全
    ↓
    URL 去重
    ↓
    保存 SQLite
    ↓
    导出 CSV

    4.3 为什么不用 Scrapy 起步

    Scrapy 很强,适合中大型采集项目。但对于一个课程目录页,尤其字段只有 5 个左右时,用 Scrapy 起步略显重。学习成本、项目文件数量、配置项都会增加。

    本文使用 requests + bs4 的理由是:

  • 代码短,容易解释;
  • 调试方便;
  • 对静态目录页足够用;
  • 对新手友好;
  • 后续迁移到 Scrapy 也不困难。
  • 如果你后面要采集多个老年大学网站、多个城市、多个栏目,那就可以升级到 Scrapy,把请求调度、去重、持久化和日志做得更规范。

    4.4 为什么不用 Playwright 起步

    Playwright 适合浏览器自动化和动态页面渲染,但它比 requests 重很多。启动浏览器、等待元素、管理上下文,都会增加复杂度。

    如果课程目录页源码里已经有数据,用 Playwright 就像用吊车搬一箱书,当然能搬,但不是最经济的选择。

    本文的策略是:

    先 requests,看源码是否有数据;
    如果没有,再看接口;
    接口找不到,再考虑 Playwright。

    这也是我在实际项目里常用的判断顺序。


    5️⃣ 环境准备与依赖安装(可复现)

    5.1 Python 版本

    建议使用 Python 3.10 或以上版本。

    查看版本:

    python –version

    示例输出:

    Python 3.11.8

    5.2 创建项目目录

    推荐目录结构如下:

    elder_course_spider/
    ├── elder_course_spider.py
    ├── requirements.txt
    ├── output/
    │ ├── courses.csv
    │ └── courses.sqlite3
    └── README.md

    为了便于复现,本文先采用单文件实现。单文件不是偷懒,而是为了让读者先完整跑通流程。等项目稳定后,再拆成 fetcher.py、parser.py、storage.py 也不迟。

    5.3 安装依赖

    创建 requirements.txt:

    requests==2.32.3
    beautifulsoup4==4.12.3
    lxml==5.3.0

    安装:

    pip install -r requirements.txt

    如果你只是临时测试,也可以直接安装:

    pip install requests beautifulsoup4 lxml

    5.4 项目运行方式预览

    运行本地示例数据:

    python elder_course_spider.py –sample

    指定真实列表页:

    python elder_course_spider.py –list-url "https://example.org/elder-university/courses.html"

    指定输出目录:

    python elder_course_spider.py –sample –output-dir output


    6️⃣ 核心实现:请求层(Fetcher)

    请求层的职责很明确:稳定、温和地把 HTML 拿回来。

    它不应该做太多解析逻辑,也不应该关心字段怎么存储。一个干净的 Fetcher 至少要考虑以下内容:

  • headers;
  • timeout;
  • session;
  • cookie;
  • 重试;
  • 退避;
  • 异常处理;
  • 请求间隔。
  • 6.1 headers 设置

    很多网站会对空 User-Agent 的请求不友好。我们可以设置一个正常浏览器 UA,同时带上 Referer。

    示例:

    DEFAULT_HEADERS = {
    "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.6",
    "Connection": "keep-alive",
    }

    这里不需要伪装得很夸张。普通 UA、合理频率、清晰边界,已经足够。

    6.2 timeout

    爬虫一定要设置 timeout。不要让一个请求无限等待。

    本文采用:

    timeout=(5, 15)

    含义是:

  • 连接超时 5 秒;
  • 读取超时 15 秒。
  • 真实项目中,如果目标站点较慢,可以适当调大读取超时。

    6.3 session/cookie

    requests.Session() 可以复用连接,也方便统一 headers 和 cookies。

    如果目标页面不需要登录,一般不需要手动设置 cookie。 如果网站依赖基础 cookie 才能展示页面,可以在合规前提下通过 Session 保持访问状态。

    本文示例代码保留了 cookies 参数,但默认不使用。

    6.4 失败处理:重试与退避

    常见失败包括:

  • 网络波动;
  • 连接超时;
  • 服务器临时 500;
  • 429 请求过快;
  • 502/503 网关错误。
  • 重试不是无限重试。推荐做法是:

    最多重试 3 次,每次失败后等待更长时间。

    例如:

    第 1 次失败后等待 1 秒
    第 2 次失败后等待 2 秒
    第 3 次失败后等待 4 秒

    这种方式叫指数退避。它能减少对目标站点的压力,也能避免脚本在临时故障中直接失败。


    7️⃣ 核心实现:解析层(Parser)

    解析层负责把 HTML 变成结构化数据。

    老年大学课程目录页通常有两种结构:

    第一种是卡片式:

    <div class="course-card">
    <a href="/course/1001.html">智能手机摄影基础班</a>
    <span class="category">数字生活</span>
    <span class="time">2026-03-10 09:00</span>
    <span class="campus">东城校区 302 教室</span>
    </div>

    第二种是表格式:

    <tr>
    <td>智能手机摄影基础班</td>
    <td>数字生活</td>
    <td>2026-03-10 09:00</td>
    <td>东城校区 302 教室</td>
    <td><a href="/course/1001.html">详情</a></td>
    </tr>

    本文的解析器同时兼容这两类结构,并提供容错逻辑。

    7.1 列表页如何拿详情链接

    一般从列表页中找到课程卡片或课程表格行,然后提取 a[href]。

    关键点是要把相对链接转成绝对链接:

    from urllib.parse import urljoin

    absolute_url = urljoin(base_url, href)

    例如:

    base_url = https://example.org/elder-university/courses.html
    href = /course/1001.html
    结果 = https://example.org/course/1001.html

    7.2 详情页如何抽字段

    详情页常见字段结构包括:

  • dl/dt/dd;
  • 表格;
  • 普通段落;
  • 标题 + 文本描述;
  • JSON 片段。
  • 本文先用 CSS/BS4 抽取常见标签,再用文本规则兜底。对于字段名,兼容一些同义写法,例如:

    课程类别 / 类别 / 分类
    开课时间 / 上课时间 / 时间
    校区 / 地点 / 上课地点

    7.3 缺失字段怎么办

    真实网页经常缺字段。我的习惯是:

  • 能从列表页拿到的字段先保存;
  • 详情页能补充就覆盖;
  • 详情页缺失时使用列表页字段;
  • 仍然缺失时填空字符串;
  • 不因为单个字段缺失导致整条数据报错。
  • 这在目录型项目里很重要。因为课程目录可能有旧数据、临时课程、取消课程、补充通知,页面格式未必完全一致。


    8️⃣ 数据存储与导出(Storage)

    本文选择 SQLite + CSV。

    SQLite 适合本地小项目,优点是:

  • 不需要安装数据库服务;
  • 单文件保存;
  • 支持 SQL 查询;
  • 支持唯一约束;
  • 比纯 CSV 更适合增量去重。
  • CSV 适合后续交给 Excel、WPS、Pandas 或 BI 工具处理。

    8.1 字段映射表
    字段名类型示例值说明
    course_name TEXT 智能手机摄影基础班 课程名称
    category TEXT 数字生活 课程分类
    start_time TEXT 2026-03-10 09:00 开课时间,先按文本保存
    campus TEXT 东城校区 302 教室 校区或上课地点
    url TEXT UNIQUE https://example.org/course/1001.html 详情页链接
    content_hash TEXT md5 值 用于内容去重
    created_at TEXT 2026-06-09 10:00:00 入库时间
    8.2 去重策略

    本文采用两层去重:

    第一层:URL 唯一。 如果同一个详情页 URL 已经入库,就更新数据,不重复插入。

    第二层:内容 hash。 将课程名、类别、开课时间、校区拼接后计算 MD5。即使 URL 不同,也可以辅助识别重复内容。

    对于课程目录型页面,URL 唯一通常已经足够。内容 hash 更多是为了后续扩展。


    9️⃣ 完整代码:可运行版本

    下面是完整的 elder_course_spider.py。复制后即可运行。

    它有两个模式:

  • –sample:使用内置示例 HTML,不需要网络;
  • –list-url:请求真实课程目录页面。
  • #!/usr/bin/env python3
    # -*- coding: utf-8 -*-

    """
    elder_course_spider.py

    采集老年大学课程目录页,提取:
    – 课程名
    – 课程类别
    – 开课时间
    – 校区/地点
    – 详情链接

    默认提供 –sample 模式,可直接离线运行。
    真实站点采集时,使用 –list-url 指定课程目录页 URL。
    """

    from __future__ import annotations

    import argparse
    import csv
    import hashlib
    import logging
    import os
    import re
    import sqlite3
    import sys
    import time
    from dataclasses import dataclass, asdict
    from datetime import datetime
    from typing import Dict, Iterable, List, Optional, Tuple
    from urllib.parse import urljoin, urlparse

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

    DEFAULT_HEADERS = {
    "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.6",
    "Connection": "keep-alive",
    }

    SAMPLE_LIST_URL = "https://example.org/elder-university/courses.html"

    SAMPLE_LIST_HTML = """
    <!doctype html>
    <html lang="zh-CN">
    <head>
    <meta charset="utf-8">
    <title>老年大学课程目录</title>
    </head>
    <body>
    <h1>2026年春季老年大学课程目录</h1>

    <div class="course-list">
    <div class="course-card">
    <a class="course-title" href="/elder-university/course/1001.html">智能手机摄影基础班</a>
    <span class="category">数字生活</span>
    <span class="time">2026-03-10 09:00</span>
    <span class="campus">东城校区 302 教室</span>
    </div>

    <div class="course-card">
    <a class="course-title" href="/elder-university/course/1002.html">国画山水入门</a>
    <span class="category">书画艺术</span>
    <span class="time">2026-03-12 14:00</span>
    <span class="campus">西城校区 201 画室</span>
    </div>

    <div class="course-card">
    <a class="course-title" href="/elder-university/course/1003.html">合唱与声乐训练</a>
    <span class="category">音乐舞台</span>
    <span class="time">2026-03-15 10:00</span>
    <span class="campus">中心校区 多功能厅</span>
    </div>
    </div>

    <table class="course-table">
    <thead>
    <tr>
    <th>课程名</th>
    <th>课程类别</th>
    <th>开课时间</th>
    <th>校区/地点</th>
    <th>链接</th>
    </tr>
    </thead>
    <tbody>
    <tr>
    <td>太极拳养生班</td>
    <td>健康养生</td>
    <td>2026-03-18 08:30</td>
    <td>南城校区 操场</td>
    <td><a href="/elder-university/course/1004.html">查看详情</a></td>
    </tr>
    <tr>
    <td>短视频剪辑体验课</td>
    <td>数字生活</td>
    <td>2026-03-20 15:00</td>
    <td>北城校区 机房A</td>
    <td><a href="/elder-university/course/1005.html">查看详情</a></td>
    </tr>
    </tbody>
    </table>
    </body>
    </html>
    """

    SAMPLE_DETAIL_HTML_MAP: Dict[str, str] = {
    "https://example.org/elder-university/course/1001.html": """
    <html><body>
    <h1>智能手机摄影基础班</h1>
    <dl class="course-info">
    <dt>课程类别</dt><dd>数字生活</dd>
    <dt>开课时间</dt><dd>2026-03-10 09:00</dd>
    <dt>校区/地点</dt><dd>东城校区 302 教室</dd>
    </dl>
    <div class="content">本课程适合希望学习手机拍照、构图与日常修图的学员。</div>
    </body></html>
    """
    ,
    "https://example.org/elder-university/course/1002.html": """
    <html><body>
    <h1>国画山水入门</h1>
    <table class="detail-table">
    <tr><th>类别</th><td>书画艺术</td></tr>
    <tr><th>上课时间</th><td>2026-03-12 14:00</td></tr>
    <tr><th>上课地点</th><td>西城校区 201 画室</td></tr>
    </table>
    </body></html>
    """
    ,
    "https://example.org/elder-university/course/1003.html": """
    <html><body>
    <h1 class="title">合唱与声乐训练</h1>
    <p>课程类别:音乐舞台</p>
    <p>开课时间:2026-03-15 10:00</p>
    <p>地点:中心校区 多功能厅</p>
    </body></html>
    """
    ,
    "https://example.org/elder-university/course/1004.html": """
    <html><body>
    <h1>太极拳养生班</h1>
    <p>分类:健康养生</p>
    <p>时间:2026-03-18 08:30</p>
    <p>校区:南城校区 操场</p>
    </body></html>
    """
    ,
    "https://example.org/elder-university/course/1005.html": """
    <html><body>
    <h1>短视频剪辑体验课</h1>
    <dl>
    <dt>课程类别</dt><dd>数字生活</dd>
    <dt>开课时间</dt><dd>2026-03-20 15:00</dd>
    <dt>校区/地点</dt><dd>北城校区 机房A</dd>
    </dl>
    </body></html>
    """
    ,
    }

    @dataclass
    class Course:
    course_name: str
    category: str
    start_time: str
    campus: str
    url: str
    content_hash: str = ""
    created_at: str = ""

    def fill_meta(self) –> "Course":
    raw = "|".join([
    self.course_name.strip(),
    self.category.strip(),
    self.start_time.strip(),
    self.campus.strip(),
    self.url.strip(),
    ])
    self.content_hash = hashlib.md5(raw.encode("utf-8")).hexdigest()
    self.created_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    return self

    def setup_logger(level: str = "INFO") –> None:
    numeric_level = getattr(logging, level.upper(), logging.INFO)
    logging.basicConfig(
    level=numeric_level,
    format="%(asctime)s | %(levelname)s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
    )

    def clean_text(value: Optional[str]) –> str:
    if not value:
    return ""
    value = re.sub(r"\\s+", " ", value)
    value = value.replace("\\u3000", " ")
    return value.strip(" \\t\\r\\n::")

    def normalize_url(url: str) –> str:
    return clean_text(url)

    def safe_join(base_url: str, href: str) –> str:
    href = clean_text(href)
    if not href:
    return ""
    return normalize_url(urljoin(base_url, href))

    def extract_by_label_text(text: str, labels: Iterable[str]) –> str:
    """
    从普通文本中按标签抽取字段。
    例如:
    课程类别:数字生活
    开课时间:2026-03-10 09:00
    """

    if not text:
    return ""

    label_pattern = "|".join(re.escape(label) for label in labels)
    pattern = rf"(?:{label_pattern})\\s*[::]\\s*([^\\n\\r;;]+)"
    match = re.search(pattern, text)
    if match:
    return clean_text(match.group(1))
    return ""

    class Fetcher:
    def __init__(
    self,
    timeout: Tuple[int, int] = (5, 15),
    delay: float = 1.0,
    headers: Optional[Dict[str, str]] = None,
    cookies: Optional[Dict[str, str]] = None,
    ) –> None:
    self.timeout = timeout
    self.delay = delay
    self.session = requests.Session()
    self.session.headers.update(headers or DEFAULT_HEADERS)
    if cookies:
    self.session.cookies.update(cookies)

    retry = Retry(
    total=3,
    connect=3,
    read=3,
    status=3,
    backoff_factor=1.0,
    status_forcelist=(429, 500, 502, 503, 504),
    allowed_methods=frozenset(["GET"]),
    raise_on_status=False,
    )
    adapter = HTTPAdapter(max_retries=retry)
    self.session.mount("http://", adapter)
    self.session.mount("https://", adapter)

    def fetch_text(self, url: str, referer: Optional[str] = None) –> str:
    if not url:
    raise ValueError("url 不能为空")

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

    logging.info("请求页面:%s", url)

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

    if self.delay > 0:
    time.sleep(self.delay)

    if response.status_code >= 400:
    raise RuntimeError(f"请求失败:{response.status_code} {url}")

    response.encoding = response.apparent_encoding or response.encoding
    return response.text

    class SampleFetcher:
    """
    离线示例 Fetcher,保证本文代码不依赖外部网站也能跑通。
    """

    def __init__(self, delay: float = 0.0) –> None:
    self.delay = delay

    def fetch_text(self, url: str, referer: Optional[str] = None) –> str:
    if self.delay > 0:
    time.sleep(self.delay)

    if url == SAMPLE_LIST_URL:
    logging.info("读取内置示例列表页:%s", url)
    return SAMPLE_LIST_HTML

    if url in SAMPLE_DETAIL_HTML_MAP:
    logging.info("读取内置示例详情页:%s", url)
    return SAMPLE_DETAIL_HTML_MAP[url]

    raise RuntimeError(f"示例数据中没有这个 URL:{url}")

    class CourseParser:
    category_labels = ["课程类别", "类别", "分类"]
    time_labels = ["开课时间", "上课时间", "时间", "授课时间"]
    campus_labels = ["校区/地点", "上课地点", "地点", "校区", "教室"]

    def parse_list(self, html: str, base_url: str) –> List[Course]:
    soup = BeautifulSoup(html, "lxml")
    courses: List[Course] = []

    card_courses = self._parse_card_list(soup, base_url)
    table_courses = self._parse_table_list(soup, base_url)

    courses.extend(card_courses)
    courses.extend(table_courses)

    deduped: Dict[str, Course] = {}
    for item in courses:
    if not item.url:
    continue
    deduped[item.url] = item

    logging.info("列表页解析到 %d 条课程链接", len(deduped))
    return list(deduped.values())

    def _parse_card_list(self, soup: BeautifulSoup, base_url: str) –> List[Course]:
    result: List[Course] = []

    selectors = [
    ".course-card",
    ".course-item",
    "li.course",
    ".item",
    ]

    cards = []
    for selector in selectors:
    selected = soup.select(selector)
    if selected:
    cards.extend(selected)

    for card in cards:
    link = card.select_one("a[href]")
    if not link:
    continue

    course_name = clean_text(link.get_text())
    url = safe_join(base_url, link.get("href", ""))

    category = self._select_text(card, [".category", ".type", ".course-category"])
    start_time = self._select_text(card, [".time", ".start-time", ".course-time"])
    campus = self._select_text(card, [".campus", ".place", ".location", ".address"])

    if not course_name:
    course_name = self._guess_title_from_card(card)

    if course_name and url:
    result.append(Course(
    course_name=course_name,
    category=category,
    start_time=start_time,
    campus=campus,
    url=url,
    ))

    return result

    def _parse_table_list(self, soup: BeautifulSoup, base_url: str) –> List[Course]:
    result: List[Course] = []

    for table in soup.select("table"):
    rows = table.select("tr")
    if len(rows) <= 1:
    continue

    headers = [clean_text(th.get_text()) for th in rows[0].select("th,td")]

    for row in rows[1:]:
    cells = row.select("td")
    if len(cells) < 2:
    continue

    cell_texts = [clean_text(td.get_text()) for td in cells]
    link = row.select_one("a[href]")
    url = safe_join(base_url, link.get("href", "")) if link else ""

    course_name = self._get_table_value(headers, cell_texts, ["课程名", "课程名称", "名称"], default_index=0)
    category = self._get_table_value(headers, cell_texts, ["课程类别", "类别", "分类"], default_index=1)
    start_time = self._get_table_value(headers, cell_texts, ["开课时间", "上课时间", "时间"], default_index=2)
    campus = self._get_table_value(headers, cell_texts, ["校区/地点", "地点", "校区"], default_index=3)

    if course_name and url:
    result.append(Course(
    course_name=course_name,
    category=category,
    start_time=start_time,
    campus=campus,
    url=url,
    ))

    return result

    def parse_detail(self, html: str, url: str, fallback: Optional[Course] = None) –> Course:
    soup = BeautifulSoup(html, "lxml")
    page_text = soup.get_text("\\n")

    course_name = self._extract_title(soup)
    category = self._extract_from_structured_blocks(soup, self.category_labels)
    start_time = self._extract_from_structured_blocks(soup, self.time_labels)
    campus = self._extract_from_structured_blocks(soup, self.campus_labels)

    if not category:
    category = extract_by_label_text(page_text, self.category_labels)
    if not start_time:
    start_time = extract_by_label_text(page_text, self.time_labels)
    if not campus:
    campus = extract_by_label_text(page_text, self.campus_labels)

    if fallback:
    course_name = course_name or fallback.course_name
    category = category or fallback.category
    start_time = start_time or fallback.start_time
    campus = campus or fallback.campus

    course = Course(
    course_name=clean_text(course_name),
    category=clean_text(category),
    start_time=clean_text(start_time),
    campus=clean_text(campus),
    url=url,
    )
    return course.fill_meta()

    def _select_text(self, root, selectors: List[str]) –> str:
    for selector in selectors:
    node = root.select_one(selector)
    if node:
    value = clean_text(node.get_text())
    if value:
    return value
    return ""

    def _guess_title_from_card(self, card) –> str:
    for selector in ["h1", "h2", "h3", ".title", ".name"]:
    node = card.select_one(selector)
    if node:
    value = clean_text(node.get_text())
    if value:
    return value
    return clean_text(card.get_text())[:80]

    def _get_table_value(
    self,
    headers: List[str],
    values: List[str],
    candidates: List[str],
    default_index: int,
    ) –> str:
    for candidate in candidates:
    for index, header in enumerate(headers):
    if candidate in header and index < len(values):
    return clean_text(values[index])

    if default_index < len(values):
    return clean_text(values[default_index])

    return ""

    def _extract_title(self, soup: BeautifulSoup) –> str:
    for selector in ["h1", ".title", ".course-title", "h2"]:
    node = soup.select_one(selector)
    if node:
    value = clean_text(node.get_text())
    if value:
    return value

    if soup.title:
    return clean_text(soup.title.get_text())

    return ""

    def _extract_from_structured_blocks(self, soup: BeautifulSoup, labels: List[str]) –> str:
    dl_value = self._extract_from_dl(soup, labels)
    if dl_value:
    return dl_value

    table_value = self._extract_from_table(soup, labels)
    if table_value:
    return table_value

    return ""

    def _extract_from_dl(self, soup: BeautifulSoup, labels: List[str]) –> str:
    for dt in soup.select("dt"):
    label = clean_text(dt.get_text())
    if not any(candidate in label for candidate in labels):
    continue

    dd = dt.find_next_sibling("dd")
    if dd:
    value = clean_text(dd.get_text())
    if value:
    return value
    return ""

    def _extract_from_table(self, soup: BeautifulSoup, labels: List[str]) –> str:
    for row in soup.select("tr"):
    cells = row.select("th,td")
    if len(cells) < 2:
    continue

    key = clean_text(cells[0].get_text())
    value = clean_text(cells[1].get_text())

    if any(candidate in key for candidate in labels) and value:
    return value

    return ""

    class CourseStorage:
    def __init__(self, db_path: str, csv_path: str) –> None:
    self.db_path = db_path
    self.csv_path = csv_path

    def init_db(self) –> None:
    os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
    with sqlite3.connect(self.db_path) as conn:
    conn.execute(
    """
    CREATE TABLE IF NOT EXISTS courses (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    course_name TEXT NOT NULL,
    category TEXT,
    start_time TEXT,
    campus TEXT,
    url TEXT NOT NULL UNIQUE,
    content_hash TEXT,
    created_at TEXT
    )
    """

    )
    conn.execute("CREATE INDEX IF NOT EXISTS idx_courses_category ON courses(category)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_courses_start_time ON courses(start_time)")
    conn.commit()

    def upsert_courses(self, courses: List[Course]) –> int:
    if not courses:
    return 0

    self.init_db()

    sql = """
    INSERT INTO courses (
    course_name,
    category,
    start_time,
    campus,
    url,
    content_hash,
    created_at
    )
    VALUES (?, ?, ?, ?, ?, ?, ?)
    ON CONFLICT(url) DO UPDATE SET
    course_name = excluded.course_name,
    category = excluded.category,
    start_time = excluded.start_time,
    campus = excluded.campus,
    content_hash = excluded.content_hash,
    created_at = excluded.created_at
    """

    rows = [
    (
    item.course_name,
    item.category,
    item.start_time,
    item.campus,
    item.url,
    item.content_hash,
    item.created_at,
    )
    for item in courses
    ]

    with sqlite3.connect(self.db_path) as conn:
    conn.executemany(sql, rows)
    conn.commit()

    return len(rows)

    def export_csv(self) –> None:
    os.makedirs(os.path.dirname(self.csv_path), exist_ok=True)

    with sqlite3.connect(self.db_path) as conn:
    conn.row_factory = sqlite3.Row
    rows = conn.execute(
    """
    SELECT
    course_name,
    category,
    start_time,
    campus,
    url,
    content_hash,
    created_at
    FROM courses
    ORDER BY start_time ASC, course_name ASC
    """

    ).fetchall()

    with open(self.csv_path, "w", newline="", encoding="utf-8-sig") as f:
    writer = csv.DictWriter(
    f,
    fieldnames=[
    "course_name",
    "category",
    "start_time",
    "campus",
    "url",
    "content_hash",
    "created_at",
    ],
    )
    writer.writeheader()
    for row in rows:
    writer.writerow(dict(row))

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

    def read_preview(self, limit: int = 5) –> List[Dict[str, str]]:
    with sqlite3.connect(self.db_path) as conn:
    conn.row_factory = sqlite3.Row
    rows = conn.execute(
    """
    SELECT course_name, category, start_time, campus, url
    FROM courses
    ORDER BY start_time ASC, course_name ASC
    LIMIT ?
    """
    ,
    (limit,),
    ).fetchall()
    return [dict(row) for row in rows]

    def crawl_courses(
    list_url: str,
    fetcher,
    parser: CourseParser,
    max_items: int = 100,
    ) –> List[Course]:
    list_html = fetcher.fetch_text(list_url)
    list_courses = parser.parse_list(list_html, list_url)

    if max_items > 0:
    list_courses = list_courses[:max_items]

    final_courses: List[Course] = []

    for index, item in enumerate(list_courses, start=1):
    try:
    logging.info("抓取详情页 %d/%d:%s", index, len(list_courses), item.url)
    detail_html = fetcher.fetch_text(item.url, referer=list_url)
    course = parser.parse_detail(detail_html, item.url, fallback=item)
    final_courses.append(course)
    except Exception as exc:
    logging.warning("详情页抓取失败,使用列表页字段兜底:%s | %s", item.url, exc)
    final_courses.append(item.fill_meta())

    return final_courses

    def print_preview(rows: List[Dict[str, str]]) –> None:
    if not rows:
    print("暂无数据")
    return

    print("\\n采集结果预览:")
    print("-" * 110)
    for row in rows:
    print(f"课程名:{row.get('course_name', '')}")
    print(f"类别:{row.get('category', '')}")
    print(f"开课时间:{row.get('start_time', '')}")
    print(f"校区/地点:{row.get('campus', '')}")
    print(f"链接:{row.get('url', '')}")
    print("-" * 110)

    def parse_args(argv: Optional[List[str]] = None) –> argparse.Namespace:
    parser = argparse.ArgumentParser(
    description="老年大学课程目录页采集工具"
    )
    parser.add_argument(
    "–sample",
    action="store_true",
    help="使用内置示例 HTML,保证离线可运行",
    )
    parser.add_argument(
    "–list-url",
    default="",
    help="真实课程目录页 URL,例如 https://example.org/courses.html",
    )
    parser.add_argument(
    "–output-dir",
    default="output",
    help="输出目录,默认 output",
    )
    parser.add_argument(
    "–delay",
    type=float,
    default=1.0,
    help="请求间隔秒数,默认 1.0",
    )
    parser.add_argument(
    "–max-items",
    type=int,
    default=100,
    help="最多抓取多少条课程,默认 100",
    )
    parser.add_argument(
    "–log-level",
    default="INFO",
    help="日志级别,默认 INFO",
    )
    return parser.parse_args(argv)

    def main(argv: Optional[List[str]] = None) –> int:
    args = parse_args(argv)
    setup_logger(args.log_level)

    output_dir = args.output_dir
    db_path = os.path.join(output_dir, "courses.sqlite3")
    csv_path = os.path.join(output_dir, "courses.csv")

    if args.sample:
    list_url = SAMPLE_LIST_URL
    fetcher = SampleFetcher(delay=0.0)
    else:
    if not args.list_url:
    print("请提供 –list-url,或使用 –sample 运行内置示例。", file=sys.stderr)
    return 2

    parsed = urlparse(args.list_url)
    if parsed.scheme not in {"http", "https"}:
    print("真实采集仅支持 http/https URL。", file=sys.stderr)
    return 2

    list_url = args.list_url
    fetcher = Fetcher(delay=args.delay)

    parser = CourseParser()
    storage = CourseStorage(db_path=db_path, csv_path=csv_path)

    courses = crawl_courses(
    list_url=list_url,
    fetcher=fetcher,
    parser=parser,
    max_items=args.max_items,
    )

    inserted = storage.upsert_courses(courses)
    storage.export_csv()

    logging.info("本次处理课程数:%d", inserted)
    logging.info("SQLite 数据库:%s", db_path)
    logging.info("CSV 文件:%s", csv_path)

    preview_rows = storage.read_preview(limit=5)
    print_preview(preview_rows)

    return 0

    if __name__ == "__main__":
    raise SystemExit(main())


    9️⃣ 运行方式与结果展示(必写)

    9.1 使用内置示例运行

    在项目目录中执行:

    python elder_course_spider.py –sample

    正常情况下,会看到类似日志:

    2026-06-09 10:00:00 | INFO | 读取内置示例列表页:https://example.org/elder-university/courses.html
    2026-06-09 10:00:00 | INFO | 列表页解析到 5 条课程链接
    2026-06-09 10:00:00 | INFO | 抓取详情页 1/5:https://example.org/elder-university/course/1001.html
    2026-06-09 10:00:00 | INFO | 抓取详情页 2/5:https://example.org/elder-university/course/1002.html
    2026-06-09 10:00:00 | INFO | CSV 已导出:output/courses.csv

    9.2 指定真实课程目录页运行

    假设真实课程目录页地址是:

    https://example.org/elder-university/courses.html

    运行:

    python elder_course_spider.py \\
    –list-url "https://example.org/elder-university/courses.html" \\
    –output-dir output \\
    –delay 2 \\
    –max-items 50

    参数说明:

    参数说明
    –list-url 课程目录页地址
    –output-dir 输出目录
    –delay 请求间隔秒数
    –max-items 最多抓取课程数量
    –sample 使用内置示例 HTML
    9.3 输出位置

    运行后会生成:

    output/courses.sqlite3
    output/courses.csv

    SQLite 表名是:

    courses

    CSV 文件可以直接用 Excel 或 WPS 打开。如果中文乱码,注意文件编码是 utf-8-sig,通常对 Excel 比较友好。

    9.4 示例结果

    示例 CSV 内容如下:

    course_namecategorystart_timecampusurl
    智能手机摄影基础班 数字生活 2026-03-10 09:00 东城校区 302 教室 https://example.org/elder-university/course/1001.html
    国画山水入门 书画艺术 2026-03-12 14:00 西城校区 201 画室 https://example.org/elder-university/course/1002.html
    合唱与声乐训练 音乐舞台 2026-03-15 10:00 中心校区 多功能厅 https://example.org/elder-university/course/1003.html
    太极拳养生班 健康养生 2026-03-18 08:30 南城校区 操场 https://example.org/elder-university/course/1004.html
    短视频剪辑体验课 数字生活 2026-03-20 15:00 北城校区 机房A https://example.org/elder-university/course/1005.html
    9.5 使用 SQLite 查询

    你也可以直接查询 SQLite:

    sqlite3 output/courses.sqlite3

    进入后执行:

    SELECT course_name, category, start_time, campus
    FROM courses
    ORDER BY start_time ASC;

    按类别统计:

    SELECT category, COUNT(*) AS total
    FROM courses
    GROUP BY category
    ORDER BY total DESC;

    查询数字生活类课程:

    SELECT course_name, start_time, campus, url
    FROM courses
    WHERE category LIKE '%数字生活%';


    🔟 常见问题与排错(强烈建议写)

    10.1 遇到 403 怎么办

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

  • User-Agent 为空或异常;
  • Referer 缺失;
  • 访问频率过高;
  • 目标路径不允许访问;
  • 需要基础 cookie;
  • 服务器策略限制。
  • 处理思路:

    第一,先用浏览器打开页面,确认页面公开可访问。 第二,检查 robots.txt,确认路径是否允许访问。 第三,设置正常 headers。 第四,降低请求频率。 第五,不要尝试绕过登录或权限限制。

    可以先把 delay 调大:

    python elder_course_spider.py \\
    –list-url "https://example.org/elder-university/courses.html" \\
    –delay 3

    如果仍然 403,就要回到合规边界和页面访问规则上判断,而不是盲目加代理。

    10.2 遇到 429 怎么办

    429 表示请求过多。解决办法很直接:

  • 增大 delay;
  • 减少 max-items;
  • 不要并发;
  • 增加缓存;
  • 稍后再运行。
  • 示例:

    python elder_course_spider.py \\
    –list-url "https://example.org/elder-university/courses.html" \\
    –delay 5 \\
    –max-items 20

    对于课程目录页,不需要频繁请求。一天一次、一周一次,很多场景已经够用。

    10.3 HTML 抓到空壳怎么办

    有时浏览器里能看到课程列表,但代码拿到的 HTML 里没有课程。这通常说明页面是动态渲染。

    你可以这样排查:

  • 浏览器右键查看网页源代码;
  • 搜索课程名;
  • 如果源代码里没有课程名,说明数据不是直接写在 HTML 里;
  • 打开开发者工具 Network;
  • 刷新页面;
  • 查看 XHR/fetch 请求;
  • 找到返回 JSON 的接口。
  • 如果能找到接口,优先请求接口。因为 JSON 比 HTML 更稳定。

    如果接口不好分析,再考虑 Playwright。简单示例:

    from playwright.sync_api import sync_playwright

    def render_html(url: str) –> str:
    with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    page = browser.new_page()
    page.goto(url, wait_until="networkidle", timeout=30000)
    html = page.content()
    browser.close()
    return html

    安装:

    pip install playwright
    playwright install chromium

    但我建议不要一开始就用 Playwright。它适合动态页面,不是所有页面都需要它。

    10.4 解析报错怎么办

    解析报错通常有几种原因:

  • 页面结构变了;
  • 选择器不匹配;
  • 某些课程没有详情页;
  • 表格列数不一致;
  • 标签嵌套不规范。
  • 解决方式是提高容错,而不是假设页面永远规整。

    例如原来选择器是:

    soup.select(".course-card")

    但真实页面可能是:

    <div class="list-item">

    这时需要添加选择器:

    selectors = [
    ".course-card",
    ".course-item",
    ".list-item",
    "li.course",
    ]

    如果表格列顺序变化,最好通过表头匹配,而不是死写索引。本文代码中的 _get_table_value() 就是这种思路。

    10.5 编码或乱码怎么办

    乱码常见于老网站,尤其是 GBK、GB2312 编码页面。

    本文代码使用:

    response.encoding = response.apparent_encoding or response.encoding

    如果仍然乱码,可以手动指定:

    response.encoding = "gbk"

    或者:

    html = response.content.decode("gb18030", errors="ignore")

    CSV 导出时,本文使用:

    encoding="utf-8-sig"

    这对 Excel 打开中文比较友好。

    10.6 详情页链接拼接错误怎么办

    如果列表页 href 是相对路径,例如:

    <a href="../course/1001.html">详情</a>

    不要自己手动拼字符串,使用:

    urljoin(base_url, href)

    它会自动处理相对路径、根路径和完整 URL。

    10.7 字段缺失怎么办

    不要因为一个字段缺失就丢弃整条课程。比较稳妥的策略是:

  • 课程名和 URL 是核心字段;
  • 类别、时间、地点可以为空;
  • 列表页字段可以作为兜底;
  • 详情页字段可以覆盖列表页字段;
  • 记录 warning 日志,后续人工检查。
  • 本文代码中,如果详情页失败,会使用列表页字段兜底:

    except Exception as exc:
    logging.warning("详情页抓取失败,使用列表页字段兜底:%s | %s", item.url, exc)
    final_courses.append(item.fill_meta())

    这类小设计很实用。真实项目里,网络失败、页面下架、临时维护都很常见,爬虫不能太脆。


    1️⃣1️⃣ 进阶优化(可选但加分)

    11.1 并发优化

    课程目录项目通常不需要高并发,但如果你采集的是多个城市、多所老年大学、多栏目页面,可以适度并发。

    线程池示例:

    from concurrent.futures import ThreadPoolExecutor, as_completed

    def fetch_detail_task(fetcher, parser, list_url, item):
    try:
    html = fetcher.fetch_text(item.url, referer=list_url)
    return parser.parse_detail(html, item.url, fallback=item)
    except Exception:
    return item.fill_meta()

    def crawl_details_concurrent(fetcher, parser, list_url, items, workers=3):
    results = []
    with ThreadPoolExecutor(max_workers=workers) as executor:
    futures = [
    executor.submit(fetch_detail_task, fetcher, parser, list_url, item)
    for item in items
    ]

    for future in as_completed(futures):
    results.append(future.result())

    return results

    注意:并发数不要太高。对于教育类公开信息页面,workers=2 或 workers=3 已经足够。

    11.2 asyncio 版本思路

    如果数据量更大,可以考虑 aiohttp。但异步代码会增加复杂度,尤其是重试、限速、异常处理要写得更细。

    伪流程:

    创建 aiohttp.ClientSession
    ↓
    使用 Semaphore 控制并发
    ↓
    异步请求详情页
    ↓
    解析 HTML
    ↓
    批量写入数据库

    不要为了异步而异步。项目规模没上来之前,requests 更容易维护。

    11.3 断点续跑

    断点续跑的核心是:已经抓过的 URL 不重复抓。

    可以在 SQLite 中先查出已有 URL:

    def load_existing_urls(db_path: str) –> set[str]:
    if not os.path.exists(db_path):
    return set()

    with sqlite3.connect(db_path) as conn:
    rows = conn.execute("SELECT url FROM courses").fetchall()

    return {row[0] for row in rows}

    然后过滤:

    existing_urls = load_existing_urls("output/courses.sqlite3")
    pending_items = [item for item in list_courses if item.url not in existing_urls]

    如果课程详情可能更新,也可以增加更新时间字段,定期重抓最近几页或最近几天的数据。

    11.4 日志与监控

    日志不要只打印“成功”。我建议至少记录:

  • 列表页 URL;
  • 解析到多少条课程;
  • 成功详情页数量;
  • 失败详情页数量;
  • CSV 输出路径;
  • SQLite 输出路径;
  • 运行耗时。
  • 可以给主流程加一个简单计时:

    start = time.time()

    # 执行采集

    cost = time.time() – start
    logging.info("任务完成,耗时 %.2f 秒", cost)

    如果定时运行,可以把日志写入文件:

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

    11.5 定时任务

    Linux/macOS 可以使用 cron:

    crontab -e

    添加:

    0 8 * * 1 cd /path/to/elder_course_spider && /usr/bin/python3 elder_course_spider.py –list-url "https://example.org/elder-university/courses.html" –output-dir output >> output/cron.log 2>&1

    含义是每周一早上 8 点运行一次。

    Windows 可以使用任务计划程序,配置 Python 路径和脚本路径即可。

    如果任务变复杂,比如多个站点、失败告警、数据入仓,可以考虑 Airflow、Prefect 或自建调度服务。

    11.6 配置化多个站点

    如果要采集多个老年大学网站,建议写一个配置文件:

    sites:
    – name: city_a_elder_university
    list_url: https://example.org/a/courses.html
    parser_type: default
    delay: 2

    – name: city_b_elder_university
    list_url: https://example.org/b/course–list.html
    parser_type: table
    delay: 3

    然后脚本循环读取配置。不同站点如果 HTML 差异较大,就给每个站点写一个独立 parser。

    不要强行用一个超级复杂的解析器兼容所有网页。维护到后面会很难受。适度抽象,适度重复,反而更稳。

    11.7 数据质量校验

    采集完成后,可以加一些检查规则:

  • 课程名不能为空;
  • URL 必须以 http 或 https 开头;
  • 开课时间为空的记录单独输出;
  • 校区为空的记录单独输出;
  • 类别过多或过乱时做归一化。
  • 示例:

    def validate_course(course: Course) –> List[str]:
    errors = []

    if not course.course_name:
    errors.append("课程名为空")

    if not course.url.startswith(("http://", "https://")):
    errors.append("URL 格式异常")

    if not course.start_time:
    errors.append("开课时间为空")

    return errors

    数据采集只是第一步,数据是否可信,往往取决于后面的校验和清洗。


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

    这篇文章完成了一个完整的老年大学课程目录页采集项目。我们从需求出发,确定了目标字段:

    课程名、课程类别、开课时间、校区/地点、链接

    然后设计了整体流程:

    采集 → 解析 → 清洗 → 存储

    技术上选择了比较轻量的组合:

    requests + BeautifulSoup + lxml + SQLite + CSV

    代码层面实现了:

  • 请求层 Fetcher;
  • 离线示例 SampleFetcher;
  • 列表页解析;
  • 详情页解析;
  • 字段容错;
  • URL 去重;
  • SQLite 入库;
  • CSV 导出;
  • 命令行运行;
  • 示例结果预览。
  • 这个项目不算花哨,但非常贴近真实工作。很多数据采集任务并不是一上来就要解决特别复杂的反爬问题,而是要先把“字段定义、页面结构、解析规则、异常兜底、存储格式”这些基本环节做好。

    下一步可以继续扩展:

  • 用 Scrapy 管理多站点采集;
  • 用 Playwright 处理动态渲染页面;
  • 用 Pandas 做课程类别统计;
  • 用 Streamlit 做一个课程检索小页面;
  • 增加定时任务,定期发现新增课程;
  • 增加数据质量报告,标记缺失字段;
  • 将 SQLite 换成 MySQL 或 PostgreSQL;
  • 将采集结果推送到内部表格或知识库。
  • 如果只把爬虫理解为“把网页内容抓下来”,那它确实很简单。但如果把它看作一个小型数据工程项目,里面就有不少值得打磨的细节。尤其是课程目录这种场景,页面温和、字段清晰、价值明确,很适合作为 Python 爬虫实战的入门到进阶案例。

    我的建议是:先把本文这份单文件代码跑通,再找一个公开课程目录页做适配。不要急着加并发,也不要急着换框架。先让数据稳定落地,再谈性能和规模。很多时候,稳比快更重要。

    🌟 文末

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

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

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

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

    ✅ 讲得清楚(原理)|✅ 跑得起来(代码)|✅ 用得上(场景)|✅ 扛得住(工程化)

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

    ✅ 互动征集

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

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


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


    ✅ 免责声明

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

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

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

    赞(0)
    未经允许不得转载:171主机测评 » Python 实战:爬取老年大学课程目录页,采集课程名、类别、开课时间、校区与详情链接
    分享到: 更多 (0)

    评论 抢沙发

    • 昵称 (必填)
    • 邮箱 (必填)
    • 网址