智能数据脱敏：大模型驱动的 PII 实体识别、FPE 保序加密与动态网关落地实战

在数据资产流通、开发测试与商业智能（BI）分析中，**个人隐私信息（PII, Personally Identifiable Information）**的合规保护是一条不可逾越的法律红线（如《个人信息保护法》与 GDPR）。

然而，传统基于静态规则与正则表达式的脱敏方案在生产落地中正面临严峻挑战：

• 结构化盲区：很多上游业务表中字段名高度混淆（如 remark_text、extend_json），内部却塞满了未经治理的真实手机号、住址与银行卡号；
• 非结构化自由文本漏判：在客服工单、风控调查备注与用户评价中，用户经常输入“我是李明，麻烦退款到我招行卡 622202...，电话 13812345678”。静态正则无法结合上下文语义进行消歧，极易漏判或误伤正常的订单号与序列号；
• 业务可用性破坏：如果把手机号全部简单粗暴地替换为 ***，下游的数仓关联分析（如按用户手机号 Hash 关联跨端行为）就会彻底瘫痪。

将轻量命名实体识别（NER）大模型与**高性能规则引擎（Presidio 架构）结合，并引入保形加密（Format-Preserving Encryption, FPE）**与动态查询网关，正在成为现代数据安全治理的标准解法。

本文深入剖析智能 PII 识别流水线、重叠实体消歧算法、FPE 保形脱敏机制与生产级 Python 安全引擎实现。

一、双路识别架构：规则快路 + 语义模型深度消歧

为了兼顾高吞吐毫秒级延迟与非结构化高准确度，工业级 PII 识别引擎必须采用双路融合架构：

+-----------------------------------------------------------------------------------+
| 原始输入文本 (结构化字段值 / 客服工单 / 日志自由文本)                             |
+-----------------------------------------------------------------------------------+
                                          |
            +-----------------------------+-----------------------------+
            | (同步快路: < 2ms)                                         | (异步/微批深度路: < 20ms)
            v                                                           v
+--------------------------------------+     +--------------------------------------+
| 1. 确定性规则引擎 (Rule Engine)       |     | 2. 上下文语义 NER 模型 (LLM / BERT)  |
| - 正则匹配 + 算法校验码 (Checksum)   |     | - 识别自由文本中的模糊实体           |
| - 身份证校验算法 (ISO 7064)          |     | - 结合句式上下文消除误判             |
| - 银行卡 Luhn 算法有效性过滤         |     | - 识别真实人名、详细家庭住址         |
+--------------------------------------+     +--------------------------------------+
            |                                                           |
            +-----------------------------+-----------------------------+
                                          |
                                          v
+-----------------------------------------------------------------------------------+
| 3. 实体区间合并与冲突消歧 (Span Overlap Resolution)                               |
|    - 解决多路识别结果在字符区间上的重叠与包含冲突 (按置信度与规则优先级对齐)      |
+-----------------------------------------------------------------------------------+
                                          |
                                          v
+-----------------------------------------------------------------------------------+
| 4. 动态脱敏与保形加密策略引擎 (FPE / Masking Strategy Engine)                     |
|    - 角色 A (外包研发): 掩码打码 (138****0000)                                    |
|    - 角色 B (算法分析师): FPE 保形加密 (密文依然是合法的 11 位数字，支持直接 JOIN)|
|    - 角色 C (合规审计): 明文或可逆密钥解密查看                                    |
+-----------------------------------------------------------------------------------+

二、脱敏手段的进阶：从暴力掩码到 FPE 保形加密

在实际业务场景中，脱敏不能一味地“一盖了之”，需根据使用场景细分四种处置策略：

特别关注保形加密（Format-Preserving Encryption, FPE / FF1 / FF3-1）：
传统的 AES 加密会将 11 位的手机号加密成 32 字节的 Base64 字符串（如 U2FsdGVkX1...），这会破坏原有数据表的字段长度约束（VARCHAR(11) 报错），且无法利用原有索引进行高速 JOIN。
FPE 算法能够保证输入是 11 位数字，加密后的密文依然是确定性的 11 位数字，在完全保护真实隐私的同时，下游数仓的 ETL Pipeline 与关联查询无需做任何结构改造即可正常运行！

三、生产级双路 PII 识别、Luhn 校验与 FPE 动态脱敏引擎实现

下面的 Python 实现结合了正则表达式、身份证与银行卡有效性校验码算法、实体位置冲突消除、以及动态角色策略改写。

"""
smart_pii_masker.py
生产级智能 PII 识别、算法校验与动态脱敏引擎
"""

import hashlib
import re
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple


@dataclass
class PIIEntity:
    entity_type: str         # PHONE, ID_CARD, BANK_CARD, NAME, ADDRESS
    raw_text: str
    start: int
    end: int
    confidence: float
    source: str              # REGEX or NER_MODEL


class ChecksumValidator:
    """算法校验码过滤器：杜绝误伤非敏感的普通序列号"""

    @staticmethod
    def is_valid_luhn(card_number: str) -> bool:
        """Luhn 算法校验银行卡号"""
        digits = [int(d) for d in card_number if d.isdigit()]
        if len(digits) < 13 or len(digits) > 19:
            return False
        checksum = 0
        reverse_digits = digits[::-1]
        for i, d in enumerate(reverse_digits):
            if i % 2 == 1:
                doubled = d * 2
                checksum += doubled - 9 if doubled > 9 else doubled
            else:
                checksum += d
        return checksum % 10 == 0

    @staticmethod
    def is_valid_chinese_id(id_card: str) -> bool:
        """ISO 7064:1983.MOD 11-2 校验中国二代身份证校验位"""
        if not re.match(r"^\d{17}[\dXx]$", id_card):
            return False
        factors = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
        check_chars = "10X98765432"
        s = sum(int(id_card[i]) * factors[i] for i in range(17))
        expected_char = check_chars[s % 11]
        return id_card[-1].upper() == expected_char


class SmartPIIDetector:
    """双路 PII 探测器"""

    PHONE_RE = re.compile(r"(?<!\d)(1[3-9]\d{9})(?!\d)")
    ID_CARD_RE = re.compile(r"(?<!\d)(\d{17}[\dXx])(?!\d)")
    BANK_CARD_RE = re.compile(r"(?<!\d)(\d{16,19})(?!\d)")

    def detect_entities(self, text: str) -> List[PIIEntity]:
        candidates: List[PIIEntity] = []

        # 1. 规则快路匹配 + 算法校验码过滤
        for m in self.PHONE_RE.finditer(text):
            candidates.append(PIIEntity("PHONE", m.group(1), m.start(1), m.end(1), 0.99, "REGEX"))

        for m in self.ID_CARD_RE.finditer(text):
            val = m.group(1)
            if ChecksumValidator.is_valid_chinese_id(val):
                candidates.append(PIIEntity("ID_CARD", val, m.start(1), m.end(1), 1.0, "REGEX"))

        for m in self.BANK_CARD_RE.finditer(text):
            val = m.group(1)
            if ChecksumValidator.is_valid_luhn(val):
                candidates.append(PIIEntity("BANK_CARD", val, m.start(1), m.end(1), 0.95, "REGEX"))

        # 2. 模拟轻量 NER 模型深度路 (识别散落在自由文本中的人名与地址)
        # 实际生产中可调用微调后的 RoBERTa-NER 或 SpaCy 管道
        if "李雷" in text:
            idx = text.find("李雷")
            candidates.append(PIIEntity("NAME", "李雷", idx, idx + 2, 0.92, "NER_MODEL"))

        # 3. 区间冲突合并 (按置信度去重)
        return self._resolve_overlaps(candidates)

    def _resolve_overlaps(self, candidates: List[PIIEntity]) -> List[PIIEntity]:
        """合并重叠的区间，优先保留置信度最高且范围完整的实体"""
        sorted_entities = sorted(candidates, key=lambda x: (x.start, -(x.end - x.start), -x.confidence))
        resolved: List[PIIEntity] = []
        last_end = -1

        for ent in sorted_entities:
            if ent.start >= last_end:
                resolved.append(ent)
                last_end = ent.end
        return resolved


class DynamicMaskingGateway:
    """动态脱敏网关：根据请求方角色执行差异化脱敏"""

    def __init__(self, detector: SmartPIIDetector, salt_key: str = "SECURE_SALT_2026"):
        self.detector = detector
        self.salt_key = salt_key

    def transform(self, raw_text: str, user_role: str) -> str:
        entities = self.detector.detect_entities(raw_text)
        if not entities:
            return raw_text

        # 从后向前替换，防止破坏字符索引偏移
        entities_rev = sorted(entities, key=lambda x: x.start, reverse=True)
        result = raw_text

        for ent in entities_rev:
            masked_val = self._apply_policy(ent, user_role)
            result = result[:ent.start] + masked_val + result[ent.end:]

        return result

    def _apply_policy(self, ent: PIIEntity, role: str) -> str:
        # 管理员/审计员查看明文
        if role == "AUDITOR":
            return ent.raw_text

        # 数据研发/外包人员：局部掩码
        if role == "OUTSOURCING":
            if ent.entity_type == "PHONE":
                return ent.raw_text[:3] + "****" + ent.raw_text[7:]
            elif ent.entity_type == "ID_CARD":
                return ent.raw_text[:6] + "********" + ent.raw_text[-4:]
            elif ent.entity_type == "BANK_CARD":
                return ent.raw_text[:4] + " **** **** " + ent.raw_text[-4:]
            elif ent.entity_type == "NAME":
                return ent.raw_text[0] + "**"
            return "***"

        # 算法分析师：保形/保序伪匿名化 (通过加盐 Hash 保持确定性映射)
        if role == "ANALYST":
            hashed = hashlib.sha256((ent.raw_text + self.salt_key).encode()).hexdigest()
            # 格式保留示例：将 Hash 映射为相同长度的伪随机数字串
            pseudo_num = "".join(str(int(c, 16) % 10) for c in hashed[:len(ent.raw_text)])
            return f"[FPE:{pseudo_num}]"

        return "***"

生产环境脱敏效果演练

detector = SmartPIIDetector()
gateway = DynamicMaskingGateway(detector)

# 测试包含混合非结构化 PII 的客服工单文本
test_ticket = "用户李雷（身份证: 110101199003072396）反馈退款未到账，要求转至招行卡 6222020200112233445，联系电话 13812345678。"

print("【原始明文工单】:")
print(test_ticket)

print("\n【外包人员视角 (OUTSOURCING - 强掩码)】:")
print(gateway.transform(test_ticket, user_role="OUTSOURCING"))

print("\n【分析师视角 (ANALYST - 保形伪匿名化，支持唯一性聚合与跨表关联)】:")
print(gateway.transform(test_ticket, user_role="ANALYST"))

print("\n【合规审计视角 (AUDITOR - 明文审计)】:")
print(gateway.transform(test_ticket, user_role="AUDITOR"))

四、生产避坑与安全合规边界

在企业级数据脱敏架构落地中，必须严守以下四条铁律：

1. 动态脱敏只在网关与查询读路径（Read-Path）生效：
   动态脱敏严禁直接覆写底层的不可变存储（如 S3/HDFS/Iceberg）。数据湖中必须保留原始密文或受控明文，脱敏是针对不同权限角色在查询下发网关（如 Trino / Knox / 代理网关）处进行的实时内存重写。
2. 防范 $k$-匿名度不足带来的“再识别攻击（Re-identification Attack）”：
   攻击者即使看不到手机号，但如果结合“出生年月 + 所在小区 + 性别 + 职业”四列准标识符（Quasi-identifiers），依然有极高概率在外部公开库中唯一锁定目标个体。必须引入差分隐私（Differential Privacy）或针对稀疏组合进行 $k$-匿名度扰动。
3. 严格隔离密钥管理系统（KMS）：
   FPE 加密算法与哈希盐值（Salt Key）必须由企业级硬件安全模块（HSM）或云 KMS 独立托管，并设置定期的密钥轮转与访问审计日志，绝不能硬编码在工程代码中。

通过构建“规则与语义双路识别 + 算法有效性过滤 + FPE 保形加密 + 角色动态策略”的立体防御架构，企业能够在彻底消除 PII 泄露风险的同时，最大化保留数据资产的分析与建模价值。