(三)从原始文本得到统一的书目
1. 先统一字段表示
第九章已用选择器提取网页。提取到的字符串还可能带全角空格、重复空白和重复标签。先处理表示方式,之后才能比较两张卡片是否讲的是同一本书。
unicodedata.normalize("NFKC", value) 来自标准库 unicodedata,会把全角英数、兼容形式等规范为相应形式。接着 split() 按空白拆词," ".join(...) 用一个普通空格接回。例如 " Python 入门 " 变成 "Python 入门"。保留词之间的空格,书名中的两个单词仍可区分。
# 项目文件:catalog_project/src/cataloghub/cleaning.py
from dataclasses import asdict, dataclass
import re
import unicodedata
def text(value):
# 全角与兼容字符规范化,再把连续空白折成一个普通空格。
return " ".join(unicodedata.normalize("NFKC", str(value)).split())
@dataclass(frozen=True)
class Book:
source_id: str
title: str
finished: int
tags: tuple[str, ...]
def data(self):
return asdict(self)
@dataclass
class Batch:
books: list[Book]
rejected: list[dict]
raw_count: int
duplicates: intBook 是清洗后的数据对象,frozen=True 使这一份结果不再被后续步骤改写。标签保存为有序元组;同样的标签集合不受网页排列次序影响。Batch 同时保存接受的书目、拒收原因和计数,入库时就能生成完整报告。
# 续写同一文件:catalog_project/src/cataloghub/cleaning.py
def clean_one(raw):
# 先统一文本,再把可接受的字段转换为 Book 的确定类型。
source_id, title = text(raw["source_id"]).lower(), text(raw["title"])
if not re.fullmatch(r"b\d{3}", source_id):
raise ValueError("来源编号应为 b 加三位数字")
if not title or len(title) > 100:
raise ValueError("书名需要 1 到 100 个字符")
finished = text(raw["finished"])
if finished not in {"0", "1"}:
raise ValueError("已读字段应为 0 或 1")
tags = tuple(sorted({text(item) for item in raw["tags"] if text(item)}))
if len(tags) > 10 or any(len(tag) > 30 for tag in tags):
raise ValueError("标签最多十项,每项最多三十个字符")
return Book(source_id, title, int(finished), tags)re.fullmatch 要求编号整段符合 b 加三位数字;书名空白、已读字段写成其他值,会抛出说明具体字段的错误。标签先清理,再用集合去重,最后排序。这里的错误信息也是拒收表中要保存的内容。
2. 分开处理重复、坏记录和冲突
不能用“后来的覆盖前面的”处理所有重复。如果同批两张卡片都叫 b101,书名却完全不同,程序需要先停下来查看来源。直接覆盖会把页面次序变成决定书名的规则。
# 续写同一文件:catalog_project/src/cataloghub/cleaning.py
def clean_cards(cards):
# 字典按来源编号保留第一份清洗结果,重复和拒收各自计数。
accepted, rejected, duplicates = {}, [], 0
for number, raw in enumerate(cards):
try:
book = clean_one(raw)
except (KeyError, TypeError, ValueError) as error:
rejected.append(
{"index": number, "page": raw.get("page", ""), "reason": str(error)}
)
continue
previous = accepted.get(book.source_id)
if previous is not None:
if previous != book:
raise ValueError(f"同一批来源编号 {book.source_id} 出现了不同内容")
duplicates += 1
else:
accepted[book.source_id] = book
return Batch(list(accepted.values()), rejected, len(cards), duplicates)accepted 的键是来源编号,值是已经清洗过的 Book。编号第一次出现就保存;再次出现且对象相等,重复计数加一;同编号不同内容则抛出错误,整批还没进入数据库。单张卡片缺少书名则记到 rejected,连同原卡片的序号和页面地址保存下来。