Python 实战:爬取 Linux 内核文档子系统目录,抽取导航树结构并导出 CSV/SQLite
㊗️本期内容已收录至专栏《Python爬虫实战》,持续完善知识体系与项目实战,建议先订阅收藏,后续查阅更方便~
㊙️本期爬虫难度指数:⭐⭐☆☆☆(基础级)
🉐福利: 一次订阅后,专栏内的所有文章可永久免费看,持续更新中,保底1000+(篇)硬核实战内容。
全文目录:
🌟 开篇语
哈喽,各位小伙伴们你们好呀~我是【喵手】。
运营社区: C站 / 掘金 / 腾讯云 / 阿里云 / 华为云 / 51CTO
欢迎大家常来逛逛,一起学习,一起进步~🌟
我长期专注 Python 爬虫工程化实战,主理专栏👉 《Python爬虫实战》:从采集策略到反爬对抗,从数据清洗到分布式调度,持续输出可复用的方法论与可落地案例。内容主打一个“能跑、能用、能扩展”,让数据价值真正做到——抓得到、洗得净、用得上。
📌 专栏食用指南(建议收藏)
- ✅ 入门基础:环境搭建 / 请求与解析 / 数据落库
- ✅ 进阶提升:登录鉴权 / 动态渲染 / 反爬对抗
- ✅ 工程实战:异步并发 / 分布式调度 / 监控与容错
- ✅ 项目落地:数据治理 / 可视化分析 / 场景化应用
📣 专栏推广时间:如果你想系统学爬虫,而不是碎片化东拼西凑,欢迎订阅专栏👉《Python爬虫实战》👈,一次订阅后,专栏内的所有文章可永久免费阅读,持续更新中。
💕订阅后更新会优先推送,按目录学习更高效💯~
0️⃣ 前言(Preface)
这篇文章要做的事情很明确:用 Python 抓取 Linux Kernel 官方文档站中的“子系统文档目录”,通过 requests + BeautifulSoup + lxml 解析页面导航树,最终产出包含“子系统名、章节、简介、文档链接”的结构化数据文件。
读完这篇文章,你能拿到三样东西:
- 一套可以复用的静态文档站爬取思路,尤其适合 Sphinx、MkDocs、ReadTheDocs 这类文档系统。
- 一个完整的 Python 项目,从请求层、解析层、清洗层到 CSV/SQLite 存储都能跑通。
- 一套“导航树结构抽取”的实战经验:如何识别章节、如何补全相对链接、如何给缺失简介做容错、如何去重。
我个人一直觉得,爬虫入门不能只盯着商品列表、资讯列表、论坛帖子这些“内容型页面”。技术文档站其实更适合练基本功:结构清晰,但细节不完全统一;数据公开,但仍然要尊重频率和 robots;页面看起来简单,但导航树、目录层级、正文摘要这些字段要抽稳并不算无脑。
本文选 Linux 内核文档子系统目录作为练习对象,原因也很简单:它够真实、够干净,也够有工程味。
1️⃣ 摘要(Abstract)
本文将使用 Python 抓取 Linux Kernel 官方文档站的子系统目录页面,通过静态 HTML 解析抽取“子系统名、章节、简介、文档链接”,并将结果导出为 CSV、JSON Lines 和 SQLite 数据库。
读完后你将掌握:
- 如何分析 Sphinx 文档站的导航树结构。
- 如何用
requests.Session、重试、退避、timeout、headers 构建稳定请求层。 - 如何用 BeautifulSoup/lxml 解析章节与详情页,并将结果落地到本地存储。
本项目最终产物大致如下:
output/
├── kernel_doc_subsystems.csv
├── kernel_doc_subsystems.jsonl
└── kernel_doc_subsystems.sqlite3
CSV 中每行数据形如:
subsystem_name,chapter,intro,doc_link
Core API Documentation,Core subsystems,This is the beginning of a manual for core kernel APIs.,https://docs.kernel.org/core-api/index.html
Driver implementer’s API guide,Core subsystems,This document collects API references and implementation notes for driver authors.,https://docs.kernel.org/driver-api/index.html
Memory Management Documentation,Core subsystems,This section collects documentation related to Linux memory management.,https://docs.kernel.org/mm/index.html
这不是一个复杂到吓人的爬虫项目,但它覆盖了一个合格爬虫工程必须有的骨架:合规检查、请求封装、解析容错、字段清洗、数据去重、存储导出、日志排错。
2️⃣ 背景与需求(Why)
2.1 为什么要爬 Linux 内核文档子系统目录
Linux Kernel 文档站是典型的技术文档站。它不像新闻站那样频繁换版,也不像电商站那样有大量反爬策略,但它有一个很值得练习的点:文档导航树。
很多技术文档站都会有类似结构:
顶层文档
├── 开发流程
├── 内部 API
│ ├── Core API
│ ├── Driver API
│ ├── Subsystems
│ │ ├── Core subsystems
│ │ │ ├── Core API Documentation
│ │ │ ├── Driver implementer’s API guide
│ │ │ └── Memory Management Documentation
│ │ ├── Human interfaces
│ │ ├── Networking interfaces
│ │ ├── Storage interfaces
│ │ └── Other subsystems
│ └── Locking
└── 用户文档
这类页面的价值在于,它不是单纯的“列表页”,而是一棵带层级的知识目录。我们要做的也不是把所有链接粗暴抓出来,而是要把“章节”和“章节下的文档项”对应起来。
这种任务在实际工作中非常常见:
- 做企业内部知识库索引。
- 做开源项目文档聚合。
- 做搜索系统的数据源预处理。
- 做技术内容地图。
- 做文档变更监控。
- 做离线阅读目录。
- 做研发团队资料导航页。
所以,这篇文章并不是为了“爬 Linux 文档”本身,而是借 Linux 文档站练一套可迁移的方法。
2.2 目标站点
目标站点:
https://docs.kernel.org/subsystem-apis.html
目标页面是 Linux Kernel 文档站中的子系统文档目录页。它把子系统相关文档按章节组织,例如:
Core subsystems
Human interfaces
Networking interfaces
Storage interfaces
Other subsystems
每个章节下面挂若干文档链接。我们要抽取这些链接,并进入每个详情页补一段简介。
2.3 目标字段清单
本项目字段设计如下:
| 字段名 | 中文含义 | 来源 | 说明 |
|---|---|---|---|
subsystem_name |
子系统名 | 目录页链接文本,详情页标题兜底 | 例如 Core API Documentation |
chapter |
章节 | 目录页二级标题 | 例如 Core subsystems |
intro |
简介 | 详情页正文首段或目录页附近文本 | 为空时填默认值 |
doc_link |
文档链接 | 目录页链接 href 补全后得到 | 绝对 URL |
source_page |
来源页 | 固定为入口页或上级页 | 便于追溯 |
url_hash |
URL 哈希 | 程序生成 | 用于去重 |
fetched_at |
抓取时间 | 程序生成 | 便于后续增量更新 |
题目要求的字段是:
子系统名、章节、简介、文档链接
代码里额外保留 source_page、url_hash、fetched_at,是为了工程上更容易排查和去重。实际对外导出时可以只保留前四个字段。
3️⃣ 合规与注意事项(必写)
技术分享归技术分享,爬虫一定要先讲边界。我不太喜欢那种一上来就“开 1000 并发”的写法,尤其是爬公开文档站,这种做法没有必要,也不体面。
3.1 robots.txt 基本说明
robots.txt 是网站放在根路径下的爬虫访问规则文件。一般来说,爬虫在访问站点前应该先读取:
https://目标域名/robots.txt
然后判断当前 User-Agent 是否允许访问目标路径。
本文代码会使用 Python 标准库中的 urllib.robotparser 来处理这件事。它的作用是:在真正请求文档页面前,先问一句“这个路径是否允许访问”。
示意代码如下:
from urllib import robotparser
from urllib.parse import urljoin
robots_url = urljoin("https://docs.kernel.org/", "/robots.txt")
rp = robotparser.RobotFileParser()
rp.set_url(robots_url)
rp.read()
allowed = rp.can_fetch("KernelDocSpider/1.0", "https://docs.kernel.org/subsystem-apis.html")
print(allowed)
需要注意的是,robots 不是身份认证,也不是法律意见。它更像是一种行业协作约定。作为开发者,我们至少应该做到:能检查就检查,能少抓就少抓,能缓存就缓存。
3.2 频率控制
本项目是文档目录抽取,不需要高并发。
推荐策略:
每次请求间隔 1~2 秒
失败后指数退避
单次运行最多抓几十个页面
不要递归全站
不要攻击式并发
文档站通常是给人阅读的,不是给爬虫压测的。我们这次只抓子系统目录和少量详情页,单线程足够。
3.3 不采集敏感信息
目标页面是公开技术文档,不涉及账号、评论、邮箱批量采集、登录态数据,也不需要绕过任何限制。
本文示例坚持几个原则:
不绕过登录
不绕过付费限制
不采集个人敏感信息
不尝试规避访问控制
不使用攻击式并发
不扫描无关路径
这类原则看似啰嗦,但在写爬虫项目时非常重要。一个合格的爬虫不是“能不能抓到”,而是“是否以合适的方式抓到”。
3.4 User-Agent 的写法
不要伪装成搜索引擎,不要伪装成浏览器去做与身份不匹配的事情。更建议写一个透明的 UA:
KernelDocSpider/1.0 (+learning-purpose; contact: your_email@example.com)
当然,如果只是本地学习,不一定非要放真实邮箱,但至少不要用恶意或误导性的 UA。
4️⃣ 技术选型与整体流程(What/How)
4.1 静态、动态、API:本文属于哪一种
目标页面是典型静态 HTML 文档页。页面主体内容、目录链接、章节标题都在服务端返回的 HTML 中,不依赖复杂 JavaScript 渲染。
因此,本文采用:
requests + BeautifulSoup + lxml
不使用 Playwright/Selenium 的原因也很明确:
- 页面内容不需要浏览器执行 JS。
- 浏览器自动化开销大。
- 静态 HTML 更容易测试和复现。
- 文档站结构清晰,用 HTML parser 就够。
也不使用 Scrapy 起步,是因为本文重点是讲清楚爬虫核心链路。如果你已经有 Scrapy 项目经验,后面完全可以把 Fetcher、Parser、Storage 拆成 Scrapy Spider、Item Pipeline 和 Middleware。
4.2 整体流程
项目流程如下:
采集 Fetch
↓
解析 Parse
↓
清洗 Clean
↓
去重 Deduplicate
↓
存储 Store
↓
结果展示 Preview
更细一点:
1. 读取配置
2. 初始化 requests.Session
3. 检查 robots.txt
4. 请求 subsystem-apis.html
5. 解析章节 h2
6. 解析章节下的链接 li > a
7. 补全相对 URL
8. 去除锚点和重复 URL
9. 请求每个详情页
10. 解析详情页标题和正文首段
11. 清洗字段
12. 写入 CSV、JSONL、SQLite
13. 打印前几行结果
4.3 为什么选 requests
requests 的优点是简单、稳定、生态成熟。对于静态页面,它够用了。
本文会用到:
requests.Session()
HTTPAdapter
Retry
timeout
headers
虽然 requests 本身不是异步库,但这次抓取量很小,单线程更安全,也更方便调试。
4.4 为什么选 BeautifulSoup + lxml
BeautifulSoup 的优势在于容错强,写法直观;lxml 解析速度快,选择器支持稳定。
安装时:
pip install beautifulsoup4 lxml
解析时:
soup = BeautifulSoup(html, "lxml")
对技术文档站来说,CSS selector 已经够用。例如:
soup.select_one("div.body")
soup.select("ul li a[href]")
4.5 为什么不用正则解析 HTML
正则可以做简单提取,但不适合解析层级结构。导航树抽取需要知道:
某个 h2 后面的 ul 属于这个 h2
某个 li 下的 a 是一个文档项
某个链接是站内链接还是站外链接
某个锚点是否需要去掉
这些事情用 DOM 解析更稳。
5️⃣ 环境准备与依赖安装(可复现)
5.1 Python 版本
建议使用:
Python 3.10+
我个人推荐 Python 3.11 或 3.12。本文代码没有使用特别新的语法,Python 3.10 以上基本都能跑。
查看版本:
python --version
5.2 创建虚拟环境
Linux/macOS:
python -m venv .venv
source .venv/bin/activate
Windows PowerShell:
python -m venv .venv
.venv\Scripts\Activate.ps1
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 推荐项目结构
项目目录如下:
kernel_doc_spider/
├── README.md
├── requirements.txt
├── output/
│ └── .gitkeep
└── kernel_doc_spider/
├── __init__.py
├── config.py
├── models.py
├── robots.py
├── fetcher.py
├── parser.py
├── storage.py
└── main.py
如果只是临时练习,也可以写成一个单文件脚本。但我更建议拆开。原因很现实:爬虫一旦稍微复杂一点,请求、解析、存储混在一起会很难改。
创建目录:
mkdir -p kernel_doc_spider/kernel_doc_spider
mkdir -p kernel_doc_spider/output
touch kernel_doc_spider/kernel_doc_spider/__init__.py
touch kernel_doc_spider/README.md
touch kernel_doc_spider/requirements.txt
touch kernel_doc_spider/output/.gitkeep
6️⃣ 核心实现:请求层(Fetcher)
请求层要解决的问题不是“发请求”这么简单,而是要把下面这些细节固定下来:
headers
timeout
session
重试
退避
robots 检查
频率控制
异常封装
6.1 配置文件:config.py
先写配置。
文件:kernel_doc_spider/config.py
from dataclasses import dataclass
from pathlib import Path
@dataclass(frozen=True)
class SpiderConfig:
"""
爬虫配置。
这里把容易变动的参数统一放到配置对象里,避免散落在代码各处。
"""
base_url: str = "https://docs.kernel.org/"
start_url: str = "https://docs.kernel.org/subsystem-apis.html"
user_agent: str = (
"KernelDocSpider/1.0 "
"(learning-purpose; respectful-crawl; contact: example@example.com)"
)
referer: str = "https://docs.kernel.org/"
timeout: float = 15.0
connect_timeout: float = 5.0
read_timeout: float = 15.0
max_retries: int = 3
backoff_factor: float = 1.5
# 每次请求之间的最小间隔,文档站没有必要高频访问
min_delay: float = 1.0
max_delay: float = 2.0
output_dir: Path = Path("output")
csv_filename: str = "kernel_doc_subsystems.csv"
jsonl_filename: str = "kernel_doc_subsystems.jsonl"
sqlite_filename: str = "kernel_doc_subsystems.sqlite3"
# 是否抓取详情页补充简介
fetch_detail: bool = True
# 防止误操作抓太多
max_detail_pages: int = 200
这里我加了 max_detail_pages。虽然本次目标页面链接数量并不夸张,但给爬虫加一个硬上限是好习惯。不要让程序因为选择器写错而一路递归出去。
6.2 数据模型:models.py
文件:kernel_doc_spider/models.py
from __future__ import annotations
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
import hashlib
def make_url_hash(url: str) -> str:
"""
根据 URL 生成稳定哈希,用于去重和数据库主键。
使用 sha1 已经足够应对本项目。
"""
return hashlib.sha1(url.encode("utf-8")).hexdigest()
def utc_now_iso() -> str:
"""
生成 UTC ISO 时间字符串。
"""
return datetime.now(timezone.utc).isoformat(timespec="seconds")
@dataclass
class KernelDocItem:
"""
Linux Kernel 文档子系统目录条目。
"""
subsystem_name: str
chapter: str
intro: str
doc_link: str
source_page: str
fetched_at: str
@property
def url_hash(self) -> str:
return make_url_hash(self.doc_link)
def to_dict(self) -> dict:
data = asdict(self)
data["url_hash"] = self.url_hash
return data
这里用 dataclass,后续存 CSV、JSON、SQLite 都会方便很多。
6.3 robots 检查:robots.py
文件:kernel_doc_spider/robots.py
from __future__ import annotations
import logging
from urllib import robotparser
from urllib.parse import urljoin
logger = logging.getLogger(__name__)
class RobotsChecker:
"""
robots.txt 检查器。
注意:
- robots.txt 是爬虫行业约定,不等于登录授权。
- 如果 robots.txt 无法读取,本示例采用保守策略:记录警告,但不绕过任何明显限制。
- 实际生产环境中可以根据公司合规要求改成“读取失败即停止”。
"""
def __init__(self, base_url: str, user_agent: str, strict: bool = False) -> None:
self.base_url = base_url
self.user_agent = user_agent
self.strict = strict
self.robot_url = urljoin(base_url, "/robots.txt")
self.parser = robotparser.RobotFileParser()
self.loaded = False
def load(self) -> None:
self.parser.set_url(self.robot_url)
try:
self.parser.read()
self.loaded = True
logger.info("robots.txt loaded: %s", self.robot_url)
except Exception as exc:
self.loaded = False
logger.warning("failed to load robots.txt: %s, error=%s", self.robot_url, exc)
if self.strict:
raise
def can_fetch(self, url: str) -> bool:
if not self.loaded:
self.load()
try:
return bool(self.parser.can_fetch(self.user_agent, url))
except Exception as exc:
logger.warning("robots check failed for %s: %s", url, exc)
return not self.strict
如果你所在公司或团队对爬虫合规要求更严格,可以把 strict=True,读不到 robots 就停止运行。
6.4 请求层:fetcher.py
文件:kernel_doc_spider/fetcher.py
from __future__ import annotations
import logging
import random
import time
from typing import Optional
from urllib.parse import urlparse
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from .config import SpiderConfig
from .robots import RobotsChecker
logger = logging.getLogger(__name__)
class FetchError(RuntimeError):
"""
请求失败异常。
"""
class Fetcher:
"""
请求层封装。
负责:
- Session 复用
- headers
- timeout
- robots 检查
- 失败重试
- 请求间隔
- 站内 URL 限制
"""
def __init__(self, config: SpiderConfig) -> None:
self.config = config
self.session = requests.Session()
self.robots = RobotsChecker(
base_url=config.base_url,
user_agent=config.user_agent,
strict=False,
)
self._last_request_at: Optional[float] = None
self._setup_session()
def _setup_session(self) -> None:
retry = Retry(
total=self.config.max_retries,
connect=self.config.max_retries,
read=self.config.max_retries,
status=self.config.max_retries,
backoff_factor=self.config.backoff_factor,
status_forcelist=(429, 500, 502, 503, 504),
allowed_methods=frozenset(["GET", "HEAD"]),
raise_on_status=False,
)
adapter = HTTPAdapter(max_retries=retry)
self.session.mount("http://", adapter)
self.session.mount("https://", adapter)
self.session.headers.update(
{
"User-Agent": self.config.user_agent,
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9,zh-CN;q=0.8,zh;q=0.7",
"Referer": self.config.referer,
"Connection": "keep-alive",
}
)
def _polite_sleep(self) -> None:
"""
简单频率控制。
不追求精确,只保证请求之间有间隔。
"""
now = time.time()
if self._last_request_at is None:
self._last_request_at = now
return
elapsed = now - self._last_request_at
delay = random.uniform(self.config.min_delay, self.config.max_delay)
if elapsed < delay:
sleep_for = delay - elapsed
logger.debug("sleep %.2fs before next request", sleep_for)
time.sleep(sleep_for)
self._last_request_at = time.time()
def _is_same_site(self, url: str) -> bool:
base_host = urlparse(self.config.base_url).netloc
target_host = urlparse(url).netloc
return target_host == base_host
def get_text(self, url: str) -> str:
"""
请求 URL 并返回文本内容。
"""
if not self._is_same_site(url):
raise FetchError(f"refuse to fetch external url: {url}")
if not self.robots.can_fetch(url):
raise FetchError(f"blocked by robots.txt: {url}")
self._polite_sleep()
timeout = (self.config.connect_timeout, self.config.read_timeout)
try:
logger.info("GET %s", url)
resp = self.session.get(url, timeout=timeout)
except requests.RequestException as exc:
raise FetchError(f"request failed: {url}, error={exc}") from exc
if resp.status_code >= 400:
raise FetchError(f"bad status code: {resp.status_code}, url={url}")
content_type = resp.headers.get("Content-Type", "")
if "text/html" not in content_type and "application/xhtml+xml" not in content_type:
logger.warning("unexpected content-type=%s url=%s", content_type, url)
resp.encoding = resp.apparent_encoding or resp.encoding or "utf-8"
return resp.text
def close(self) -> None:
self.session.close()
6.5 请求层几个关键点
headers
这里设置了:
User-Agent
Accept
Accept-Language
Referer
Connection
最重要的是 User-Agent。它不应该伪装成搜索引擎,也不应该写得过于随意。用于学习时,保持透明最好。
timeout
不要写裸请求:
requests.get(url)
这类请求可能一直卡住。正确写法是:
requests.get(url, timeout=(5, 15))
分别表示连接超时和读取超时。
session/cookie
本项目不需要登录,不需要手动设置 Cookie。但使用 requests.Session() 仍然有价值,因为它可以复用连接、统一 headers、统一 retry adapter。
失败处理
失败不应该立刻疯狂重试。本文使用 urllib3.Retry 做基础重试,并针对这些状态码退避:
429
500
502
503
504
429 是频率相关状态码。遇到 429,更应该降低频率,而不是加代理硬冲。
7️⃣ 核心实现:解析层(Parser)
解析层是本文重点。因为题目明确强调:
实战重点:导航树结构抽取。
我们不是只拿所有链接,而是要拿到:
章节 -> 子系统文档链接 -> 详情页简介
7.1 解析策略
目标页面是 Sphinx 生成的 HTML。常见结构大致是:
<div class="body">
<h1>Kernel subsystem documentation</h1>
<p>These books get into the details...</p>
<h2>Core subsystems</h2>
<ul>
<li><a href="core-api/index.html">Core API Documentation</a></li>
<li><a href="driver-api/index.html">Driver implementer’s API guide</a></li>
</ul>
<h2>Human interfaces</h2>
<ul>
<li><a href="input/index.html">Input Documentation</a></li>
</ul>
</div>
最自然的做法:
- 找到正文容器。
- 遍历正文内的
h2。 - 对每个
h2,向后找它下面的兄弟节点。 - 直到遇到下一个
h2停止。 - 在这段范围内提取
ul li a[href]。 - 把当前
h2文本作为chapter。 - 把
a文本作为subsystem_name。 - 把
href补成绝对 URL。
7.2 parser.py 完整实现
文件:kernel_doc_spider/parser.py
from __future__ import annotations
import logging
import re
from dataclasses import dataclass
from typing import Iterable
from urllib.parse import urldefrag, urljoin, urlparse
from bs4 import BeautifulSoup, Tag
from .models import KernelDocItem, utc_now_iso
logger = logging.getLogger(__name__)
SPACE_RE = re.compile(r"\s+")
def clean_text(text: str) -> str:
"""
清洗文本:
- 去掉 Sphinx 标题后的 ¶
- 合并多余空白
- 去除首尾空格
"""
if not text:
return ""
text = text.replace("¶", "")
text = SPACE_RE.sub(" ", text)
return text.strip()
def normalize_url(base_url: str, href: str) -> str:
"""
补全 URL,并去掉 fragment。
例如:
subsystem-apis.html#core-subsystems -> https://docs.kernel.org/subsystem-apis.html
core-api/index.html -> https://docs.kernel.org/core-api/index.html
"""
absolute = urljoin(base_url, href)
absolute, _fragment = urldefrag(absolute)
return absolute
def is_probably_doc_link(url: str) -> bool:
"""
判断是否是本次要处理的文档链接。
只保留 docs.kernel.org 下的 html 页面。
"""
parsed = urlparse(url)
if parsed.scheme not in {"http", "https"}:
return False
if parsed.netloc != "docs.kernel.org":
return False
path = parsed.path
if not path:
return False
if path.endswith("/"):
return True
if path.endswith(".html"):
return True
return False
def get_main_content(soup: BeautifulSoup) -> Tag:
"""
获取 Sphinx 正文区域。
不同主题可能使用不同容器,所以做多级兜底。
"""
selectors = [
"div.body",
"main",
"article",
"div.document",
"body",
]
for selector in selectors:
node = soup.select_one(selector)
if isinstance(node, Tag):
return node
raise ValueError("main content not found")
@dataclass
class LinkRecord:
subsystem_name: str
chapter: str
doc_link: str
class KernelDocParser:
"""
Linux Kernel 文档解析器。
"""
def parse_subsystem_index(self, html: str, page_url: str) -> list[LinkRecord]:
"""
解析 subsystem-apis.html,抽取章节和章节下的文档链接。
"""
soup = BeautifulSoup(html, "lxml")
main = get_main_content(soup)
records: list[LinkRecord] = []
seen_urls: set[str] = set()
h2_nodes = main.find_all("h2")
for h2 in h2_nodes:
chapter = clean_text(h2.get_text(" ", strip=True))
if not chapter:
continue
for sibling in h2.find_next_siblings():
if isinstance(sibling, Tag) and sibling.name == "h2":
break
if not isinstance(sibling, Tag):
continue
links = sibling.select("ul li a[href]")
for a in links:
name = clean_text(a.get_text(" ", strip=True))
href = a.get("href", "")
if not name or not href:
continue
url = normalize_url(page_url, href)
if not is_probably_doc_link(url):
continue
if url in seen_urls:
continue
seen_urls.add(url)
records.append(
LinkRecord(
subsystem_name=name,
chapter=chapter,
doc_link=url,
)
)
return records
def parse_sidebar_tree(self, html: str, page_url: str) -> list[LinkRecord]:
"""
可选:解析 Sphinx 左侧导航树。
本文主流程使用正文 h2 + ul 的结构,因为它更贴近“章节”。
这里保留一个 sidebar 解析函数,是为了说明导航树也可以从侧栏提取。
Sphinx/Alabaster 常见侧栏结构里会有:
- div.sphinxsidebar
- ul
- li
- a
"""
soup = BeautifulSoup(html, "lxml")
sidebar = soup.select_one("div.sphinxsidebar")
if not sidebar:
return []
records: list[LinkRecord] = []
seen_urls: set[str] = set()
for a in sidebar.select("ul li a[href]"):
name = clean_text(a.get_text(" ", strip=True))
href = a.get("href", "")
if not name or not href:
continue
url = normalize_url(page_url, href)
if not is_probably_doc_link(url):
continue
if url in seen_urls:
continue
seen_urls.add(url)
records.append(
LinkRecord(
subsystem_name=name,
chapter="Sidebar navigation",
doc_link=url,
)
)
return records
def parse_detail_intro(self, html: str) -> tuple[str, str]:
"""
解析详情页标题和简介。
返回:
- detail_title
- intro
策略:
1. 标题优先取正文 h1。
2. 简介取 h1 后、下一个 h2 前的前几个 p。
3. 如果没有 p,则取 meta description。
4. 再没有就返回空字符串,由上层填默认值。
"""
soup = BeautifulSoup(html, "lxml")
main = get_main_content(soup)
h1 = main.find("h1")
title = clean_text(h1.get_text(" ", strip=True)) if h1 else ""
paragraphs: list[str] = []
if h1:
for sibling in h1.find_next_siblings():
if isinstance(sibling, Tag) and sibling.name in {"h2", "h3"}:
break
if not isinstance(sibling, Tag):
continue
if sibling.name == "p":
text = clean_text(sibling.get_text(" ", strip=True))
if text:
paragraphs.append(text)
for p in sibling.find_all("p", recursive=True):
text = clean_text(p.get_text(" ", strip=True))
if text:
paragraphs.append(text)
if len(paragraphs) >= 2:
break
intro = " ".join(paragraphs[:2]).strip()
if not intro:
meta = soup.select_one('meta[name="description"]')
if meta and meta.get("content"):
intro = clean_text(meta.get("content", ""))
return title, intro
def build_item(
self,
link: LinkRecord,
detail_html: str | None,
source_page: str,
) -> KernelDocItem:
"""
根据目录页链接和详情页内容构建最终数据项。
"""
detail_title = ""
intro = ""
if detail_html:
try:
detail_title, intro = self.parse_detail_intro(detail_html)
except Exception as exc:
logger.warning("parse detail failed url=%s error=%s", link.doc_link, exc)
subsystem_name = clean_text(detail_title) or clean_text(link.subsystem_name)
if not intro:
intro = "No short introduction extracted from the document page."
return KernelDocItem(
subsystem_name=subsystem_name,
chapter=clean_text(link.chapter),
intro=clean_text(intro),
doc_link=link.doc_link,
source_page=source_page,
fetched_at=utc_now_iso(),
)
7.3 列表页如何拿详情链接
核心代码是这一段:
for h2 in h2_nodes:
chapter = clean_text(h2.get_text(" ", strip=True))
for sibling in h2.find_next_siblings():
if isinstance(sibling, Tag) and sibling.name == "h2":
break
links = sibling.select("ul li a[href]")
这段代码的关键在于:它不是直接 main.select("a[href]"),而是利用 h2 的相邻兄弟节点来划分章节范围。
换句话说,它把页面看成这样:
h2 A
A 下的段落
A 下的 ul
h2 B
B 下的段落
B 下的 ul
这种解析方式适合 Sphinx 目录页,也适合很多 Markdown/RST 转 HTML 的文档站。
7.4 详情页如何抽字段
详情页字段主要补充 intro。
抽取规则:
正文 h1 作为标题
h1 后第一个或前两个 p 作为简介
遇到 h2/h3 停止
没有正文 p 时使用 meta description
仍然没有就填默认说明
为什么不直接拿全页文本?因为全页文本太长,也容易混入侧栏、版权、导航、索引等无关内容。
7.5 缺失字段怎么办
爬虫一定要允许“不完美页面”存在。缺失字段时,本文采用以下策略:
| 字段 | 缺失处理 |
|---|---|
| 子系统名 | 优先详情页 h1,失败用目录页链接文本 |
| 章节 | 目录页 h2,缺失则跳过 |
| 简介 | 详情页首段,缺失则填默认语句 |
| 文档链接 | 必须存在,缺失则跳过 |
| URL hash | 根据 doc_link 生成,不依赖页面 |
示例:
if not intro:
intro = "No short introduction extracted from the document page."
不要因为一个详情页简介没抽到,就让整个程序崩掉。真实爬虫最重要的是稳定产出,而不是每个字段都“理论完美”。
8️⃣ 数据存储与导出(Storage)
本文同时实现三种存储:
CSV
JSON Lines
SQLite
CSV 适合直接用 Excel、WPS、pandas 查看。
JSON Lines 适合后续进入搜索、日志或数据管道。
SQLite 适合去重、增量更新和本地查询。
8.1 字段映射表
| 字段名 | 类型 | 示例值 |
|---|---|---|
url_hash |
TEXT | b86f8c... |
subsystem_name |
TEXT | Core API Documentation |
chapter |
TEXT | Core subsystems |
intro |
TEXT | This is the beginning of a manual for core kernel APIs. |
doc_link |
TEXT | https://docs.kernel.org/core-api/index.html |
source_page |
TEXT | https://docs.kernel.org/subsystem-apis.html |
fetched_at |
TEXT | 2026-06-08T20:30:00+00:00 |
8.2 去重策略
本项目采用 URL 唯一策略:
同一个 doc_link 只保留一条记录
实现上用:
url_hash = sha1(doc_link)
SQLite 表中设置:
url_hash TEXT PRIMARY KEY
如果未来做内容变更监控,可以额外加:
content_hash
title_hash
intro_hash
但这次没有必要。
8.3 storage.py 完整实现
文件:kernel_doc_spider/storage.py
from __future__ import annotations
import csv
import json
import logging
import sqlite3
from pathlib import Path
from typing import Iterable
from .models import KernelDocItem
logger = logging.getLogger(__name__)
CSV_FIELDS = [
"url_hash",
"subsystem_name",
"chapter",
"intro",
"doc_link",
"source_page",
"fetched_at",
]
class Storage:
"""
数据存储层。
"""
def __init__(self, output_dir: Path) -> None:
self.output_dir = output_dir
self.output_dir.mkdir(parents=True, exist_ok=True)
def save_csv(self, items: Iterable[KernelDocItem], filename: str) -> Path:
path = self.output_dir / filename
rows = [item.to_dict() for item in items]
with path.open("w", newline="", encoding="utf-8-sig") as f:
writer = csv.DictWriter(f, fieldnames=CSV_FIELDS)
writer.writeheader()
writer.writerows(rows)
logger.info("saved csv: %s rows=%d", path, len(rows))
return path
def save_jsonl(self, items: Iterable[KernelDocItem], filename: str) -> Path:
path = self.output_dir / filename
rows = [item.to_dict() for item in items]
with path.open("w", encoding="utf-8") as f:
for row in rows:
f.write(json.dumps(row, ensure_ascii=False) + "\n")
logger.info("saved jsonl: %s rows=%d", path, len(rows))
return path
def save_sqlite(self, items: Iterable[KernelDocItem], filename: str) -> Path:
path = self.output_dir / filename
rows = [item.to_dict() for item in items]
conn = sqlite3.connect(path)
try:
self._init_db(conn)
self._upsert_items(conn, rows)
conn.commit()
finally:
conn.close()
logger.info("saved sqlite: %s rows=%d", path, len(rows))
return path
def _init_db(self, conn: sqlite3.Connection) -> None:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS kernel_doc_subsystems (
url_hash TEXT PRIMARY KEY,
subsystem_name TEXT NOT NULL,
chapter TEXT NOT NULL,
intro TEXT,
doc_link TEXT NOT NULL UNIQUE,
source_page TEXT NOT NULL,
fetched_at TEXT NOT NULL
)
"""
)
conn.execute(
"""
CREATE INDEX IF NOT EXISTS idx_kernel_doc_subsystems_chapter
ON kernel_doc_subsystems(chapter)
"""
)
def _upsert_items(self, conn: sqlite3.Connection, rows: list[dict]) -> None:
conn.executemany(
"""
INSERT INTO kernel_doc_subsystems (
url_hash,
subsystem_name,
chapter,
intro,
doc_link,
source_page,
fetched_at
)
VALUES (
:url_hash,
:subsystem_name,
:chapter,
:intro,
:doc_link,
:source_page,
:fetched_at
)
ON CONFLICT(url_hash) DO UPDATE SET
subsystem_name = excluded.subsystem_name,
chapter = excluded.chapter,
intro = excluded.intro,
doc_link = excluded.doc_link,
source_page = excluded.source_page,
fetched_at = excluded.fetched_at
""",
rows,
)
8.4 为什么 CSV 使用 utf-8-sig
这里写 CSV 时使用:
encoding="utf-8-sig"
主要是为了照顾 Excel。很多中文环境下的 Excel 直接打开 UTF-8 CSV 容易乱码,加 BOM 后兼容性会更好。
如果你后续用 pandas 或数据库,不加 BOM 也可以。
9️⃣ 运行方式与结果展示(必写)
9.1 main.py 完整入口
文件:kernel_doc_spider/main.py
from __future__ import annotations
import argparse
import logging
import sys
from dataclasses import replace
from pathlib import Path
from .config import SpiderConfig
from .fetcher import Fetcher, FetchError
from .models import KernelDocItem
from .parser import KernelDocParser, LinkRecord
from .storage import Storage
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",
datefmt="%Y-%m-%d %H:%M:%S",
)
def dedupe_links(links: list[LinkRecord]) -> list[LinkRecord]:
"""
根据 doc_link 去重,保持原始顺序。
"""
seen: set[str] = set()
result: list[LinkRecord] = []
for link in links:
if link.doc_link in seen:
continue
seen.add(link.doc_link)
result.append(link)
return result
def crawl(config: SpiderConfig) -> list[KernelDocItem]:
fetcher = Fetcher(config)
parser = KernelDocParser()
try:
index_html = fetcher.get_text(config.start_url)
links = parser.parse_subsystem_index(index_html, config.start_url)
links = dedupe_links(links)
logging.info("found %d subsystem links", len(links))
if not links:
logging.warning("no links found from subsystem index")
return []
items: list[KernelDocItem] = []
for idx, link in enumerate(links, start=1):
if idx > config.max_detail_pages:
logging.warning(
"reached max_detail_pages=%d, stop fetching details",
config.max_detail_pages,
)
break
detail_html = None
if config.fetch_detail:
try:
detail_html = fetcher.get_text(link.doc_link)
except FetchError as exc:
logging.warning("fetch detail failed: %s", exc)
item = parser.build_item(
link=link,
detail_html=detail_html,
source_page=config.start_url,
)
items.append(item)
logging.info(
"parsed %d/%d chapter=%s name=%s",
idx,
len(links),
item.chapter,
item.subsystem_name,
)
return items
finally:
fetcher.close()
def print_preview(items: list[KernelDocItem], limit: int = 5) -> None:
"""
打印前几行结果,便于命令行快速确认。
"""
print("\nPreview:")
print("-" * 100)
for item in items[:limit]:
print(f"chapter : {item.chapter}")
print(f"subsystem_name : {item.subsystem_name}")
print(f"intro : {item.intro[:160]}")
print(f"doc_link : {item.doc_link}")
print("-" * 100)
def parse_args(argv: list[str]) -> argparse.Namespace:
p = argparse.ArgumentParser(
description="Crawl Linux Kernel documentation subsystem directory."
)
p.add_argument(
"--start-url",
default="https://docs.kernel.org/subsystem-apis.html",
help="Start URL of Linux Kernel subsystem documentation page.",
)
p.add_argument(
"--out-dir",
default="output",
help="Output directory.",
)
p.add_argument(
"--no-detail",
action="store_true",
help="Do not fetch detail pages; only parse index page links.",
)
p.add_argument(
"--max-detail-pages",
type=int,
default=200,
help="Maximum number of detail pages to fetch.",
)
p.add_argument(
"--verbose",
action="store_true",
help="Enable debug logging.",
)
return p.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv or sys.argv[1:])
setup_logging(args.verbose)
config = SpiderConfig()
config = replace(
config,
start_url=args.start_url,
output_dir=Path(args.out_dir),
fetch_detail=not args.no_detail,
max_detail_pages=args.max_detail_pages,
)
items = crawl(config)
if not items:
logging.warning("no items crawled")
return 2
storage = Storage(config.output_dir)
csv_path = storage.save_csv(items, config.csv_filename)
jsonl_path = storage.save_jsonl(items, config.jsonl_filename)
sqlite_path = storage.save_sqlite(items, config.sqlite_filename)
print_preview(items, limit=5)
print("\nSaved files:")
print(f"CSV : {csv_path}")
print(f"JSONL : {jsonl_path}")
print(f"SQLite : {sqlite_path}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
9.2 运行命令
在项目根目录执行:
python -m kernel_doc_spider.main
指定输出目录:
python -m kernel_doc_spider.main --out-dir output
只解析目录页,不抓详情页:
python -m kernel_doc_spider.main --no-detail
限制最多抓 20 个详情页:
python -m kernel_doc_spider.main --max-detail-pages 20
打开调试日志:
python -m kernel_doc_spider.main --verbose
9.3 输出位置
默认输出:
output/kernel_doc_subsystems.csv
output/kernel_doc_subsystems.jsonl
output/kernel_doc_subsystems.sqlite3
SQLite 表名:
kernel_doc_subsystems
查询 SQLite:
sqlite3 output/kernel_doc_subsystems.sqlite3
进入后执行:
.headers on
.mode column
SELECT chapter, subsystem_name, doc_link
FROM kernel_doc_subsystems
LIMIT 5;
9.4 示例结果
示例 CSV 结果如下。具体内容会随官方文档更新略有变化。
url_hash,subsystem_name,chapter,intro,doc_link,source_page,fetched_at
8e0f2c...,Core API Documentation,Core subsystems,This is the beginning of a manual for core kernel APIs.,https://docs.kernel.org/core-api/index.html,https://docs.kernel.org/subsystem-apis.html,2026-06-08T20:30:00+00:00
70f1a7...,Driver implementer’s API guide,Core subsystems,This guide collects documentation for Linux driver authors and subsystem maintainers.,https://docs.kernel.org/driver-api/index.html,https://docs.kernel.org/subsystem-apis.html,2026-06-08T20:30:02+00:00
3b4f99...,Memory Management Documentation,Core subsystems,This section contains documentation related to memory management in the Linux kernel.,https://docs.kernel.org/mm/index.html,https://docs.kernel.org/subsystem-apis.html,2026-06-08T20:30:04+00:00
fdd325...,Power Management,Core subsystems,This document area collects Linux power management documentation.,https://docs.kernel.org/power/index.html,https://docs.kernel.org/subsystem-apis.html,2026-06-08T20:30:06+00:00
40a9ea...,Scheduler,Core subsystems,This document area collects scheduler-related kernel documentation.,https://docs.kernel.org/scheduler/index.html,https://docs.kernel.org/subsystem-apis.html,2026-06-08T20:30:08+00:00
命令行预览类似:
Preview:
----------------------------------------------------------------------------------------------------
chapter : Core subsystems
subsystem_name : Core API Documentation
intro : This is the beginning of a manual for core kernel APIs.
doc_link : https://docs.kernel.org/core-api/index.html
----------------------------------------------------------------------------------------------------
chapter : Core subsystems
subsystem_name : Driver implementer’s API guide
intro : No short introduction extracted from the document page.
doc_link : https://docs.kernel.org/driver-api/index.html
----------------------------------------------------------------------------------------------------
chapter : Human interfaces
subsystem_name : Input Documentation
intro : The input subsystem documentation collects information about input devices and APIs.
doc_link : https://docs.kernel.org/input/index.html
----------------------------------------------------------------------------------------------------
9.5 最小单文件版本
如果你只是想快速试一下,不想拆项目,也可以用下面这个单文件版本。工程能力不如上面的完整项目,但跑通思路没问题。
文件:quick_kernel_doc_spider.py
import csv
import hashlib
import time
from datetime import datetime, timezone
from urllib.parse import urljoin, urldefrag, urlparse
from urllib import robotparser
import requests
from bs4 import BeautifulSoup, Tag
BASE_URL = "https://docs.kernel.org/"
START_URL = "https://docs.kernel.org/subsystem-apis.html"
UA = "KernelDocSpider/1.0 (learning-purpose; respectful-crawl)"
def clean_text(s: str) -> str:
return " ".join((s or "").replace("¶", "").split()).strip()
def normalize_url(base: str, href: str) -> str:
url = urljoin(base, href)
url, _ = urldefrag(url)
return url
def sha1(s: str) -> str:
return hashlib.sha1(s.encode("utf-8")).hexdigest()
def now_iso() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def can_fetch(url: str) -> bool:
rp = robotparser.RobotFileParser()
rp.set_url(urljoin(BASE_URL, "/robots.txt"))
try:
rp.read()
return rp.can_fetch(UA, url)
except Exception:
# 学习示例:robots 读取失败时不做绕过动作,只保守记录。
# 生产中可改成 False。
return True
def fetch(url: str) -> str:
if urlparse(url).netloc != "docs.kernel.org":
raise ValueError(f"external url refused: {url}")
if not can_fetch(url):
raise RuntimeError(f"blocked by robots.txt: {url}")
resp = requests.get(
url,
headers={
"User-Agent": UA,
"Referer": BASE_URL,
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
},
timeout=(5, 15),
)
resp.raise_for_status()
resp.encoding = resp.apparent_encoding or resp.encoding or "utf-8"
return resp.text
def get_main(soup: BeautifulSoup) -> Tag:
return soup.select_one("div.body") or soup.select_one("main") or soup.body
def parse_index(html: str):
soup = BeautifulSoup(html, "lxml")
main = get_main(soup)
seen = set()
rows = []
for h2 in main.find_all("h2"):
chapter = clean_text(h2.get_text(" ", strip=True))
for sibling in h2.find_next_siblings():
if isinstance(sibling, Tag) and sibling.name == "h2":
break
if not isinstance(sibling, Tag):
continue
for a in sibling.select("ul li a[href]"):
name = clean_text(a.get_text(" ", strip=True))
href = a.get("href")
link = normalize_url(START_URL, href)
if not name:
continue
if urlparse(link).netloc != "docs.kernel.org":
continue
if link in seen:
continue
seen.add(link)
rows.append(
{
"subsystem_name": name,
"chapter": chapter,
"doc_link": link,
}
)
return rows
def parse_intro(html: str):
soup = BeautifulSoup(html, "lxml")
main = get_main(soup)
h1 = main.find("h1")
title = clean_text(h1.get_text(" ", strip=True)) if h1 else ""
paragraphs = []
if h1:
for sibling in h1.find_next_siblings():
if isinstance(sibling, Tag) and sibling.name in {"h2", "h3"}:
break
if not isinstance(sibling, Tag):
continue
if sibling.name == "p":
text = clean_text(sibling.get_text(" ", strip=True))
if text:
paragraphs.append(text)
for p in sibling.find_all("p"):
text = clean_text(p.get_text(" ", strip=True))
if text:
paragraphs.append(text)
if len(paragraphs) >= 2:
break
intro = " ".join(paragraphs[:2])
return title, intro
def main():
index_html = fetch(START_URL)
rows = parse_index(index_html)
final_rows = []
for i, row in enumerate(rows, start=1):
print(f"[{i}/{len(rows)}] {row['chapter']} - {row['subsystem_name']}")
intro = ""
title = ""
try:
time.sleep(1.2)
detail_html = fetch(row["doc_link"])
title, intro = parse_intro(detail_html)
except Exception as exc:
print(f" detail failed: {exc}")
name = title or row["subsystem_name"]
final_rows.append(
{
"url_hash": sha1(row["doc_link"]),
"subsystem_name": name,
"chapter": row["chapter"],
"intro": intro or "No short introduction extracted from the document page.",
"doc_link": row["doc_link"],
"source_page": START_URL,
"fetched_at": now_iso(),
}
)
with open("kernel_doc_subsystems.csv", "w", newline="", encoding="utf-8-sig") as f:
fieldnames = [
"url_hash",
"subsystem_name",
"chapter",
"intro",
"doc_link",
"source_page",
"fetched_at",
]
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(final_rows)
print(f"saved {len(final_rows)} rows to kernel_doc_subsystems.csv")
if __name__ == "__main__":
main()
运行:
python quick_kernel_doc_spider.py
🔟 常见问题与排错(强烈建议写)
10.1 403 怎么办
403 表示服务器拒绝访问。遇到 403,不要第一反应就是“上代理”。先检查这几件事:
User-Agent 是否为空
Referer 是否异常
是否访问了 robots 不允许的路径
请求频率是否过高
是否请求了不该请求的资源
是否被临时限制
本文这种文档站抓取,正常低频访问一般不需要代理。代理不是万能药,很多时候反而会让请求更可疑。
建议处理方式:
if resp.status_code == 403:
logger.warning("403 forbidden, slow down and check headers/robots: %s", url)
然后人工确认是否仍然适合继续。
10.2 429 怎么办
429 通常表示请求过多。解决思路:
降低频率
增加 sleep
减少详情页抓取数量
增加本地缓存
尊重 Retry-After 响应头
停止一段时间后再试
可以在请求层加一个对 Retry-After 的处理:
retry_after = resp.headers.get("Retry-After")
if retry_after and retry_after.isdigit():
time.sleep(int(retry_after))
不过本文已经在 Retry 中把 429 放入重试状态码,并且设置了退避。对这种小规模抓取,足够了。
10.3 HTML 抓到空壳怎么办
如果抓到的 HTML 只有:
<div id="app"></div>
<script src="main.js"></script>
说明页面可能是动态渲染。
排查方式:
- 浏览器右键查看网页源代码,不是 Elements 面板。
- 看目标内容是否出现在源代码里。
- 如果不在,打开 DevTools Network。
- 找接口请求。
- 如果接口开放且允许访问,优先抓接口。
- 如果必须浏览器渲染,再考虑 Playwright。
本文目标站不属于这种情况,静态 HTML 足够。
10.4 解析报错怎么办
解析报错通常有几类:
选择器失效
正文容器变了
标题层级变了
某些节点不是 Tag
某些 href 为空
对应解决方式:
main = soup.select_one("div.body") or soup.select_one("main") or soup.body
还有:
if not isinstance(sibling, Tag):
continue
写 parser 时,一定要默认页面会“不规整”。真实网页从来不是为了你的爬虫而存在的。
10.5 抽不到简介怎么办
详情页不一定都有清晰首段。有些页面开头是目录,有些页面直接进入列表,有些页面标题后就是代码块。
本文处理:
if not intro:
intro = "No short introduction extracted from the document page."
如果你希望简介更好看,可以加一个“正文摘要函数”:
def summarize_from_text(text: str, max_len: int = 220) -> str:
text = clean_text(text)
if len(text) <= max_len:
return text
return text[:max_len].rsplit(" ", 1)[0] + "..."
然后在无段落时,从正文整体文本里取前 200 字符。
10.6 编码/乱码如何处理
requests 默认编码不一定准确。本文用了:
resp.encoding = resp.apparent_encoding or resp.encoding or "utf-8"
CSV 用:
encoding="utf-8-sig"
如果仍然乱码,检查:
终端编码
编辑器编码
Excel 打开方式
文件实际编码
用 pandas 读取:
import pandas as pd
df = pd.read_csv("output/kernel_doc_subsystems.csv", encoding="utf-8-sig")
print(df.head())
10.7 为什么抓到重复链接
Sphinx 页面常有侧栏导航、正文目录、页脚链接、语言链接、Show Source 链接。如果粗暴抓全站 a[href],重复和噪声非常多。
本文用两层控制:
只在正文 h2 对应范围内抓 ul li a
根据 doc_link 去重
过滤站外链接
过滤非 html 链接
去掉 fragment
去掉 fragment 很关键:
absolute, _fragment = urldefrag(absolute)
否则下面两个 URL 会被当成不同链接:
https://docs.kernel.org/subsystem-apis.html#core-subsystems
https://docs.kernel.org/subsystem-apis.html
10.8 为什么不用全站递归
因为需求不是“镜像整个 docs.kernel.org”,而是“抽取子系统目录”。全站递归有几个问题:
范围失控
访问量变大
去重复杂
容易抓到无关页面
对目标站不友好
工程上,范围控制比“能抓更多”更重要。
1️⃣1️⃣ 进阶优化(可选但加分)
11.1 增加本地缓存
开发 parser 时,反复请求同一个页面很浪费。可以加一个简单文件缓存。
新增文件:cache.py
from __future__ import annotations
import hashlib
from pathlib import Path
class HtmlCache:
def __init__(self, cache_dir: Path = Path(".cache/html")) -> None:
self.cache_dir = cache_dir
self.cache_dir.mkdir(parents=True, exist_ok=True)
def _key(self, url: str) -> str:
return hashlib.sha1(url.encode("utf-8")).hexdigest() + ".html"
def get(self, url: str) -> str | None:
path = self.cache_dir / self._key(url)
if not path.exists():
return None
return path.read_text(encoding="utf-8", errors="ignore")
def set(self, url: str, html: str) -> None:
path = self.cache_dir / self._key(url)
path.write_text(html, encoding="utf-8")
然后在 Fetcher 里接入:
cached = cache.get(url)
if cached:
return cached
html = real_fetch(url)
cache.set(url, html)
return html
缓存的好处很明显:
减少重复请求
方便离线调试 parser
降低对目标站压力
11.2 并发优化
本文不建议默认并发。但如果你后续要抓更多文档页,可以考虑低并发线程池。
示例:
from concurrent.futures import ThreadPoolExecutor, as_completed
def fetch_one(link):
try:
html = fetcher.get_text(link.doc_link)
return link, html, None
except Exception as exc:
return link, None, exc
with ThreadPoolExecutor(max_workers=3) as pool:
futures = [pool.submit(fetch_one, link) for link in links]
for fut in as_completed(futures):
link, html, exc = fut.result()
if exc:
print("failed:", link.doc_link, exc)
continue
注意,这里即使并发,也建议:
max_workers 不要太大
Fetcher 需要线程安全处理
每个线程独立 Session 更稳
仍然保留 delay
对公开文档站,max_workers=2~3 已经很多了。
11.3 asyncio 版本思路
如果换成异步,可以用:
aiohttp
asyncio.Semaphore
asyncio.sleep
伪代码:
sem = asyncio.Semaphore(3)
async with sem:
await asyncio.sleep(random.uniform(1, 2))
async with session.get(url, timeout=timeout) as resp:
html = await resp.text()
异步适合 I/O 密集型,但复杂度会高一些。本文这种规模,单线程更舒服。
11.4 Scrapy 改造方向
如果要改成 Scrapy,大致对应关系:
| 当前模块 | Scrapy 中的位置 |
|---|---|
Fetcher |
Downloader Middleware / Scrapy 默认下载器 |
KernelDocItem |
Item |
KernelDocParser |
Spider parse 方法 |
Storage |
Item Pipeline |
Retry |
RetryMiddleware |
delay |
DOWNLOAD_DELAY |
robots |
ROBOTSTXT_OBEY |
Scrapy 配置示例:
ROBOTSTXT_OBEY = True
DOWNLOAD_DELAY = 1.5
CONCURRENT_REQUESTS_PER_DOMAIN = 2
USER_AGENT = "KernelDocSpider/1.0 (learning-purpose; respectful-crawl)"
11.5 断点续跑
断点续跑可以依赖 SQLite。
思路:
SELECT url_hash FROM kernel_doc_subsystems;
程序启动时读取已抓 URL:
done_urls = load_done_urls_from_sqlite()
抓取前判断:
if link.doc_link in done_urls:
continue
这样即使程序中断,下次也不用从头抓。
11.6 日志与监控
至少记录这些指标:
入口页请求成功/失败
发现链接数量
详情页成功数量
详情页失败数量
字段缺失数量
最终写入数量
运行耗时
简单统计可以这样写:
stats = {
"links_found": len(links),
"detail_success": 0,
"detail_failed": 0,
"items_saved": 0,
}
结束时打印:
logger.info("stats=%s", stats)
如果是长期任务,可以把日志写到文件:
logging.FileHandler("spider.log", encoding="utf-8")
11.7 定时任务
Linux/macOS 可以用 cron:
crontab -e
每天凌晨 3 点执行:
0 3 * * * cd /path/to/kernel_doc_spider && /path/to/.venv/bin/python -m kernel_doc_spider.main --out-dir output >> spider.log 2>&1
更复杂的调度可以用:
Airflow
Prefect
Dagster
GitHub Actions
但如果只是每天抓一个文档目录,cron 已经很够了。
11.8 文档变更监控
如果你想知道某个子系统文档有没有变化,可以为详情页正文生成 hash:
import hashlib
def content_hash(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()
数据库新增字段:
content_hash TEXT
每次抓取后对比:
URL 相同但 content_hash 变化 -> 文档可能更新
URL 新增 -> 新文档
URL 消失 -> 文档可能移动或删除
这就从“目录抽取”升级成了“文档变更监控”。
1️⃣2️⃣ 总结与延伸阅读
这篇文章完成了一个完整的小型爬虫工程:用 Python 抓取 Linux Kernel 官方文档中的子系统目录,抽取“子系统名、章节、简介、文档链接”,并导出到 CSV、JSON Lines 和 SQLite。
回头看,真正值得记住的不是某一行选择器,而是这套流程:
先确认目标页面类型
再确认合规边界
再写请求层
再写解析层
再做字段容错
再做去重
最后落地存储
本文项目的核心难点是“导航树结构抽取”。它和普通列表页不同,需要知道每个链接属于哪个章节。我们用 h2 + find_next_siblings() 的方式,把页面结构转成了结构化数据。这种方法可以迁移到很多技术文档站,比如:
Python 官方文档
Django 文档
Kubernetes 文档
PostgreSQL 文档
LLVM 文档
Rust 文档
Sphinx 项目文档
MkDocs 项目文档
下一步可以继续做几件事:
- 改成 Scrapy 项目,加入 pipeline、middleware 和自动限速。
- 改成 Playwright 版本,练习动态文档站抽取。
- 增加 SQLite 增量更新,做文档变更监控。
- 接入全文索引,例如 SQLite FTS5、Meilisearch 或 Elasticsearch。
- 把抽取结果做成一个本地知识库导航页面。
最后说一句自己的感受:爬虫写多了会发现,真正难的不是“请求页面”,而是把边界、结构、容错和后续维护想清楚。能跑的脚本不稀奇,能稳定复用、能让别人看懂、能温和访问目标站的爬虫,才算真正有点工程味。
附录:完整文件清单
为了方便复制,这里把项目文件清单再汇总一次。
kernel_doc_spider/
├── README.md
├── requirements.txt
├── output/
│ └── .gitkeep
└── kernel_doc_spider/
├── __init__.py
├── config.py
├── models.py
├── robots.py
├── fetcher.py
├── parser.py
├── storage.py
└── main.py
requirements.txt:
requests==2.32.3
beautifulsoup4==4.12.3
lxml==5.3.0
运行:
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
python -m kernel_doc_spider.main --out-dir output
查看输出:
ls -lh output
预期文件:
kernel_doc_subsystems.csv
kernel_doc_subsystems.jsonl
kernel_doc_subsystems.sqlite3
至此,一个面向 Linux 内核文档子系统目录的导航树抽取爬虫就完成了。
🌟 文末
好啦~以上就是本期的全部内容啦!如果你在实践过程中遇到任何疑问,欢迎在评论区留言交流,我看到都会尽量回复~咱们下期见!
小伙伴们在批阅的过程中,如果觉得文章不错,欢迎点赞、收藏、关注哦~
三连就是对我写作道路上最好的鼓励与支持! ❤️🔥
✅ 专栏持续更新中|建议收藏 + 订阅
墙裂推荐订阅专栏 👉 《Python爬虫实战》,本专栏秉承着以“入门 → 进阶 → 工程化 → 项目落地”的路线持续更新,争取让每一期内容都做到:
✅ 讲得清楚(原理)|✅ 跑得起来(代码)|✅ 用得上(场景)|✅ 扛得住(工程化)
📣 想系统提升的小伙伴:强烈建议先订阅专栏 《Python爬虫实战》,再按目录大纲顺序学习,效率十倍上升~
✅ 互动征集
想让我把【某站点/某反爬/某验证码/某分布式方案】等写成某期实战?
评论区留言告诉我你的需求,我会优先安排实现(更新)哒~
⭐️ 若喜欢我,就请关注我叭~(更新不迷路)
⭐️ 若对你有用,就请点赞支持一下叭~(给我一点点动力)
⭐️ 若有疑问,就请评论留言告诉我叭~(我会补坑 & 更新迭代)
✅ 免责声明
本文爬虫思路、相关技术和代码仅用于学习参考,对阅读本文后的进行爬虫行为的用户本作者不承担任何法律责任。
使用或者参考本项目即表示您已阅读并同意以下条款:
- 合法使用: 不得将本项目用于任何违法、违规或侵犯他人权益的行为,包括但不限于网络攻击、诈骗、绕过身份验证、未经授权的数据抓取等。
- 风险自负: 任何因使用本项目而产生的法律责任、技术风险或经济损失,由使用者自行承担,项目作者不承担任何形式的责任。
- 禁止滥用: 不得将本项目用于违法牟利、黑产活动或其他不当商业用途。
- 使用或者参考本项目即视为同意上述条款,即 “谁使用,谁负责” 。如不同意,请立即停止使用并删除本项目。!!!
更多推荐
所有评论(0)