使用 Pydantic 验证抓取的数据：捕捉选择器隐藏的错误
Senior Web Scraping Engineer
TL;DR:
- 一个返回数据的爬虫并不是一个返回正确数据的爬虫。 实时收集的二十条记录通过了一个宽松的 Pydantic 模型，而每个库存数量默默地设置为
0。 - 危险的代码是回退，而不是解析。
return int(match.group(1)) if match else 0看起来很小心，但正是这样的错误数字进入了数据库。抛出异常的验证器把这20个静默通过变成了20个明确拒绝。 - 拒绝指出了真正的错误。 列表页面上带有字符串
In stock，但根本没有数量——数量仅存在于每个产品的页面上，提取器从未读取这些页面。 - 验证不仅检查类型，还转换类型。
'£51.77'变为Decimal，'Three'变为3，而'In stock (22 available)'变为22，因此下游代码从不重新解析字符串。 - 字段约束捕捉到验证器遗漏的内容。
min_length、gt和le在没有一行自定义代码的情况下拒绝了空标题、非数字价格和超范围评分。 - 开始是免费的。 抓取阶段在通用爬虫 API 的免费层运行。
一个崩溃的爬虫告诉你它出错了。一个用自信的错误数据填充表格的爬虫什么也不告诉你，因为每个字段都有正确的类型，但没有人检查这些值是否有意义。
这个管道收集了一个实时目录页面，用 Pydantic 对记录建模，然后收紧模型，直到发现一个已存在的缺陷。
管道一览
抓取页面 → 提取原始字符串 → 验证和强制转换 → 收紧直到失败 → 修复源头
有趣的阶段是第四个。前三个是普通的，它们造成了这个错误。
阶段 1：抓取页面
Pydantic 没有 HTTP 客户端和 HTML 解析器——它验证 Python 对象，因此必须给它一些对象。这个管道通过 通用爬虫 API 抓取，返回渲染的文档作为字符串：
python
import json
import os
import urllib.request
UNLOCKER = "https://api.scrapeless.com/api/v2/unlocker/request"
def fetch(url):
payload = json.dumps({
"actor": "unlocker.webunlocker",
"input": {"url": url, "proxy_country": "US", "js_render": False},
}).encode()
request = urllib.request.Request(
UNLOCKER, data=payload,
headers={"Content-Type": "application/json",
"x-api-token": os.environ["SCRAPELESS_API_KEY"]},
)
with urllib.request.urlopen(request, timeout=120) as response:
return json.loads(response.read().decode())["data"]关键保存在 SCRAPELESS_API_KEY 环境变量中，从不存储在文件里。
阶段 2：提取原始字符串
页面给你的全部都是文本。这就是整个问题所在：
python
from bs4 import BeautifulSoup
def listing_records(html):
soup = BeautifulSoup(html, "html.parser")
return [{
"title": pod.h3.a["title"],
"price_gbp": pod.select_one("p.price_color").get_text(strip=True),
"in_stock": pod.select_one("p.instock.availability").get_text(strip=True),
"rating": pod.select_one("p.star-rating")["class"][1],
} for pod in soup.select("article.product_pod")]一次实时运行中的一条记录看起来像这样：
text
{"title": "A Light in the Attic", "price_gbp": "£51.77", "in_stock": "In stock", "rating": "Three"}带有货币符号的价格、可用性的句子，以及用英语单词表示的评分，均来自 CSS 类。这里的内容都无法被求和、排序或比较，直到它被转换为适当的格式。
阶段 3：验证和强制转换
Pydantic 模型声明你想要的类型及其转换。类型注解是声明机制，遵循 PEP 484 类型提示规范，并且在 before 模式下运行的 field_validator 在输入时转换原始值——这是在 Pydantic 验证器参考 中记录的钩子：
python
import re
from decimal import Decimal
from pydantic import BaseModel, field_validator
RATING_WORDS = {"One": 1, "Two": 2, "Three": 3, "Four": 4, "Five": 5}
class LenientBook(BaseModel):
title: str
price_gbp: Decimal
in_stock: int
rating: int
@field_validator("price_gbp", mode="before")
@classmethod
def parse_price(cls, value):
match = re.search(r"\d+\.\d{2}", value) if isinstance(value, str) else None
return Decimal(match.group()) if match else value
@field_validator("in_stock", mode="before")
@classmethod
def parse_availability(cls, value):
if not isinstance(value, str):
return value
match = re.search(r"\((\d+) available\)", value)
return int(match.group(1)) if match else 0 # <-- the silent fallback
@field_validator("rating", mode="before")
@classmethod
def parse_rating(cls, value):
return RATING_WORDS.get(value, value) if isinstance(value, str) else valueprice_gbp 是一个 Decimal 而不是一个 float，这是故意的。二进制浮点数无法准确表示大多数十进制分数，这就是为什么 Python十进制模块参考 的存在；将会被相加或比较的钱属于 Decimal。
对这20条实时记录运行模型并报告完全成功：20个有效，0个被拒绝。
这是错误的。
第4阶段：收紧直到失败
再看看标记的行。当可用性字符串不匹配时，验证器返回 0。这完美地满足了 in_stock: int —— 0 是一个整数，模型很高兴，记录被写入。
将每个回退替换为异常，并添加描述有效记录实际是什么的约束：
python
from pydantic import BaseModel, Field, field_validator
class StrictBook(BaseModel):
title: str = Field(min_length=1)
price_gbp: Decimal = Field(gt=0)
in_stock: int = Field(ge=0)
rating: int = Field(ge=1, le=5)
@field_validator("price_gbp", mode="before")
@classmethod
def parse_price(cls, 值):
如果 not isinstance(值, str):
返回 值
匹配 = re.search(r"\d+\.\d{2}", 值)
如果 not 匹配:
raise ValueError(f"在 {值!r} 中没有价格")
返回 Decimal(匹配.group())
@field_validator("in_stock", mode="before")
@classmethod
def parse_availability(cls, 值):
如果 not isinstance(值, str):
返回 值
匹配 = re.search(r"\((\d+) available\)", 值)
如果 not 匹配:
raise ValueError(f"在 {值!r} 中没有可用数量")
返回 int(匹配.group(1))
@field_validator("rating", mode="before")
@classmethod
def parse_rating(cls, 值):
如果 not isinstance(值, str):
返回 值
如果 值 not in RATING_WORDS:
raise ValueError(f"未知评级 {值!r}")
返回 RATING_WORDS[值]这些约束来自 Pydantic字段自定义参考，添加这些约束没有成本：标题必须有字符，价格必须为正，评级必须在1到5之间。
相同的20条记录现在产生0个有效和20个被拒绝，每一条都有相同的消息：
text
field=in_stock msg=值错误，在 'In stock' 中没有可用数量这不是模型过于苛刻。列表页面确实没有公布库存数量——它只说 In stock，没有更多。宽松的模型曾记录了一个实际数字0，这是页面从未说明的数量。
第5阶段：修复来源
数量是存在的，只是不在提取器查看的地方。每个产品的页面上都带有它：
python
def detail_record(html):
soup = BeautifulSoup(html, "html.parser")
返回 {
"title": soup.h1.get_text(strip=True),
"price_gbp": soup.select_one("p.price_color").get_text(strip=True),
"in_stock": soup.select_one("p.instock.availability").get_text(strip=True),
"rating": soup.select_one("p.star-rating")["class"][1],
}在详细页面上，相同的选择器返回 'In stock (22 available)'，StrictBook 接受它——in_stock=22，将 price_gbp=51.77 作为 Decimal，rating=3 作为 int。
修复是对管道的更改，而不是对模型的更改。这就是重点：模型的工作是拒绝猜测，而拒绝的结果使缺失的页面浮出水面。
开始不需要任何卡——免费计划 足以覆盖这个规模的运行。
整个管道
python
import json
import os
import re
import urllib.request
from decimal import Decimal
from bs4 import BeautifulSoup
from pydantic import BaseModel, Field, ValidationError, field_validator
UNLOCKER = "https://api.scrapeless.com/api/v2/unlocker/request"
LISTING = "https://books.toscrape.com/"
DETAIL = "https://books.toscrape.com/catalogue/a-light-in-the-attic_1000/index.html"
RATING_WORDS = {"One": 1, "Two": 2, "Three": 3, "Four": 4, "Five": 5}
def fetch(url):
payload = json.dumps({
"actor": "unlocker.webunlocker",
"input": {"url": url, "proxy_country": "US", "js_render": False},
}).encode()
request = urllib.request.Request(
UNLOCKER, data=payload,
headers={"Content-Type": "application/json",
"x-api-token": os.environ["SCRAPELESS_API_KEY"]},
)
with urllib.request.urlopen(request, timeout=120) as response:
return json.loads(response.read().decode())["data"]
class LenientBook(BaseModel):
title: str
price_gbp: Decimal
in_stock: int
rating: int
@field_validator("price_gbp", mode="before")
@classmethod
def parse_price(cls, value):
match = re.search(r"\d+\.\d{2}", value) if isinstance(value, str) else None
return Decimal(match.group()) if match else value
@field_validator("in_stock", mode="before")
@classmethod
def parse_availability(cls, value):
if not isinstance(value, str):
return value
match = re.search(r"\((\d+) available\)", value)
return int(match.group(1)) if match else 0 # <-- 哑元回退
@field_validator("rating", mode="before")
@classmethod
def parse_rating(cls, value):
return RATING_WORDS.get(value, value) if isinstance(value, str) else value
class StrictBook(BaseModel):
title: str = Field(min_length=1)
price_gbp: Decimal = Field(gt=0)
in_stock: int = Field(ge=0)
rating: int = Field(ge=1, le=5)
@field_validator("price_gbp", mode="before")
@classmethod
def parse_price(cls, value):
if not isinstance(value, str):
return value
match = re.search(r"\d+\.\d{2}", value)
if not match:
raise ValueError(f"no price in {value!r}")
return Decimal(match.group())
@field_validator("in_stock", mode="before")
@classmethod
def parse_availability(cls, value):
if not isinstance(value, str):
return value
match = re.search(r"\((\d+) available\)", value)
if not match:
raise ValueError(f"no availability count in {value!r}")
return int(match.group(1))
@field_validator("rating", mode="before")
@classmethod
def parse_rating(cls, value):
if not isinstance(value, str):
return value
if value not in RATING_WORDS:
raise ValueError(f"unknown rating {value!r}")
return RATING_WORDS[value]
def listing_records(html):
soup = BeautifulSoup(html, "html.parser")
return [{
"title": pod.h3.a["title"],
"price_gbp": pod.select_one("p.price_color").get_text(strip=True),
"in_stock": pod.select_one("p.instock.availability").get_text(strip=True),
"rating": pod.select_one("p.star-rating")["class"][1],
} for pod in soup.select("article.product_pod")]
def detail_record(html):
soup = BeautifulSoup(html, "html.parser")
return {
"title": soup.h1.get_text(strip=True),
"price_gbp": soup.select_one("p.price_color").get_text(strip=True),
"in_stock": soup.select_one("p.instock.availability").get_text(strip=True),
"rating": soup.select_one("p.star-rating")["class"][1],
}
def run(model, records):
valid, errors = [], []
for record in records:
try:
valid.append(model(**record))
except ValidationError as exc:
errors.append(exc)
return valid, errors
def main():
records = listing_records(fetch(LISTING))
print(f"提取的列表记录数量：{len(records)}")
print(f"原始示例：{json.dumps(records[0], ensure_ascii=False)}")
lenient, lenient_errors = run(LenientBook, records)
zeroed = sum(1 for book in lenient if book.in_stock == 0)
print(f"宽松模式模型：{len(lenient)} 个有效, {len(lenient_errors)} 个被拒绝")
print(f" 默默地将库存归零：{zeroed} 中的 {len(lenient)}")
strict, strict_errors = run(StrictBook, records)
print(f"严格模式模型：{len(strict)} 个有效, {len(strict_errors)} 个被拒绝")
first = strict_errors[0].errors()[0]
print(f" 第一个错误：字段={first['loc'][0]} 消息={first['msg']}")
detail = detail_record(fetch(DETAIL))
print(f"详细原始可用性：{detail['in_stock']!r}")
book = StrictBook(**detail)
print(f"详细验证：标题={book.title!r} 价格_gbp={book.price_gbp} "
f"库存={book.in_stock} 评分={book.rating}")
print(f" 类型：价格={type(book.price_gbp).__name__} "
f"库存={type(book.in_stock).__name__} 评分={type(book.rating).__name__}")
malformed = [
{**detail, "price_gbp": "按要求提供价格"},
{**detail, "rating": "十一"},
{**detail, "title": ""},
]
print("格式错误的输入：")
for record in malformed:
try:
StrictBook(**record)
print(" 意外有效")
except ValidationError as exc:
err = exc.errors()[0]
print(f" 拒绝 字段={err['loc'][0]} 类型={err['type']}")
if __name__ == "__main__":
main()其输出：
text
提取的列表记录数量：20
原始示例：{"title": "A Light in the Attic", "price_gbp": "£51.77", "in_stock": "有货", "rating": "三"}
宽松模式模型：20 个有效, 0 个被拒绝
默默地将库存归零：20 中的 20
严格模式模型：0 个有效, 20 个被拒绝
第一个错误：字段=in_stock 消息=值错误，在“库存中”没有可用数量
详细原始可用性：'库存中（22个可用）'
详细验证：标题='阁楼里的光' 价格_英镑=51.77 在库存中=22 评分=3
类型：价格=Decimal 在库存中=int 评分=int
格式错误的输入：
拒绝的字段=价格_英镑 类型=值错误
拒绝的字段=评分 类型=值错误
拒绝的字段=标题 类型=字符串过短最后三行显示了输入的约束，这些是验证器单独无法捕捉到的。非数字价格和未知评分词被引发的验证器阻止；空标题被min_length=1阻止，机器可读代码为string_too_short，没有自定义代码。
验证可以和不能告诉你的事情
模型证明一个值具有正确的类型并满足你编写的规则。它无法证明值是真实的。
in_stock=0 对于所有20条记录来说都是一个完全有效的整数。使其错误的是页面从未表示为零，而没有类型系统可以看到这一点。捕捉它的规则更窄：该字段必须源于实际存在的模式。编写引发而不是替代的验证器就是你如何编码的。
这种区别还决定了验证应在何处进行。在边界处，对每个记录进行验证，这样拒绝会指向一个页面。在完成的数据集上聚合检查——这正是数据框擅长的，参考pandas管道指南——会捕捉到不同的问题，并在发现的问题页面很难识别时捕捉到它们。
如果你的爬虫在一个框架内运行，请在记录离开解析器时挂载模型。在Scrapy中，这是一个项目管道，它在写入任何内容之前看到来自每个爬虫的每个项目。
故障排除
网站重新设计后每条记录都被拒绝。 选择器仍然匹配一些东西，只是与之前不同。在验证之前打印原始字典——错误消息引用了有问题的值，通常会立即识别出错误的元素。
验证器从未运行。 mode="before" 的验证器在原始输入上运行；不使用它的情况下，验证器在Pydantic尝试强制转换之后运行，因此像 '£51.77' 这样的字符串在你的代码看到它之前在类型转换时失败。
Decimal 与浮点数比较时行为异常。 不要混合它们。将货币保持在Decimal中端到端，仅在显示边界进行转换。
某些页面的字段根据设计缺失。 将其建模为 int | None = None，而不是默认值为 0。None 表示缺失；0 则虚构了一个测量值。
被拒绝的记录消失。 收集 ValidationError 和原始字典，并将两者写入某处。拒绝是运行中最有价值的输出——它们是结构发生更改的页面列表。
结论
验证在失败时赢得它的位置。一个通过你提供的所有数据的模型只是装饰；这个模型拒绝了20条活跃记录，并在此过程中报告了管道正在从一个没有发布库存计数的页面读取库存计数。
值得养成的习惯是小的：当解析不匹配时，抛出异常而不是替代。默认值将明显的失败转化为无声的失败，而无声的失败是到达生产的那些。
准备在自己的管道上应用这个吗？创建一个免费Scrapeless帐户，导出你的密钥，并在扩展爬取之前逐个记录类型建模。计划限制在定价页面上。
常见问题
问：为什么使用Pydantic而不是自己编写检查？
因为检查是与类型声明一起出现的，而不是分散在爬虫中。一个模型声明了什么是有效记录，将原始字符串转换为可用类型，并生成带有字段名和错误代码的机器可读错误。手动编写的检查在与编写数据的代码不同步时会漂移；模型不能，因为没有什么在未通过之前被构造。
问：何时使用 field_validator，何时使用 Field 约束？
对于任何可以作为关于完成值的规则表达的内容使用约束——min_length、gt、le、模式。当原始输入必须在可以检查之前被转换时，使用验证器，对爬取的数据来说，这适用于大多数字符串字段。这两者可以组合：验证器将 '£51.77' 转换为 Decimal，然后 gt=0 检查结果。
问：我应该如何处理未通过验证的记录？
将它们写入拒绝文件，包含原始字典和错误，并保持运行进行。拒绝是关于页面的证据，而不是停止的理由。对拒绝率的警报比对单个失败的警报更有用，因为从少量跳至全部意味着网站发生了变化。
问：这是否替代了检查我的选择器是否有效？
不，它捕捉到的是不同的失败。选择器测试告诉你是否找到了元素；验证告诉你所包含的内容是否有意义。这个帖子中的错误通过了选择器测试——p.instock.availability在每个记录上都匹配——而是通过询问匹配文本实际内容才被发现。
问：Pydantic的速度对于大规模抓取是否足够快？
它的验证核心是编译的，而不是纯Python，而且对于抓取成本主要由网络时间主导——获取一个页面的时间比验证它所产生的记录慢几个数量级。如果一个配置文件显示结果不是这样，应该批量验证，而不是放弃检查。
问：我可以根据数据生成模型而不是手动编写吗？
可以，出于探索的目的这是有用的，但从一个样本推断出的模型编码了该样本所包含的内容。这里的价值在于将记录应该是什么写下来，并让现实与之不符——一个推导出的模型将学习到in_stock总是0且从未反对。
在Scrapeless，我们仅访问公开可用的数据，并严格遵循适用的法律、法规和网站隐私政策。本博客中的内容仅供演示之用，不涉及任何非法或侵权活动。我们对使用本博客或第三方链接中的信息不做任何保证，并免除所有责任。在进行任何抓取活动之前，请咨询您的法律顾问，并审查目标网站的服务条款或获取必要的许可。
