3.3 清洗车间：数据清洗与预处理

3.3 清洗车间：数据清洗与预处理

本节摘要：清洗分三层规整——字符层、结构层、语义层——加两级去重（精确指纹与近似指纹）。本节给出每层规则的职责边界、可复用的流水线实现，以及 SimHash 类近似去重的直觉与落地代码，并处理 1.5 节交下来的匿名化工位。

清洗车间的三层规整

清洗最容易犯的错是把规则糊成一团：一个函数里既去空白又去广告又改格式，两周后没人说得清哪条规则改了什么。车间的做法是分三层，每层职责单一、规则显式：

字符层管编码与符号：全角半角统一、不可见控制字符剔除、连续空白折叠、异常 Unicode 替换。它处理的是"字符世界的杂质"，不关心内容含义。

结构层管字段形态：HTML 残留标签剥离、日期解析成统一格式、数值字段去掉货币符号与千分位、枚举字段映射到备料单规定的取值。它保证每列数据长得像备料单说的那样。

语义层管内容取舍：语言识别过滤（只要中文语料却混进越南文页面）、长度过滤（三个字的"好评。"对训练无贡献）、模板残留识别（"登录后查看更多"这类站点水印）、广告与导航句剔除。它动的是内容本身，所以规则最需要抽样验证。

图 3-3 清洗流水线：三层规整加两级去重的漏斗

漏斗右侧的"典型损耗"不是标准答案，是校准锚点：字符层损耗突然涨到百分之五，多半是上游站点换了编码或混入了异常页面；语义层损耗腰斩，可能是过滤规则把正常内容也杀了。每层损耗率是流水线的体检指标，3.4 节入库时它们要随版本一起归档。

三层规整的流水线实现

import re, unicodedata

def clean_character(text: str) -> str:
    """字符层：编码统一、控制字符、空白折叠"""
    text = unicodedata.normalize("NFKC", text)        # 全角转半角，兼容字符统一
    text = "".join(ch for ch in text if unicodedata.category(ch)[0] != "C")  # 剔除控制字符
    return re.sub(r"\s+", " ", text).strip()           # 连续空白折叠为一个

def clean_structure(date_str: str) -> str:
    """结构层示例：多种日期写法归一到 ISO 格式"""
    from datetime import datetime
    for fmt in ("%Y-%m-%d", "%Y年%m月%d日", "%Y/%m/%d %H:%M", "%Y%m%d"):
        try:
            return datetime.strptime(date_str.strip(), fmt).strftime("%Y-%m-%d")
        except ValueError:
            continue
    return ""                                          # 解析失败置空，交给完整率统计暴露

def clean_semantic(text: str, min_len: int = 5,
                   templates: tuple = ("登录后查看", "点击展开", "广告")) -> str:
    """语义层：模板残留剔除与长度过滤"""
    for t in templates:
        if t in text:
            return ""                                  # 含站点水印的整条弃用
    if len(text) < min_len:
        return ""
    return text

raw = "\u3000\u3000好评\u3000！！\u3000"              # 全角空格加全角感叹号
step1 = clean_character(raw)
print(repr(step1))                     # 输出：'好评!!'
print(clean_structure("2025年11月03日"))   # 输出：2025-11-03
print(clean_structure("2025/11/03 09:30")) # 输出：2025-11-03
print(repr(clean_semantic("登录后查看更多内容")))   # 输出：''（模板残留被剔除）
​

三段函数各管一层，组合方式就是流水线的编排顺序：字符层最便宜先跑，语义层最贵最后跑——让便宜规则先砍掉体量，昂贵规则只处理幸存者。顺序反过来不是不行，是费钱。

两级去重：精确指纹加近似指纹

去重要拦两种重复。字面重复：同一条数据被多个源收录、或翻页异常抓了双份——精确指纹（归一化后哈希）就能拦住。近似重复：转载时改了标点、加了"转载自"前缀、换了段首——字面不同但语义同源，需要近似指纹。近似去重的经典思路是把文本切成词块（shingle），只保留满足哈希条件的一部分作为指纹，两个指纹的重合度超过阈值判为重复：

import hashlib, re

def shingles(text: str, k: int = 3) -> set[tuple]:
    """中文按字符切3克，英文风格文本可换成按词切"""
    text = re.sub(r"\s+", "", text)
    return {tuple(text[i:i+k]) for i in range(max(1, len(text) - k + 1))}

def simhash_fingerprint(text: str) -> int:
    """简化版 SimHash：词块加权投票合成一个64位指纹"""
    v = [0] * 64
    for sh in shingles(text):
        h = int(hashlib.md5("".join(sh).encode()).hexdigest(), 16)
        for bit in range(64):
            v[bit] += 1 if (h >> bit) & 1 else -1
    fp = 0
    for bit in range(64):
        if v[bit] > 0:
            fp |= (1 << bit)
    return fp

def hamming(a: int, b: int) -> int:
    return bin(a ^ b).count("1")

def is_near_duplicate(a: str, b: str, threshold: int = 12) -> bool:
    """汉明距离小于阈值判为近似重复"""
    return hamming(simhash_fingerprint(a), simhash_fingerprint(b)) <= threshold

t1 = "这款扫地机器人的续航表现非常出色，一次充电可以完成全屋清扫。"
t2 = "这款扫地机器人的续航表现非常出色，一次充电可以完成全屋清扫！（转载）"
t3 = "扫地机器人的避障算法这几年进步明显，家具保护做得越来越好。"
print(is_near_duplicate(t1, t2))   # 输出：True（标点与前缀变化不改变指纹）
print(is_near_duplicate(t1, t3))   # 输出：False（主题不同，距离远大于阈值）
​

工程上的分工：精确指纹用字典或数据库唯一索引兜底，成本近乎为零，全量执行；近似指纹计算贵，通常只在"同主题分桶"内两两比较（桶按 URL 域名或主题分类预分组），把平方级比较压到可承受。阈值 12 不是圣旨——转载改写越狠的领域阈值要越松，标定方法还是那句话：抽样，人工看。

常见坑：去重键只用了正文，忽略了"正文相同但回答不同"的场景。问答语料里同一条问题配两个版本的答案完全合法，去重键应该包含问题字段或整体 schema，否则会把有效数据当重复误杀。

关键直觉：清洗规则的目标不是"干净"这个状态，而是"可复现"这个过程。同一份规则跑两次结果不同（比如用了随机采样但没固定种子），三个月后你就无法回答"这批数据是怎么来的"——而这个问题在数据出问题时一定会被问到。