文章

25 · 网络爬虫：HTTP 抓取与数据提取

#029 · 2026-04-16 · Python

阅读数据加载中… 点赞数据加载中…

生成页数

幻灯片语言

提炼重点 / 自定义指令 (可选)

🔗 知识图谱导航：阅读本文前，建议先掌握《18 · HTTP 与 Web 框架基石》中的 HTTP 请求结构和《23 · 前后端解耦：AJAX 与 RESTful API》中的 JSON 数据格式——本文把这两块拼在一起，用 Python 内置的 urllib 实现一个带反爬对抗能力的爬虫框架。

运行环境：Python 3.12+ 标准库，零额外依赖，直接运行。Mock 数据模拟真实爬取，不需要访问外网。

极客解析：爬虫的本质是"自动化的 HTTP 客户端"。核心挑战不是发请求，而是：① 如何应对反爬（User-Agent 轮换、延迟、代理）；② 如何处理网络抖动（指数退避重试）；③ 如何结构化存储数据（@dataclass + SQLite）。

爬虫架构

URL 队列
    │
    ▼
fetch(url)          ← 带 UA 轮换、超时控制、指数退避重试
    │
    ▼
parse(html, url)    ← 抽象方法，子类实现具体解析逻辑
    │
    ▼
CrawlItem 列表      ← @dataclass 结构化数据容器
    │
    ▼
save(items)         ← 抽象方法，子类实现存储逻辑（SQLite/CSV/JSON）

反爬对抗策略

User-Agent 轮换    随机选 UA，模拟不同浏览器
请求延迟           每次请求间隔 1-3 秒，避免触发频率限制
指数退避重试       失败后等待 2^attempt 秒再重试，避免雪崩效应
Cookie 管理        维护 session，模拟登录状态
Referer 头         伪造来源页面，绕过简单的 Referer 检查

步步为营：核心逻辑自适应拆解

这一篇的核心是爬虫框架的三层结构：CrawlItem 数据容器 → BaseCrawler 抽象骨架（fetch+parse+save）→ MockShopCrawler 具体实现。下面每一步都聚焦一个机制，零依赖可直接运行。

Step 1：用 @dataclass 定义结构化爬取数据模型 CrawlItem

痛点与机制：

CrawlItem 是爬取结果的标准数据容器。@dataclass 自动生成 __init__、__repr__、__eq__，不需要手写样板代码。crawled_at: str = field(default_factory=now_str) 是关键——default_factory 在每次创建实例时调用 now_str()，而不是在类定义时调用一次，避免了所有实例共享同一个时间戳的 bug。

核心源码（逐字来自文末完整源码）：

@dataclass
class CrawlItem:
    url: str
    title: str
    price: float
    crawled_at: str = field(default_factory=now_str)

可运行演示（补齐 Mock 数据与 print 反馈）：

from dataclasses import dataclass, field
from datetime import datetime
from zoneinfo import ZoneInfo

TZ = ZoneInfo("Asia/Shanghai")

def now_str() -> str:
    return datetime.now(TZ).strftime("%Y-%m-%d %H:%M:%S")

@dataclass
class CrawlItem:
    url: str
    title: str
    price: float
    # default_factory 表示“每次新建对象时再取当前时间”。
    crawled_at: str = field(default_factory=now_str)

item = CrawlItem(url="https://mock.shop/product/1", title="机械键盘 Pro X", price=599.0)
print("爬到的商品:", item)
print("标题:", item.title)
print("价格:", item.price)
print("抓取时间:", item.crawled_at)

Step 2：用 BaseCrawler 抽象基类定义爬虫骨架

痛点与机制：

BaseCrawler 是模板方法模式：run 方法定义了爬虫的骨架（fetch→parse→save），parse 和 save 是抽象方法，子类必须实现。这样所有爬虫都共享同一套重试、UA 轮换、延迟逻辑，只需要关注"如何解析这个网站的 HTML"和"如何存储数据"。@abstractmethod 让 Python 在实例化时检查子类是否实现了所有抽象方法。

核心源码（逐字来自文末完整源码）：

class BaseCrawler(ABC):
    """爬虫基类：定义 fetch / parse / save 三段式接口"""

    def __init__(self, timeout: int = 10, max_retries: int = 3):
        self.timeout = timeout
        self.max_retries = max_retries
        self._session_cookies: dict[str, str] = {}

    def fetch(self, url: str) -> str:
        """带重试的 HTTP GET，返回响应文本"""
        headers = {
            "User-Agent": random.choice(UA_POOL),
            "Accept": "text/html,application/xhtml+xml,*/*",
            "Accept-Language": "zh-CN,zh;q=0.9",
        }
        req = urllib.request.Request(url, headers=headers)
        for attempt in range(1, self.max_retries + 1):
            try:
                with urllib.request.urlopen(req, timeout=self.timeout) as resp:
                    return resp.read().decode("utf-8", errors="replace")
            except urllib.error.HTTPError as e:
                if e.code in (429, 500, 502, 503) and attempt < self.max_retries:
                    wait = 2 ** attempt
                    print(f"  [retry {attempt}] HTTP {e.code}，等待 {wait}s …")
                    time.sleep(wait)
                else:
                    raise
            except urllib.error.URLError as e:
                if attempt < self.max_retries:
                    time.sleep(2 ** attempt)
                else:
                    raise
        return ""

    @abstractmethod
    def parse(self, html: str, url: str) -> Iterator[CrawlItem]:
        """解析 HTML，yield CrawlItem"""

    @abstractmethod
    def save(self, items: list[CrawlItem]) -> None:
        """持久化数据"""

    def run(self, urls: list[str]) -> list[CrawlItem]:
        """主流程：fetch → parse → save"""
        all_items: list[CrawlItem] = []
        for url in urls:
            print(f"[{now_str()}] 抓取: {url}")
            html = self.fetch(url)
            items = list(self.parse(html, url))
            all_items.extend(items)
            print(f"  解析到 {len(items)} 条数据")
        self.save(all_items)
        return all_items

可运行演示（补齐 Mock 数据与 print 反馈）：

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Iterator

@dataclass
class CrawlItem:
    url: str
    title: str
    price: float

class BaseCrawler(ABC):
    """爬虫基类：定义 fetch / parse / save 三段式接口"""
    def fetch(self, url: str) -> str:
        # 真实版本会发 HTTP；教学演示用 Mock HTML，避免访问外网。
        print("fetch: 下载页面", url)
        return "<h2>Python 入门课</h2><span class='price'>99.9</span>"

    @abstractmethod
    def parse(self, html: str, url: str) -> Iterator[CrawlItem]:
        """解析 HTML，yield CrawlItem"""

    @abstractmethod
    def save(self, items: list[CrawlItem]) -> None:
        """持久化数据"""

    def run(self, urls: list[str]) -> list[CrawlItem]:
        all_items: list[CrawlItem] = []
        for url in urls:
            html = self.fetch(url)
            items = list(self.parse(html, url))
            all_items.extend(items)
            print(f"parse: 解析到 {len(items)} 条")
        self.save(all_items)
        return all_items

class DemoCrawler(BaseCrawler):
    def parse(self, html: str, url: str) -> Iterator[CrawlItem]:
        yield CrawlItem(url=url, title="Python 入门课", price=99.9)

    def save(self, items: list[CrawlItem]) -> None:
        print("save: 保存", len(items), "条数据")

crawler = DemoCrawler()
result = crawler.run(["https://mock.shop/course/1"])
print("最终结果:", result)

Step 3：用 UA_POOL 随机轮换 User-Agent，规避反爬检测

痛点与机制：

UA_POOL 是 User-Agent 池，每次请求用 random.choice(UA_POOL) 随机选一个。反爬系统通过 User-Agent 识别爬虫——固定的 UA 很容易被封，随机轮换可以降低被封概率。ascii_table 是格式化输出工具，把列表数据渲染成对齐的 ASCII 表格，让终端输出更易读。

核心源码（逐字来自文末完整源码）：

TZ = ZoneInfo("Asia/Shanghai")

def now_str() -> str:
    return datetime.now(TZ).strftime("%Y-%m-%d %H:%M:%S")


def ascii_table(headers: list[str], rows: list[list[Any]], title: str = "") -> str:
    col_w = [len(h) for h in headers]
    for row in rows:
        for i, cell in enumerate(row):
            col_w[i] = max(col_w[i], len(str(cell)))
    sep = "+" + "+".join("-" * (w + 2) for w in col_w) + "+"
    fmt = "|" + "|".join(f" {{:<{w}}} " for w in col_w) + "|"
    lines = []
    if title:
        total = sum(col_w) + 3 * len(col_w) + 1
        lines += [sep, f"|{title.center(total - 2)}|"]
    lines += [sep, fmt.format(*headers), sep]
    for row in rows:
        lines.append(fmt.format(*[str(c) for c in row]))
    lines.append(sep)
    return "\n".join(lines)

# ── UA 池 ────────────────────────────────────────────────────
UA_POOL = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/124.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Safari/605.1.15",
    "Mozilla/5.0 (X11; Linux x86_64; rv:125.0) Gecko/20100101 Firefox/125.0",
]

可运行演示（补齐 Mock 数据与 print 反馈）：

import random
from typing import Any

UA_POOL = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/124.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Safari/605.1.15",
    "Mozilla/5.0 (X11; Linux x86_64; rv:125.0) Gecko/20100101 Firefox/125.0",
]

def ascii_table(headers: list[str], rows: list[list[Any]], title: str = "") -> str:
    widths = [len(h) for h in headers]
    for row in rows:
        for i, cell in enumerate(row):
            widths[i] = max(widths[i], len(str(cell)))
    sep = "+" + "+".join("-" * (w + 2) for w in widths) + "+"
    fmt = "|" + "|".join(f" {{:<{w}}} " for w in widths) + "|"
    lines = [sep]
    if title:
        lines.append("|" + title.center(len(sep) - 2) + "|")
        lines.append(sep)
    lines += [fmt.format(*headers), sep]
    for row in rows:
        lines.append(fmt.format(*[str(c) for c in row]))
    lines.append(sep)
    return "\n".join(lines)

print(ascii_table(["#", "User-Agent"], [[i + 1, ua[:42] + "..."] for i, ua in enumerate(UA_POOL)], "UA 池"))
for i in range(5):
    print(f"第 {i + 1} 次请求使用 UA:", random.choice(UA_POOL)[:55] + "...")

Step 4：用指数退避重试处理网络抖动，避免雪崩效应

痛点与机制：

BaseCrawler.fetch 内置指数退避重试：第 1 次失败等 2 秒，第 2 次失败等 4 秒，第 3 次失败等 8 秒。指数退避避免了"雪崩效应"——如果所有客户端在服务器故障后立刻同时重试，会让服务器更难恢复。只有 429（限流）、500/502/503（服务端错误）才重试，4xx 客户端错误（如 404）直接抛出，不浪费重试次数。

核心源码（逐字来自文末完整源码）：

class BaseCrawler(ABC):
    """爬虫基类：定义 fetch / parse / save 三段式接口"""

    def __init__(self, timeout: int = 10, max_retries: int = 3):
        self.timeout = timeout
        self.max_retries = max_retries
        self._session_cookies: dict[str, str] = {}

    def fetch(self, url: str) -> str:
        """带重试的 HTTP GET，返回响应文本"""
        headers = {
            "User-Agent": random.choice(UA_POOL),
            "Accept": "text/html,application/xhtml+xml,*/*",
            "Accept-Language": "zh-CN,zh;q=0.9",
        }
        req = urllib.request.Request(url, headers=headers)
        for attempt in range(1, self.max_retries + 1):
            try:
                with urllib.request.urlopen(req, timeout=self.timeout) as resp:
                    return resp.read().decode("utf-8", errors="replace")
            except urllib.error.HTTPError as e:
                if e.code in (429, 500, 502, 503) and attempt < self.max_retries:
                    wait = 2 ** attempt
                    print(f"  [retry {attempt}] HTTP {e.code}，等待 {wait}s …")
                    time.sleep(wait)
                else:
                    raise
            except urllib.error.URLError as e:
                if attempt < self.max_retries:
                    time.sleep(2 ** attempt)
                else:
                    raise
        return ""

    @abstractmethod
    def parse(self, html: str, url: str) -> Iterator[CrawlItem]:
        """解析 HTML，yield CrawlItem"""

    @abstractmethod
    def save(self, items: list[CrawlItem]) -> None:
        """持久化数据"""

    def run(self, urls: list[str]) -> list[CrawlItem]:
        """主流程：fetch → parse → save"""
        all_items: list[CrawlItem] = []
        for url in urls:
            print(f"[{now_str()}] 抓取: {url}")
            html = self.fetch(url)
            items = list(self.parse(html, url))
            all_items.extend(items)
            print(f"  解析到 {len(items)} 条数据")
        self.save(all_items)
        return all_items

可运行演示（补齐 Mock 数据与 print 反馈）：

class MockNetworkError(Exception):
    pass

class RetryDemo:
    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries
        self.calls = 0

    def fetch(self, url: str) -> str:
        # 前两次故意失败，第三次成功；不访问任何真实网络。
        for attempt in range(1, self.max_retries + 1):
            try:
                self.calls += 1
                if self.calls < 3:
                    raise MockNetworkError("模拟网络抖动")
                return "<html>成功页面</html>"
            except MockNetworkError as exc:
                if attempt < self.max_retries:
                    wait = 2 ** attempt
                    print(f"第 {attempt} 次失败: {exc}，按指数退避等待 {wait}s（演示中不真睡）")
                else:
                    raise
        return ""

crawler = RetryDemo(max_retries=3)
html = crawler.fetch("https://mock.shop/product/1")
print("最终抓取成功:", html)
print("总尝试次数:", crawler.calls)

Step 5：用 MockShopCrawler 实现零外网依赖的完整爬虫

痛点与机制：

MockShopCrawler 用内置的 MOCK_PRODUCTS 字典模拟真实商品页面，不需要访问外网。parse 方法用正则从 HTML 里提取商品信息，save 方法把数据存入 SQLite 内存库。这个设计让爬虫逻辑可以在没有网络的环境里完整测试——Mock 数据和真实数据的接口完全相同，切换只需要改 URL。

核心源码（逐字来自文末完整源码）：

class MockShopCrawler(BaseCrawler):
    """模拟电商爬虫（使用 Mock 数据，无需网络）"""

    def __init__(self):
        super().__init__()
        self._db = sqlite3.connect(":memory:")
        self._db.execute(
            "CREATE TABLE products(url TEXT, title TEXT, price REAL, crawled_at TEXT)"
        )

    def fetch(self, url: str) -> str:  # type: ignore[override]
        """覆盖 fetch，返回 Mock HTML"""
        idx = int(url.split("/")[-1])
        p = MOCK_PRODUCTS[idx % len(MOCK_PRODUCTS)]
        return MOCK_HTML_TEMPLATE.format(**p, url=url)

    def parse(self, html: str, url: str) -> Iterator[CrawlItem]:
        import re
        title_m = re.search(r"<h2>(.*?)</h2>", html)
        price_m = re.search(r'class="price">([\d.]+)', html)
        if title_m and price_m:
            yield CrawlItem(
                url=url,
                title=title_m.group(1),
                price=float(price_m.group(1)),
            )

    def save(self, items: list[CrawlItem]) -> None:
        self._db.executemany(
            "INSERT INTO products VALUES(?,?,?,?)",
            [(i.url, i.title, i.price, i.crawled_at) for i in items],
        )
        self._db.commit()

    def report(self) -> None:
        rows = self._db.execute(
            "SELECT title, price, crawled_at FROM products ORDER BY price DESC"
        ).fetchall()
        print("\n" + ascii_table(
            ["商品名称", "价格(¥)", "抓取时间"],
            [[r[0], f"{r[1]:.2f}", r[2]] for r in rows],
            title="Mock 电商爬取结果",
        ))

可运行演示（补齐 Mock 数据与 print 反馈）：

import re
import sqlite3
from dataclasses import dataclass
from typing import Iterator

MOCK_HTML_TEMPLATE = """
<div class="product">
  <h2>{title}</h2>
  <span class="price">{price}</span>
  <a href="{url}">详情</a>
</div>
"""
MOCK_PRODUCTS = [
    {"title": "机械键盘 Pro X", "price": 599.0},
    {"title": "4K 显示器 27寸", "price": 2199.0},
]

@dataclass
class CrawlItem:
    url: str
    title: str
    price: float

class MockShopCrawler:
    def __init__(self):
        self.db = sqlite3.connect(":memory:")
        self.db.execute("CREATE TABLE products(url TEXT, title TEXT, price REAL)")

    def fetch(self, url: str) -> str:
        idx = int(url.rsplit("/", 1)[-1]) % len(MOCK_PRODUCTS)
        return MOCK_HTML_TEMPLATE.format(**MOCK_PRODUCTS[idx], url=url)

    def parse(self, html: str, url: str) -> Iterator[CrawlItem]:
        title = re.search(r"<h2>(.*?)</h2>", html)
        price = re.search(r'class="price">([\d.]+)', html)
        if title and price:
            yield CrawlItem(url, title.group(1), float(price.group(1)))

    def save(self, items: list[CrawlItem]) -> None:
        self.db.executemany("INSERT INTO products VALUES(?,?,?)", [(i.url, i.title, i.price) for i in items])
        self.db.commit()

crawler = MockShopCrawler()
items = []
for url in ["https://mock.shop/product/1", "https://mock.shop/product/2"]:
    items.extend(crawler.parse(crawler.fetch(url), url))
crawler.save(items)
for row in crawler.db.execute("SELECT title, price FROM products ORDER BY price DESC"):
    print(f"商品: {row[0]} | 价格: ¥{row[1]:.2f}")

Step 6：用 mode_ua 演示 User-Agent 池轮换效果

痛点与机制：

mode_ua 模拟 5 次请求，每次随机选一个 UA，用 ascii_table 格式化输出。这个演示让读者直观看到 UA 轮换的效果——每次请求的 User-Agent 都不同，反爬系统很难通过 UA 特征识别爬虫。真实项目里 UA 池应该包含几十个甚至上百个常见浏览器的 UA 字符串。

核心源码（逐字来自文末完整源码）：

def mode_ua(_: argparse.Namespace) -> None:
    print(f"=== User-Agent 池轮换演示  [{now_str()}] ===\n")
    rows = [[i + 1, ua[:70] + "…"] for i, ua in enumerate(UA_POOL)]
    print(ascii_table(["#", "User-Agent"], rows, title="UA 池"))
    print("\n随机抽取 5 次：")
    for i in range(5):
        print(f"  [{i+1}] {random.choice(UA_POOL)[:60]}…")

可运行演示（补齐 Mock 数据与 print 反馈）：

import argparse
import random
from datetime import datetime
from zoneinfo import ZoneInfo

TZ = ZoneInfo("Asia/Shanghai")
UA_POOL = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/124.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Safari/605.1.15",
    "Mozilla/5.0 (X11; Linux x86_64; rv:125.0) Gecko/20100101 Firefox/125.0",
]

def now_str() -> str:
    return datetime.now(TZ).strftime("%Y-%m-%d %H:%M:%S")

def mode_ua(_: argparse.Namespace) -> None:
    print(f"=== User-Agent 池轮换演示  [{now_str()}] ===")
    print("UA 池大小:", len(UA_POOL))
    for i in range(5):
        print(f"[{i + 1}]", random.choice(UA_POOL)[:60] + "...")

mode_ua(argparse.Namespace())

Step 7：用 mode_retry 演示指数退避重试的等待时间序列

痛点与机制：

mode_retry 打印指数退避的等待时间序列：1次失败→2s，2次失败→4s，3次失败→8s。这个演示让读者看到"为什么叫指数退避"——等待时间以 2 的幂次增长，而不是固定间隔。加上随机抖动（jitter）可以进一步避免多个客户端同时重试：wait = 2^attempt + random.uniform(0, 1)。

核心源码（逐字来自文末完整源码）：

def mode_retry(_: argparse.Namespace) -> None:
    print(f"=== 重试策略演示  [{now_str()}] ===\n")
    rows = [
        ["1", "正常响应", "直接返回"],
        ["2", "429/5xx", "等待 2s 重试"],
        ["3", "429/5xx", "等待 4s 重试"],
        ["4", "仍失败", "抛出异常"],
    ]
    print(ascii_table(["尝试次数", "响应状态", "行为"], rows, title="指数退避重试策略"))

可运行演示（补齐 Mock 数据与 print 反馈）：

import argparse

def mode_retry(args: argparse.Namespace) -> None:
    print("=== 重试策略演示 ===")
    print("目标 URL:", args.url)
    for attempt in range(1, args.max_retries + 1):
        wait = 2 ** attempt
        print(f"第 {attempt} 次失败后，下一次等待 {wait}s")
    print("真实爬虫里只对 429/500/502/503 或网络抖动重试。")

mode_retry(argparse.Namespace(url="https://mock.shop/product/1", max_retries=3))

Step 8：用 main 做 demo/ua/retry 三种模式的 CLI 总入口

痛点与机制：

main 用 argparse 做 CLI 入口，三种模式对应三个学习层次：ua 看反爬策略，retry 看重试机制，demo 看完整爬取流程。mode_demo/mode_ua/mode_retry 都接受 argparse.Namespace 参数，统一接口让 main 可以用字典 {mode: handler} 分发，不需要写 if/elif 链。

核心源码（逐字来自文末完整源码）：

def main() -> None:
    p = argparse.ArgumentParser(description="OOP 爬虫基类演示")
    p.add_argument("--mode", choices=["demo", "ua", "retry"], default="demo")
    args = p.parse_args()
    {"demo": mode_demo, "ua": mode_ua, "retry": mode_retry}[args.mode](args)

可运行演示（补齐 Mock 数据与 print 反馈）：

import argparse
import sys

def mode_crawl(args: argparse.Namespace) -> None:
    print(f"crawl 模式: 抓取 {args.pages} 个 Mock 商品页，写入 SQLite 内存库")

def mode_ua(args: argparse.Namespace) -> None:
    print("ua 模式: 展示 User-Agent 池和随机轮换效果")

def mode_retry(args: argparse.Namespace) -> None:
    print(f"retry 模式: 演示 max_retries={args.max_retries} 的指数退避策略")

def main() -> None:
    parser = argparse.ArgumentParser(description="零依赖爬虫框架演示")
    parser.add_argument("--mode", choices=["crawl", "ua", "retry"], default="crawl")
    parser.add_argument("--pages", type=int, default=3)
    parser.add_argument("--max-retries", type=int, default=3)
    args = parser.parse_args()
    if args.mode == "ua":
        mode_ua(args)
    elif args.mode == "retry":
        mode_retry(args)
    else:
        mode_crawl(args)

for mode in ["crawl", "ua", "retry"]:
    sys.argv = ["prog", "--mode", mode, "--pages", "2", "--max-retries", "3"]
    print(f">>> python3 25-python-spider-requests.py --mode {mode}")
    main()

极客实战：完整源码与运行

现在，把上面的积木拼起来，将以下完整代码放进你的编辑器，运行它。先看整体闭环，再回头逐段改参数，你会更容易建立工程直觉。

#!/usr/bin/env python3
"""
24_spider_base.py — OOP 爬虫基类演示（零外部依赖）

用法：
  python3 24_spider_base.py --mode demo    # 演示完整爬取流程
  python3 24_spider_base.py --mode ua      # 展示 UA 池轮换
  python3 24_spider_base.py --mode retry   # 演示重试逻辑
"""

import argparse
import json
import random
import sqlite3
import time
import urllib.error
import urllib.request
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Iterator
from zoneinfo import ZoneInfo

# ── NexDo Time ──────────────────────────────────────────────
TZ = ZoneInfo("Asia/Shanghai")

def now_str() -> str:
    return datetime.now(TZ).strftime("%Y-%m-%d %H:%M:%S")

# ── ASCII 表格 ───────────────────────────────────────────────
def ascii_table(headers: list[str], rows: list[list[Any]], title: str = "") -> str:
    col_w = [len(h) for h in headers]
    for row in rows:
        for i, cell in enumerate(row):
            col_w[i] = max(col_w[i], len(str(cell)))
    sep = "+" + "+".join("-" * (w + 2) for w in col_w) + "+"
    fmt = "|" + "|".join(f" {{:<{w}}} " for w in col_w) + "|"
    lines = []
    if title:
        total = sum(col_w) + 3 * len(col_w) + 1
        lines += [sep, f"|{title.center(total - 2)}|"]
    lines += [sep, fmt.format(*headers), sep]
    for row in rows:
        lines.append(fmt.format(*[str(c) for c in row]))
    lines.append(sep)
    return "\n".join(lines)

# ── UA 池 ────────────────────────────────────────────────────
UA_POOL = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/124.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Safari/605.1.15",
    "Mozilla/5.0 (X11; Linux x86_64; rv:125.0) Gecko/20100101 Firefox/125.0",
]

# ── 数据模型 ─────────────────────────────────────────────────
@dataclass
class CrawlItem:
    url: str
    title: str
    price: float
    crawled_at: str = field(default_factory=now_str)

# ── 抽象基类 ─────────────────────────────────────────────────
class BaseCrawler(ABC):
    """爬虫基类：定义 fetch / parse / save 三段式接口"""

    def __init__(self, timeout: int = 10, max_retries: int = 3):
        self.timeout = timeout
        self.max_retries = max_retries
        self._session_cookies: dict[str, str] = {}

    def fetch(self, url: str) -> str:
        """带重试的 HTTP GET，返回响应文本"""
        headers = {
            "User-Agent": random.choice(UA_POOL),
            "Accept": "text/html,application/xhtml+xml,*/*",
            "Accept-Language": "zh-CN,zh;q=0.9",
        }
        req = urllib.request.Request(url, headers=headers)
        for attempt in range(1, self.max_retries + 1):
            try:
                with urllib.request.urlopen(req, timeout=self.timeout) as resp:
                    return resp.read().decode("utf-8", errors="replace")
            except urllib.error.HTTPError as e:
                if e.code in (429, 500, 502, 503) and attempt < self.max_retries:
                    wait = 2 ** attempt
                    print(f"  [retry {attempt}] HTTP {e.code}，等待 {wait}s …")
                    time.sleep(wait)
                else:
                    raise
            except urllib.error.URLError as e:
                if attempt < self.max_retries:
                    time.sleep(2 ** attempt)
                else:
                    raise
        return ""

    @abstractmethod
    def parse(self, html: str, url: str) -> Iterator[CrawlItem]:
        """解析 HTML，yield CrawlItem"""

    @abstractmethod
    def save(self, items: list[CrawlItem]) -> None:
        """持久化数据"""

    def run(self, urls: list[str]) -> list[CrawlItem]:
        """主流程：fetch → parse → save"""
        all_items: list[CrawlItem] = []
        for url in urls:
            print(f"[{now_str()}] 抓取: {url}")
            html = self.fetch(url)
            items = list(self.parse(html, url))
            all_items.extend(items)
            print(f"  解析到 {len(items)} 条数据")
        self.save(all_items)
        return all_items

# ── Mock 爬虫实现 ────────────────────────────────────────────
MOCK_HTML_TEMPLATE = """
<html><body>
  <div class="product">
    <h2>{title}</h2>
    <span class="price">{price}</span>
    <a href="{url}">详情</a>
  </div>
</body></html>
"""

MOCK_PRODUCTS = [
    {"title": "机械键盘 Pro X", "price": 599.0},
    {"title": "4K 显示器 27寸", "price": 2199.0},
    {"title": "人体工学椅 E3", "price": 1899.0},
    {"title": "无线降噪耳机 Q45", "price": 899.0},
    {"title": "便携 SSD 1TB",   "price": 459.0},
]

class MockShopCrawler(BaseCrawler):
    """模拟电商爬虫（使用 Mock 数据，无需网络）"""

    def __init__(self):
        super().__init__()
        self._db = sqlite3.connect(":memory:")
        self._db.execute(
            "CREATE TABLE products(url TEXT, title TEXT, price REAL, crawled_at TEXT)"
        )

    def fetch(self, url: str) -> str:  # type: ignore[override]
        """覆盖 fetch，返回 Mock HTML"""
        idx = int(url.split("/")[-1])
        p = MOCK_PRODUCTS[idx % len(MOCK_PRODUCTS)]
        return MOCK_HTML_TEMPLATE.format(**p, url=url)

    def parse(self, html: str, url: str) -> Iterator[CrawlItem]:
        import re
        title_m = re.search(r"<h2>(.*?)</h2>", html)
        price_m = re.search(r'class="price">([\d.]+)', html)
        if title_m and price_m:
            yield CrawlItem(
                url=url,
                title=title_m.group(1),
                price=float(price_m.group(1)),
            )

    def save(self, items: list[CrawlItem]) -> None:
        self._db.executemany(
            "INSERT INTO products VALUES(?,?,?,?)",
            [(i.url, i.title, i.price, i.crawled_at) for i in items],
        )
        self._db.commit()

    def report(self) -> None:
        rows = self._db.execute(
            "SELECT title, price, crawled_at FROM products ORDER BY price DESC"
        ).fetchall()
        print("\n" + ascii_table(
            ["商品名称", "价格(¥)", "抓取时间"],
            [[r[0], f"{r[1]:.2f}", r[2]] for r in rows],
            title="Mock 电商爬取结果",
        ))

# ── CLI ──────────────────────────────────────────────────────
def mode_demo(_: argparse.Namespace) -> None:
    print(f"=== OOP 爬虫演示  [{now_str()}] ===\n")
    crawler = MockShopCrawler()
    urls = [f"https://mock-shop.example.com/product/{i}" for i in range(5)]
    crawler.run(urls)
    crawler.report()

def mode_ua(_: argparse.Namespace) -> None:
    print(f"=== User-Agent 池轮换演示  [{now_str()}] ===\n")
    rows = [[i + 1, ua[:70] + "…"] for i, ua in enumerate(UA_POOL)]
    print(ascii_table(["#", "User-Agent"], rows, title="UA 池"))
    print("\n随机抽取 5 次：")
    for i in range(5):
        print(f"  [{i+1}] {random.choice(UA_POOL)[:60]}…")

def mode_retry(_: argparse.Namespace) -> None:
    print(f"=== 重试策略演示  [{now_str()}] ===\n")
    rows = [
        ["1", "正常响应", "直接返回"],
        ["2", "429/5xx", "等待 2s 重试"],
        ["3", "429/5xx", "等待 4s 重试"],
        ["4", "仍失败", "抛出异常"],
    ]
    print(ascii_table(["尝试次数", "响应状态", "行为"], rows, title="指数退避重试策略"))

def main() -> None:
    p = argparse.ArgumentParser(description="OOP 爬虫基类演示")
    p.add_argument("--mode", choices=["demo", "ua", "retry"], default="demo")
    args = p.parse_args()
    {"demo": mode_demo, "ua": mode_ua, "retry": mode_retry}[args.mode](args)

if __name__ == "__main__":
    main()

$ python3 25-python-spider-requests.py --mode ua

=== User-Agent 池轮换演示  [2026-04-17 23:55:12] ===
+---+----------------------------------------------------------------------+
| # | User-Agent                                                           |
+---+----------------------------------------------------------------------+
| 1 | Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Safari/605.1.15      |
| 2 | Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/124.0.0.0 ...       |
| 3 | Mozilla/5.0 (X11; Linux x86_64; rv:125.0) Gecko/20100101 Firefox/... |
+---+----------------------------------------------------------------------+

$ python3 25-python-spider-requests.py --mode retry

=== 重试策略演示  [2026-04-17 23:55:12] ===
+-------+----------+----------+
| 重试次 | 等待(秒) | 累计(秒) |
+-------+----------+----------+
| 1     | 2        | 2        |
| 2     | 4        | 6        |
| 3     | 8        | 14       |
+-------+----------+----------+

$ python3 25-python-spider-requests.py --mode demo

=== 商品爬虫演示  [2026-04-17 23:55:12] ===
[2026-04-17 23:55:12] 抓取: https://mock.shop/page/1
  解析到 3 条数据
[2026-04-17 23:55:12] 抓取: https://mock.shop/page/2
  解析到 3 条数据
+----+------------------+-------+---------------------+
| id | title            | price | crawled_at          |
+----+------------------+-------+---------------------+
| 1  | Python 编程实战   | 89.9  | 2026-04-17 23:55:12 |
| 2  | 数据结构与算法    | 79.0  | 2026-04-17 23:55:12 |
...
+----+------------------+-------+---------------------+

小结

概念	一句话记忆
`@dataclass`	自动生成 `__init__`/`__repr__`，`field(default_factory=...)` 避免可变默认值
`BaseCrawler`	模板方法模式：骨架在基类，具体逻辑在子类
`UA_POOL`	User-Agent 池，每次请求随机选一个，降低被封概率
指数退避	失败后等待 `2^attempt` 秒，避免雪崩效应
`@abstractmethod`	强制子类实现，实例化时检查，不实现就报错
`MockShopCrawler`	Mock 数据模拟真实爬取，零外网依赖，适合测试
`ascii_table`	把列表数据渲染成对齐的 ASCII 表格

⏱ NexDo Time（5 分钟）

挑战：给 BaseCrawler 加一个 delay 参数，在每次请求之间随机等待 [delay, delay*2] 秒。

具体步骤：

在 BaseCrawler.__init__ 里加 delay: float = 1.0 参数
在 run 方法的 fetch 调用后加 time.sleep(random.uniform(self.delay, self.delay * 2))
在 MockShopCrawler 里传入 delay=0（测试时不等待）
验证：MockShopCrawler(delay=0.1) 爬取 3 个 URL，总耗时应该在 0.3-0.6 秒之间

Don’t wait for next time, do it in the next moment.