# masking_engine.py
import re
import hashlib
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum
class DataLevel(Enum):
L1 = "公开"
L2 = "内部"
L3 = "敏感"
L4 = "高度敏感"
L5 = "机密"
@dataclass
class MaskingRule:
name: str
pattern: str
method: str
replacement: str
level: DataLevel
enabled: bool = True
class MaskingEngine:
def __init__(self, rules: List[MaskingRule]):
self.rules = {rule.name: rule for rule in rules if rule.enabled}
def detect_level(self, text: str) -> DataLevel:
"""检测文本中的数据级别"""
max_level = DataLevel.L1
for rule in self.rules.values():
if re.search(rule.pattern, text):
if rule.level.value > max_level.value:
max_level = rule.level
return max_level
def mask(self, text: str, target_level: DataLevel = DataLevel.L2) -> str:
"""脱敏处理"""
result = text
for rule in self.rules.values():
if rule.level.value >= target_level.value:
if rule.method == "replace":
result = re.sub(rule.pattern, rule.replacement, result)
elif rule.method == "truncate":
result = self._truncate_match(result, rule)
elif rule.method == "hash":
result = self._hash_match(result, rule)
elif rule.method == "generalize":
result = self._generalize_match(result, rule)
return result
def _truncate_match(self, text: str, rule: MaskingRule) -> str:
"""截断脱敏"""
def replace_func(match):
original = match.group(0)
keep_prefix = getattr(rule, 'keep_prefix', 0)
keep_suffix = getattr(rule, 'keep_suffix', 0)
replacement_char = getattr(rule, 'replacement_char', '*')
if len(original) <= keep_prefix + keep_suffix:
return original
masked = (
original[:keep_prefix] +
replacement_char * (len(original) - keep_prefix - keep_suffix) +
original[-keep_suffix:] if keep_suffix > 0 else ""
)
return masked
return re.sub(rule.pattern, replace_func, text)
def _hash_match(self, text: str, rule: MaskingRule) -> str:
"""哈希脱敏"""
def replace_func(match):
original = match.group(0)
return hashlib.sha256(original.encode()).hexdigest()[:16]
return re.sub(rule.pattern, replace_func, text)
def _generalize_match(self, text: str, rule: MaskingRule) -> str:
"""泛化脱敏"""
# 实现泛化逻辑
return re.sub(rule.pattern, "[已脱敏]", text)
# 使用示例
if __name__ == "__main__":
rules = [
MaskingRule(
name="id_card",
pattern=r"\\d{17}[\\dXx]",
method="replace",
replacement="****************",
level=DataLevel.L4
),
MaskingRule(
name="mobile",
pattern=r"1\\d{10}",
method="truncate",
replacement="*",
level=DataLevel.L3,
keep_prefix=1,
keep_suffix=4
),
]
engine = MaskingEngine(rules)
text = "客户张三，身份证 110101199001011234，电话 13800138000"
masked = engine.mask(text, DataLevel.L2)
print(masked)
# 输出：客户**，身份证 ****************，电话 1****8000