Python实现字符串敏感信息（名词、数字）掩码与反掩码需求

核心思路是建立原始敏感信息与掩码的一一映射关系：先对英文查询中的敏感实体（如产品名、客户ID关联数字）进行掩码并记录映射，待模型生成含掩码的SQL后，再通过映射表将掩码还原为原始数据。

步骤1：识别并掩码英文查询中的敏感信息 #

通过实体识别工具定位需要保护的敏感实体，替换为唯一标识的掩码（或指定长度的x串），同时保存每个掩码对应的原始值。

代码实现 #

import spacy
import re

# 加载spaCy英文模型（需先执行安装命令：pip install spacy && python -m spacy download en_core_web_sm）
nlp = spacy.load("en_core_web_sm")

def mask_sensitive_query(query):
    doc = nlp(query)
    mask_mapping = {}
    masked_query = query
    mask_index = 1

    # 定义需要掩码的实体类型：产品、ID关联数字
    target_entity_types = ["PRODUCT", "CARDINAL"]

    # 倒序替换避免索引混乱
    for entity in reversed(doc.ents):
        # 掩码产品实体
        if entity.label_ == "PRODUCT":
            mask_key = f"[MASK_{mask_index}]"
            mask_mapping[mask_key] = entity.text
            masked_query = masked_query[:entity.start_char] + mask_key + masked_query[entity.end_char:]
            mask_index += 1
        # 掩码ID后的数字
        elif entity.label_ == "CARDINAL" and query[entity.start_char-3:entity.start_char].strip() == "ID":
            mask_key = f"[MASK_{mask_index}]"
            mask_mapping[mask_key] = entity.text
            masked_query = masked_query[:entity.start_char] + mask_key + masked_query[entity.end_char:]
            mask_index += 1

    # 若需替换为示例中的固定长度x串，可将mask_key替换为：
    # mask_key = "x" * len(entity.text)

    return masked_query, mask_mapping

# 测试示例
original_query = "How many Chocolate Orders for a customer with ID 123456?"
masked_query, mask_map = mask_sensitive_query(original_query)
print("掩码后查询：", masked_query)
print("掩码映射表：", mask_map)

步骤2：生成含掩码的SQL #

将上一步得到的masked_query传入你的SQL生成模型，得到含掩码的SQL语句（如示例中的Select count(1) from sample-bucket as d where d.Type ='[MASK_1]' and d.CustId = '[MASK_2]'）。

步骤3：反掩码还原SQL中的敏感信息 #

利用之前保存的mask_map，将SQL中的掩码替换回原始敏感数据。

代码实现 #

def unmask_generated_sql(masked_sql, mask_mapping):
    unmasked_sql = masked_sql
    # 遍历映射表替换所有掩码
    for mask_key, original_value in mask_mapping.items():
        # 处理带引号的掩码（SQL字符串常量）
        unmasked_sql = unmasked_sql.replace(f"'{mask_key}'", f"'{original_value}'")
        # 处理无引号的掩码（若存在）
        unmasked_sql = unmasked_sql.replace(mask_key, original_value)
    return unmasked_sql

# 测试示例
masked_sql = "Select count(1) from `sample-bucket` as d where d.Type ='[MASK_1]' and d.CustId = '[MASK_2]'"
unmasked_sql = unmask_generated_sql(masked_sql, mask_map)
print("反掩码后SQL：", unmasked_sql)

注意事项 #

• 若使用固定长度的x串作为掩码，需注意相同长度的敏感信息会导致映射冲突（如两个6位数字会被替换为相同的xxxxxx，反掩码时无法区分），因此更推荐使用带唯一编号的掩码（如[MASK_1]）。
• 可根据实际需求扩展实体类型（如添加PERSON、ORG等），或通过关键词匹配进一步精准定位敏感信息。

内容的提问来源于stack exchange，提问作者Aadhi Verma