179 lines
5.5 KiB
Python
179 lines
5.5 KiB
Python
"""脱敏工具模块。
|
||
|
||
提供名称脱敏功能，用于 PPT/海报导出时替换真实保司和产品名称。
|
||
"""
|
||
import re
|
||
import logging
|
||
from difflib import SequenceMatcher
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 保司/产品名称中应保留的常见后缀
|
||
_PRESERVE_SUFFIXES = [
|
||
"保险", "人寿", "财险", "资产", "金融", "集团", "控股",
|
||
"储蓄保险计划", "保险计划", "储蓄计划", "保障计划", "危疾保障",
|
||
"终身寿险", "定期寿险", "万用寿险",
|
||
]
|
||
|
||
|
||
def fallback_mask_name(name: str) -> str:
|
||
"""兜底脱敏规则：对未配置脱敏字段的名称进行智能脱敏。
|
||
|
||
规则：
|
||
- 长度 <= 2：保留首字，后面用 X
|
||
- 长度 3-5：保留首尾，中间用 X
|
||
- 长度 > 5：保留前 1-2 个关键字，替换中间 1-2 字
|
||
- 引号、括号、后缀（保险/计划等）尽量保留
|
||
"""
|
||
if not name or len(name.strip()) <= 1:
|
||
return name
|
||
|
||
name = name.strip()
|
||
|
||
# 处理带引号的产品名，如「财富盈活」储蓄保险计划
|
||
quote_match = re.match(r'^([「『"\'（(].*?[」』"\'）)])\s*(.*)$', name)
|
||
if quote_match:
|
||
inner = quote_match.group(1)
|
||
suffix = quote_match.group(2)
|
||
# 对引号内部分脱敏
|
||
inner_clean = inner[1:-1] # 去掉引号
|
||
masked_inner = _mask_core(inner_clean)
|
||
return f"{inner[0]}{masked_inner}{inner[-1]}{suffix}"
|
||
|
||
# 普通名称
|
||
return _mask_core(name)
|
||
|
||
|
||
def _mask_core(text: str) -> str:
|
||
"""对核心文字进行脱敏。"""
|
||
if len(text) <= 1:
|
||
return text
|
||
if len(text) == 2:
|
||
return text[0] + "X"
|
||
if len(text) <= 5:
|
||
return text[0] + "X" + text[-1]
|
||
# 长度 > 5：保留前 2 字和后 2 字，中间用 X 替换
|
||
return text[:2] + "X" + text[-2:]
|
||
|
||
|
||
def apply_company_mask(company_dict: dict, use_masked: bool) -> dict:
|
||
"""对公司信息字典应用脱敏。
|
||
|
||
优先使用 maskedDisplayName，否则使用兜底脱敏规则。
|
||
"""
|
||
if not use_masked:
|
||
return company_dict
|
||
|
||
masked_name = company_dict.get("maskedDisplayName", "")
|
||
if masked_name:
|
||
company_dict["displayName"] = masked_name
|
||
else:
|
||
original = company_dict.get("displayName", "")
|
||
if original:
|
||
company_dict["displayName"] = fallback_mask_name(original)
|
||
|
||
return company_dict
|
||
|
||
|
||
def apply_product_mask(product_dict: dict, use_masked: bool) -> dict:
|
||
"""对产品信息字典应用脱敏。"""
|
||
if not use_masked:
|
||
return product_dict
|
||
|
||
masked_name = product_dict.get("maskedDisplayName", "")
|
||
if masked_name:
|
||
product_dict["displayName"] = masked_name
|
||
else:
|
||
original = product_dict.get("displayName", "")
|
||
if original:
|
||
product_dict["displayName"] = fallback_mask_name(original)
|
||
|
||
return product_dict
|
||
|
||
|
||
def mask_text(text: str, replacements: dict[str, str]) -> str:
|
||
"""对文本中的名称进行替换。
|
||
|
||
参数:
|
||
text: 需要替换的文本
|
||
replacements: {真实名: 脱敏名} 映射
|
||
"""
|
||
if not text or not replacements:
|
||
return text
|
||
for real_name, masked_name in replacements.items():
|
||
if real_name and masked_name and real_name != masked_name:
|
||
text = text.replace(real_name, masked_name)
|
||
return text
|
||
|
||
|
||
def build_name_replacements(companies: list[dict] = None, products: list[dict] = None,
|
||
use_masked: bool = False) -> dict[str, str]:
|
||
"""构建名称替换映射。
|
||
|
||
返回: {真实名: 脱敏名} 字典
|
||
"""
|
||
if not use_masked:
|
||
return {}
|
||
|
||
replacements = {}
|
||
|
||
if companies:
|
||
for c in companies:
|
||
real = c.get("displayName", "")
|
||
masked = c.get("maskedDisplayName", "")
|
||
if not masked:
|
||
masked = fallback_mask_name(real) if real else ""
|
||
if real and masked and real != masked:
|
||
replacements[real] = masked
|
||
# 也处理中文名
|
||
real_zh = c.get("nameZh", "")
|
||
if real_zh and real_zh != real and masked:
|
||
replacements[real_zh] = masked
|
||
|
||
if products:
|
||
for p in products:
|
||
real = p.get("displayName", "")
|
||
masked = p.get("maskedDisplayName", "")
|
||
if not masked:
|
||
masked = fallback_mask_name(real) if real else ""
|
||
if real and masked and real != masked:
|
||
replacements[real] = masked
|
||
|
||
return replacements
|
||
|
||
|
||
def match_product_by_name(product_name: str, all_products: list[dict]) -> dict | None:
|
||
"""通过名称和别名匹配产品表中的产品。
|
||
|
||
返回匹配到的产品 dict 或 None。
|
||
"""
|
||
if not product_name or not all_products:
|
||
return None
|
||
|
||
product_name_lower = product_name.strip().lower()
|
||
|
||
for p in all_products:
|
||
# 精确匹配
|
||
if p.get("displayName", "").strip().lower() == product_name_lower:
|
||
return p
|
||
# 别名匹配
|
||
aliases = p.get("aliases", [])
|
||
if isinstance(aliases, list):
|
||
for alias in aliases:
|
||
if isinstance(alias, str) and alias.strip().lower() == product_name_lower:
|
||
return p
|
||
|
||
# 模糊匹配：相似度 > 0.7
|
||
best_match = None
|
||
best_score = 0.0
|
||
for p in all_products:
|
||
name = p.get("displayName", "")
|
||
if not name:
|
||
continue
|
||
score = SequenceMatcher(None, product_name_lower, name.strip().lower()).ratio()
|
||
if score > best_score and score > 0.7:
|
||
best_score = score
|
||
best_match = p
|
||
|
||
return best_match
|