网络爬虫最佳实践：包含技巧和可运行代码的生命周期指南

如果您只能养成十个习惯，请务必做到以下几点：编写数据契约、优先使用 API 和 JSON 而非 HTML、正确遵守 robots.txt 规则、标识您的爬虫、控制每个主机的访问频率、遵守 Retry-After 规则、设置超时、在保存前验证记录、对空页面或内容稀疏的页面发出警报，以及将检查点与它们所描述的数据一起提交。

这十项网络爬虫最佳实践中的每一项都附有相关说明章节的链接。

1. 在编写任何代码之前，先制定一份数据契约，其中应明确字段、类型、唯一键和数据新鲜度。
2. 优先使用官方 API 或页面自身的 JSON，而非解析 HTML。
3. 应按照 RFC 9309 的定义来应用 robots.txt，而非依赖于您所使用的库的实现方式。
4. 使用真实的 User-Agent 和联系地址来标识您的爬虫。
5. 对每个主机分别进行速率控制，从低速开始，并根据服务器情况进行调整。
6. 遵守 Retry-After 规则，并采用带抖动（jitter）的指数级退避策略。
7. 为每个请求设置连接和读取超时。
8. 在保存记录前进行验证，并保留每次拒绝的原因。
9. 不仅在发生崩溃时，还应在遇到空页面或内容稀疏的页面时发出警报。
10. 将检查点与其所描述的数据一同提交，确保崩溃时页面绝不会处于半保存状态。

从数据出发进行逆向规划。先列出所需的字段、类型、时效性和数据量，然后选择最经济且可靠的来源。按成本对网页抓取技术进行排序：优先使用官方 API，其次是页面已加载的 JSON 数据，然后是用于发现的站点地图，接着是静态 HTML，而无头浏览器仅作为最后手段。 在编写代码前先制定预算。

如果你还在犹豫是否要进行数据抓取，请先从编写抓取程序前需要解答的问题入手。

数据契约是一张简短的表格，明确列出爬虫必须交付的内容。以下是一个图书价格监控工具的数据契约示例：

监控器每天需要最新数据，每次运行约 1,000 行。

解析器负责履行此约定，验证机制确保其执行，当某次运行未达标时，监控系统会触发警报。

按此列表依次筛选，并在第一个包含合同中所有字段的数据源处停止：

1. 官方 API。当网站提供 API 时，从 API 获取数据优于解析 HTML：您将获得稳定的字段名称、有文档记录的限制以及明确的使用条款。
2. 页面已加载的 JSON 数据。打开开发者工具，在“网络”选项卡中按“Fetch/XHR”过滤，然后刷新页面。许多网站在将数据渲染到屏幕上之前，会先以 JSON 格式获取数据。
3. 站点地图，用于发现。利用网站的 XML 站点地图来发现 URL，而不是爬取每个链接。其 lastmod 日期信息还能显示自上次运行以来哪些页面发生了变化。
4. 静态 HTML。当数据位于源代码中时，请获取并解析该页面。
5. 已渲染的页面。仅当数据在其他任何地方都不存在时，才运行无头浏览器。

每向下走一步，都会增加请求次数、解析代码量，并且当网站发生变化时更容易出现故障。

以下是步骤 2 的实际应用：quotes.toscrape.com 的 JavaScript 版本从分页的 JSON 接口加载报价数据，因此你完全无需接触 HTML。

json_endpoint.py

"""Prefer the JSON a page already loads over parsing its HTML."""
import requests

url = "https://quotes.toscrape.com/api/quotes"
page, quotes = 1, []
while True:
    data = requests.get(url, params={"page": page}, timeout=(5, 30),
                        headers={"User-Agent": "acme-quotes-bot/0.1 (+mailto:data@example.com)"}).json()
    quotes += [(q["text"], q["author"]["name"]) for q in data["quotes"]]
    if not data["has_next"]:
        break
    page += 1
print(len(quotes), "quotes from", page, "JSON pages; first:", quotes[0][1])

• /api/quotes?page=N 与页面在浏览器中发出的请求相同，可在“网络”选项卡中找到。
• has_next 这是停止条件，因此你无需猜测最后一页，也不会获取空页面。
• 这里不需要 HTML 解析器，因为字段在传输时已带有名称和类型。

粗略的预算可以判断计划是否可行：页面数 × 频率 ÷ 允许速率 = 运行时长。

假设你每天需要抓取 10,000 个产品页面，而该网站允许每两秒发送一次请求。这意味着 10,000 × 2 = 20,000 秒，即每天大约 5.6 小时的抓取时间。 如果无法满足，则需要减少页面数量、采用增量爬取，或者使用每次请求包含更多记录的数据源。加快爬取速度并非可行方案，因为网站已设定了速率限制。

将抓取到的原始 HTML 或 JSON 数据压缩后，按 URL 和抓取时间作为键值存储几天。这样，当选择器失效或出现新字段时，只需重新解析存储的副本，而无需再次爬取网站。

仅抓取你有正当理由收集的公开数据，仔细阅读网站的使用条款，除非获得许可，否则请勿进入需要登录的区域。 获取 robots.txt 文件并按 RFC 9309 的定义加以应用，通过联系 URL 标识您的机器人，避免抓取不需要的个人数据，并且切勿冒充搜索引擎爬虫。

自 2022 年 IETF 发布 RFC 9309《机器人排除协议》以来，robots.txt 文件已成为一项正式标准。关键规则如下：

• 首先找到所属组。各组以 User-agent 行。您的爬虫将遵循以产品令牌命名的组，而 * 该组仅在无其他规则匹配时才生效。
• 最长的匹配规则优先。当多条规则匹配同一路径时，最具体的规则（即字符数最多的规则）决定最终结果。
• 允许规则在平局时胜出。如果一个 Allow 和 Disallow 规则匹配长度相同时，该路径被允许。
• 两个特殊字符。 * 匹配任意字符序列，而 $ 将路径末尾作为锚点，因此 Disallow: /*.pdf$ 会阻止所有路径以 .pdf.
• 状态码会改变其含义。robots.txt 中的 4xx 错误表示爬虫可以访问所有内容。5xx 错误或请求失败则表示必须将其视为完全禁止。
• 缓存存在时效限制。请勿将缓存副本使用超过 24 小时。
• Crawl-delay 不在标准范围内，且 Google 会忽略它。但若存在该参数，仍应予以遵守，因为它体现了网站所有者的意愿。

RFC 还指出，这些规则并非一种访问授权形式：该文件是网站所有者发布的要求，而非一种锁定机制。

对于 Python 爬虫而言，urllib.robotparser 显然是首选工具，但它会按文件顺序检查规则，并在遇到第一个匹配项时停止，而非选择有效期最长的规则。此外，它将规则路径视为普通前缀，因此 * 和 $ 并不具有特殊含义。

下面的代码片段将一个小型 robots.txt 文件分别通过这两种方法进行处理，并将处理结果并排打印出来。该“礼貌”抓取器随后导入了其 parse_robots() 和 rfc9309_allowed() 函数。

robots_match.py

"""robots.txt per RFC 9309: pick the group for your user agent, then the
longest matching rule wins, Allow wins ties, '*' matches any characters and
'$' anchors the end of the path."""
import re
import urllib.robotparser


def parse_robots(text, user_agent):
    """Return (rules, crawl_delay) for the group that applies to user_agent."""
    groups, current, in_agent_lines = [], None, False
    for raw in text.splitlines():
        line = raw.split("#", 1)[0].strip()
        if ":" not in line:
            continue
        key, value = (part.strip() for part in line.split(":", 1))
        key = key.lower()
        if key == "user-agent":
            if not in_agent_lines:  # a new group starts
                current = {"agents": [], "rules": [], "delay": None}
                groups.append(current)
            current["agents"].append(value.lower())
            in_agent_lines = True
            continue
        in_agent_lines = False
        if current is None:
            continue
        if key in ("allow", "disallow"):
            current["rules"].append((key, value))
        elif key == "crawl-delay":  # not in RFC 9309, but a clear signal
            try:
                current["delay"] = float(value)
            except ValueError:
                pass
    token = user_agent.split("/")[0].lower()  # product token, e.g. "acme-price-monitor"
    matched = [g for g in groups if token in g["agents"]] or \
              [g for g in groups if "*" in g["agents"]]
    rules = [rule for g in matched for rule in g["rules"]]
    delay = next((g["delay"] for g in matched if g["delay"] is not None), None)
    return rules, delay


def _pattern(rule_path):
    regex = re.escape(rule_path).replace(r"\*", ".*")
    if regex.endswith(r"\$"):
        regex = regex[:-2] + "$"
    return re.compile(regex)


def rfc9309_allowed(rules, path):
    """rules: list of ("allow"|"disallow", path) for the group that applies to you."""
    best_len, verdict = -1, True  # no matching rule means allowed
    for kind, rule_path in rules:
        if not rule_path:
            continue  # an empty Disallow allows everything
        if _pattern(rule_path).match(path):
            length = len(rule_path)
            if length > best_len or (length == best_len and kind == "allow"):
                best_len, verdict = length, kind == "allow"
    return verdict


if __name__ == "__main__":
    robots_txt = """User-agent: *
Disallow: /catalogue/
Allow: /catalogue/category/
Disallow: /*.pdf$

User-agent: acme-price-monitor
Disallow: /checkout/
Crawl-delay: 2
"""
    stdlib = urllib.robotparser.RobotFileParser()
    stdlib.parse(robots_txt.splitlines())
    rules, _ = parse_robots(robots_txt, "mybot/1.0")
    for path in ["/catalogue/category/books_1/index.html", "/catalogue/item_1.html",
                 "/files/report.pdf", "/files/report.pdf?v=2", "/index.html"]:
        print(f"{path:42} RFC 9309: {str(rfc9309_allowed(rules, path)):5}  "
              f"urllib.robotparser: {stdlib.can_fetch('mybot', path)}")
    rules, delay = parse_robots(robots_txt, "acme-price-monitor/1.2 (+https://example.com/bot)")
    print("acme-price-monitor group:", rules, "crawl-delay:", delay)
    print("/catalogue/item_1.html for acme-price-monitor:", rfc9309_allowed(rules, "/catalogue/item_1.html"))

比较结果表明：

• /files/report.pdf 被 urllib.robotparser，尽管 Disallow: /*.pdf$ 标准中禁止这种写法，因为标准库解析器会将 * 作为字面量进行解析。
• /catalogue/category/books_1/index.html 被 urllib.robotparser，因为 Disallow: /catalogue/ 在文件中排在前面。根据标准，较长的 Allow: /catalogue/category/ 优先，因此该页面被允许。
• acme-price-monitor 拥有自己的组，该组完全替换了 * 组。对于该机器人， /catalogue/item_1.html 允许以两秒的爬行延迟访问，因为 * 规则会延续下来。

请使用符合标准的匹配器，或者至少根据您所依赖的具体规则对解析器进行测试。

几条规则即可涵盖大多数情况：

• 请先阅读服务条款。登录通常意味着接受这些条款，且许多条款禁止自动化访问，因此请将经过身份验证的区域视为未经许可不得进入的区域。
• 仅收集最低限度的个人数据，或完全不收集。跳过数据协议中未列出的姓名、电子邮件或个人资料详情；无论您是否抓取了个人数据，GDPR 等数据保护法律均适用。
• 不要绕过付费墙、登录验证或验证码（CAPTCHA）。这些措施表明所有者希望限制访问。

这些是工作习惯，而非法律建议。有关法律细节，请参阅我们的《网络爬虫是否合法》指南。

使用一个稳定的 User-Agent，其中应包含您的产品名称、版本号以及联系方式：

acme-price-monitor/1.2 (+https://example.com/bot; data@example.com)

网站所有者若发现来自该字符串的异常流量，可直接联系您而非封锁整个 IP 范围；而 robots.txt 的编写者也能获得一个用于定位的产品标识符。

切勿发送 Googlebot User-Agent。谷歌在文档中说明了网站所有者如何通过反向和正向 DNS 查询或其公布的 IP 范围来验证其爬虫，因此伪造的 User-Agent 无法通过验证，一旦被发现就会显得欺骗性。

没有放之四海皆准的安全速率。建议从每个主机一到两个并发请求开始，并间隔约一秒，随后让服务器决定节奏：延迟增加时减速，收到 429 或 503 状态码时停止并等待，遵守 Retry-After 指令，并在出现抖动时呈指数级降低请求频率。

十个工作线程，只要每个都礼貌地暂停，仍可向同一个域名发送十个并行请求。在每个请求经过的节点控制速率：一个抓取器对象，或跨机器的共享限流器。

延迟上升意味着服务器负载加重，因此应延长触发错误前的延迟时间。将大规模爬取安排在该网站的非高峰时段，通常是其主要市场时区的夜间。

429 请求过多（RFC 6585）表示你在指定时间内发送了过多请求，而 503 则表示服务器过载或因维护而停机。 这两种情况都可能携带 Retry-After 头部，而《HTTP 语义》（RFC 9110）允许其值采用两种格式：

Retry-After: 120
Retry-After: Fri, 09 Oct 2026 07:30:00 GMT

第一种是秒，第二种是 HTTP 日期；若爬虫仅能处理其中一种格式，在遇到另一种格式时，程序将崩溃或忽略该标头。请将该值视为最小等待时间。

当不存在 Retry-After 头，或重试再次失败时，每次等待时间应相应延长：

wait = min(cap, base × 2^attempt) × random(0.5 to 1.5)

采用该等待时间与任何 Retry-After 值中的较大值。随机因子即为抖动：若无抖动，同时失败的任务将同步重试，导致服务器在同一波次中承受集中请求。

对幂等 GET 请求中的 429、500、502、503 和 504 响应以及网络超时进行重试。 不要重试 404 状态码，因为页面不会恢复；也不要反复请求 403 状态码，因为相同的请求会得到相同的响应。限制重试次数，并记录每次放弃的情况，以便卡住的 URL 能显示在你的指标中。

下文的 PoliteFetcher 类将本节与下一节的内容结合在一起，导入 parse_robots() 和 rfc9309_allowed() 来自 robots.txt 代码片段，因此请将这两个文件保存在同一个文件夹中。

polite_fetch.py

"""A polite, resilient fetcher: robots.txt, per-host pacing, timeouts,
retries with backoff and jitter, Retry-After, and conditional requests."""
import email.utils
import random
import time
from urllib.parse import urlsplit

import requests

from robots_match import parse_robots, rfc9309_allowed  # Snippet 2

USER_AGENT = "acme-price-monitor/1.2 (+https://example.com/bot; data@example.com)"
RETRYABLE = {429, 500, 502, 503, 504}
ROBOTS_TTL = 24 * 3600  # RFC 9309: don't trust a cached copy for more than 24 h


class PoliteFetcher:
    def __init__(self, min_delay=1.0, max_retries=4, timeout=(5, 30)):
        self.session = requests.Session()
        self.session.headers["User-Agent"] = USER_AGENT
        self.min_delay, self.max_retries, self.timeout = min_delay, max_retries, timeout
        self.robots, self.last_hit, self.validators = {}, {}, {}

    def _robots(self, host):
        cached = self.robots.get(host)
        if cached and time.monotonic() - cached[0] < ROBOTS_TTL:
            return cached[1], cached[2]
        try:
            resp = self.session.get(host + "/robots.txt", timeout=self.timeout)
        except requests.RequestException:
            resp = None
        if resp is None or resp.status_code >= 500:
            rules, delay = [("disallow", "/")], None  # unreachable: assume full disallow
        elif resp.status_code >= 400:
            rules, delay = [], None  # unavailable (e.g. 404): no restrictions
        else:
            rules, delay = parse_robots(resp.text, USER_AGENT)
        self.robots[host] = (time.monotonic(), rules, delay)
        return rules, delay

    def allowed(self, url):
        parts = urlsplit(url)
        rules, _ = self._robots(f"{parts.scheme}://{parts.netloc}")
        path = (parts.path or "/") + (f"?{parts.query}" if parts.query else "")
        return rfc9309_allowed(rules, path)

    def _pace(self, host):
        _, crawl_delay = self._robots(host)
        delay = max(self.min_delay, crawl_delay or 0)
        wait = self.last_hit.get(host, 0) + delay - time.monotonic()
        if wait > 0:
            time.sleep(wait)
        self.last_hit[host] = time.monotonic()

    @staticmethod
    def _retry_after(value):
        if not value:
            return None
        if value.isdigit():
            return int(value)
        when = email.utils.parsedate_to_datetime(value)  # HTTP-date form
        return max(0, when.timestamp() - time.time())

    def get(self, url):
        if not self.allowed(url):
            raise PermissionError(f"robots.txt disallows {url}")
        parts = urlsplit(url)
        host = f"{parts.scheme}://{parts.netloc}"
        headers = {}
        etag, modified = self.validators.get(url, (None, None))
        if etag:
            headers["If-None-Match"] = etag
        if modified:
            headers["If-Modified-Since"] = modified
        for attempt in range(self.max_retries + 1):
            self._pace(host)
            try:
                resp = self.session.get(url, headers=headers, timeout=self.timeout)
            except (requests.ConnectionError, requests.Timeout):
                resp = None
            if resp is not None and resp.status_code not in RETRYABLE:
                if resp.status_code == 200:
                    self.validators[url] = (resp.headers.get("ETag"),
                                            resp.headers.get("Last-Modified"))
                return resp  # 200, 304, or a non-retryable error for the caller
            if attempt == self.max_retries:
                break
            backoff = min(60, 2 ** attempt) * random.uniform(0.5, 1.5)
            server_hint = self._retry_after(resp.headers.get("Retry-After")) if resp is not None else None
            time.sleep(max(backoff, server_hint or 0))
        raise RuntimeError(f"gave up on {url} after {self.max_retries + 1} attempts")


if __name__ == "__main__":
    f = PoliteFetcher(min_delay=1.0, max_retries=2)
    for url in ["https://books.toscrape.com/", "https://httpbin.org/etag/abc123",
                "https://httpbin.org/etag/abc123", "https://httpbin.org/status/404"]:
        r = f.get(url)
        print(r.status_code, len(r.content), "bytes", url)
    for url in ["https://httpbin.org/deny", "https://httpbin.org/status/503"]:
        try:
            f.get(url)
        except (PermissionError, RuntimeError) as exc:
            print("expected:", exc)

关键代码行：

• _robots() 使用您自己的 User-Agent 抓取 robots.txt，将 4xx 状态码视为“无限制”，将 5xx 或网络错误视为“禁止访问”，并在缓存副本超过 24 小时后重新抓取。
• _pace() 强制执行按主机设定的延迟，当网站的 Crawl-delay 值长于您设定的最小值时，则采用该值。
• RETRYABLE 将重试限制在速率限制和服务器错误范围内。遇到 404 状态码时，会直接返回给调用方。
• _retry_after() 同时支持秒和 HTTP 日期两种格式，循环等待时间取决于回退时间与服务器提示时间中较长者。

下方的演示代码在重复请求时收到 304 响应，会直接返回 404 而不进行重试；在发送请求前会拒绝被 robots 规则禁止的路径；并在达到重试次数上限后，面对永久性 503 状态码会放弃请求。

大多数失败源于粗心的请求，而非高明的防御机制。请务必设置连接和读取超时，每个主机复用一个会话，发送一致且真实的请求头集，接受压缩，跳过不需要的图片和字体，规范化 URL 以避免重复，并使用条件请求，使未更改的页面返回空的 304 状态码。

• 为每个请求分别设置连接和读取超时，例如 5 秒和 30 秒。在 requests，未设置超时的请求可能会无限期等待，从而导致工作进程卡死。
• 使用一个 requests.Session 。这可复用连接，避免重复的 TLS 握手，从而减轻服务器负载。
• 在每次请求中发送相同、真实且规范的标头集，并包含可识别您身份的 User-Agent。我们的网络爬虫 HTTP 标头指南详细说明了哪些标头重要以及它们的排列顺序。
• 保持压缩功能开启。 requests 默认请求 gzip 压缩；如果您手动构建标头，请保留 Accept-Encoding，因为压缩后的HTML在传输中的体积仅为原始大小的几分之一。
• 在 URL 进入队列前进行清理：去除跟踪参数并 #fragments，遵循 rel="canonical" 链接，并去除重复项。
• 跳过跟踪像素和重复的分页视图；仅获取你将要解析的内容。

当响应中包含 ETag 或 Last-Modified 标头，请将其存储起来，并在下次请求时作为 If-None-Match 或 If-Modified-Since。通过这些条件请求，未更改的页面将返回不含正文的 304 Not Modified 状态码，这既为您节省了带宽，也减轻了网站的服务负担。

PoliteFetcher.get() 已经实现了这一点：其 validators 字典中存储了每个 URL 的 ETag 和 Last-Modified ，并在下次请求时直接返回。304响应会原样返回，因此您的代码可以复用已存储的副本。

在升级处理前先进行诊断。解读状态码（403、429、 503，或是返回验证码或空页面的200状态码），首先检查自身的请求速率和请求头，然后循序渐进地解决问题：降低请求速率、修正请求头、添加经过测试的代理、渲染页面，最后将难以处理的目标交由托管式爬虫API处理。

不确定代码的含义？请参阅我们的代理和 HTTP 状态错误列表。

软阻塞最为危险：返回 200 状态码但包含验证页面或无数据的响应看似成功，只有本指南后文提到的数据验证才能发现此类问题。

网站会封锁爬虫，因为大量的自动化流量会给服务器带来负担，并可能损害业务。OWASP 的自动化威胁列表将爬取列为独立威胁（OAT-011），而大多数防御措施针对的是流量规模和滥用模式，因此礼貌且身份明确的流量触发这些防御机制的概率较低。

请按以下顺序逐步升级：

1. 降低受影响主机的请求速度并减少并发数。
2. 修正请求头，确保其一致且完整。
3. 添加代理，首先根据目标选择合适的代理类型。在扩大规模前测试代理，然后通过轮换代理来分散负载。
4. 如果数据仅在 JavaScript 运行后才显示，请在无头浏览器中渲染页面。
5. 将最难处理的目标交由托管式爬取 API 处理。

针对指纹识别、验证码和隐身浏览器，请参考我们的操作指南，以避免被封锁。

仅当数据既不在 HTML 中，也不在页面发起的 JSON 调用中时才需要。将页面源代码与开发者工具显示的内容进行对比；如果数据是通过 XHR 请求传来的，请直接调用该端点。仅渲染真正需要 JavaScript 的页面，屏蔽图片和字体，并复用浏览器上下文，因为渲染速度远慢且成本更高。

无头浏览器会在没有可见窗口的情况下运行页面的 JavaScript：这是获取页面最全面的方式，但也是资源消耗最大的方式。以下两个快速测试可帮助您判断是否需要使用无头浏览器：

• 将“查看源代码”与“元素”面板进行对比。在页面源代码中（Ctrl+U 或 Cmd+Option+U）搜索你想要的值，例如价格。如果该值存在于源代码中，则普通 HTTP 请求就足够了；如果只有开发者工具的“元素”面板显示该值，则说明它是通过 JavaScript 生成的。
• 检查“网络”选项卡中是否存在携带该数据的 Fetch/XHR 请求（如前文提到的 JSON 端点示例），并直接调用该请求。

如果确实需要渲染，请仅将未能通过上述测试的 URL 发送至浏览器，并屏蔽图片、字体和媒体资源。使用 Playwright 或类似工具时，建议为每个任务运行一个浏览器进程并创建新的上下文，而非为每个页面启动一个新浏览器。

Scrapy 用户可以仅对需要 JavaScript 的请求进行渲染，而将爬取过程的其余部分保留在 Scrapy 的快速 HTTP 下载器上。

将选择器锚定在极少变化的元素上（如 ID、数据属性、语义结构、嵌入式 JSON-LD），而非样式类；为每个字段维护一个有序的备用选择器列表；正确解码原始字节；并确保必填字段缺失时会明确报错。一个默默返回空值的解析器，比直接崩溃的解析器更糟糕。

诸如 .text-lg.font-bold 每当设计师修改页面时就会改变。ID、 data-* 属性、 itemprop 微数据、语义标签和 JSON-LD 块的变更频率要低得多，因为其他代码依赖于它们。同时应保持选择器简短，因为当任何包裹 div 移动时，从 <body> 在任何包裹 div 移动时就会失效。

CSS 选择器更易于阅读，而 XPath 既能按文本选择元素，也能沿树结构向上遍历。XPath 与 CSS 选择器的对比分析了二者的权衡。

为每个字段提供一组选择器，按从最具体到最不具体的顺序排列，并取第一个匹配的选择器。 SELECTORS 下文代码片段中的字典即按此方式工作。

注意“£”编码漏洞。如果服务器发送 Content-Type: text/html 未指定字符集， requests 在构建时会回退到 ISO-8859-1 resp.text时，会回退到 ISO-8859-1，因此 UTF-8 页面会被错误解码，“£”会变成“£”，所有价格都无法解析。请将 resp.content原始字节；BeautifulSoup会读取页面自身的 <meta charset>.

将几个真实页面保存为测试数据。当选择器发生变化时，重新解析存储的原始响应，并比较变更前后的输出结果。

这段代码将 books.toscrape.com 目录页面上的图书卡片转换为带类型的记录，验证每一条，并打印质量报告；随后向解析器传入一张价格类已重命名的卡片，以展示布局变更已被捕获。

parse_validate.py

"""Parse defensively, normalize, validate, and report data quality."""
import json
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from urllib.parse import urljoin

import requests
from bs4 import BeautifulSoup

# Ordered fallbacks: the first selector that matches wins.
SELECTORS = {
    "title": ["h3 a[title]", "h3 a"],
    "price": ["p.price_color", "[itemprop=price]"],
    "availability": ["p.availability", ".instock"],
    "url": ["h3 a[href]"],
}
RATINGS = {"One": 1, "Two": 2, "Three": 3, "Four": 4, "Five": 5}


@dataclass
class Book:
    url: str
    title: str
    price: Decimal
    currency: str
    in_stock: bool
    rating: int | None
    fetched_at: str  # provenance: when this record was fetched (UTC)


def first(card, field):
    for css in SELECTORS[field]:
        node = card.select_one(css)
        if node is not None:
            return node
    return None


def parse_card(card, page_url, fetched_at=""):
    errors = []
    link, title_node = first(card, "url"), first(card, "title")
    price_node, stock_node = first(card, "price"), first(card, "availability")
    title = (title_node.get("title") or title_node.get_text(strip=True)) if title_node else ""
    if not title:
        errors.append("title missing")
    raw_price = price_node.get_text(strip=True) if price_node else ""
    currency = "GBP" if "£" in raw_price else "UNKNOWN"
    try:
        price = Decimal(raw_price.replace("£", "").replace(",", ""))
        if price <= 0:
            errors.append("price not positive")
    except InvalidOperation:
        price = Decimal("NaN")
        errors.append(f"price unparseable: {raw_price!r}")
    rating_node = card.select_one("p.star-rating")
    rating = next((RATINGS[c] for c in (rating_node.get("class", []) if rating_node else [])
                   if c in RATINGS), None)
    if link is None:
        errors.append("url missing")
    book = Book(url=urljoin(page_url, link["href"]) if link else "", title=title, price=price,
                currency=currency, in_stock="in stock" in (stock_node.get_text(" ", strip=True).lower()
                                                          if stock_node else ""),
                rating=rating, fetched_at=fetched_at)
    return book, errors


def scrape(page_url):
    resp = requests.get(page_url, timeout=(5, 30),
                        headers={"User-Agent": "acme-catalogue-bot/0.1 (+mailto:data@example.com)"})
    resp.raise_for_status()
    fetched_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
    # Pass bytes, not resp.text: with no charset in Content-Type, requests guesses
    # ISO-8859-1 and "£" becomes "£". BeautifulSoup reads the page's <meta charset>.
    soup = BeautifulSoup(resp.content, "html.parser")
    cards = soup.select("article.product_pod")
    seen, good, rejected = set(), [], []
    for card in cards:
        book, errors = parse_card(card, page_url, fetched_at)
        if book.url in seen:
            errors.append("duplicate url")
        seen.add(book.url)
        (rejected if errors else good).append((book, errors))
    report = {
        "page": page_url,
        "cards_found": len(cards),
        "valid": len(good),
        "rejected": len(rejected),
        "missing_rating": sum(1 for b, _ in good if b.rating is None),
        "sample": {k: str(v) for k, v in asdict(good[0][0]).items()} if good else None,
    }
    return good, rejected, report


if __name__ == "__main__":
    _, rejected, report = scrape("https://books.toscrape.com/catalogue/page-1.html")
    print(json.dumps(report, indent=2))
    for book, errors in rejected:
        print("REJECTED", book.url, errors)
    # Simulate a layout change: the price class is renamed.
    html = '<article class="product_pod"><h3><a href="x.html" title="T">T</a></h3><p class="cost">£1.00</p></article>'
    card = BeautifulSoup(html, "html.parser").select_one("article")
    print("after layout change:", parse_card(card, "https://books.toscrape.com/")[1])

• first() 会遍历每个字段的备用列表，因此一个重命名的类不会导致该列变为空。
• BeautifulSoup(resp.content, ...) 传递字节数组，从而避免了上述编码错误。
• parse_card() 收集错误而非触发异常，因此单张有问题的卡片不会导致页面处理中断，且每条被拒绝的记录都有明确原因。
• 模拟的版面布局变更会返回“价格无法解析”而非空白价格，这正是您希望在日志中看到的信号。

关于将原始页面转换为字段的更多内容，请参阅我们的数据解析指南。

在数据通过模式验证之前，应将每条记录视为不可信：必填字段是否存在、类型是否正确、数值范围是否合理，以及主键是否唯一。货币数据应使用 Decimal 类型，对货币和日期进行标准化处理，基于主键进行去重，存储源 URL 和抓取时间，并生成每轮抓取的有效值、被拒绝值和缺失值的报告。

数据质量是数据抓取工具常会“静默失败”的环节：抓取成功、解析运行，但错误值却会持续数周流入报告中。以下数据抓取最佳实践可有效防止此类问题：

• 在代码中定义模式。 Book 解析代码片段中的 `dataclass` 即为 Python 中的数据契约；pydantic 虽能提供更多检查，但 `dataclass` 本身无需额外依赖。
• 宁弃勿强型。当价格无法解析时，应附带原因拒绝该记录，而非存储零值；拒绝原因是你最早发现布局变更的预警信号。
• 将货币以 Decimal，切勿使用浮点数，因为浮点数无法精确表示大多数价格。将货币标准化为 ISO 4217 代码，将时间戳标准化为 UTC。
• 关注字段级空值率，观察其与上次运行相比是否出现突变。
• 基于规范 URL 或 SKU 等业务键进行去重。重复项通常意味着分页循环，或是同一商品被归入两个类别。
• 为每行赋予 source_url 和 fetched_at，以便任何值都能追溯到其对应的页面和时间点。
• 生成每轮运行的质量报告：发现的卡片、有效卡片、被拒绝的卡片以及缺失的可选字段。解析代码片段会精确输出这些内容，下一节将对其进行监控。

跟踪若干运行层指标，并在发生变化时触发警报（而不仅仅是崩溃时）：按状态码划分的成功率、每页记录数、空页面、字段空值率、重试次数、延迟、数据新鲜度以及每条有效记录的成本。当指标超出阈值时，让任务退出时返回非零值，这样调度程序（而非客户）能率先察觉异常。

发生崩溃的爬虫很容易被察觉；但那些悄无声息地收集错误数据的爬虫则难以被发现。请将下文列出的阈值视为调优的起点，而非基准。

养成以下两个习惯可提高这些指标的实用性：

• 每次运行时，都抓取几个具有已知值的“金丝雀”URL，例如标题固定的产品。如果“金丝雀”结果有误，说明解析器出问题了，而不是数据本身有问题。
• 基于变化幅度而非固定数值触发警报。与7天中位数相比，每页记录数的下降，比任何绝对计数更能有效预警布局变更。

下一节中的爬取代码片段以一个 health() 一个函数，该函数会标记抓取错误、空页面和内容稀疏的页面，然后以状态码 1 退出。cron、systemd、Airflow 和 CI 运行器等调度程序会将此视为失败，因此警报将由您已运行的工具触发。

关于调度、交付和重放，请参阅如何构建可恢复的抓取任务。

请在工作进程之前扩展后台管理机制，因为分布式爬取等高级网络爬取技术都依赖于此。 维护一个去重后的 URL 前沿数据集，将每页的记录及其“完成”标志在单个事务中提交，采用幂等写入以确保重跑时不会重复行，在所有 worker 中强制执行按域的速率限制，并利用站点地图和条件请求进行增量爬取，而非重新抓取所有内容。

可扩展网页抓取的最佳实践主要与状态管理有关，因为工作进程越多，进度追踪中存在的任何问题都会被放大。

• 使用“前沿”（frontier）而非列表：这是一个去重后的 URL 队列，允许爬虫在运行过程中动态发现新页面。初次接触“前沿”概念？不妨从 Python 网页爬取入门。
• 让检查点具有事务性。在一项事务中提交页面的记录、发现的链接以及“完成”标志，这样即使发生崩溃，该页面的所有工作要么全部保留，要么全部作废。
• 采用幂等写入。主键或 upsert 操作会使重复写入成为无操作，因此重跑和重试都是安全的。
• 恢复而非重启。下次运行将从第一个未完成的 URL 继续。

要查看其运行效果，请使用下面的代码片段爬取 quotes.toscrape.com，该片段维护一个 SQLite 前沿，并将每页的工作提交到一个事务中。运行 python resumable_crawl.py 3 将停止在第三页后，然后再次运行 python resumable_crawl.py 再次运行：第二次运行将从第一次停止处继续，且不会产生重复行。

resumable_crawl.py

"""Resumable pagination crawl with a SQLite checkpoint and a run health check."""
import json
import sqlite3
import sys
import time
from urllib.parse import urljoin

import requests
from bs4 import BeautifulSoup

START = "https://quotes.toscrape.com/page/1/"
HEADERS = {"User-Agent": "acme-quotes-bot/0.1 (+mailto:data@example.com)"}


def open_db(path="crawl.db"):
    db = sqlite3.connect(path)
    db.executescript("""
        CREATE TABLE IF NOT EXISTS frontier (url TEXT PRIMARY KEY, done INTEGER DEFAULT 0);
        CREATE TABLE IF NOT EXISTS quotes (text TEXT, author TEXT, tags TEXT,
                                           page TEXT, PRIMARY KEY (text, author));
    """)
    db.execute("INSERT OR IGNORE INTO frontier(url) VALUES (?)", (START,))
    db.commit()
    return db


def crawl(db, max_pages):
    stats = {"pages": 0, "errors": 0, "rows": 0, "empty_pages": 0, "rejected": 0}
    while stats["pages"] < max_pages:
        row = db.execute("SELECT url FROM frontier WHERE done = 0 LIMIT 1").fetchone()
        if row is None:
            break  # frontier exhausted
        url = row[0]
        try:
            resp = requests.get(url, headers=HEADERS, timeout=(5, 30))
            resp.raise_for_status()
        except requests.RequestException:
            stats["errors"] += 1
            break  # leave it undone; the next run resumes here
        soup = BeautifulSoup(resp.content, "html.parser")
        quotes = soup.select("div.quote")
        stats["empty_pages"] += not quotes
        with db:  # rows, link discovery and the checkpoint commit together
            for q in quotes:
                text, author = q.select_one("span.text"), q.select_one("small.author")
                if text is None or author is None:
                    stats["rejected"] += 1  # malformed card: reject it, don't coerce
                    continue
                cur = db.execute(
                    "INSERT OR IGNORE INTO quotes VALUES (?, ?, ?, ?)",
                    (text.get_text(strip=True), author.get_text(strip=True),
                     ",".join(t.get_text(strip=True) for t in q.select("a.tag")), url))
                stats["rows"] += cur.rowcount
            nxt = soup.select_one("li.next a[href]")
            if nxt:
                db.execute("INSERT OR IGNORE INTO frontier(url) VALUES (?)",
                           (urljoin(url, nxt["href"]),))
            db.execute("UPDATE frontier SET done = 1 WHERE url = ?", (url,))
        stats["pages"] += 1
        time.sleep(1.0)  # pace requests to one host
    return stats


def health(stats, db, min_rows_per_page=5):
    problems = []
    if stats["errors"]:
        problems.append(f"{stats['errors']} fetch errors")
    if stats["empty_pages"]:
        problems.append(f"{stats['empty_pages']} pages had no records (layout change?)")
    if stats["rejected"]:
        problems.append(f"{stats['rejected']} cards missing text or author")
    if stats["pages"] and stats["rows"] / stats["pages"] < min_rows_per_page:
        problems.append("rows per page below threshold")
    total = db.execute("SELECT COUNT(*) FROM quotes").fetchone()[0]
    pending = db.execute("SELECT COUNT(*) FROM frontier WHERE done = 0").fetchone()[0]
    return {**stats, "total_rows": total, "pending_urls": pending, "problems": problems}


if __name__ == "__main__":
    max_pages = int(sys.argv[1]) if len(sys.argv) > 1 else 100
    db = open_db()
    report = health(crawl(db, max_pages), db)
    print(json.dumps(report))
    sys.exit(1 if report["problems"] else 0)

• with db: 将插入的引文、新发现的下一页链接以及 done = 1 更新操作封装在一个事务中。
• INSERT OR IGNORE 对主键的更新使每次写入都具有幂等性，无论是前沿 URL 还是引文。
• 获取失败会将其 URL 标记为未完成并停止当前运行，因此下一次运行会首先重试该 URL。
• 缺少引文或作者的卡片会被跳过并计数，而不是以半空状态保存。
• health() 将运行的计数器转换为问题列表，并且 sys.exit(1) 使这些问题对任何调度该任务的系统可见。

一旦记账机制稳固，还有三项做法可安全地提升容量：

• 共享按域名划分的速率限制。在多台机器上，本地限流已无法保障网站安全，因此应将每个域名的配额集中存储在共享位置，例如 Redis 中的令牌桶。
• 增量抓取。利用站点地图 lastmod 日期仅将新URL或已更改的URL加入队列，并通过条件请求将未更改的页面转换为空的304响应。
• 最后进行水平扩展。只有在集中实施了每台主机限额后，才增加工作进程，这样更多的工作进程意味着更多主机并行处理，而不是给单个主机增加压力。

应衡量每条有效记录的成本，而非每次请求的成本。大部分节省来自减少数据获取：使用缓存和条件请求、优先选择 JSON 接口、仅在必要时渲染、限制重试次数、仅对需要代理的主机使用代理，并采用增量爬取。在构建更多功能之前，请将该数值与托管式爬取 API 的成本进行客观对比。

计算公式很简单：

cost per valid record = (infrastructure + proxies + engineering hours) ÷ valid records delivered

工程工时应计入分子，因为一个每周都需要开发人员维护的爬虫绝非廉价。分母仅统计通过验证的记录。

对于少数稳定且安全防护要求较低的目标，自主开发是合理的选择。当代理、重试、渲染和屏蔽所耗费的工程时间超过数据的实际价值时，请参考我们汇总的网页爬虫和爬取API选项，比较每条有效记录的成本。 如果您完全不想运行此类基础设施，WebScrapingAPI 的 Scraper API 是一款托管式抓取 API，可自动处理代理、重试以及可选的 JavaScript 渲染。该 API 上的 JavaScript 渲染和地理定位无需额外消耗积分，且失败的请求不会产生费用。

请忽略那些以短期访问权限为代价却带来风险的建议：伪造 Googlebot User-Agent、在每次请求中轮换 User-Agent 并使用不匹配的请求头、默认使用脚本模拟“随机”鼠标移动、采用无视服务器信号的固定延迟、对每个错误无限次重试，以及“以防万一”地抓取所有内容。 上述每种做法都会增加成本、风险或导致数据质量下降。

“使用 Googlebot User-Agent 以便网站放行。”网站所有者可以通过 DNS 查询识破伪造的 Googlebot。请按照身份识别部分的说明，使用包含联系信息的真实 User-Agent。

“每次请求都轮换 User-Agent。”一个带有真实浏览器绝不会发送的标头的浏览器 User-Agent，比保持一致的身份更显眼，因此请在每个会话中保持一套连贯的标头。

“添加随机的鼠标移动和点击以模拟人类行为。”对于大多数目标网站，这只会增加复杂性并拖慢每个页面的加载速度，却无法解决根本问题——请求量过大，因此请先调整请求速率和请求头。

“请求间隔 5 到 10 秒就安全了。”对于运行良好的网站，固定延迟太慢；对于返回 429 状态码的网站，则又太快。请采用根据延迟和 Retry-After 值动态调整的按主机限速策略。

“一直重试直到成功。”无限次重试会将一次服务中断演变成重试风暴。仅对可重试的错误进行重试，采用抖动策略进行退避，并限制重试次数。

“为了以防万一，把所有内容都抓取下来。”多余的页面会消耗带宽、存储空间和服务器时间，而如果额外字段包含个人数据，还会增加你的风险。只收集数据协议中列出的内容。

“Crawl-delay 是人人遵循的标准。”它并非 RFC 9309 的一部分，且 Google 会忽略它。若存在该参数请予以遵守，但不要指望每个爬虫或解析器都会以相同方式解读它。

“我的库会自动处理 robots.txt。” requests 和 BeautifulSoup 根本不会获取或应用 robots.txt。Scrapy 则可以通过其 ROBOTSTXT_OBEY 设置实现，该设置在启用 scrapy startproject 启用该功能。请检查您的设置，并根据您所依赖的规则测试匹配器。

在爬虫投入生产环境之前，请确认它具备以下要素：书面数据契约、符合标准的 robots.txt 检查、真实的 User-Agent、支持 Retry-After 的按主机限流、超时机制、条件请求、备用选择器、模式验证、运行健康警报、事务性检查点，以及对每条有效记录成本的追踪。

每项都有“完成条件”测试，因此您可以判断爬虫是否已准备好投入生产环境。

从零开始掌握这些技能？请从我们最优质的网页抓取课程中挑选。

网页抓取最佳实践是指确保抓取行为合法、礼貌、准确且经济实惠的一系列规范。其中包括：仅收集所需数据、遵守 robots.txt 文件和网站条款、标识您的机器人、控制每个主机的抓取频率、遵守 Retry-After 指令、验证数据记录、监控抓取进程、设置进度检查点，以及追踪每条有效记录的成本。

没有一个数值能适用于所有情况。建议从每个主机一到两个并发请求开始，请求间隔约一秒，然后根据实际情况调整：当响应时间增加时降低请求频率；遇到 429 或 503 响应时暂停；遵守 Retry-After 指令；并在非高峰时段进行大规模爬取。

从技术上讲没有强制要求，但 robots.txt 是网站所有者针对自动化客户端发布的规则，无视它可能会导致被封禁。请按照 RFC 9309 的定义来应用该规则，如果存在，请使用您爬虫自身的组别，同时也要阅读网站的条款。

并不完全符合。Python 的 urllib.robotparser 会按文件中的规则顺序进行处理，而非选择最长的匹配项，因此可能会允许网站禁止的 URL，或屏蔽网站允许的 URL。此外，它还会忽略 RFC 9309 中定义的 * 和 $ 通配符。

请使用真实、稳定的 User-Agent，其中应注明您的机器人名称并提供联系网址或电子邮箱，以便网站所有者能与您联系，而非直接封禁您。切勿冒充 Googlebot，因为 Google 已公开了验证其爬虫的方法。

停止向该主机发送请求，读取 Retry-After 字段，并至少等待该时间长度；该值可以是秒数，也可以是 HTTP 日期格式。如果没有该标头，则在恢复请求前，应采用带抖动（jitter）的指数级退避策略，限制重试次数，并降低每个主机的并发请求数。

虽然无法阻止网站改版，但可以确保爬虫在改版后仍能正常运行：将锚点选择器绑定到 ID、data 属性或 JSON-LD 等稳定锚点上，为每个字段设置备用方案，在缺少必填字段时明确报错，并在每页记录数减少时发出警报。

在存储前，请根据模式对每条记录进行验证：必填字段、正确的数据类型、合理的数值范围以及唯一键。价格字段使用 Decimal 类型，基于键进行去重，记录拒绝原因，并比较不同运行中的空值率和行数。

通常不需要。如果数据位于页面源代码中，或“网络”选项卡中的 JSON 请求中，请直接获取。仅当页面通过 JavaScript 生成内容时，才使用无头浏览器，并屏蔽图片和字体。

减少请求次数而非增加请求强度：使用 JSON 接口而非渲染后的页面，采用条件请求使未更改的页面返回 304 状态码，基于网站地图的增量爬取，以及连接复用。然后在不同主机之间进行并行处理（而非单个主机内部），并设置每台主机的请求限制。

请务必谨慎。登录通常意味着接受网站的条款，其中许多条款禁止自动化访问，因此在未获得许可或未签订合同的情况下，应将经过身份验证的区域视为禁区。若获得许可，请将凭据保存在密钥管理器中，限制个人数据的采集，并遵守常规的速率限制。

当目标网站数量较少、状态稳定且安全防护较弱时，建议自行开发。当代理、重试、渲染和屏蔽等操作所耗费的工程时间超过数据本身的价值，或者需要覆盖多个国家时，应使用托管式抓取 API，并综合考虑每条有效记录的成本（包括工程工时）。

上述所有操作均可由内部团队完成，对于许多团队而言，这是正确的选择。 若情况并非如此，您可以尝试提供 5,000 次免费请求的 Scraper API，让它处理代理、重试和可选的 JavaScript 渲染；通过 Browser API 仅需一次 REST 请求即可运行动态页面；添加按量付费的住宅代理（每 GB 8.50 美元起）， 或者通过 Data API 从 116 个受支持的数据源获取结构化记录。