高级篇 / 第 19 章

第 19 章 · 爬虫开发与数据采集

爬虫第一课：合规、robots.txt 与伦理边界

爬虫（web crawler / spider / scraper）是指按照一定规则，自动访问网页并把内容抽取成结构化数据的自动化程序。它本身是一个中性技术：搜索引擎的索引、竞品价格监控、学术数据归档、舆情监测都依赖它。但「能爬」不等于「可以爬」，这是本章必须放在最前面讲的内容。

1.1 必须先看的三份文件

拿到一个目标站点，做任何事之前先访问这三个地址：

1.2 robots.txt 怎么读

# 示例：一个典型的 robots.txt
User-agent: *
Disallow: /admin/
Disallow: /api/private/
Disallow: /search?
Crawl-delay: 10

User-agent: BadBot
Disallow: /

User-agent: Googlebot
Disallow: /tmp/
Crawl-delay: 5

Sitemap: https://example.com/sitemap.xml

1.3 不要手写解析器，用 urllib.robotparser

Python 标准库自带 robots.txt 解析器，它实现了最长匹配规则，可以直接判断某个 URL 是否允许抓取：

robots_check.py · 抓取前的合规闸门

import urllib.robotparser as robotparser
import time
from urllib.parse import urlparse

RP_CACHE: dict[str, robotparser.RobotFileParser] = {}


def get_robot_parser(site_root: str) -> robotparser.RobotFileParser:
    """按站点缓存解析器结果，避免每次请求都去拉 robots.txt。"""
    if site_root in RP_CACHE:
        return RP_CACHE[site_root]

    rp = robotparser.RobotFileParser()
    rp.set_url(site_root.rstrip("/") + "/robots.txt")
    try:
        rp.read()          # 内部用 urllib.request，带超时与 User-Agent
    except Exception as e:
        # 取不到 robots.txt（404/超时）时，标准做法是「视为不限制」
        print(f"[robots] 读取失败 {e}，按不限制处理")
        rp.allow_all = True
    RP_CACHE[site_root] = rp
    return rp


def can_fetch(url: str, user_agent: str = "MyStudyBot") -> bool:
    p = urlparse(url)
    rp = get_robot_parser(f"{p.scheme}://{p.netloc}")
    return rp.can_fetch(user_agent, url)


# ---- 自检演示（离线，不发请求）----
rp = robotparser.RobotFileParser()
rp.parse([
    "User-agent: *",
    "Disallow: /private/",
    "Disallow: /bin",
    "Crawl-delay: 5",
].splitlines())

tests = [
    "https://x.com/public/a.html",   # 允许
    "https://x.com/private/a.html",  # 禁止（匹配 /private/）
    "https://x.com/binary",          # 禁止（匹配 /bin 前缀）
    "https://x.com/binocular",       # 允许（前缀 /bin 虽匹配开头，但路径不同）
]
for t in tests:
    print(t, "->", rp.can_fetch("MyStudyBot", t), rp.crawl_delay("MyStudyBot"))
# https://x.com/public/a.html -> True 5
# https://x.com/private/a.html -> False 5
# https://x.com/binary -> False 5
# https://x.com/binocular -> True 5

# 真实使用：
# if not can_fetch("https://example.com/news/1"):
#     raise PermissionError("robots.txt 不允许抓取该路径")
# time.sleep(rp.crawl_delay("*") or 1)

1.4 法律与伦理清单（工程化的底线）

1. 优先找官方 API。很多站点提供开放接口（开放平台、RSS、sitemap、数据开放门户），用 API 永远比解析 HTML 更稳定、更合规、更高效。本教程配套的天气/汇率/新闻 API 就是典型例子。
2. 绝不采集个人隐私数据。身份证号、手机号、住址、病历、财务信息属于个人敏感信息，采集、入库、上传都可能违法。爬到就脱敏或丢弃。
3. 不绕过登录墙与付费墙。用未授权账号抓取会员内容，可能构成不正当竞争甚至侵犯著作权。
4. 不破解验证码、不伪造身份高频轰炸。伪装 User-Agent 伪装成搜索引擎属于欺骗行为，且极易被封 IP。
5. 限速。单 IP 请求间隔建议 ≥1 秒，并发不要太高。一个 10 万页的任务跑 3 小时没问题，5 分钟打完是在给对方服务器制造事故。
6. 标注来源与时间。每条数据保留 source_url 与 fetched_at，这是数据可信度的基础，也是很多合规审计的硬要求。
7. 遵守 robots 与 User-Agent 标识。把爬虫名字和联系方式写进 User-Agent：MyStudyBot/1.0 (+https://example.com/bot)。

HTTP 基础与 requests：把网页抓下来

2.1 必须理解的 HTTP 语义

爬虫的每一步都是 HTTP 请求/响应。状态码决定你怎么走：

请求的关键请求头（Header）字段：

2.2 requests 核心用法

fetch_basic.py · 单次请求与状态码处理

import requests
from requests import Response

TIMEOUT = (5, 20)   # (连接超时, 读取超时) —— 一定要设！

HEADERS = {
    "User-Agent": ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
                   "(KHTML, like Gecko) Chrome/124.0 Safari/537.36"),
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
}

def fetch(url: str, **kw) -> Response:
    resp = requests.get(url, headers=HEADERS, timeout=TIMEOUT, **kw)
    if resp.status_code == 200:
        resp.encoding = resp.apparent_encoding   # 交给 chardet 猜编码
        return resp
    raise requests.HTTPError(f"{url} -> HTTP {resp.status_code}")

r = fetch("https://httpbin.org/get", params={"q": "python", "page": 2})
print(r.status_code, r.url)          # 200 https://httpbin.org/get?q=python&page=2
print(r.headers["Content-Type"])
print(r.json())

2.3 Session：连接复用与 Cookie 管理

Session（会话）是 requests 的核心优化点。它维护 CookieJar，自动为同一域名携带 Cookie，并通过 urllib3 连接池（connection pool）复用 TCP+TLS 连接。对于上百次请求的爬虫，这一项能省掉 50% 以上的时间。

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

def make_session(pool_size: int = 10) -> requests.Session:
    s = requests.Session()
    s.headers.update(HEADERS)

    # 自动重试：只对「连接层失败」和部分 5xx 重试，
    # 绝不要对 4xx 重试（重试 403 只会让对方更快封你）
    retries = Retry(
        total=3,
        connect=3,
        read=3,
        status=3,
        backoff_factor=0.5,           # 0.5s, 1s, 2s 指数退避
        status_forcelist=[500, 502, 503, 504],
        allowed_methods=frozenset(["GET", "HEAD"]),
    )
    adapter = HTTPAdapter(
        max_retries=retries,
        pool_connections=pool_size,   # 连接池大小
        pool_maxsize=pool_size,
    )
    s.mount("http://", adapter)
    s.mount("https://", adapter)
    return s

session = make_session()

# 第一次访问会种下 cookie，后续请求自动带上
session.get("https://httpbin.org/cookies/set?site=demo")
r = session.get("https://httpbin.org/cookies")
print(r.json())          # {"cookies": {"site": "demo"}}

print(session.cookies.get_dict())   # {'site': 'demo'}
session.close()

2.4 限速：做一个有礼貌的爬虫

import random
import time
from urllib.parse import urlparse

class RateLimiter:
    """按域名限速：对每个 host 独立控制最小请求间隔 + 抖动。"""

    def __init__(self, min_interval: float = 1.0, jitter: float = 0.4):
        self.min_interval = min_interval
        self.jitter = jitter
        self._last: dict[str, float] = {}

    def wait(self, url: str) -> None:
        host = urlparse(url).netloc
        gap = self.min_interval + random.uniform(0, self.jitter)
        elapsed = time.monotonic() - self._last.get(host, 0.0)
        if elapsed < gap:
            time.sleep(gap - elapsed)
        self._last[host] = time.monotonic()

limiter = RateLimiter(1.0)
for i in range(3):
    limiter.wait("https://example.com/a")
    print(f"第 {i+1} 次请求发出")

2.5 POST、表单与 JSON 接口

# 1) POST 表单（登录、搜索）
s = make_session()
data = {"username": "alice", "password": "pwd", "submit": "登录"}
r = s.post("https://httpbin.org/post", data=data, timeout=TIMEOUT)
print(r.json()["form"])          # 提交上去的键值

# 2) POST JSON（绝大多数现代接口的形式）
import json
r = s.post("https://httpbin.org/post",
           json={"name": "张三", "tags": ["a", "b"]}, timeout=TIMEOUT)
print(r.json()["json"])          # {"name": "\u5f20\u4e09", "tags": ["a", "b"]}

# 3) GET 带参数：params 交给 requests 编码，不要自己拼字符串
r = s.get("https://httpbin.org/get", params={"kw": "机器学习", "pn": 3}, timeout=TIMEOUT)
print(r.url)     # 参数被正确 percent-encode，中文也不会炸

# 4) 上传文件
with open("report.csv", "rb") as f:
    r = s.post("https://httpbin.org/post", files={"file": f}, timeout=TIMEOUT)
print(r.status_code)

2.6 代理与错误处理骨架

import logging, random, time
import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("crawler")

def fetch_with_retry(session, url, tries=3, base=1.0):
    for attempt in range(1, tries + 1):
        try:
            r = session.get(url, timeout=(5, 20))
            if r.status_code in (429, 500, 502, 503, 504):
                wait = r.headers.get("Retry-After")
                delay = float(wait) if wait and wait.isdigit() else base * (2 ** (attempt - 1))
                log.warning("HTTP %s，第 %d 次，退避 %.1fs", r.status_code, attempt, delay)
                time.sleep(delay + random.uniform(0, 1))
                continue
            r.raise_for_status()
            return r
        except requests.RequestException as e:
            log.warning("第 %d 次失败：%s", attempt, e)
            time.sleep(base * (2 ** (attempt - 1)))
    log.error("彻底失败：%s", url)
    return None

HTML 结构与 BeautifulSoup：正则之外的正确姿势

3.1 为什么不要用正则解析 HTML

HTML 是有嵌套结构的文档，不是文本。用正则解析典型结果：

例外：从一段纯文本里抽电话号码/正则匹配 SKU、以及解析 JSON 字符串时，正则依然是对的工具。

3.2 安装与解析器选择

pip 安装

pip install requests beautifulsoup4 lxml parsel selenium
# lxml 是 C 扩展，性能比内置 html.parser 快 3~5 倍，强烈建议装
python -c "import lxml; print(lxml.__version__)"

3.3 BeautifulSoup 常用操作

parse_bs4.py · 在一段离线 HTML 上练习（无需联网）

from bs4 import BeautifulSoup

HTML = """
<html><body>
  <div class="book-list">
    <div class="item" data-id="101">
      <h3 class="title">Python 编程：从入门到实践（第 3 版）</h3>
      <span class="author">埃里克·马瑟斯</span>
      <span class="price">￥89.00</span>
      <a class="link" href="/book/101">详情</a>
    </div>
    <div class="item" data-id="102">
      <h3 class="title">流畅的 Python（第 2 版）</h3>
      <span class="author"> Luciano Ramalho</span>
      <span class="price">￥139.00</span>
      <a class="link" href="/book/102">详情</a>
    </div>
  </div>
</body></html>
"""

soup = BeautifulSoup(HTML, "lxml")

# 1) CSS 选择器：最直观
items = soup.select("div.book-list > div.item")
print(len(items))                       # 2

# 2) 按 class 找（注意：class 是多值列表，必须用 class_）
h3s = soup.find_all("h3", class_="title")
print(h3s[0].get_text(strip=True))      # Python 编程：从入门到实践（第 3 版）

# 3) 第一个匹配
first = soup.select_one("div.item[data-id='101']")
print(first["data-id"])                 # 101

# 4) 在子节点内继续找（避免选到别的区块）
title = first.select_one("h3.title").get_text(strip=True)
author = first.select_one("span.author").get_text(strip=True)
price = float(first.select_one("span.price").get_text(strip=True).replace("￥", ""))
href = first.select_one("a.link")["href"]
print(title, "|", author, "|", price, "|", href)

# 5) 遍历直接子节点
print([c.name for c in first.find_all(recursive=False)])
# ['h3', 'span', 'span', 'a']

# 6) 提取全部链接并转绝对地址
from urllib.parse import urljoin
BASE = "https://books.example.com"
for a in soup.select("a.link"):
    print(urljoin(BASE, a["href"]))     # https://books.example.com/book/101

3.4 文本清洗的四个常用套路

import re

RAW = "  ¥1,299.00  \n 促销价  \t"

# 1) 去首尾空白 + 合并内部空白
def clean(s: str) -> str:
    return re.sub(r"\s+", " ", s).strip()

print(clean("Python   编程\n\n入门"))     # Python 编程 入门

# 2) 抽取数字（处理千分位与货币符号）
num = re.search(r"[\d,]+(?:\.\d+)?", clean(RAW)).group().replace(",", "")
print(float(num))                        # 1299.0

# 3) 处理全角/不可见字符
text = "Ｐｙｔｈｏｎ\xa0入门\u200b"
text = text.translate(str.maketrans("ＰＹＴＨＯＮ", "PYTHON"))
print(clean(text.replace("\xa0", " ").replace("\u200b", "")))
# PYTHON 入门

# 4) 用正则把实体反转义（bs4 已自动处理 html.unescape，但要小心二次转义）
from bs4 import BeautifulSoup
s = BeautifulSoup("<b>A &amp; B</b>", "lxml")
print(s.get_text())        # <b>A &amp; B</b>  —— 注意这是「文本」，不是标签

3.5 选择器失效的排查手法

1. 先看 print(soup.prettify()[:3000])，确认解析器把页面结构还原对了（有没有把整页吞进 body、有没有一堆乱码）。
2. 用浏览器开发者工具（F12）确认节点是否在 HTML 里 —— 有些内容是 JS 动态插入的，原始 HTML 里根本没有。
3. 优先选稳定特征：data-id、itemprop、aria-label、语义标签（article/time）；避免用 .c3a2b1 这类自动生成的哈希 class，网站一改版全废。
4. 写一个「断言」确保解析到了预期数量：assert len(items) > 0。数量为 0 立刻报警，而不是静悄悄地写入空结果。
5. 把原始 HTML 存档。页面改版后，用存档 + 新 HTML 对比，能快速定位是选择器问题还是反爬问题。

XPath 与 parsel：结构化选择的高效方案

XPath（XML Path Language）是用于在 XML/HTML 树中定位节点的语言。HTML5 解析后的 DOM 本质上是 XML 树，因此 XPath 可以直接用于网页解析。它的表达力比 CSS 选择器强得多，尤其是按属性条件、按文本、按序号、按祖先/兄弟关系定位。

4.1 XPath 语法速查

4.2 用 parsel 跑起来（它同时支持 CSS 和 XPath）

parse_parsel.py · 离线练习

from parsel import Selector

HTML = """
<ul id="rank">
  <li class="row" data-no="1">
    <span class="num">1</span>
    <a href="/x/1" title="书籍A">书籍 A</a>
    <span class="author">作者甲</span>
    <span class="hot">  9,876 人评价  </span>
  </li>
  <li class="row" data-no="2">
    <span class="num">2</span>
    <a href="/x/2">书籍 B</a>
    <span class="author">作者乙</span>
    <span class="hot">8,765 人评价</span>
  </li>
</ul>
"""
sel = Selector(HTML)

# --- XPath 写法 ---
rows = sel.xpath("//ul[@id='rank']/li")
print(len(rows))                            # 2

for r in rows:
    no = r.xpath("./span[@class='num']/text()").get()
    href = r.xpath("./a/@href").get()
    title = r.xpath("./a/@title").get() or r.xpath("./a/text()").get()
    author = r.xpath("./span[@class='author']/text()").get()
    hot = r.xpath("./span[@class='hot']/text()").get().strip()
    print(no, title, author, href, hot)

# --- CSS 写法（parsel 的 select 等价于 cssselect）---
print(sel.css("#rank li a::attr(href)").getall())
print(sel.css("li.row span.author::text").getall())
print(sel.css("li[data-no='2'] a::text").get())

# --- getall() vs get() vs get(default) ---
print(sel.xpath("//span[@class='none']/text()").get())        # None
print(sel.xpath("//span[@class='none']/text()").get(""))      # ''
print(sel.xpath("//span[@class='none']/text()").getall())      # []

4.3 纯 lxml 的等价写法

from lxml import html, etree

doc = html.fromstring(HTML)

# 方式 A：xpath 表达式（lxml 原生）
rows = doc.xpath("//ul[@id='rank']/li")
print(doc.xpath("//a/@href"))

# 方式 B：CSS 选择器（需要 cssselect 包）
rows2 = doc.cssselect("ul#rank > li")
print([a.get("href") for a in doc.cssselect("li.row a")])

# 导出结构，便于调试
print(etree.tostring(doc, pretty_print=True, encoding="unicode")[:400])

4.4 三种方案怎么选

Selenium：浏览器自动化与 JS 渲染页面

很多站点的数据不在 HTML 里，而是由 JavaScript 执行后插入 DOM，requests 拿到的只是一堆 <script>。这时需要真正的浏览器环境，Selenium 就是驱动浏览器的标准方案。

5.1 原理与安装

Selenium通过 WebDriver 协议与浏览器驱动通信：你的 Python 代码 → WebDriver → chromedriver → 真实浏览器。每次 HTTP 请求都在本地完成，不依赖 Selenium Grid 也能跑。

# Selenium 4.6+ 自带 Selenium Manager，会自动下载匹配版本的驱动
pip install selenium

# 想手动控制浏览器版本，下载对应 chromedriver 放到 PATH：
# https://googlechromelabs.github.io/chrome-for-testing/

5.2 核心 API 最小可用集

selenium_demo.py · 建议在本地浏览器手动运行一次感受时序

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.keys import Keys

def build_driver(headless: bool = True) -> webdriver.Chrome:
    opt = Options()
    if headless:
        opt.add_argument("--headless=new")
    opt.add_argument("--window-size=1440,900")
    opt.add_argument("--disable-gpu")
    opt.add_argument("--no-sandbox")            # Linux 容器里常需要
    opt.add_argument("--disable-dev-shm-usage") # Docker 里 /dev/shm 只有 64M，不加会崩
    opt.add_argument("--lang=zh-CN")
    # 反检测的最基础一条：让 UA 与真实 Chrome 一致
    opt.add_argument(
        "--user-agent=Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
        "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36")
    d = webdriver.Chrome(options=opt)
    d.set_page_load_timeout(30)
    return d

driver = build_driver(headless=True)
try:
    driver.get("https://httpbin.org/forms")

    # 显式等待：直到元素出现才继续，比 time.sleep 稳
    WebDriverWait(driver, 10).until(
        EC.presence_of_element_located((By.NAME, "custname")))

    driver.find_element(By.NAME, "custname").send_keys("张三")
    driver.find_element(By.NAME, "custtel").send_keys("13800000000")
    driver.find_element(By.NAME, "custemail").send_keys("a@b.com")
    driver.find_element(By.CSS_SELECTOR, "input[type=submit]").click()

    WebDriverWait(driver, 10).until(
        EC.text_to_be_present_in_element((By.TAG_NAME, "body"), "We've got your order"))

    print(driver.find_element(By.TAG_NAME, "body").text[:200])
    print("当前 URL：", driver.current_url)
    driver.save_screenshot("order_ok.png")   # 留证据，便于排查
finally:
    driver.quit()

5.3 显式等待 vs 隐式等待

from selenium.webdriver.support.ui import WebDriverWait

wait = WebDriverWait(driver, 10, poll_frequency=0.5)

# 常见条件
wait.until(EC.visibility_of_element_located((By.CSS_SELECTOR, ".item")))
wait.until(EC.element_to_be_clickable((By.CSS_SELECTOR, ".next")))
wait.until(EC.presence_of_all_elements_located((By.CSS_SELECTOR, "li.item")))
wait.until(EC.staleness_of(first_element))          # 元素被刷新/替换
wait.until(lambda d: len(d.find_elements(By.CLASS_NAME, "row")) > 20)
wait.until(EC.alert_is_present())

# ⚠️ 隐式等待与显式等待不要混用：混用后等待时间会叠加，表现是莫名变慢
# driver.implicitly_wait(5)   ← 注释掉它

5.4 无头浏览器、指纹与常见检测点

5.5 Selenium 的性能现实

• 启动一次 Chrome 约 1~3 秒，每次点击/跳转都有 IPC 开销，比 requests 慢 1~2 个数量级。
• 只把 Selenium 用于「拿到初始数据/建立会话」，之后能换 requests 就换：打开页面一次 → 拿 Cookie → 关浏览器 → 后续用 requests 复用 Cookie 抓数据。这是业界最常用的混合策略。
• 需要更高并发时，优先考虑 第 15 章的异步思路或后文的 Playwright。Chrome 原生 DevTools 协议（Playwright/CDP）比 Selenium WebDriver 层更轻，通常快 2~3 倍。

# 混合策略：Selenium 拿会话 → requests 狂抓
from urllib.parse import urljoin

session = requests.Session()
session.headers.update(HEADERS)

driver = build_driver(headless=True)
try:
    driver.get("https://example.com/login")
    WebDriverWait(driver, 10).until(
        EC.presence_of_element_located((By.ID, "username")))
    driver.find_element(By.ID, "username").send_keys("alice")
    driver.find_element(By.ID, "password").send_keys("pwd")
    driver.find_element(By.CSS_SELECTOR, "button[type=submit]").click()
    WebDriverWait(driver, 15).until(
        EC.presence_of_element_located((By.CSS_SELECTOR, ".profile")))

    # 关键一步：把浏览器里的 Cookie 搬进 requests
    for c in driver.get_cookies():
        session.cookies.set(c["name"], c["value"], domain=c["domain"], path=c.get("path", "/"))
finally:
    driver.quit()

# 之后全部用轻量的 requests
r = session.get("https://example.com/api/feed?page=1", timeout=20)
print(r.json()["items"][:1])

分页遍历：URL 发现、去重与增量

6.1 四种分页形式与对应处理

6.2 三种翻页循环写法

import time

# ---- 写法 A：页码参数 ----
def crawl_by_page(session, base: str, max_pages: int = 20, pause: float = 1.2):
    for page in range(1, max_pages + 1):
        r = session.get(base, params={"page": page}, timeout=20)
        r.raise_for_status()
        rows = parse_items(r.text)
        if not rows:
            print(f"第 {page} 页无数据，停止")
            break
        yield from rows
        time.sleep(pause)

# ---- 写法 B：跟随「下一页」链接（最通用） ----
def crawl_follow_next(session, start_url: str, seen: set[str] | None = None):
    seen = seen if seen is not None else set()
    url, n = start_url, 0
    while url and url not in seen and n < 100:
        seen.add(url)
        n += 1
        r = session.get(url, timeout=20)
        r.raise_for_status()
        rows, next_url = parse_page_and_next(r.text, base=url)
        yield from rows
        url = next_url
        time.sleep(1.2)

# ---- 写法 C：游标分页（接口首选） ----
def crawl_cursor(session, api: str):
    cursor = None
    while True:
        params = {"limit": 100}
        if cursor:
            params["cursor"] = cursor
        r = session.get(api, params=params, timeout=20)
        data = r.json()
        yield from data["items"]
        cursor = data.get("next_cursor")
        if not cursor:
            break
        time.sleep(1.0)

6.3 URL 归一化与去重

真实爬虫里，同一个页面会以几十种形式出现在链接里：带不同追踪参数、大小写不同、末尾斜杠不同、片段（#section）不同。URL 归一化（canonicalization）是去重的第一步。

from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode, urlencode as _ue

TRACKING_PARAMS = {
    "utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content",
    "spm", "from", "ref", "referer", "share_token", "_t", "timestamp",
}

def canonicalize(url: str) -> str:
    """把 URL 归一化成唯一标识，用于去重。"""
    p = urlsplit(url.strip())

    scheme = p.scheme.lower() or "https"
    netloc = p.netloc.lower()
    if netloc.endswith(":443") and scheme == "https":
        netloc = netloc[:-4]
    elif netloc.endswith(":80") and scheme == "http":
        netloc = netloc[:-3]
    if netloc.startswith("www."):
        netloc = netloc[4:]

    path = p.path or "/"
    if len(path) > 1 and path.endswith("/"):
        path = path[:-1]        # /a/ 与 /a 视为同一资源

    query = [(k, v) for k, v in parse_qsl(p.query, keep_blank_values=True)
             if k.lower() not in TRACKING_PARAMS]
    query.sort()                # 参数顺序不影响语义
    return urlunsplit((scheme, netloc, path, urlencode(query), ""))  # 丢弃 fragment

samples = [
    "HTTP://WWW.Example.com/a/?utm_source=x&id=7#top",
    "https://example.com/a?id=7",
    "https://example.com/a?utm_campaign=y&id=7",
]
print({canonicalize(u) for u in samples})
# {'https://example.com/a?id=7'}  ← 三个 URL 归一化为同一个

6.4 增量抓取：只采新增与变更

全量重爬是反爬触发器，也是浪费。做增量的常用信号：

import hashlib

def content_fingerprint(text: str) -> str:
    """正文指纹：用于识别内容变更，忽略空白与无关噪声。"""
    import re
    norm = re.sub(r"\s+", "", text)
    norm = re.sub(r"(今天|昨天)\d{1,2}:\d{2}", "", norm)   # 去掉相对时间
    return hashlib.sha256(norm.encode("utf-8")).hexdigest()[:16]

def should_refetch(url: str, html: str, known: dict[str, str]) -> bool:
    fp = content_fingerprint(html)
    return known.get(url) != fp

# 条件请求示例
import email.utils
last_mod = "Wed, 21 Oct 2024 07:28:00 GMT"
r = session.get("https://example.com/a", headers={"If-Modified-Since": last_mod}, timeout=20)
if r.status_code == 304:
    print("未变化，跳过解析")

数据落地：存储、去重与断点续爬

爬虫最怕的不是抓不到，而是「跑了三小时最后崩溃，什么都没留下」。工程上必须做到边抓边存、随时可恢复。

7.1 用 SQLite 做任务与状态管理

crawler_store.py · 一个可复用的最小状态层

import sqlite3
from contextlib import contextmanager

SCHEMA = """
CREATE TABLE IF NOT EXISTS pages (
    url         TEXT PRIMARY KEY,
    url_hash    TEXT UNIQUE NOT NULL,   -- 归一化 URL 的哈希
    status      TEXT NOT NULL DEFAULT 'pending',  -- pending/done/failed/skipped
    tries       INTEGER NOT NULL DEFAULT 0,
    http_status INTEGER,
    content_fp  TEXT,
    fetched_at  TEXT,
    error       TEXT
);
CREATE INDEX IF NOT EXISTS idx_pages_status ON pages(status);
CREATE TABLE IF NOT EXISTS items (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    title      TEXT NOT NULL,
    author     TEXT,
    price      REAL,
    source_url TEXT NOT NULL,
    item_key   TEXT UNIQUE NOT NULL,   -- 业务唯一键，防重复
    fetched_at TEXT NOT NULL
);
"""

class Store:
    def __init__(self, path: str = "crawl.db"):
        self.conn = sqlite3.connect(path, timeout=30)
        self.conn.row_factory = sqlite3.Row
        self.conn.execute("PRAGMA journal_mode=WAL")   # 并发读写更友好
        self.conn.executescript(SCHEMA)
        self.conn.commit()

    @contextmanager
    def tx(self):
        try:
            yield self.conn
            self.conn.commit()
        except Exception:
            self.conn.rollback()
            raise

    # ---- 任务队列 ----
    def add_task(self, url: str, url_hash: str) -> bool:
        """入队；已存在返回 False。"""
        with self.tx() as c:
            cur = c.execute(
                "INSERT OR IGNORE INTO pages (url, url_hash) VALUES (?, ?)",
                (url, url_hash))
            return cur.rowcount > 0

    def next_batch(self, limit: int = 10) -> list[sqlite3.Row]:
        return self.conn.execute(
            "SELECT * FROM pages WHERE status='pending' ORDER BY url LIMIT ?",
            (limit,)).fetchall()

    def mark_done(self, url: str, http_status: int, fp: str):
        with self.tx() as c:
            c.execute("UPDATE pages SET status='done', tries=tries+1, "
                      "http_status=?, content_fp=?, fetched_at=datetime('now'), "
                      "error=NULL WHERE url=?", (http_status, fp, url))

    def mark_failed(self, url: str, err: str, retry: bool = True):
        with self.tx() as c:
            if retry:
                c.execute("UPDATE pages SET status='pending', tries=tries+1, "
                          "error=? WHERE url=? AND tries < 3", (err, url))
            else:
                c.execute("UPDATE pages SET status='failed', tries=tries+1, "
                          "error=? WHERE url=?", (err, url))

    # ---- 结果入库 ----
    def save_item(self, title, author, price, source_url, item_key) -> None:
        with self.tx() as c:
            c.execute(
                "INSERT OR REPLACE INTO items "
                "(title, author, price, source_url, item_key, fetched_at) "
                "VALUES (?,?,?,?,?,datetime('now'))",
                (title, author, price, source_url, item_key))

    def done_count(self) -> int:
        return self.conn.execute(
            "SELECT COUNT(*) FROM pages WHERE status='done'").fetchone()[0]


if __name__ == "__main__":
    s = Store()
    print("已入库条目：", s.conn.execute("SELECT COUNT(*) FROM items").fetchone()[0])

7.2 断点续爬的循环结构

import time, traceback

def run(store: Store, session, seeds: list[str], pause: float = 1.2):
    # 1) 种子入队（归一化后去重）
    for u in seeds:
        store.add_task(u, canonicalize(u))

    # 2) 主循环：每轮取一批，失败自动回到 pending（最多 3 次）
    while True:
        batch = store.next_batch(10)
        if not batch:
            break
        for row in batch:
            url = row["url"]
            try:
                r = fetch_with_retry(session, url)
                if r is None:
                    store.mark_failed(url, "重试耗尽")
                    continue
                if r.status_code == 404:
                    store.mark_failed(url, "404", retry=False)   # 永久失败
                    continue
                fp = content_fingerprint(r.text)
                if row["content_fp"] == fp:
                    store.mark_done(url, r.status_code, fp)      # 内容未变
                    continue
                items = parse_items(r.text, base=url)
                for it in items:
                    store.save_item(it["title"], it["author"], it["price"], url, it["key"])
                    # 发现的详情页继续入队
                    if it.get("detail_url"):
                        store.add_task(it["detail_url"], canonicalize(it["detail_url"]))
                store.mark_done(url, r.status_code, fp)
                print(f"OK {url} -> {len(items)} 条（累计 {store.done_count()}）")
            except Exception as e:
                traceback.print_exc()
                store.mark_failed(url, repr(e))
            finally:
                time.sleep(pause)      # 无论成功失败都限速

    print("全部完成")

7.3 原始数据存档：给未来的自己留后路

import gzip, hashlib, json, pathlib

ARCHIVE = pathlib.Path("raw_pages")

def archive_html(url: str, html: str) -> pathlib.Path:
    """按内容哈希存一份 gzip，文件名可反查来源。"""
    ARCHIVE.mkdir(exist_ok=True)
    key = hashlib.sha1(url.encode()).hexdigest()[:16]
    path = ARCHIVE / f"{key}.html.gz"
    with gzip.open(path, "wt", encoding="utf-8") as f:
        f.write(html)
    (ARCHIVE / f"{key}.meta.json").write_text(
        json.dumps({"url": url}, ensure_ascii=False), encoding="utf-8")
    return path

# 好处：网站半年后改版，你还能拿旧 HTML 重新解析，验证新解析器的正确性
p = archive_html("https://example.com/a", "<html>...")
print(p, p.stat().st_size)

并发、异步与分布式

8.1 并发模型选择

8.2 线程池版：最实用的并发写法

from concurrent.futures import ThreadPoolExecutor, as_completed

def crawl_concurrent(seed_urls: list[str], workers: int = 8, pause: float = 1.0):
    results, errors = [], []

    def task(url: str):
        time.sleep(pause)                       # 每个线程独立限速
        r = fetch_with_retry(session, url, tries=2)
        if r is None:
            raise RuntimeError("fetch failed")
        return url, len(parse_items(r.text, base=url))

    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(task, u): u for u in seed_urls}
        for fut in as_completed(futures):
            url = futures[fut]
            try:
                u, n = fut.result()
                results.append((u, n))
                print(f"[OK] {u} -> {n}")
            except Exception as e:
                errors.append((url, repr(e)))
                print(f"[FAIL] {url}: {e}")
    return results, errors

ok, bad = crawl_concurrent(
    [f"https://example.com/p/{i}" for i in range(20)], workers=6)
print(f"成功 {len(ok)}，失败 {len(bad)}")

8.3 异步版：httpx + asyncio

# pip install httpx
import asyncio, random, time
import httpx

HEADERS_ASYNC = {
    "User-Agent": "MyStudyBot/1.0 (+https://example.com/bot)",
    "Accept": "text/html",
}

async def fetch_one(client: httpx.AsyncClient, sem: asyncio.Semaphore,
                    url: str) -> tuple[str, int | None]:
    async with sem:                      # 限流：信号量控制并发数
        try:
            r = await client.get(url, timeout=20.0)
            if r.status_code != 200:
                return url, None
            return url, len(parse_items(r.text, base=url))
        except httpx.HTTPError as e:
            return url, None

async def main(urls: list[str], concurrency: int = 10):
    sem = asyncio.Semaphore(concurrency)
    limits = httpx.Limits(max_connections=concurrency,
                          max_keepalive_connections=concurrency)
    async with httpx.AsyncClient(headers=HEADERS_ASYNC,
                                 limits=limits,
                                 follow_redirects=True) as client:
        tasks = [fetch_one(client, sem, u) for u in urls]
        t0 = time.perf_counter()
        out = await asyncio.gather(*tasks)
        ok = [x for x in out if x[1] is not None]
        print(f"成功 {len(ok)}/{len(urls)}，耗时 {time.perf_counter()-t0:.2f}s")
        return ok

if __name__ == "__main__":
    asyncio.run(main([f"https://example.com/p/{i}" for i in range(50)]))

8.4 分布式：任务队列 + 多 worker

当任务量到达十万级，单机多线程已不够（连接数、内存、IP 都会成为瓶颈），这时需要把任务分片（sharding）给多个 worker。核心组件只有三个：

# 用 Redis List 做最简任务队列（pip install redis）
import redis, json, uuid

rds = redis.Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True)
QUEUE = "crawl:queue"

def push_url(url: str):
    rds.sadd("crawl:seen", url)          # Set 去重：原子且高效
    rds.rpush(QUEUE, json.dumps({"url": url, "job_id": str(uuid.uuid4())}))

def pop_url(timeout: int = 5):
    item = rds.blpop(QUEUE, timeout=timeout)   # 阻塞式出队，队列空时 CPU 空转 0
    if item is None:
        return None
    _, payload = item
    return json.loads(payload)

# Worker 端（可多开）
while True:
    job = pop_url()
    if not job:
        continue
    url = job["url"]
    if not can_fetch(url):
        continue
    resp = fetch_with_retry(session, url, tries=2)
    if resp is None:
        continue
    for it in parse_items(resp.text, base=url):
        store.save_item(it["title"], it["author"], it["price"], url, it["key"])
        if it.get("detail_url"):
            push_url(it["detail_url"])
    time.sleep(random.uniform(0.8, 1.5))

实战：把前面所有零件拼成一个可用爬虫

下面是一个完整的、可直接改造为生产版本的骨架。它包含：robots 检查、Session 复用、限速、归一化去重、线程池并发、SQLite 断点续爬、增量指纹、原始数据存档、优雅退出。

"""crawler.py —— 通用型合规爬虫骨架（Python 3.12）

用法：
    python crawler.py --seed https://example.com/list --db data.db --workers 8
设计原则：
    1. 合规优先：robots + 限速 + 可识别 UA
    2. 抓取与解析分离
    3. 状态入库，随时中断可恢复
    4. 无网络也能跑测试：把 fetch 换成读本地 HTML
"""
from __future__ import annotations

import argparse
import logging
import random
import signal
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, asdict
from urllib.parse import urljoin, urlparse

from parsel import Selector

log = logging.getLogger("crawler")

BOT_UA = ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
          "(KHTML, like Gecko) Chrome/124.0 Safari/537.36 "
          "MyStudyBot/1.0 (+https://example.com/bot)")

# ---------- 1. 配置 ----------
@dataclass
class Config:
    seeds: list[str]
    db: str = "data.db"
    workers: int = 8
    interval: float = 1.2
    max_pages: int = 500
    respect_robots: bool = True

# ---------- 2. 解析函数（页面改版时只改这里） ----------
def parse_items(html: str, base: str) -> list[dict]:
    """从列表页 HTML 抽取条目。真实项目请按目标站改选择器。"""
    sel = Selector(html)
    out = []
    for li in sel.css("div.book-item"):
        href = li.css("a.detail::attr(href)").get()
        title = (li.css("h3.title::text").get() or "").strip()
        price_txt = (li.css("span.price::text").get() or "0").strip()
        try:
            price = float("".join(ch for ch in price_txt if ch.isdigit() or ch == "."))
        except ValueError:
            price = 0.0
        if not href or not title:
            continue
        detail = urljoin(base, href)
        out.append({
            "title": title,
            "author": (li.css("span.author::text").get() or "").strip(),
            "price": price,
            "detail_url": detail,
            "key": canonicalize(detail),      # 业务唯一键
        })
    return out

# ---------- 3. 断点控制 ----------
STOP = False
def _handle(signum, frame):
    global STOP
    STOP = True
    log.warning("收到信号 %s，本轮结束后优雅退出", signum)

# ---------- 4. 主流程 ----------
def main(cfg: Config):
    signal.signal(signal.SIGINT, _handle)
    signal.signal(signal.SIGTERM, _handle)

    store = Store(cfg.db)
    session = make_session(pool_size=cfg.workers * 2)
    session.headers["User-Agent"] = BOT_UA
    limiter = RateLimiter(cfg.interval)

    for u in cfg.seeds:
        if store.add_task(u, canonicalize(u)):
            log.info("种子入队：%s", u)

    def handle_one(url: str) -> int:
        if STOP:
            return 0
        if cfg.respect_robots and not can_fetch(url):
            log.info("robots 禁止：%s", url)
            store.mark_failed(url, "robots-disallow", retry=False)
            return 0
        limiter.wait(url)
        row = store.conn.execute(
            "SELECT content_fp FROM pages WHERE url=?", (url,)).fetchone()
        resp = fetch_with_retry(session, url, tries=3, base=cfg.interval)
        if resp is None:
            store.mark_failed(url, "network")
            return 0
        fp = content_fingerprint(resp.text)
        if row and row["content_fp"] == fp:
            store.mark_done(url, resp.status_code, fp)
            return 0
        items = parse_items(resp.text, url)
        archive_html(url, resp.text)
        for it in items:
            store.save_item(it["title"], it["author"], it["price"], url, it["key"])
            if store.add_task(it["detail_url"], it["key"]):
                pass
        store.mark_done(url, resp.status_code, fp)
        return len(items)

    total = 0
    while not STOP:
        batch = store.next_batch(cfg.workers)
        if not batch:
            break
        with ThreadPoolExecutor(max_workers=cfg.workers) as pool:
            futs = [pool.submit(handle_one, r["url"]) for r in batch]
            for f in futs:
                try:
                    total += f.result()
                except Exception:
                    log.exception("任务异常")
        if total >= cfg.max_pages:
            log.info("达到 max_pages，停止")
            break

    log.info("本轮共解析 %d 条；已完成页面 %d", total, store.done_count())
    session.close()

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO,
                        format="%(asctime)s %(levelname)s %(message)s")
    ap = argparse.ArgumentParser()
    ap.add_argument("--seed", action="append", required=True, help="可重复传入")
    ap.add_argument("--db", default="data.db")
    ap.add_argument("--workers", type=int, default=8)
    ap.add_argument("--interval", type=float, default=1.2)
    ap.add_argument("--max-pages", type=int, default=500)
    a = ap.parse_args()
    main(Config(seeds=a.seed, db=a.db, workers=a.workers,
               interval=a.interval, max_pages=a.max_pages))

9.1 项目组织建议

进阶主题与常见问题排查

10.1 编码与中文乱码

import chardet, requests

r = requests.get(url, headers=HEADERS, timeout=20)
print("header 声明：", r.encoding)
print("探测结果：", chardet.detect(r.content[:4096]))
r.encoding = chardet.detect(r.content)["encoding"]   # 或 r.apparent_encoding
print(r.text[:100])

# 响应头里 charset 缺失或写错（常见于 GBK 老站）时，上面这行是必须的
# 极端情况：meta 标签在 body 中间、chardet 也猜错，可强制指定
r.encoding = "gb18030"    # 比 gbk 覆盖面更广

10.2 症状 → 原因速查表

10.3 抓包：用浏览器理解真实请求

前面反复强调「优先找接口」，怎么找？用浏览器开发者工具：

1. F12 → Network（网络） 面板，勾选 Preserve log，勾选 Fetch/XHR 过滤。
2. 在页面上做一次翻页/搜索，观察新出现的 XHR 请求。
3. 点开请求，记录：URL、Query、请求方法、请求头（Cookie/Authorization/X-Requested-With）、请求体、响应 JSON 结构、翻页参数名。
4. 把关键请求「Copy as cURL」，粘贴到终端跑一遍验证；再翻译成 Python。
5. 注意签名参数：有的接口带 sign/timestamp/token，这是接口本身的安全设计。正确做法是走官方 SDK 或申请 API Key，而不是去逆向签名算法。

# 从浏览器抄来的请求，翻译成 Python
import requests, json

API = "https://api.example.com/v1/items"
s = requests.Session()
s.headers.update({
    "User-Agent": BOT_UA,
    "Accept": "application/json, text/plain, */*",
    "Referer": "https://example.com/list",
    "Authorization": "Bearer <从官方渠道申请的 Token>",
})

r = s.get(API, params={"page": 1, "page_size": 50, "order": "desc"}, timeout=20)
r.raise_for_status()
data = r.json()
print(data.keys())                 # dict_keys(['code', 'data', 'total'])
print(data["total"], len(data["data"]["list"]))
for it in data["data"]["list"][:3]:
    print(it["id"], it["name"], it["price"])

10.4 需要知道的进阶库

本章小结

• 合规是前提：先看 robots.txt、服务条款、隐私政策；优先用官方 API；不碰个人隐私数据、不绕验证码、不伪造身份；限速并带可识别 UA。
• robots.txt 是前缀匹配，用标准库 urllib.robotparser 解析即可，注意 Crawl-delay。
• 抓取层：requests.Session 复用连接与 Cookie，务必设 timeout，用 HTTPAdapter+Retry 做指数退避，raise_for_status() 处理非 200。
• 解析层：用解析器不用正则。CSS 选择器（BeautifulSoup）可读性最好；XPath（parsel/lxml）表达力最强，正反两套方案互补使用；urljoin 把相对链接转绝对。
• 文本清洗是解析的一半工作量：全角字符、NBSP、零宽字符、千分位数字都要处理，清洗逻辑写成可测试的纯函数。
• Selenium 用于 JS 渲染页面与登录会话建立；务必用显式等待而非 time.sleep；拿到会话后切回 requests 是性能最佳实践。对抗反爬的正当手段是慢与稳，不是伪装。
• 翻页要优先找游标式接口，其次「下一页」链接，最后才猜页码；连续两页为空即终止。
• 去重先做 URL 归一化（去追踪参数、去片段、统一大小写与端口），再落唯一键；内容变更用指纹 + Last-Modified 做增量。
• 存储：用 SQLite 做任务状态表（pending/done/failed + 重试次数），业务表用业务唯一键防重；边抓边存，进程随时可中断恢复；原始 HTML 存档便于事后重解析。
• 并发选型：线程池最适合 requests I/O 场景；asyncio + httpx 追求吞吐；分布式靠队列分片，但必须有全局限速。
• 排查：症状→原因速查表 + 抓包定位接口，是爬虫调试的两把钥匙。

练习题

1. robots 检查器。写一个函数 check_all(urls, user_agent="MyBot")，批量判断一组 URL 是否允许抓取，输出「允许/禁止」清单。对 Disallow: /private 和 Disallow: /private/ 两种写法分别用测试 URL 验证差异。
   参考答案要点核心是 robotparser.RobotFileParser：rp.parse(lines) 支持离线喂内容，rp.can_fetch(ua, url) 返回布尔。多 URL 时按 host 分组缓存解析器，避免重复下载 robots.txt。注意 Disallow: /private 会连带禁止 /private.php（前缀匹配），而 /private/ 不会。
2. URL 归一化与去重。实现 canonicalize(url)：去掉 utm/spm/from 等追踪参数、统一协议与主机大小写、去 www、去末尾斜杠、丢弃 fragment、参数排序。给 10 个测试 URL，断言归一化后只剩 3 个不同值。
   参考答案要点用 urlsplit/urlunsplit + parse_qsl/urlencode；排序用 query.sort()；fragment 直接置空。写 pytest 参数化测试最省事：@pytest.mark.parametrize("raw,expected", [...])。
3. 解析器单元测试。准备一份包含 5 个商品条目的本地 HTML（可用 <li class="item"> 结构），写 parse_items(html, base) 并用 unittest 断言：条目数为 5、第一条标题为指定文本、价格能正确转成 float、相对链接被转成绝对 URL。
   参考答案要点用 parsel.Selector(html).css("li.item")；价格清洗写成 re.sub(r"[^\d.]", "", price_txt)；urljoin(BASE, href) 处理相对路径。测试文件放 tests/test_parsers.py，HTML 放 tests/fixtures/，不依赖网络。
4. 线程池并发抓取。用 ThreadPoolExecutor 并发请求 20 个 URL（可用 httpbin 之类的合法测试站点或本地 HTTP 服务），要求：共用一个 Session、每线程限速 ≥1 秒、逐个调用 fut.result() 收集异常、最后打印成功率与总耗时。对比串行版耗时。
   参考答案要点关键点：Session 必须在主线程创建后共享；as_completed 拿结果；time.sleep(pause) 放在任务函数内模拟限速；except Exception 收集到 errors 列表而不是让整个循环中断。注意并发数别超过 8。
5. 断点续爬存储层。实现 Store 类：add_task（去重入队）、next_batch(n)、mark_done、mark_failed(retry)（失败自动回 pending，最多 3 次）。写一段脚本：入队 5 个 URL、处理 2 个后「崩溃」退出，重新运行后能从断点继续且不重复处理。
   参考答案要点用 SQLite，url_hash TEXT UNIQUE 配合 INSERT OR IGNORE；next_batch 用 WHERE status='pending'；mark_failed 里用 WHERE tries < 3 决定回 pending 还是置 failed。可用 raise RuntimeError 模拟崩溃验证恢复能力。
6. 增量抓取。实现 content_fingerprint(html)（去空白后哈希）和判定逻辑：本地已有指纹则相同则跳过。构造两次「页面抓取」：第一次写入，第二次内容完全相同应跳过，第三次改了标题应更新。再用 If-Modified-Since 演示 304 跳过的分支。
   参考答案要点fingerprint = sha256(re.sub(r"\s+", "", text).encode()).hexdigest()[:16]；DB 存 url -> content_fp；相同则只更新 last_checked 不重解析。304 分支用 requests.get(headers={"If-Modified-Since": ...}) 判断 status_code == 304。
7. Selenium 显式等待改造。写一个脚本：打开一个需要 JS 渲染才能出现列表的页面（可用本地构造的 data: 测试页或公开测试站点），要求用 WebDriverWait + EC 等待列表出现后再抓取，禁止使用 time.sleep；抓完后截图存档并把 Cookie 迁移到 requests.Session。
   参考答案要点核心是 WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.CSS_SELECTOR, ".item")))；抓取可用 driver.page_source 交给 parsel 复用同一套解析函数；Cookie 迁移用 driver.get_cookies() 遍历 session.cookies.set(...)；finally: driver.quit() 必须写。
8. 异步采集与限流对比。用 asyncio+httpx 写异步抓取器，用 asyncio.Semaphore(5) 限制并发为 5，抓 30 个测试 URL，统计总耗时与请求速率（RPS）。与线程池版本对比，说明在 I/O 等待多时异步的优势来源。
   参考答案要点async with httpx.AsyncClient(limits=httpx.Limits(...))；async with sem 做并发闸门；asyncio.gather(*tasks) 汇总；用 time.perf_counter() 计时、len/耗时 算 RPS。结论要点：等待时间被摊薄，线程切换开销更小，但解析是 CPU 密集时应放回 run_in_executor。
9. （选做）小型分布式爬虫。用 Redis List 作队列，编写 producer（从种子页发现 URL 入队）与 worker（出队抓取解析、入库）两个脚本，用 Redis Set 做 URL 去重。启动 2 个 worker 进程共同消费 200 个任务，对比单进程耗时，并给出一个全局令牌桶限流的实现。
   参考答案要点队列 rpush/blpop，去重 sadd 返回值为 0 则说明已存在直接跳过；限流用 Redis INCR + EXPIRE 实现固定窗口计数，或 Lua 脚本实现令牌桶；入库仍复用第 5 题的 Store。务必说明：多 worker 不等于可以更快，要压低全局 RPS。