Python 基础体系 · 第 93/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

Python 网络采集：Requests、HTTPX、BeautifulSoup、限速和合规

网络采集不是“发一个 GET 请求，再用正则表达式找字符串”。一个可维护的采集程序至少要处理四类问题：

  1. HTTP 客户端：如何建立连接、复用连接、设置超时、处理 TLS 和响应流。
  2. 内容解析：如何从 HTML、JSON 或二进制响应中提取结构化数据。
  3. 并发与限速：如何提高吞吐量，同时避免给目标服务制造突发流量。
  4. 合规与可审计性：如何判断采集是否被允许，如何留下足够的请求、响应和决策记录。

本文以 Python 3.14 为范围，重点介绍 Requests、HTTPX 和 BeautifulSoup，并把它们放入一个完整的采集流程中。

一、先建立正确的模型：采集系统由哪些阶段组成

一次典型的 HTML 采集可以抽象为：

flowchart LR
    A[任务输入 URL] --> B[URL 校验与规范化]
    B --> C[合规检查]
    C --> D[限速器]
    D --> E[HTTP 客户端]
    E --> F[连接池/TLS/超时]
    F --> G[HTTP 响应]
    G --> H[状态码与内容类型检查]
    H --> I[BeautifulSoup 解析]
    I --> J[字段提取与校验]
    J --> K[去重与持久化]
    E --> L[日志与指标]
    C --> L
    H --> L
    J --> L

各组件的职责不能混淆：

  • Requests 或 HTTPX 负责把 HTTP 请求发送出去，并返回响应对象。
  • BeautifulSoup 负责解析已经获得的 HTML，不负责网络连接，也不负责执行 JavaScript。
  • 限速器 控制“什么时候可以发请求”，连接池控制“最多同时占用多少连接”，二者不是一回事。
  • 合规检查 决定“是否应该发请求”，不是请求失败后的异常处理。
  • 持久化和审计 负责记录采集结果以及为什么采集或跳过某个 URL。

如果把这些职责混在一个函数里，常见结果是：没有超时、没有重试边界、并发数不可控、解析异常被吞掉，最后无法判断数据错误来自网络、页面变化，还是程序本身。

二、Requests 和 HTTPX：都是 HTTP 客户端，但抽象重点不同

2.1 Requests：同步阻塞模型

Requests 提供同步 API。最简单的调用是：

import requests

response = requests.get(
    "https://example.com/",
    timeout=(3.05, 10),
)

response.raise_for_status()
print(response.status_code)
print(response.text[:100])

这里有三个重要步骤：

  1. requests.get() 发起同步请求。当前线程会等待 DNS、连接、响应头和响应体。
  2. timeout=(3.05, 10) 分别设置连接超时和读取超时。
  3. raise_for_status() 把 4xx、5xx 响应转成异常。

Requests 默认不会自动设置超时；如果不显式传入 timeout，请求可能长时间等待。Requests 的读取超时更准确地说是“等待底层连接继续收到数据的最长间隔”，不是整个下载过程的总墙上时钟限制。(requests.readthedocs.io)

因此，下面这种代码在生产采集中是不完整的：

response = requests.get(url)

它的问题不是“偶尔慢”，而是失败路径没有时间上限：

DNS 长时间阻塞
    ↓
TCP 连接等待
    ↓
服务器接受连接但不返回数据
    ↓
程序线程持续占用
    ↓
任务堆积，进程看似存活但没有进展

2.2 Session：连接池、Cookie 和默认配置的容器

如果要访问同一个站点的多个 URL，应使用 Session：

import requests

with requests.Session() as session:
    session.headers.update({
        "User-Agent": "wrblog-research-bot/1.0 (+contact@example.org)",
        "Accept": "text/html,application/xhtml+xml",
    })

    for url in [
        "https://example.com/",
        "https://example.com/about",
    ]:
        response = session.get(url, timeout=(3.05, 10))
        response.raise_for_status()
        print(response.url, len(response.content))

Session 会保存 Cookie、默认请求配置，并通过底层连接池复用连接。复用连接可以减少重复的 TCP/TLS 握手，但它并不意味着请求自动变成异步，也不意味着可以无限并发。(requests.readthedocs.io)

Requests 的连接池通过 HTTPAdapter 配置：

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

retry = Retry(
    total=3,
    connect=3,
    read=0,
    status=3,
    backoff_factor=0.5,
    status_forcelist=(429, 500, 502, 503, 504),
    allowed_methods=frozenset({"GET", "HEAD"}),
    respect_retry_after_header=True,
)

adapter = HTTPAdapter(
    pool_connections=10,
    pool_maxsize=10,
    max_retries=retry,
    pool_block=True,
)

session = requests.Session()
session.mount("https://", adapter)
session.mount("http://", adapter)

这里需要区分两个概念：

  • pool_maxsize=10 限制适配器保存或使用的连接规模；
  • Retry 决定某些网络失败或响应状态是否再次尝试；
  • pool_block=True 使连接不足时等待，而不是继续制造更多连接。

Requests 文档明确说明，HTTPAdapter 的默认重试主要针对连接建立阶段；一旦请求数据已经送达服务器，就不能简单假设重试是安全的。更细粒度的状态码和方法控制应通过 urllib3.util.Retry 配置。(requests.readthedocs.io)

2.3 HTTPX：同步和异步 API

HTTPX 同时支持同步客户端和异步客户端：

import httpx

with httpx.Client(
    timeout=httpx.Timeout(10.0, connect=5.0),
    limits=httpx.Limits(
        max_connections=10,
        max_keepalive_connections=5,
    ),
    follow_redirects=True,
) as client:
    response = client.get("https://example.com/")
    response.raise_for_status()
    print(response.http_version)

HTTPX 的 Client 对应 Requests 的 Session。顶层 httpx.get() 适合一次性实验；需要访问多个 URL 时，应复用同一个客户端，因为 Client 才能有效利用连接池。HTTPX 文档还说明，客户端支持 Cookie 持久化、HTTP/2 等能力。(python-httpx.org)

HTTPX 默认会对网络不活动设置超时，默认行为是网络连续约 5 秒没有进展时抛出超时异常。它把超时细分为四类：

  • connect：建立连接；
  • read：读取响应数据块；
  • write：发送请求数据块；
  • pool：等待从连接池获取连接。

这比只写一个总超时更接近实际故障模型。(python-httpx.org)

timeout = httpx.Timeout(
    connect=5.0,
    read=15.0,
    write=10.0,
    pool=3.0,
)

limits = httpx.Limits(
    max_connections=20,
    max_keepalive_connections=10,
    keepalive_expiry=30.0,
)

with httpx.Client(timeout=timeout, limits=limits) as client:
    response = client.get("https://example.com/")

2.4 何时选 Requests，何时选 HTTPX

可以按调用模型选择：

场景 适合的客户端
单线程、批处理脚本、同步代码 Requests
已有成熟 Requests 生态 Requests
asyncio、FastAPI、ASGI 应用 HTTPX AsyncClient
需要异步并发多个独立 URL HTTPX AsyncClient
需要 HTTP/2 或统一同步/异步接口 HTTPX

异步不是“更快的同步请求”。异步的核心是：当一个请求等待网络时，事件循环可以运行其他协程。HTTPX 的异步客户端使用 await client.get()，并且应在多个任务之间复用同一个 AsyncClient。在循环内部反复创建客户端，会破坏连接池的收益。(python-httpx.org)

三、响应处理：状态码成功，不等于业务成功

HTTP 响应至少包含：

状态码 + 响应头 + 响应体

例如：

  • 200 OK：服务器正常返回，但内容可能是错误页面、登录页或验证码页；
  • 301/302：发生重定向；
  • 403：服务器拒绝访问；
  • 404：资源不存在；
  • 429：请求过于频繁；
  • 500/502/503/504：服务器或网关异常。

所以采集程序应分层判断：

content_type = response.headers.get("content-type", "").lower()

if response.status_code == 429:
    # 读取 Retry-After，进入退避流程
    ...
elif response.status_code in {403, 404}:
    # 通常不应盲目重试
    ...
else:
    response.raise_for_status()

if "text/html" not in content_type:
    raise ValueError(f"unexpected content type: {content_type}")

raise_for_status() 只能说明 HTTP 状态码属于失败范围，不能验证页面是否真的是目标内容。业务层还应检查：

from bs4 import BeautifulSoup

soup = BeautifulSoup(response.content, "html.parser")

title = soup.title.get_text(" ", strip=True) if soup.title else ""
if "登录" in title or "验证码" in title:
    raise RuntimeError("received an authentication or challenge page")

一个常见反例是：

data = soup.select_one(".price").get_text(strip=True)

当页面结构变化或响应变成登录页时，select_one() 可能返回 None，随后触发：

AttributeError: 'NoneType' object has no attribute 'get_text'

更稳妥的做法是把“字段缺失”作为可诊断状态，而不是简单吞掉：

price_node = soup.select_one(".price")

if price_node is None:
    raise ValueError({
        "reason": "required_field_missing",
        "selector": ".price",
        "url": str(response.url),
        "title": title,
    })

price = price_node.get_text(" ", strip=True)

四、BeautifulSoup：解析 HTML，不执行网页应用

4.1 解析器和输入类型

BeautifulSoup 的职责是把 HTML 或 XML 文本解析成树。它可以使用不同解析器：

from bs4 import BeautifulSoup

html = """
<html>
  <head><title>示例页面</title></head>
  <body>
    <article class="post">
      <h1>网络采集</h1>
      <p>正文。</p>
      <a href="/next">下一页</a>
    </article>
  </body>
</html>
"""

soup = BeautifulSoup(html, "html.parser")

html.parser 是 Python 标准库提供的 HTML 解析器，适合基础场景。页面严重损坏、需要更强容错或有性能要求时，可以选择其他已安装解析器，但解析结果可能因解析器不同而不同。

4.2 查找元素：find、find_all 和 CSS 选择器

BeautifulSoup 支持传统查找接口：

article = soup.find("article", class_="post")
heading = article.find("h1") if article else None

if heading:
    print(heading.get_text(" ", strip=True))

也支持 CSS 选择器：

heading = soup.select_one("article.post > h1")
links = soup.select("article.post a[href]")

for link in links:
    print(link.get_text(" ", strip=True), link["href"])

select() 使用 SoupSieve 执行 CSS 选择器；get_text() 用于提取节点及其后代节点中的文本。(beautiful-soup-4.readthedocs.io)

选择器的工程含义是：它是一个对页面结构的假设。例如：

article.post > h1

表示：

  1. 存在 article 元素；
  2. 它的 class 包含 post；
  3. h1 是直接子元素。

如果网页改成：

<div class="post">
  <header><h1>网络采集</h1></header>
</div>

原选择器就会失效。可以放宽为：

soup.select_one("article.post h1, div.post h1")

但选择器越宽，误匹配的可能性越大。因此实际项目中应同时做：

  • 必填字段检查；
  • 页面版本或模板识别；
  • 样本页面回归测试；
  • 结果字段类型校验。

4.3 相对 URL 必须规范化

页面中的链接通常是相对路径：

<a href="/news/1">详情</a>
<a href="../archive">归档</a>

不能直接把它们当完整 URL 使用：

from urllib.parse import urljoin

base_url = "https://example.com/news/list"
for link in soup.select("a[href]"):
    absolute_url = urljoin(base_url, link["href"])
    print(absolute_url)

urljoin() 会依据基准 URL 解析路径。采集程序还应检查：

from urllib.parse import urlparse

parsed = urlparse(absolute_url)

if parsed.scheme not in {"http", "https"}:
    continue

这样可以跳过 mailto:、javascript: 等不适合 HTTP 采集的链接。

五、一个完整的同步采集器

下面的示例把客户端、robots 检查、限速、重试、解析和审计组合起来。它只针对公开页面，并且把站点域名限制为 example.com，适合本地演示。

from __future__ import annotations

import json
import time
from dataclasses import dataclass
from urllib.parse import urlparse
from urllib.robotparser import RobotFileParser

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


@dataclass
class Article:
    url: str
    title: str
    text: str


class RateLimiter:
    """最小间隔限速器：相邻请求至少间隔 interval 秒。"""

    def __init__(self, interval: float) -> None:
        if interval < 0:
            raise ValueError("interval must be non-negative")
        self.interval = interval
        self.next_allowed = 0.0

    def wait(self) -> None:
        now = time.monotonic()
        delay = self.next_allowed - now

        if delay > 0:
            time.sleep(delay)

        # 使用单调时钟，避免系统时间回拨影响限速。
        self.next_allowed = max(self.next_allowed, time.monotonic()) + self.interval


def build_session() -> requests.Session:
    retry = Retry(
        total=3,
        connect=3,
        read=0,
        status=3,
        backoff_factor=0.5,
        status_forcelist=(429, 500, 502, 503, 504),
        allowed_methods=frozenset({"GET", "HEAD"}),
        respect_retry_after_header=True,
    )

    adapter = HTTPAdapter(
        max_retries=retry,
        pool_connections=4,
        pool_maxsize=4,
        pool_block=True,
    )

    session = requests.Session()
    session.mount("https://", adapter)
    session.headers.update({
        "User-Agent": "wrblog-example-collector/1.0 (+contact@example.org)",
        "Accept": "text/html,application/xhtml+xml",
    })
    return session


def load_robots(base_url: str) -> RobotFileParser:
    parsed = urlparse(base_url)
    robots_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt"

    parser = RobotFileParser(robots_url)
    parser.read()
    return parser


def fetch_article(
    session: requests.Session,
    limiter: RateLimiter,
    robots: RobotFileParser,
    url: str,
) -> Article:
    parsed = urlparse(url)

    if parsed.scheme not in {"http", "https"}:
        raise ValueError(f"unsupported scheme: {parsed.scheme}")

    if parsed.netloc != "example.com":
        raise ValueError(f"unexpected host: {parsed.netloc}")

    user_agent = session.headers["User-Agent"]
    if not robots.can_fetch(user_agent, url):
        raise PermissionError(f"blocked by robots.txt: {url}")

    limiter.wait()

    started = time.monotonic()
    response = session.get(
        url,
        timeout=(3.05, 15),
        allow_redirects=True,
    )
    elapsed = time.monotonic() - started

    content_type = response.headers.get("content-type", "").lower()

    audit = {
        "url": url,
        "final_url": response.url,
        "status_code": response.status_code,
        "content_type": content_type,
        "elapsed_seconds": round(elapsed, 3),
        "content_length": len(response.content),
    }
    print(json.dumps(audit, ensure_ascii=False))

    response.raise_for_status()

    if "text/html" not in content_type:
        raise ValueError(f"expected HTML, got {content_type}")

    soup = BeautifulSoup(response.content, "html.parser")

    title_node = soup.select_one("title")
    article_node = soup.select_one("article")

    if title_node is None or article_node is None:
        raise ValueError("page template does not contain required nodes")

    title = title_node.get_text(" ", strip=True)
    text = article_node.get_text("\n", strip=True)

    if not text:
        raise ValueError("article text is empty")

    return Article(
        url=str(response.url),
        title=title,
        text=text,
    )


def main() -> None:
    urls = ["https://example.com/"]

    with build_session() as session:
        robots = load_robots(urls[0])
        limiter = RateLimiter(interval=1.5)

        for url in urls:
            try:
                article = fetch_article(session, limiter, robots, url)
            except requests.Timeout as exc:
                print(f"timeout: {url}: {exc}")
            except requests.RequestException as exc:
                print(f"network error: {url}: {exc}")
            except (PermissionError, ValueError) as exc:
                print(f"collector rejected: {url}: {exc}")
            else:
                print(article.title)
                print(article.text[:200])


if __name__ == "__main__":
    main()

安装依赖：

python -m pip install requests beautifulsoup4
python collector.py

这个程序的关键路径如下：

  1. build_session() 创建可复用的同步客户端。
  2. load_robots() 读取站点的 robots.txt。
  3. can_fetch() 判断当前 User-Agent 是否允许访问目标 URL。
  4. RateLimiter.wait() 在发送请求前控制节奏。
  5. session.get() 设置连接和读取超时。
  6. raise_for_status() 处理 HTTP 层失败。
  7. BeautifulSoup 解析 HTML。
  8. 选择器提取字段，并对必填字段进行验证。
  9. 输出状态码、最终 URL、响应类型、耗时和字节数。

urllib.robotparser.RobotFileParser 可以解析 robots.txt，并通过 can_fetch(useragent, url) 判断某个 User-Agent 是否被规则允许抓取。(docs.python.org)

但要注意：robots.txt 是自动化访问规则的一部分，不是法律授权书，也不是绕过合同、登录控制或访问权限的依据。如果站点条款禁止自动化访问，即使 robots.txt 没有禁止，也不能据此认为采集必然合规。

六、限速：限制的是请求调度，不只是 sleep()

6.1 固定间隔模型

假设限制器要求相邻请求至少间隔 Δ \Delta 秒。第 ii 个请求的发送时刻为 tit_i，则必须满足：

ti−ti−1≥Δt_i - t_{i-1} \geq \Delta

如果代码是：

for url in urls:
    fetch(url)
    time.sleep(1)

它实际保证的是“请求完成后等待 1 秒”，因此请求开始间隔是：

ti−ti−1=di−1+1t_i - t_{i-1} = d_{i-1} + 1

其中 di−1d_{i-1} 是上一个请求耗时。

这通常比要求更慢，而且在多线程中完全不成立：每个线程都各自 sleep(1)，多个线程仍然可能同时发起请求。

前面的 RateLimiter 采用下一个允许时刻：

ai=max⁡(ai−1,now)+Δa_i = \max(a_{i-1}, now) + \Delta

这样即使某次请求耗时很长，下一次请求也不会因为多个线程或任务同时醒来而产生突发。

6.2 令牌桶模型

如果希望允许短时突发，但限制长期平均速率，可以使用令牌桶。

设：

  • 桶容量为 BB；
  • 令牌产生速率为 rr 个/秒；
  • 每个请求消耗一个令牌；
  • 当前令牌数为 TT。

经过 ee 秒后：

T′=min⁡(B,T+r×e)T' = \min(B, T + r \times e)

若 T′≥1T' \geq 1，请求立即发送并令：

T′′=T′−1T'' = T' - 1

否则等待：

w=1−T′rw = \frac{1 - T'}{r}

令牌桶的直觉是：平时积累令牌，允许有限突发；持续请求时，平均速度不能超过 rr。

无论固定间隔还是令牌桶，都应按目标主机分别限速。例如：

example.com       1.0 请求/秒
api.example.net   5.0 请求/秒

全局只有一个限速器会让一个站点的限制错误地影响另一个站点；每个 URL 一个限速器又可能失去主机级控制。

6.3 并发数和限速必须同时约束

假设有 NN 个并发任务，平均每个请求耗时 LL 秒，理论吞吐量上限近似为：

λ≤NL\lambda \leq \frac{N}{L}

但如果站点允许的速率是 RR 请求/秒，则实际吞吐量还必须满足：

λ≤R\lambda \leq R

所以有效上限是：

λ≤min⁡(NL,R)\lambda \leq \min\left(\frac{N}{L}, R\right)

增加并发数只能提高“等待网络期间的利用率”，不能合理地突破站点允许的访问速率。连接池中的 max_connections 解决资源占用问题，速率限制器解决时间分布问题。

七、重试：只对可重试、可重复的操作重试

7.1 失败不等于可以重试

重试可能产生重复副作用。考虑：

POST /orders

客户端发送请求后发生读取超时。客户端不知道服务器是否已经创建订单：

客户端发送 POST
    ↓
服务器创建订单成功
    ↓
响应在网络中丢失
    ↓
客户端看到 ReadTimeout
    ↓
客户端重试 POST
    ↓
可能创建第二个订单

这就是“结果未知”状态，而不是“操作一定失败”。

对于 GET、HEAD，通常更容易重试，因为它们通常是幂等的。幂等表示：同一个请求执行一次或多次，最终资源状态相同。它不表示每次响应都完全相同，也不表示服务器不会记录访问日志。

重试决策至少需要考虑：

情况 通常处理
DNS、连接建立失败 可有限重试
连接超时 可有限重试
429 遵循 Retry-After
502/503/504 可退避重试
400 参数错误 不重试
401/403 权限问题 不靠重试解决
404 资源不存在 通常不重试
POST 结果未知 需要幂等键或业务确认

7.2 指数退避和随机抖动

简单指数退避可以写为：

dk=min⁡(dmax⁡,d0×2k)d_k = \min(d_{\max}, d_0 \times 2^k)

其中：

  • kk 是第 kk 次重试；
  • d0d_0 是初始等待时间；
  • dmax⁡d_{\max} 是最大等待时间。

如果所有客户端同时失败，它们会同时按照相同时间表重试，形成“惊群”。因此通常加入随机抖动：

dk=random(0,min⁡(dmax⁡,d0×2k))d_k = \text{random}(0, \min(d_{\max}, d_0 \times 2^k))

重试不是越多越可靠。重试次数过多会延长任务失败时间，还可能加重目标服务故障。采集器应记录：

原始 URL
第几次尝试
异常类型
状态码
Retry-After
实际等待时间
最终结果

八、流式响应：大文件不能无条件读入内存

response.content 或 response.text 会让程序把响应体作为整体处理。对于大文件或长响应，应使用流式读取。

Requests：

import requests

url = "https://example.com/large.bin"

with requests.get(
    url,
    stream=True,
    timeout=(5, 30),
) as response:
    response.raise_for_status()

    with open("large.bin", "wb") as output:
        for chunk in response.iter_content(chunk_size=64 * 1024):
            if chunk:
                output.write(chunk)

HTTPX：

import httpx

with httpx.stream(
    "GET",
    "https://example.com/large.bin",
    timeout=httpx.Timeout(30.0, connect=5.0),
) as response:
    response.raise_for_status()

    with open("large.bin", "wb") as output:
        for chunk in response.iter_bytes(64 * 1024):
            output.write(chunk)

HTTPX 文档建议对大下载使用流式响应，避免一次性将整个响应体加载到内存。流式响应必须正确关闭，否则连接可能无法归还连接池。(python-httpx.org)

下载时还应设置资源边界：

MAX_BYTES = 50 * 1024 * 1024
written = 0

for chunk in response.iter_bytes(64 * 1024):
    written += len(chunk)
    if written > MAX_BYTES:
        raise ValueError("response exceeds configured size limit")
    output.write(chunk)

Content-Length 不能完全作为安全依据，因为响应可能使用分块传输编码，也可能声明错误。真正写入的字节数仍然需要检查。

九、TLS、证书和 URL 安全边界

HTTPS 提供的是加密传输和服务器身份验证，但前提是客户端验证证书链。Requests 默认验证 HTTPS 证书；HTTPX 也提供证书验证配置。

不要为了绕过证书错误写：

requests.get(url, verify=False)

或者：

httpx.get(url, verify=False)

这会使中间人攻击更容易成功。正确做法是：

  • 修复系统 CA 证书；
  • 指定可信 CA bundle；
  • 对内部服务使用明确配置的企业 CA；
  • 在测试环境使用受控的本地证书，而不是在生产关闭验证。

还应限制目标 URL 的协议和主机。尤其在“采集用户提供的 URL”或“服务端代替用户抓取 URL”时，需要防范 SSRF：

from urllib.parse import urlparse

def validate_url(url: str) -> None:
    parsed = urlparse(url)

    if parsed.scheme not in {"http", "https"}:
        raise ValueError("only HTTP and HTTPS are allowed")

    if not parsed.hostname:
        raise ValueError("missing hostname")

    if parsed.username or parsed.password:
        raise ValueError("credentials in URL are not allowed")

仅检查字符串前缀是不够的：

if url.startswith("https://example.com"):
    ...

以下 URL 可能并非目标主机：

https://example.com.attacker.test/
https://example.com@attacker.test/

生产系统还需要解析 DNS 后限制内网地址、回环地址、链路本地地址和云元数据地址。这个问题属于服务端 URL 获取安全，而不只是普通爬虫逻辑。

十、HTTPX 异步采集：并发、信号量和客户端生命周期

下面是一个受并发数限制的异步示例：

import asyncio
import httpx
from bs4 import BeautifulSoup

URLS = [
    "https://example.com/",
    "https://example.com/about",
]

semaphore = asyncio.Semaphore(5)


async def fetch_one(
    client: httpx.AsyncClient,
    url: str,
) -> dict[str, str]:
    async with semaphore:
        response = await client.get(url)
        response.raise_for_status()

        soup = BeautifulSoup(response.content, "html.parser")
        title = soup.title.get_text(" ", strip=True) if soup.title else ""

        return {
            "url": str(response.url),
            "title": title,
        }


async def main() -> None:
    timeout = httpx.Timeout(
        connect=5.0,
        read=15.0,
        write=10.0,
        pool=3.0,
    )
    limits = httpx.Limits(
        max_connections=5,
        max_keepalive_connections=5,
    )

    async with httpx.AsyncClient(
        timeout=timeout,
        limits=limits,
        follow_redirects=True,
        headers={
            "User-Agent": "wrblog-async-collector/1.0 (+contact@example.org)",
        },
    ) as client:
        results = await asyncio.gather(
            *(fetch_one(client, url) for url in URLS),
            return_exceptions=True,
        )

        for result in results:
            if isinstance(result, Exception):
                print("failed:", repr(result))
            else:
                print(result)


if __name__ == "__main__":
    asyncio.run(main())

这里有三个独立约束：

  1. Semaphore(5) 限制同时运行的采集任务；
  2. max_connections=5 限制 HTTPX 连接池中的连接规模；
  3. timeout.pool=3.0 限制任务等待连接池的时间。

asyncio.gather(..., return_exceptions=True) 使一个 URL 失败时，其他 URL 的结果仍然可以返回。若不使用它，某个异常可能直接使 gather() 抛出，任务聚合策略就需要另外设计。

但这个示例还没有实现每个主机的速率限制。并发数为 5 不等于每秒最多 5 个请求；如果请求耗时很短，可能产生远高于目标站点允许值的速率。真实程序需要把异步限速器放在 client.get() 之前。

十一、在 FastAPI 或 ASGI 中使用异步客户端

ASGI 应用不是传统的“调用函数并返回一个完整结果”。ASGI 应用接收 scope、receive 和 send 三个异步对象，通过事件与服务器通信；HTTP 请求体也可以通过事件分块接收。(asgi.readthedocs.io)

因此，在 FastAPI 这类 ASGI 框架中，不应在每个请求处理函数里反复创建外部 HTTP 客户端：

# 不推荐：每次 API 请求都创建、销毁一个客户端
@app.get("/proxy")
async def proxy():
    async with httpx.AsyncClient() as client:
        return await client.get("https://example.com/")

更合理的是让客户端具有应用级生命周期。伪代码如下：

from contextlib import asynccontextmanager

import httpx
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(
        timeout=httpx.Timeout(15.0, connect=5.0),
        limits=httpx.Limits(max_connections=20),
    )
    try:
        yield
    finally:
        await app.state.http.aclose()


app = FastAPI(lifespan=lifespan)


@app.get("/fetch")
async def fetch():
    response = await app.state.http.get("https://example.com/")
    response.raise_for_status()
    return {
        "status": response.status_code,
        "length": len(response.content),
    }

客户端在启动阶段创建，在关闭阶段释放；请求处理函数只复用它。这样连接池的生命周期与应用生命周期一致。HTTPX 官方文档也建议复用同一个异步客户端，避免在热循环中不断创建新客户端。(python-httpx.org)

如果要把外部响应直接转发给下游客户端，应使用流式模式，并确保下游断开时外部响应最终被关闭。HTTPX 文档特别提醒，手动 send(..., stream=True) 时，关闭响应的责任由应用承担。(python-httpx.org)

十二、动态网页：BeautifulSoup 看不到 JavaScript 生成的内容

BeautifulSoup 解析的是已经下载到本地的 HTML：

服务器返回 HTML
    ↓
HTTP 客户端获取字节
    ↓
BeautifulSoup 构建 DOM 树
    ↓
选择器查找节点

它不会：

  • 执行 JavaScript；
  • 点击按钮；
  • 等待前端异步请求；
  • 模拟浏览器渲染；
  • 自动加载滚动后才出现的内容。

如果浏览器开发者工具中看到：

<div id="app"></div>

而页面显示了完整列表，列表可能来自后续 API 请求。此时应优先确认：

  1. 页面源代码是否已经包含数据；
  2. 浏览器 Network 面板中是否存在公开的 JSON API；
  3. API 是否有明确的使用条件、认证要求或速率限制；
  4. 是否可以使用官方 API，而不是模拟浏览器行为。

只有在确有授权和必要性时，才考虑浏览器自动化。浏览器自动化的成本更高，故障也更复杂：浏览器版本、脚本执行、资源加载、验证码和会话状态都会进入故障路径。

十三、合规：允许访问、技术可行和可以使用不是一回事

合规判断至少要分成三个层次。

13.1 访问授权

需要确认：

  • 目标资源是否公开；
  • 是否需要登录、绕过访问控制或使用个人账号；
  • 站点条款是否允许自动化访问；
  • 是否存在官方 API 或数据下载渠道；
  • robots.txt 是否禁止当前 User-Agent 访问路径。

其中任何一项不清楚，都不应简单地把“能请求成功”当成“可以采集”。

13.2 数据使用边界

即使页面公开，数据仍可能包含：

  • 个人信息；
  • 版权内容；
  • 商业敏感信息；
  • 用户生成内容；
  • 受地域、行业或合同限制的数据。

采集范围应遵循最小化原则：

不需要的字段不采集
不需要的页面不访问
不需要的历史版本不长期保存
不需要的个人标识不写入日志

日志尤其容易泄露数据。不要把完整 Cookie、Authorization、身份证号、手机号或响应正文直接写入普通日志。

13.3 负载和行为边界

以下行为会显著增加风险：

  • 高并发请求；
  • 无限制翻页；
  • 反复下载相同资源；
  • 不遵循 Retry-After；
  • 伪造或频繁更换身份标识；
  • 绕过验证码、登录限制或 IP 封禁；
  • 对非公开接口进行猜测式探测。

合规采集器的 User-Agent 应清晰、稳定，并提供可联系的信息：

wrblog-research-bot/1.0 (+mailto:contact@example.org)

User-Agent 不是授权凭证，但它有助于站点识别流量来源，也便于出现问题时沟通。

十四、诊断：根据失败位置定位问题

14.1 连接阶段失败

典型异常：

requests.exceptions.ConnectionError
httpx.ConnectError
httpx.ConnectTimeout

排查顺序：

  1. DNS 是否能解析；
  2. TCP 端口是否可达；
  3. 代理环境变量是否生效；
  4. TLS 握手是否失败；
  5. 是否所有 URL 都失败，还是只有某个主机失败。

如果所有目标都在同一时间失败，优先检查本地网络、代理、DNS 和证书，而不是立刻修改解析器。

14.2 读取阶段失败

典型异常：

requests.exceptions.ReadTimeout
httpx.ReadTimeout

这说明连接可能已经建立，但服务器没有在规定时间内继续提供数据。可能原因包括：

  • 服务器处理慢；
  • 响应体很大；
  • 服务端故意以极慢速度发送数据；
  • 代理或网关中断；
  • read 超时设置过于严格。

读取超时不是下载总时长限制，因此大文件下载还需要额外的总时长、字节数和磁盘空间控制。

14.3 解析阶段失败

典型表现：

AttributeError: 'NoneType' object has no attribute 'get_text'

不要只把选择器改得更宽。先保存或抽样记录：

最终 URL
状态码
Content-Type
页面标题
HTML 字节数
页面模板版本
缺失字段名称

很多“解析失败”实际是：

  • 被重定向到登录页面；
  • 收到验证码页面；
  • 服务器返回 JSON 错误；
  • 页面语言或区域发生变化；
  • A/B 测试切换了模板；
  • JavaScript 内容尚未出现在初始 HTML 中。

14.4 结果阶段失败

解析成功也不代表数据正确。例如价格字段可能包含：

¥1,299
暂无
登录后查看

应把文本清理和类型转换分开：

import re
from decimal import Decimal

raw = "¥1,299"
normalized = re.sub(r"[^\d.]", "", raw)

if not normalized:
    raise ValueError("price is not numeric")

price = Decimal(normalized)

不要静默地把“暂无”转换成 0。这会把缺失数据伪装成合法数据，后续统计结果会被污染。

十五、一个可审计的采集任务应记录什么

至少记录以下元数据：

{
  "requested_url": "https://example.com/",
  "final_url": "https://example.com/",
  "method": "GET",
  "user_agent": "wrblog-example-collector/1.0",
  "attempt": 1,
  "status_code": 200,
  "content_type": "text/html; charset=UTF-8",
  "bytes": 12543,
  "elapsed_seconds": 0.842,
  "robots_allowed": true,
  "parser": "html.parser",
  "result": "success"
}

审计记录的目标不是保存所有内容，而是回答：

  • 什么时候访问的？
  • 访问了哪个 URL，最终被重定向到哪里？
  • 用什么身份标识访问的？
  • 服务器返回了什么状态？
  • 程序为什么重试？
  • 为什么跳过或拒绝某个 URL？
  • 提取结果来自哪个页面版本？

对于可重复任务，还应保存：

  • 输入 URL 集合的版本；
  • 选择器配置版本；
  • 代码版本或提交 ID；
  • 运行批次 ID；
  • 输出文件校验和。

这样当页面结构变化时，可以区分“源站变化”和“采集程序回归”。

十六、常见误解和反例

误解一：加了 User-Agent 就可以爬

User-Agent 只是 HTTP 请求头中的身份声明。它不能替代授权，也不能绕过访问控制。

误解二：使用代理就解决了合规问题

代理改变的是网络出口，不改变访问权限、数据权利和目标站点条款。代理还会增加 TLS、来源识别和审计复杂度。

误解三：并发越高，采集越快

吞吐量受限于目标服务、连接池、网络延迟、限速规则和本地资源。并发过高可能导致：

429 增多
连接池耗尽
本地文件描述符耗尽
目标服务封禁
重试风暴

误解四：所有失败都应该重试三次

403、404、参数错误和解析模板错误通常不会因为重试而改善。盲目重试只会增加负载，并掩盖真正的问题。

误解五：HTTP 200 就说明拿到了正确页面

登录页、验证码页、错误提示页都可能返回 200。必须结合最终 URL、Content-Type、页面标志和字段完整性判断业务成功。

误解六：BeautifulSoup 能爬动态网页

BeautifulSoup 只能处理传给它的内容。它不会执行 JavaScript；动态数据应先确定数据来源和授权边界，再选择 API 客户端或浏览器自动化。

结语

一个可靠的 Python 网络采集器，核心不是某个神奇的选择器，而是清晰的边界：

  • Requests 适合成熟的同步脚本，Session 用于连接池和会话配置。
  • HTTPX 适合同时覆盖同步、异步、连接池、细粒度超时和 HTTP/2 的场景。
  • BeautifulSoup 负责解析 HTML，不负责请求和浏览器渲染。
  • 超时限制故障等待时间，连接池限制连接资源，限速器限制请求时间分布，并发信号量限制同时执行的任务数。
  • 重试必须结合异常类型、状态码和幂等性。
  • TLS 验证不能为了方便关闭。
  • robots.txt、站点条款、授权、数据用途和个人信息保护属于不同层面的约束，不能互相替代。
  • 审计日志应记录访问决策和结果元数据，而不是无差别保存敏感响应内容。

当这些机制被拆开并明确连接起来，采集程序才会从“能跑一次的脚本”变成可以诊断、限流、恢复和审查的工程系统。

系列导航与关联阅读

官方资料

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。