Python 爬取“专业认证结果公开目录”：从公开目录到机构-专业双维度数据表的完整实战

喵手

285人浏览 · 2026-06-29 16:25:56

喵手 · 2026-06-29 16:25:56 发布

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

全文目录：

🌟 开篇语

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

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

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

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

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

💕订阅后更新会优先推送，按目录学习更高效💯～

0️⃣ 前言（Preface）

这篇文章要做的事很明确：用 Python 抓取一个“专业认证结果公开目录”类公开页面，把页面里的学校、专业、认证状态、有效期、年份、详情链接等字段抽取出来，最后导出为 CSV，同时写入 SQLite，方便后续做机构-专业双维度分析。

读完之后，你能拿到三样东西：

第一，知道这类“公开目录”页面应该怎么拆解：列表页、详情页、字段、分页、去重、存储。

第二，拿到一套可以直接落地的 Python 爬虫项目结构，代码包含请求层、解析层、存储层、日志、重试、频率控制、断点思路。

第三，理解为什么“学校-专业”不能简单当成一行文本保存，而是应该建成机构表、专业表、认证事实表三层结构。

我写这类目录爬虫时，一般不会一上来就追求高并发。公开目录数据通常规模不算夸张，更重要的是稳定、可复查、可重复运行。尤其是认证类数据，字段看似简单，但“有效期”“认证状态”“年份”“备注”这些内容经常混在同一个单元格里，如果前期不把结构设计好，后面做统计时会非常难受。

1️⃣ 摘要（Abstract）

本文基于 Python、requests、BeautifulSoup、lxml、SQLite 和 CSV，完成一个面向“专业认证结果公开目录”的采集程序：从公开列表页采集详情链接，再进入详情页解析学校、专业、认证状态、有效期、年份、链接等字段，最后导出为可分析的数据文件和数据库表。

读完本文，你可以获得：

一套适合公开目录页的 Python 爬虫工程模板。
一个“采集 → 解析 → 清洗 → 存储”的完整落地流程。
一个机构-专业双维度建表方案，适合后续做学校维度、专业维度、年份维度的统计分析。

本文默认目标页面是公开可访问的目录页，不涉及登录、不涉及个人隐私、不绕过权限限制。所有实现都围绕“技术学习、公开数据整理、低频访问、可复现采集”展开。

2️⃣ 背景与需求（Why）

2.1 为什么要爬专业认证结果公开目录

专业认证结果公开目录通常用于展示某些学校、某些专业在某个年份或某个周期内的认证状态。对于数据分析来说，这类数据有几个很典型的价值：

首先，它适合做信息聚合。不同学校、不同专业、不同年份的认证结果往往散落在公告页、附件页、详情页里，人工查找成本高。把它整理成结构化数据后，可以快速按学校、专业、年份进行检索。

其次，它适合做数据分析。例如，统计某一年通过认证的专业数量，统计某所学校拥有多少个有效认证专业，或者观察某一类专业在不同年份的认证趋势。

再次，它适合做自动化更新。认证目录可能会不定期更新，如果每次都靠人工打开网页、复制表格、清洗字段，很容易出错。用脚本低频执行，可以把重复劳动压缩到很低。

我个人处理这类数据时，最在意两件事：第一，字段要干净；第二，数据要能追溯。所谓可追溯，就是每条记录最好保留来源链接、详情页链接和抓取时间。后面如果有人问“这条数据从哪来的”，你能回到源页面核对，而不是只剩下一张孤零零的表。

2.2 目标站点类型

本文不绑定某一个固定网站，而是面向常见的“公开目录页 + 详情页”结构。目标页面一般有以下几种形态：

静态 HTML 表格：最理想，列表页里直接包含学校、专业、认证状态、有效期、年份。
列表页 + 详情页：列表页只给标题或链接，详情页里展示完整字段。
HTML 页面中嵌入附件：比如 PDF、Excel、图片表格。
动态渲染页面：浏览器能看到数据，但 requests 抓到的是空壳，需要分析接口或使用 Playwright。
接口型页面：前端通过 API 返回 JSON，后端接口可以直接提供结构化数据。

本文主线选择“静态 HTML / 半结构化详情页”方案，使用 requests + BeautifulSoup + lxml。原因很简单：这一组合足够轻量、可读性高、调试成本低，适合多数公开目录页。如果后续目标站点是动态渲染，再扩展到 Playwright 或 Scrapy。

2.3 目标字段清单

本项目要抽取的核心字段如下：

字段	说明	示例
学校	学校或机构名称	某某大学
专业	专业名称	计算机科学与技术
认证状态	当前认证结果	通过认证
有效期	认证有效期文本	2023 年 1 月至 2028 年 12 月
年份	认证年份或发布年份	2023
链接	详情页或来源页 URL	https://example.com/detail/1001

除了这些核心字段，我建议额外保存几个技术字段：

字段	说明
source_url	列表页来源
detail_url	详情页来源
scraped_at	抓取时间
record_hash	记录唯一哈希
valid_start	有效期开始时间，若能解析
valid_end	有效期结束时间，若能解析

这些字段看似多余，但后期排错和增量更新时非常有用。

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

爬虫不是“能抓就抓”。尤其是公开目录、公告、认证结果这类页面，本身可能是为了公众查询而设置，但这并不代表可以高频访问、攻击式并发或绕过限制。

3.1 robots.txt 基本说明

robots.txt 是网站给爬虫看的访问建议文件，通常位于网站根路径下，例如：

https://example.com/robots.txt

它可能声明哪些路径允许访问，哪些路径不建议抓取。正式采集前，建议先查看 robots.txt。即使 robots.txt 不是法律文本，也应该把它当成技术边界和礼貌规则来看待。

本文的程序不会自动绕过 robots 限制，也不会使用任何规避手段。实际使用时，如果 robots.txt 明确不希望抓取某些路径，就不要抓这些路径。

3.2 频率控制

本文代码默认加入请求间隔。公开目录页通常没有必要高并发访问。我的建议是：

单线程起步。
每次请求之间 sleep 1 到 3 秒。
遇到 429 或明显限流时主动降低频率。
不要为了“快”而开几十上百个并发。

认证目录这类数据变化频率不高，慢一点并不影响结果，但太快可能影响对方服务，也可能让自己拿到不稳定数据。

3.3 不采集敏感信息

本文只采集公开目录中的机构、专业、认证状态、有效期、年份、链接等非敏感字段。不采集个人姓名、身份证号、手机号、邮箱、账号、登录态信息，也不处理任何非公开数据。

3.4 不绕过付费、登录或权限限制

如果目标站点需要登录、付费、验证码或权限校验，本文方案不提供绕过方法。遇到这种情况，合理做法是：

使用官方公开下载入口。
向数据发布方申请授权。
使用站点提供的开放 API。
放弃自动化采集，只做人工整理。

技术分享的边界要清楚。爬虫本质是自动化访问工具，不应该被用来突破权限边界。

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

4.1 静态、动态、API：本文属于哪种

本文默认目标页面属于“静态 HTML 或半结构化 HTML”。也就是说，用 requests 请求页面后，可以在返回 HTML 中看到目标字段或详情链接。

如果你打开浏览器能看到数据，但 requests 返回的 HTML 里没有数据，这通常说明页面是动态渲染。这时有两条路：

第一，打开浏览器开发者工具，查看 Network 面板，寻找返回 JSON 的接口。

第二，使用 Playwright 模拟浏览器渲染页面，再读取渲染后的 DOM。

不过，本文先不走浏览器自动化路线。原因是 requests + bs4 的维护成本更低，而且对公开目录这种数据量不大的场景更合适。

4.2 整体流程

完整流程可以写成一句话：

采集列表页 → 解析详情链接 → 请求详情页 → 抽取字段 → 清洗字段 → 去重 → 写入 CSV 和 SQLite

也可以用流程图表示：

4.3 为什么选 requests + bs4 + lxml

我选 requests 是因为它足够稳定，适合处理常规 HTTP 请求；选 BeautifulSoup 是因为它对不规范 HTML 的容错很好；选 lxml 是因为它解析速度更快，也支持 XPath。

这套组合的优点是：

代码轻，依赖少。
调试直观。
适合公开目录页。
容错能力不错。
后续容易改造成 Scrapy 项目。

如果数据量很大，或者页面层级复杂，再考虑 Scrapy。如果页面强依赖 JavaScript，再考虑 Playwright。

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

5.1 Python 版本

建议使用 Python 3.10 或以上版本。本文代码在 Python 3.10+ 语法下编写。

查看版本：

python --version

5.2 创建虚拟环境

mkdir cert-directory-spider
cd cert-directory-spider

python -m venv .venv

# Windows
.venv\Scripts\activate

# macOS / Linux
source .venv/bin/activate

5.3 安装依赖

创建 requirements.txt：

requests==2.32.3
beautifulsoup4==4.12.3
lxml==5.3.0
tenacity==9.0.0
python-dateutil==2.9.0.post0

安装：

pip install -r requirements.txt

5.4 推荐项目结构

cert-directory-spider/
├── README.md
├── requirements.txt
├── run.py
├── cert_spider/
│   ├── __init__.py
│   ├── settings.py
│   ├── models.py
│   ├── fetcher.py
│   ├── parser.py
│   ├── cleaner.py
│   ├── storage.py
│   └── crawler.py
├── examples/
│   ├── list.html
│   ├── detail_1001.html
│   └── detail_1002.html
└── data/
    ├── output.csv
    └── cert_results.sqlite3

为什么要拆这么细？因为爬虫最怕后期改不动。请求、解析、清洗、存储混在一个文件里，刚开始很爽，三天后维护就很痛苦。拆成模块后，某个页面结构变了，只改 parser；数据库表要调整，只改 storage；请求策略要加代理或重试，只改 fetcher。

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

请求层负责做四件事：

设置 headers，比如 User-Agent 和 Referer。
设置 timeout，避免请求一直卡住。
使用 session 复用连接。
失败后进行重试和退避。

6.1 settings.py

# cert_spider/settings.py

from dataclasses import dataclass
from pathlib import Path


BASE_DIR = Path(__file__).resolve().parent.parent
DATA_DIR = BASE_DIR / "data"
EXAMPLES_DIR = BASE_DIR / "examples"

DATA_DIR.mkdir(exist_ok=True)


@dataclass(frozen=True)
class SpiderSettings:
    user_agent: str = (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/124.0.0.0 Safari/537.36"
    )
    timeout: int = 15
    delay_seconds: float = 1.5
    max_retries: int = 3
    csv_path: str = str(DATA_DIR / "output.csv")
    sqlite_path: str = str(DATA_DIR / "cert_results.sqlite3")
    default_encoding: str = "utf-8"


SETTINGS = SpiderSettings()

6.2 fetcher.py

# cert_spider/fetcher.py

from __future__ import annotations

import logging
import random
import time
from pathlib import Path
from typing import Optional
from urllib.parse import urlparse

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

from cert_spider.settings import SETTINGS


logger = logging.getLogger(__name__)


class FetchError(RuntimeError):
    """请求失败时抛出的异常。"""


class Fetcher:
    """
    负责 HTTP 请求和本地示例文件读取。

    设计点：
    1. 统一 headers。
    2. 统一 timeout。
    3. 统一重试。
    4. 统一日志。
    5. 支持读取本地 examples HTML，便于没有目标站点时调试解析逻辑。
    """

    def __init__(
        self,
        user_agent: str = SETTINGS.user_agent,
        timeout: int = SETTINGS.timeout,
        delay_seconds: float = SETTINGS.delay_seconds,
    ) -> None:
        self.timeout = timeout
        self.delay_seconds = delay_seconds
        self.session = requests.Session()
        self.session.headers.update(
            {
                "User-Agent": user_agent,
                "Accept": (
                    "text/html,application/xhtml+xml,application/xml;"
                    "q=0.9,image/avif,image/webp,*/*;q=0.8"
                ),
                "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
                "Connection": "keep-alive",
            }
        )

    def polite_sleep(self) -> None:
        """
        礼貌等待。

        加一点随机抖动，避免所有请求间隔完全一致。
        这不是为了规避风控，而是为了让访问节奏更接近正常低频访问。
        """
        jitter = random.uniform(0, 0.6)
        sleep_time = self.delay_seconds + jitter
        logger.debug("sleep %.2f seconds before next request", sleep_time)
        time.sleep(sleep_time)

    def get_text(self, url_or_path: str, referer: Optional[str] = None) -> str:
        """
        获取 HTML 文本。

        支持两种输入：
        1. http/https URL。
        2. 本地文件路径，例如 examples/list.html。
        """
        parsed = urlparse(url_or_path)
        if parsed.scheme in {"http", "https"}:
            self.polite_sleep()
            return self._get_http_text(url_or_path, referer=referer)

        path = Path(url_or_path)
        if not path.exists():
            raise FetchError(f"Local file not found: {url_or_path}")

        logger.info("read local html: %s", path)
        return path.read_text(encoding="utf-8")

    @retry(
        stop=stop_after_attempt(SETTINGS.max_retries),
        wait=wait_exponential(multiplier=1, min=1, max=8),
        retry=retry_if_exception_type((requests.RequestException, FetchError)),
        reraise=True,
    )
    def _get_http_text(self, url: str, referer: Optional[str] = None) -> str:
        headers = {}
        if referer:
            headers["Referer"] = referer

        logger.info("GET %s", url)
        resp = self.session.get(url, headers=headers, timeout=self.timeout)
        self._raise_for_bad_status(resp)

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

        return resp.text

    @staticmethod
    def _raise_for_bad_status(resp: Response) -> None:
        """
        对异常状态码做统一处理。

        403：可能是访问频率、UA、Referer 或权限问题。
        429：通常是请求过快。
        5xx：服务端暂时异常，可重试。
        """
        status = resp.status_code

        if status == 200:
            return

        if status in {403, 429}:
            raise FetchError(
                f"HTTP {status}: access denied or rate limited. url={resp.url}"
            )

        if 500 <= status < 600:
            raise FetchError(f"HTTP {status}: server error. url={resp.url}")

        resp.raise_for_status()

6.3 请求层几个细节

headers 不是越复杂越好。一般设置 User-Agent、Accept、Accept-Language 就够了。如果目标站点明确要求 Referer，可以在请求详情页时传入列表页 URL。不要伪造登录态，不要带不属于自己的 Cookie。

timeout 一定要写。很多新手爬虫卡死，不是解析问题，而是请求没有超时设置。本文默认 15 秒，实际可以按站点速度调整。

重试也不要无限重试。请求失败时，重试 2 到 3 次基本够了。无限重试不仅浪费时间，也可能给对方服务器造成额外压力。

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

解析层是这类项目最容易出问题的地方。原因是公开目录页面并不总是规整。有的字段在表格里，有的在段落里，有的详情页用“学校：某某大学”，有的写成“院校名称：某某大学”。

所以解析层要做两件事：

第一，支持多种字段别名。

第二，缺失字段时不直接报错，而是保留空值，并在日志里记录。

7.1 models.py

# cert_spider/models.py

from __future__ import annotations

import hashlib
from dataclasses import dataclass, asdict
from datetime import datetime
from typing import Dict, Optional


@dataclass
class CertificationRecord:
    school: str = ""
    major: str = ""
    status: str = ""
    valid_period: str = ""
    year: str = ""
    link: str = ""
    source_url: str = ""
    detail_url: str = ""
    valid_start: str = ""
    valid_end: str = ""
    scraped_at: str = ""
    record_hash: str = ""

    def finalize(self) -> "CertificationRecord":
        if not self.scraped_at:
            self.scraped_at = datetime.now().isoformat(timespec="seconds")
        if not self.record_hash:
            self.record_hash = self.make_hash()
        return self

    def make_hash(self) -> str:
        """
        生成记录唯一哈希。

        优先考虑 school + major + year + valid_period + detail_url。
        如果 detail_url 缺失，也能通过内容字段生成稳定 hash。
        """
        raw = "|".join(
            [
                self.school.strip(),
                self.major.strip(),
                self.status.strip(),
                self.valid_period.strip(),
                self.year.strip(),
                self.detail_url.strip() or self.link.strip(),
            ]
        )
        return hashlib.sha1(raw.encode("utf-8")).hexdigest()

    def to_dict(self) -> Dict[str, str]:
        return asdict(self)

7.2 cleaner.py

# cert_spider/cleaner.py

from __future__ import annotations

import re
from typing import Tuple


SPACE_RE = re.compile(r"\s+")
YEAR_RE = re.compile(r"(20\d{2}|19\d{2})")
RANGE_RE = re.compile(
    r"(?P<start>(?:19|20)\d{2}\s*年?\s*(?:\d{1,2}\s*月?)?)"
    r".{0,8}?"
    r"(?P<end>(?:19|20)\d{2}\s*年?\s*(?:\d{1,2}\s*月?)?)"
)


def clean_text(value: str | None) -> str:
    if not value:
        return ""
    value = value.replace("\u3000", " ")
    value = value.replace("\xa0", " ")
    value = SPACE_RE.sub(" ", value)
    return value.strip(" ：:;；,\n\r\t")


def normalize_status(value: str) -> str:
    text = clean_text(value)
    if not text:
        return ""

    if "通过" in text and "未" not in text and "不予" not in text:
        return "通过认证"
    if "不予" in text or "未通过" in text:
        return "不予认证"
    if "有效" in text:
        return text

    return text


def extract_year(*values: str) -> str:
    for value in values:
        text = clean_text(value)
        match = YEAR_RE.search(text)
        if match:
            return match.group(1)
    return ""


def parse_valid_period(value: str) -> Tuple[str, str]:
    """
    尝试从有效期文本中解析开始和结束。

    注意：
    有效期格式可能非常多，例如：
    2023 年 1 月至 2028 年 12 月
    2023.01-2028.12
    2023年-2028年
    因此这里只做轻量解析，不强行转换为日期对象。
    """
    text = clean_text(value)
    if not text:
        return "", ""

    text = text.replace("—", "-").replace("～", "-").replace("至", "-")
    match = RANGE_RE.search(text)
    if match:
        return clean_text(match.group("start")), clean_text(match.group("end"))

    parts = re.split(r"[-~到至]", text)
    if len(parts) >= 2:
        return clean_text(parts[0]), clean_text(parts[1])

    return "", ""

7.3 parser.py

# cert_spider/parser.py

from __future__ import annotations

import logging
import re
from dataclasses import replace
from typing import Dict, Iterable, List, Optional
from urllib.parse import urljoin

from bs4 import BeautifulSoup, Tag

from cert_spider.cleaner import (
    clean_text,
    extract_year,
    normalize_status,
    parse_valid_period,
)
from cert_spider.models import CertificationRecord


logger = logging.getLogger(__name__)


FIELD_ALIASES: Dict[str, List[str]] = {
    "school": ["学校", "学校名称", "院校", "院校名称", "机构", "机构名称", "高校"],
    "major": ["专业", "专业名称", "认证专业", "本科专业", "项目名称"],
    "status": ["认证状态", "认证结果", "状态", "结论", "认证结论"],
    "valid_period": ["有效期", "认证有效期", "有效期限", "有效期起止", "有效期截止时间"],
    "year": ["年份", "认证年份", "发布年份", "年度", "通过年份"],
    "link": ["链接", "详情链接", "来源链接"],
}


class DirectoryParser:
    """
    解析公开目录页和详情页。

    这个解析器不是为某一个页面写死的，而是尽量适配几类常见结构：
    1. 列表页 table。
    2. 详情页 table。
    3. 详情页 dl/dt/dd。
    4. 详情页中“字段：值”形式的段落。
    """

    def parse_list_page(self, html: str, base_url: str) -> tuple[List[CertificationRecord], Optional[str]]:
        soup = BeautifulSoup(html, "lxml")
        records: List[CertificationRecord] = []

        table_records = self._parse_table_records(soup, base_url)
        if table_records:
            records.extend(table_records)
        else:
            records.extend(self._parse_link_list_records(soup, base_url))

        next_url = self._find_next_page(soup, base_url)
        logger.info("parse list page: records=%d next=%s", len(records), next_url)
        return records, next_url

    def parse_detail_page(
        self,
        html: str,
        detail_url: str,
        base_record: Optional[CertificationRecord] = None,
    ) -> CertificationRecord:
        soup = BeautifulSoup(html, "lxml")
        kv = {}

        kv.update(self._extract_table_kv(soup))
        kv.update(self._extract_dl_kv(soup))
        kv.update(self._extract_colon_text_kv(soup))

        record = base_record or CertificationRecord()
        record.detail_url = detail_url
        record.link = record.link or detail_url

        mapped = self._map_kv_to_record(kv)
        record = self._merge_record(record, mapped)

        if not record.year:
            record.year = extract_year(record.valid_period, record.detail_url, soup.get_text(" "))

        record.status = normalize_status(record.status)
        record.valid_start, record.valid_end = parse_valid_period(record.valid_period)

        return record.finalize()

    def _parse_table_records(self, soup: BeautifulSoup, base_url: str) -> List[CertificationRecord]:
        records: List[CertificationRecord] = []

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

            headers = [clean_text(cell.get_text(" ")) for cell in rows[0].find_all(["th", "td"])]
            if not headers:
                continue

            for tr in rows[1:]:
                cells = tr.find_all(["td", "th"])
                if not cells:
                    continue

                row_data = {}
                for index, cell in enumerate(cells):
                    header = headers[index] if index < len(headers) else f"col_{index}"
                    row_data[header] = clean_text(cell.get_text(" "))

                detail_url = self._extract_first_link(tr, base_url)
                mapped = self._map_kv_to_record(row_data)

                if detail_url:
                    mapped.detail_url = detail_url
                    mapped.link = detail_url

                if self._looks_like_record(mapped):
                    mapped.source_url = base_url
                    records.append(mapped.finalize())

        return records

    def _parse_link_list_records(self, soup: BeautifulSoup, base_url: str) -> List[CertificationRecord]:
        """
        当列表页不是表格时，尝试从链接列表中提取详情页。

        例如：
        <ul>
          <li><a href="/detail/1001">某某大学 计算机科学与技术 通过认证</a></li>
        </ul>
        """
        records: List[CertificationRecord] = []
        for a in soup.find_all("a", href=True):
            text = clean_text(a.get_text(" "))
            href = clean_text(a.get("href"))
            if not text or not href:
                continue

            if not self._maybe_detail_link(text, href):
                continue

            detail_url = urljoin(base_url, href)
            record = CertificationRecord(
                school="",
                major="",
                status="",
                valid_period="",
                year=extract_year(text, href),
                link=detail_url,
                source_url=base_url,
                detail_url=detail_url,
            )
            records.append(record.finalize())

        return records

    def _extract_table_kv(self, soup: BeautifulSoup) -> Dict[str, str]:
        kv: Dict[str, str] = {}
        for table in soup.find_all("table"):
            for tr in table.find_all("tr"):
                cells = tr.find_all(["td", "th"])
                if len(cells) == 2:
                    key = clean_text(cells[0].get_text(" "))
                    value = clean_text(cells[1].get_text(" "))
                    if key and value:
                        kv[key] = value

                elif len(cells) >= 4:
                    # 兼容一行两组字段：学校：A 专业：B
                    for i in range(0, len(cells) - 1, 2):
                        key = clean_text(cells[i].get_text(" "))
                        value = clean_text(cells[i + 1].get_text(" "))
                        if key and value:
                            kv[key] = value

        return kv

    def _extract_dl_kv(self, soup: BeautifulSoup) -> Dict[str, str]:
        kv: Dict[str, str] = {}
        for dl in soup.find_all("dl"):
            dts = dl.find_all("dt")
            dds = dl.find_all("dd")
            for dt, dd in zip(dts, dds):
                key = clean_text(dt.get_text(" "))
                value = clean_text(dd.get_text(" "))
                if key and value:
                    kv[key] = value
        return kv

    def _extract_colon_text_kv(self, soup: BeautifulSoup) -> Dict[str, str]:
        """
        解析类似：
        学校：某某大学
        专业：计算机科学与技术
        认证状态：通过认证
        """
        kv: Dict[str, str] = {}
        text = soup.get_text("\n")

        for line in text.splitlines():
            line = clean_text(line)
            if not line:
                continue

            match = re.match(r"^(.{2,20}?)[：:]\s*(.+)$", line)
            if not match:
                continue

            key = clean_text(match.group(1))
            value = clean_text(match.group(2))
            if key and value:
                kv[key] = value

        return kv

    def _map_kv_to_record(self, kv: Dict[str, str]) -> CertificationRecord:
        normalized = {clean_text(k): clean_text(v) for k, v in kv.items()}
        data = {}

        for field_name, aliases in FIELD_ALIASES.items():
            value = ""
            for key, raw_value in normalized.items():
                if self._key_match(key, aliases):
                    value = raw_value
                    break
            data[field_name] = value

        record = CertificationRecord(
            school=data.get("school", ""),
            major=data.get("major", ""),
            status=data.get("status", ""),
            valid_period=data.get("valid_period", ""),
            year=data.get("year", ""),
            link=data.get("link", ""),
        )

        if not record.year:
            record.year = extract_year(record.valid_period)

        record.status = normalize_status(record.status)
        record.valid_start, record.valid_end = parse_valid_period(record.valid_period)

        return record

    def _merge_record(
        self,
        base: CertificationRecord,
        detail: CertificationRecord,
    ) -> CertificationRecord:
        """
        详情页字段优先，但如果详情页缺字段，就沿用列表页字段。
        """
        return CertificationRecord(
            school=detail.school or base.school,
            major=detail.major or base.major,
            status=detail.status or base.status,
            valid_period=detail.valid_period or base.valid_period,
            year=detail.year or base.year,
            link=detail.link or base.link,
            source_url=base.source_url,
            detail_url=detail.detail_url or base.detail_url,
            valid_start=detail.valid_start or base.valid_start,
            valid_end=detail.valid_end or base.valid_end,
            scraped_at=base.scraped_at,
            record_hash="",
        )

    @staticmethod
    def _key_match(key: str, aliases: Iterable[str]) -> bool:
        key = clean_text(key)
        return any(alias == key or alias in key for alias in aliases)

    @staticmethod
    def _looks_like_record(record: CertificationRecord) -> bool:
        return bool(record.school or record.major or record.detail_url or record.link)

    @staticmethod
    def _extract_first_link(node: Tag, base_url: str) -> str:
        a = node.find("a", href=True)
        if not a:
            return ""
        return urljoin(base_url, clean_text(a.get("href")))

    @staticmethod
    def _maybe_detail_link(text: str, href: str) -> bool:
        tokens = ["详情", "查看", "认证", "结果", "专业", "detail", "show", "info"]
        raw = f"{text} {href}".lower()
        return any(token.lower() in raw for token in tokens)

    @staticmethod
    def _find_next_page(soup: BeautifulSoup, base_url: str) -> Optional[str]:
        """
        寻找下一页链接。
        """
        candidates = soup.find_all("a", href=True)
        for a in candidates:
            text = clean_text(a.get_text(" "))
            rel = " ".join(a.get("rel", [])) if a.get("rel") else ""
            if text in {"下一页", "下页", ">", "Next", "next"} or "next" in rel.lower():
                return urljoin(base_url, clean_text(a.get("href")))

        return None

7.4 列表页如何拿详情链接

列表页通常有两种情况。

第一种是表格：

<tr>
  <td>某某大学</td>
  <td>计算机科学与技术</td>
  <td>通过认证</td>
  <td>2023 年 1 月至 2028 年 12 月</td>
  <td><a href="/detail/1001">详情</a></td>
</tr>

这种最好处理，直接按表头映射字段，再取 <a href="">。

第二种是新闻列表：

<li>
  <a href="/show/1001.html">某某大学计算机科学与技术专业认证结果公告</a>
</li>

这种列表页拿不到完整字段，只能先拿详情链接，再去详情页解析。

7.5 详情页如何抽字段

详情页常见结构有三种：

第一种是字段表格：

<table>
  <tr><td>学校</td><td>某某大学</td></tr>
  <tr><td>专业</td><td>计算机科学与技术</td></tr>
</table>

第二种是 definition list：

<dl>
  <dt>学校</dt><dd>某某大学</dd>
  <dt>专业</dt><dd>计算机科学与技术</dd>
</dl>

第三种是普通文本：

学校：某某大学
专业：计算机科学与技术
认证状态：通过认证

本文解析器都做了兼容。

7.6 缺失字段怎么办

我的建议是：缺失字段不要让程序崩掉，而是保留空字符串，并在后续数据质量检查中单独统计。

例如：

school 为空：可能详情页结构变了
major 为空：可能列表页没有字段，详情页也没有抽到
year 为空：可能有效期没有年份，或者字段写法特殊

后面可以加一个质量报告：

总记录数：100
school 缺失：2
major 缺失：0
valid_period 缺失：5
year 缺失：8

这比程序中途报错停止更适合目录类采集。

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

本文同时实现两种存储：

CSV：方便打开、查看、交给别人。
SQLite：方便去重、查询、后续分析。

8.1 为什么要做机构-专业双维度建表

很多人会直接建一张表：

学校 | 专业 | 认证状态 | 有效期 | 年份 | 链接

这当然能用，但不够适合长期分析。比如你想统计某所学校所有通过认证的专业，或者统计某个专业在不同学校的认证情况，平铺表会产生大量重复文本。

更好的方式是：

institutions 机构表
majors 专业表
certifications 认证事实表

也就是把“学校”和“专业”拆成两个维度，把“认证结果”作为事实记录。

结构如下：

institutions
- id
- school

majors
- id
- major

certifications
- id
- institution_id
- major_id
- status
- valid_period
- valid_start
- valid_end
- year
- link
- source_url
- detail_url
- record_hash
- scraped_at

这样做的好处是，后续分析会舒服很多。

8.2 字段映射表

字段名	类型	示例值	说明
school	TEXT	某某大学	学校或机构名称
major	TEXT	软件工程	专业名称
status	TEXT	通过认证	认证状态
valid_period	TEXT	2023 年 1 月至 2028 年 12 月	原始有效期文本
valid_start	TEXT	2023 年 1 月	解析出的有效期开始
valid_end	TEXT	2028 年 12 月	解析出的有效期结束
year	TEXT	2023	认证年份
link	TEXT	https://example.com/detail/1001	详情链接
source_url	TEXT	https://example.com/list	列表页
detail_url	TEXT	https://example.com/detail/1001	详情页
record_hash	TEXT	sha1 hash	去重键
scraped_at	TEXT	2026-06-10T10:30:00	抓取时间

8.3 storage.py

# cert_spider/storage.py

from __future__ import annotations

import csv
import logging
import sqlite3
from pathlib import Path
from typing import Iterable, List

from cert_spider.models import CertificationRecord
from cert_spider.settings import SETTINGS


logger = logging.getLogger(__name__)


CSV_FIELDS = [
    "school",
    "major",
    "status",
    "valid_period",
    "valid_start",
    "valid_end",
    "year",
    "link",
    "source_url",
    "detail_url",
    "record_hash",
    "scraped_at",
]


class CsvStorage:
    def __init__(self, path: str = SETTINGS.csv_path) -> None:
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)

    def save(self, records: Iterable[CertificationRecord]) -> None:
        records = list(records)
        if not records:
            logger.warning("no records to write csv")
            return

        with self.path.open("w", newline="", encoding="utf-8-sig") as f:
            writer = csv.DictWriter(f, fieldnames=CSV_FIELDS)
            writer.writeheader()
            for record in records:
                row = record.to_dict()
                writer.writerow({field: row.get(field, "") for field in CSV_FIELDS})

        logger.info("csv saved: %s rows=%d", self.path, len(records))


class SQLiteStorage:
    def __init__(self, path: str = SETTINGS.sqlite_path) -> None:
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.conn = sqlite3.connect(str(self.path))
        self.conn.execute("PRAGMA foreign_keys = ON")
        self.create_tables()

    def create_tables(self) -> None:
        sql = """
        CREATE TABLE IF NOT EXISTS institutions (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            school TEXT NOT NULL UNIQUE
        );

        CREATE TABLE IF NOT EXISTS majors (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            major TEXT NOT NULL UNIQUE
        );

        CREATE TABLE IF NOT EXISTS certifications (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            institution_id INTEGER NOT NULL,
            major_id INTEGER NOT NULL,
            status TEXT,
            valid_period TEXT,
            valid_start TEXT,
            valid_end TEXT,
            year TEXT,
            link TEXT,
            source_url TEXT,
            detail_url TEXT,
            record_hash TEXT NOT NULL UNIQUE,
            scraped_at TEXT,
            FOREIGN KEY (institution_id) REFERENCES institutions(id),
            FOREIGN KEY (major_id) REFERENCES majors(id)
        );

        CREATE INDEX IF NOT EXISTS idx_cert_year
        ON certifications(year);

        CREATE INDEX IF NOT EXISTS idx_cert_status
        ON certifications(status);

        CREATE INDEX IF NOT EXISTS idx_cert_institution_major
        ON certifications(institution_id, major_id);
        """

        self.conn.executescript(sql)
        self.conn.commit()

    def save(self, records: Iterable[CertificationRecord]) -> None:
        count = 0
        for record in records:
            record.finalize()

            if not record.school:
                record.school = "未知学校"
            if not record.major:
                record.major = "未知专业"

            institution_id = self._get_or_create_institution(record.school)
            major_id = self._get_or_create_major(record.major)

            self.conn.execute(
                """
                INSERT OR IGNORE INTO certifications (
                    institution_id,
                    major_id,
                    status,
                    valid_period,
                    valid_start,
                    valid_end,
                    year,
                    link,
                    source_url,
                    detail_url,
                    record_hash,
                    scraped_at
                )
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    institution_id,
                    major_id,
                    record.status,
                    record.valid_period,
                    record.valid_start,
                    record.valid_end,
                    record.year,
                    record.link,
                    record.source_url,
                    record.detail_url,
                    record.record_hash,
                    record.scraped_at,
                ),
            )
            count += 1

        self.conn.commit()
        logger.info("sqlite saved: %s rows=%d", self.path, count)

    def _get_or_create_institution(self, school: str) -> int:
        self.conn.execute(
            "INSERT OR IGNORE INTO institutions(school) VALUES (?)",
            (school,),
        )
        cur = self.conn.execute(
            "SELECT id FROM institutions WHERE school = ?",
            (school,),
        )
        row = cur.fetchone()
        return int(row[0])

    def _get_or_create_major(self, major: str) -> int:
        self.conn.execute(
            "INSERT OR IGNORE INTO majors(major) VALUES (?)",
            (major,),
        )
        cur = self.conn.execute(
            "SELECT id FROM majors WHERE major = ?",
            (major,),
        )
        row = cur.fetchone()
        return int(row[0])

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


def deduplicate_records(records: Iterable[CertificationRecord]) -> List[CertificationRecord]:
    """
    内存去重。

    规则：
    1. 优先用 record_hash。
    2. hash 相同则保留第一条。
    """
    result: List[CertificationRecord] = []
    seen = set()

    for record in records:
        record.finalize()
        key = record.record_hash
        if key in seen:
            continue
        seen.add(key)
        result.append(record)

    return result

8.4 去重策略

本文使用两层去重：

第一层，内存去重。每条记录生成 record_hash，相同 hash 只保留一条。

第二层，SQLite 去重。数据库里对 record_hash 加 UNIQUE 约束，重复插入时自动忽略。

实际项目中也可以使用 URL 唯一去重：

detail_url unique

但有些目录页会出现同一详情页包含多条专业记录，或者同一专业结果在不同公告中重复出现，所以我更偏向使用字段组合 hash：

school + major + status + valid_period + year + detail_url

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

9.1 crawler.py

# cert_spider/crawler.py

from __future__ import annotations

import logging
from typing import List, Optional, Set

from cert_spider.fetcher import Fetcher, FetchError
from cert_spider.models import CertificationRecord
from cert_spider.parser import DirectoryParser
from cert_spider.storage import CsvStorage, SQLiteStorage, deduplicate_records


logger = logging.getLogger(__name__)


class CertificationCrawler:
    def __init__(
        self,
        start_url: str,
        max_pages: int = 10,
        fetch_detail: bool = True,
    ) -> None:
        self.start_url = start_url
        self.max_pages = max_pages
        self.fetch_detail = fetch_detail
        self.fetcher = Fetcher()
        self.parser = DirectoryParser()
        self.visited_pages: Set[str] = set()
        self.visited_details: Set[str] = set()

    def crawl(self) -> List[CertificationRecord]:
        all_records: List[CertificationRecord] = []
        current_url: Optional[str] = self.start_url
        page_count = 0

        while current_url and page_count < self.max_pages:
            if current_url in self.visited_pages:
                logger.warning("skip visited list page: %s", current_url)
                break

            self.visited_pages.add(current_url)
            page_count += 1

            try:
                html = self.fetcher.get_text(current_url)
            except Exception as exc:
                logger.exception("failed to fetch list page: %s error=%s", current_url, exc)
                break

            base_records, next_url = self.parser.parse_list_page(html, current_url)

            if not self.fetch_detail:
                all_records.extend(base_records)
            else:
                for base_record in base_records:
                    detail_url = base_record.detail_url or base_record.link
                    if not detail_url:
                        all_records.append(base_record.finalize())
                        continue

                    if detail_url in self.visited_details:
                        continue

                    self.visited_details.add(detail_url)

                    try:
                        detail_html = self.fetcher.get_text(detail_url, referer=current_url)
                        detail_record = self.parser.parse_detail_page(
                            detail_html,
                            detail_url=detail_url,
                            base_record=base_record,
                        )
                        all_records.append(detail_record)
                    except FetchError as exc:
                        logger.warning("fetch detail failed: %s error=%s", detail_url, exc)
                        all_records.append(base_record.finalize())
                    except Exception as exc:
                        logger.exception("parse detail failed: %s error=%s", detail_url, exc)
                        all_records.append(base_record.finalize())

            current_url = next_url

        all_records = deduplicate_records(all_records)
        logger.info("crawl finished. total=%d", len(all_records))
        return all_records

    def save(self, records: List[CertificationRecord]) -> None:
        CsvStorage().save(records)

        sqlite_storage = SQLiteStorage()
        try:
            sqlite_storage.save(records)
        finally:
            sqlite_storage.close()

9.2 run.py

# run.py

from __future__ import annotations

import argparse
import logging
from pathlib import Path

from cert_spider.crawler import CertificationCrawler
from cert_spider.settings import EXAMPLES_DIR


def setup_logging(verbose: bool = False) -> None:
    level = logging.DEBUG if verbose else logging.INFO
    logging.basicConfig(
        level=level,
        format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    )


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Crawl professional certification result public directory."
    )
    parser.add_argument(
        "--start-url",
        default="",
        help="公开目录列表页 URL。也可以传本地 examples/list.html。",
    )
    parser.add_argument(
        "--max-pages",
        type=int,
        default=5,
        help="最多采集多少个列表页。",
    )
    parser.add_argument(
        "--no-detail",
        action="store_true",
        help="只解析列表页，不进入详情页。",
    )
    parser.add_argument(
        "--demo",
        action="store_true",
        help="使用 examples/list.html 演示。",
    )
    parser.add_argument(
        "--verbose",
        action="store_true",
        help="输出 DEBUG 日志。",
    )
    return parser


def main() -> None:
    args = build_parser().parse_args()
    setup_logging(verbose=args.verbose)

    if args.demo:
        start_url = str(EXAMPLES_DIR / "list.html")
    else:
        if not args.start_url:
            raise SystemExit("Please provide --start-url or use --demo.")
        start_url = args.start_url

    crawler = CertificationCrawler(
        start_url=start_url,
        max_pages=args.max_pages,
        fetch_detail=not args.no_detail,
    )
    records = crawler.crawl()
    crawler.save(records)

    print(f"Done. Records: {len(records)}")
    print("CSV: data/output.csv")
    print("SQLite: data/cert_results.sqlite3")


if __name__ == "__main__":
    main()

9.3 本地演示页面

为了让代码不依赖某个外部站点也能跑，我们准备几个本地 HTML 示例文件。实际采集时，把 --demo 换成 --start-url 即可。

创建 examples/list.html：

<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8">
  <title>专业认证结果公开目录</title>
</head>
<body>
  <h1>专业认证结果公开目录</h1>
  <table>
    <tr>
      <th>学校</th>
      <th>专业</th>
      <th>认证状态</th>
      <th>有效期</th>
      <th>年份</th>
      <th>链接</th>
    </tr>
    <tr>
      <td>东湖理工大学</td>
      <td>计算机科学与技术</td>
      <td>通过认证</td>
      <td>2023 年 1 月至 2028 年 12 月</td>
      <td>2023</td>
      <td><a href="detail_1001.html">详情</a></td>
    </tr>
    <tr>
      <td>南山工业大学</td>
      <td>软件工程</td>
      <td>通过认证</td>
      <td>2022 年 1 月至 2027 年 12 月</td>
      <td>2022</td>
      <td><a href="detail_1002.html">详情</a></td>
    </tr>
    <tr>
      <td>江北科技学院</td>
      <td>数据科学与大数据技术</td>
      <td>通过认证</td>
      <td>2024 年 1 月至 2029 年 12 月</td>
      <td>2024</td>
      <td><a href="detail_1003.html">详情</a></td>
    </tr>
  </table>
</body>
</html>

创建 examples/detail_1001.html：

<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8">
  <title>东湖理工大学计算机科学与技术专业认证结果</title>
</head>
<body>
  <h1>认证结果详情</h1>
  <table>
    <tr><td>学校名称</td><td>东湖理工大学</td></tr>
    <tr><td>专业名称</td><td>计算机科学与技术</td></tr>
    <tr><td>认证结论</td><td>通过认证</td></tr>
    <tr><td>认证有效期</td><td>2023 年 1 月至 2028 年 12 月</td></tr>
    <tr><td>认证年份</td><td>2023</td></tr>
  </table>
</body>
</html>

创建 examples/detail_1002.html：

<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8">
  <title>南山工业大学软件工程专业认证结果</title>
</head>
<body>
  <h1>认证结果详情</h1>
  <dl>
    <dt>院校名称</dt>
    <dd>南山工业大学</dd>
    <dt>认证专业</dt>
    <dd>软件工程</dd>
    <dt>状态</dt>
    <dd>通过认证</dd>
    <dt>有效期限</dt>
    <dd>2022 年 1 月至 2027 年 12 月</dd>
    <dt>年度</dt>
    <dd>2022</dd>
  </dl>
</body>
</html>

创建 examples/detail_1003.html：

<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8">
  <title>江北科技学院数据科学与大数据技术专业认证结果</title>
</head>
<body>
  <h1>认证结果详情</h1>
  <p>学校：江北科技学院</p>
  <p>专业：数据科学与大数据技术</p>
  <p>认证状态：通过认证</p>
  <p>有效期：2024 年 1 月至 2029 年 12 月</p>
  <p>年份：2024</p>
</body>
</html>

9.4 启动方式

使用本地演示文件运行：

python run.py --demo --verbose

采集真实公开目录页：

python run.py --start-url "https://example.com/certification/results/index.html" --max-pages 10

只解析列表页，不进入详情页：

python run.py --start-url "https://example.com/certification/results/index.html" --no-detail

9.5 输出位置

CSV 输出：

data/output.csv

SQLite 输出：

data/cert_results.sqlite3

SQLite 里会有三张核心表：

institutions
majors
certifications

9.6 示例结果

CSV 示例：

school	major	status	valid_period	year	link
东湖理工大学	计算机科学与技术	通过认证	2023 年 1 月至 2028 年 12 月	2023	examples/detail_1001.html
南山工业大学	软件工程	通过认证	2022 年 1 月至 2027 年 12 月	2022	examples/detail_1002.html
江北科技学院	数据科学与大数据技术	通过认证	2024 年 1 月至 2029 年 12 月	2024	examples/detail_1003.html

SQLite 查询示例：

SELECT
    i.school,
    m.major,
    c.status,
    c.valid_period,
    c.year,
    c.detail_url
FROM certifications c
JOIN institutions i ON c.institution_id = i.id
JOIN majors m ON c.major_id = m.id
ORDER BY c.year DESC;

按学校统计：

SELECT
    i.school,
    COUNT(*) AS cert_count
FROM certifications c
JOIN institutions i ON c.institution_id = i.id
GROUP BY i.school
ORDER BY cert_count DESC;

按专业统计：

SELECT
    m.major,
    COUNT(*) AS school_count
FROM certifications c
JOIN majors m ON c.major_id = m.id
GROUP BY m.major
ORDER BY school_count DESC;

🔟 常见问题与排错（强烈建议写）

10.1 遇到 403 怎么办

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

User-Agent 缺失。
Referer 不符合要求。
请求过快。
目标路径不允许程序访问。
页面需要登录或权限。

处理思路：

第一，确认页面是否公开可访问。

第二，降低频率。

第三，设置合理 User-Agent。

第四，如果页面明确需要登录或权限，不要绕过，应该停止采集或申请授权。

不要把 403 简单理解成“换代理就能解决”。如果对方不希望自动访问，继续绕只会让问题变复杂。

10.2 遇到 429 怎么办

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

增加 delay。
减少并发。
加指数退避。
暂停一段时间再采集。
如果数据量很小，直接手动下载也可以。

本文 fetcher 已经使用 tenacity 做了指数退避，但我还是建议从低频开始，不要把重试当成“硬怼”。

10.3 HTML 抓到空壳怎么办

如果浏览器能看到数据，但 requests 抓到的 HTML 只有一个根节点、几个 JS 文件，那就是动态渲染页面。

排查方式：

打开浏览器开发者工具，进入 Network 面板，刷新页面，看有没有返回 JSON 的接口。常见接口路径可能包含：

/api/
ajax
list
search
query
page

如果能找到接口，优先抓接口，因为 JSON 比 HTML 好解析。

如果找不到接口，或者接口参数复杂，可以使用 Playwright：

pip install playwright
playwright install chromium

简单示例：

from playwright.sync_api import sync_playwright


def fetch_rendered_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

不过 Playwright 成本更高，运行更慢，不建议一开始就用。

10.4 解析报错怎么办

解析报错通常有三类：

第一，选择器变化。比如原来是 table，现在改成 div。

第二，字段名变化。比如“学校名称”改成“院校”。

第三，结构不稳定。比如部分详情页没有有效期。

应对方式：

字段别名要做成配置。
缺失字段不要直接 raise。
保存原始 detail_url，方便回看页面。
对空字段做质量报告。
定期抽样人工核对。

10.5 编码和乱码怎么办

乱码一般来自编码识别错误。requests 默认可能会把页面识别成 ISO-8859-1。本文 fetcher 里做了处理：

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

导出 CSV 时使用：

encoding="utf-8-sig"

这样 Excel 打开中文更稳。

10.6 PDF 或附件型目录怎么办

有些认证目录不是 HTML 表格，而是 PDF 附件。这时有三种方案：

第一，如果 PDF 是文本型，可以用 pdfplumber 抽表格。

pip install pdfplumber

示例：

import pdfplumber


def extract_tables_from_pdf(pdf_path: str):
    rows = []
    with pdfplumber.open(pdf_path) as pdf:
        for page in pdf.pages:
            tables = page.extract_tables()
            for table in tables:
                for row in table:
                    rows.append(row)
    return rows

第二，如果 PDF 是扫描图片，可能需要 OCR。OCR 成本高，错误率也更高，不建议作为第一选择。

第三，如果页面同时提供 Excel 或 Word 附件，优先选择结构化附件。Excel 通常比 PDF 更好处理。

10.7 字段混在备注里怎么办

认证数据经常出现这种情况：

有效期截止时间：2028 年 12 月（有条件）

或者：

部分年份不在认证有效期，详见备注

我的建议是：不要过度清洗。原始有效期字段要完整保留，同时额外解析 valid_start 和 valid_end。像“有条件”“备注说明”这类信息，可以单独加 remark 字段，而不是直接丢掉。

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

11.1 并发采集

公开目录类数据一般不需要高并发。如果确实数据量大，可以考虑线程池，但要控制并发数量。

示例：

from concurrent.futures import ThreadPoolExecutor, as_completed


def fetch_detail_task(fetcher, parser, detail_url, base_record):
    html = fetcher.get_text(detail_url, referer=base_record.source_url)
    return parser.parse_detail_page(html, detail_url, base_record)


def fetch_details_concurrently(fetcher, parser, records, max_workers=3):
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_map = {
            executor.submit(
                fetch_detail_task,
                fetcher,
                parser,
                record.detail_url or record.link,
                record,
            ): record
            for record in records
            if record.detail_url or record.link
        }

        for future in as_completed(future_map):
            base_record = future_map[future]
            try:
                results.append(future.result())
            except Exception:
                results.append(base_record.finalize())

    return results

我个人建议并发数从 2 或 3 开始，不要上来就开很大。

11.2 asyncio

如果目标站点响应慢、页面数量较多，可以考虑 aiohttp。但 asyncio 代码复杂度会高一些。对于认证目录这种场景，除非确实有性能瓶颈，否则没必要急着改。

11.3 Scrapy 改造

当项目变大后，可以改成 Scrapy。Scrapy 的优势是：

请求调度成熟。
去重机制成熟。
中间件体系完整。
日志和 pipeline 好用。
适合大量页面采集。

Scrapy item 可以这样设计：

import scrapy


class CertificationItem(scrapy.Item):
    school = scrapy.Field()
    major = scrapy.Field()
    status = scrapy.Field()
    valid_period = scrapy.Field()
    valid_start = scrapy.Field()
    valid_end = scrapy.Field()
    year = scrapy.Field()
    link = scrapy.Field()
    source_url = scrapy.Field()
    detail_url = scrapy.Field()
    record_hash = scrapy.Field()
    scraped_at = scrapy.Field()

11.4 断点续跑

断点续跑是目录爬虫非常实用的能力。最简单的做法是维护两个集合：

visited_pages.txt
visited_details.txt

每抓完一个 URL 就写入文件。下次启动时先读取这些文件，已经抓过的跳过。

示例：

from pathlib import Path


class UrlCheckpoint:
    def __init__(self, path: str):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.visited = set()
        if self.path.exists():
            self.visited = {
                line.strip()
                for line in self.path.read_text(encoding="utf-8").splitlines()
                if line.strip()
            }

    def contains(self, url: str) -> bool:
        return url in self.visited

    def add(self, url: str) -> None:
        if url in self.visited:
            return
        self.visited.add(url)
        with self.path.open("a", encoding="utf-8") as f:
            f.write(url + "\n")

然后在 crawler 里使用：

page_checkpoint = UrlCheckpoint("data/visited_pages.txt")
detail_checkpoint = UrlCheckpoint("data/visited_details.txt")

11.5 日志与监控

不要只 print。正式一点的爬虫应该有日志，包括：

请求了哪个 URL。
解析到多少条记录。
失败了多少详情页。
缺失字段数量。
最终写入多少条。

字段质量报告可以这样写：

def build_quality_report(records):
    total = len(records)
    fields = ["school", "major", "status", "valid_period", "year", "detail_url"]
    report = {"total": total}

    for field in fields:
        missing = sum(1 for r in records if not getattr(r, field, ""))
        report[f"{field}_missing"] = missing

    return report

打印：

report = build_quality_report(records)
for key, value in report.items():
    print(key, value)

11.6 定时任务

如果目录会定期更新，可以用 cron 做低频任务。

Linux 示例：

crontab -e

每周一上午 9 点执行：

0 9 * * 1 cd /path/to/cert-directory-spider && /path/to/.venv/bin/python run.py --start-url "https://example.com/certification/results/index.html" >> data/cron.log 2>&1

如果任务很多，可以考虑 Airflow。但对于一个目录页，每周跑一次 cron 已经够用。

11.7 数据版本管理

认证结果这类数据有时间属性。建议每次采集都保留一份原始输出：

data/snapshots/output_20260610.csv
data/snapshots/output_20260617.csv

这样可以比较两次之间新增了哪些记录，哪些记录发生了变化。

简单对比思路：

import csv


def read_hashes(path):
    with open(path, newline="", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        return {row["record_hash"] for row in reader}


old_hashes = read_hashes("data/snapshots/output_20260610.csv")
new_hashes = read_hashes("data/snapshots/output_20260617.csv")

added = new_hashes - old_hashes
removed = old_hashes - new_hashes

print("新增", len(added))
print("减少", len(removed))

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

这篇文章完成了一个面向“专业认证结果公开目录”的 Python 爬虫项目，从需求分析、合规说明、技术选型，到请求层、解析层、清洗层、存储层、运行方式和排错方案，基本覆盖了一个可落地目录采集项目所需要的主干内容。

最终我们实现了：

使用 requests 进行低频、可重试的 HTTP 请求。
使用 BeautifulSoup 和 lxml 解析列表页与详情页。
支持表格、dl、冒号文本三种常见详情结构。
抽取学校、专业、认证状态、有效期、年份、链接等字段。
保存 CSV，方便人工查看。
写入 SQLite，并按照机构-专业双维度建表。
使用 record_hash 做去重。
提供本地 examples，保证代码可以直接运行和调试。
给出 403、429、动态渲染、乱码、PDF 附件等常见问题处理思路。

如果下一步继续优化，我建议按下面这个顺序来：

第一步，把目标站点的字段选择器做成 YAML 配置，这样换站点时不用改代码。

第二步，增加 PDF/Excel 附件解析能力，尤其是认证名单经常以附件形式发布。

第三步，引入 Scrapy，把请求调度、去重、pipeline、日志管理做得更标准。

第四步，对于动态渲染页面，再接入 Playwright。不要一上来就 Playwright，能抓接口就抓接口，能用静态 HTML 就用静态 HTML。

第五步，做一个简单的数据看板，比如用 Streamlit 展示学校维度、专业维度、年份维度统计。

最后说一点实话：爬虫写起来不难，难的是长期稳定。真正能用的爬虫，不是跑一次很快，而是半年后页面小改、字段缺失、网络波动时，它还能给出可解释的日志和可复查的数据。认证目录这种数据尤其如此，宁愿慢一点，也要稳一点；宁愿字段保守一点，也不要把不确定的信息强行清洗成“看起来很漂亮”的假结构。

技术只是工具，数据才是结果。把公开目录整理成一套干净、可追溯、可分析的机构-专业双维度数据表，这就是本文这套方案最核心的价值。

🌟 文末

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

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

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

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

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

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

✅ 互动征集

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

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

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

✅ 免责声明

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

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

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

亚马逊云科技技术品牌专区

更多推荐

2026年量化工具选择，要跟着能力基础走

读者应理解，选择工具前要先判断自己的能力基础和当前任务。工具应该帮助自己补上当前最关键的缺口，而不是替代学习顺序本身。

亚马逊云科技技术品牌专区

加州理工 CS367 C 语言系统编程笔记（一）

C语言是一种强大且广泛使用的编程语言，尤其在系统编程领域。本节课我们将学习C语言的基础语法，包括如何编写“Hello, World!”程序、声明变量以及使用基本数据类型。我们将通过实际的代码示例来加深理解。本节课中我们一起学习了C语言的基础语法，包括如何编写和运行一个简单的C程序、声明变量以及使用基本数据类型。我们还介绍了如何使用Shell环境来编译和运行程序。通过对比Java，你可以看到C语言在

亚马逊云科技技术品牌专区

近期零基础学量化，先按基础差异拆学习顺序

读者应明白，拆解学习顺序时要先看自己的基础缺口，再决定哪一段需要放慢。不同流程阶段都有不同检查任务，学习路线要能反映这些差异。

亚马逊云科技技术品牌专区

所有评论(0)

查看更多评论

喵手

@weixin_66592566

已为社区贡献20条内容

Python 爬取“专业认证结果公开目录”：从公开目录到机构-专业双维度数据表的完整实战

喵手

全文目录：

🌟 开篇语

0️⃣ 前言（Preface）

1️⃣ 摘要（Abstract）

2️⃣ 背景与需求（Why）

2.1 为什么要爬专业认证结果公开目录

2.2 目标站点类型

2.3 目标字段清单

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

3.1 robots.txt 基本说明

3.2 频率控制

3.3 不采集敏感信息

3.4 不绕过付费、登录或权限限制

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

4.1 静态、动态、API：本文属于哪种

4.2 整体流程

4.3 为什么选 requests + bs4 + lxml

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

5.1 Python 版本

5.2 创建虚拟环境

5.3 安装依赖

5.4 推荐项目结构

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

6.1 settings.py

6.2 fetcher.py

6.3 请求层几个细节

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

7.1 models.py

7.2 cleaner.py

7.3 parser.py

7.4 列表页如何拿详情链接

7.5 详情页如何抽字段

7.6 缺失字段怎么办

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

8.1 为什么要做机构-专业双维度建表

8.2 字段映射表

8.3 storage.py

8.4 去重策略

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

9.1 crawler.py

9.2 run.py

9.3 本地演示页面

9.4 启动方式

9.5 输出位置

9.6 示例结果

🔟 常见问题与排错（强烈建议写）

10.1 遇到 403 怎么办

10.2 遇到 429 怎么办

10.3 HTML 抓到空壳怎么办

10.4 解析报错怎么办

10.5 编码和乱码怎么办

10.6 PDF 或附件型目录怎么办

10.7 字段混在备注里怎么办

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

11.1 并发采集

11.2 asyncio

11.3 Scrapy 改造

11.4 断点续跑

11.5 日志与监控

11.6 定时任务

11.7 数据版本管理

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

🌟 文末

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

✅ 互动征集

✅ 免责声明

所有评论(0)

温馨提示：您尚未绑定手机号

喵手