wsb1224
ed9f0327d2
OCR 表格模式改为更适合计划书的 PSM 4。 分别支持斜杠年度、独立年度/年龄列、纵向压缩表格、保证/非保证双栏。 防止错误的坐标解析结果覆盖正确 OCR 数据。 身故利益表与退保价值表按保单年度合并。 修复吸烟状态、保额、年缴/单缴金额和缴费年期。 补齐 SIUL3、SBIUL2、GIUL3、FWD IF、AIA PIL2 产品配置。 多份计划书现在保留全部保司，不再只取第一家公司。 用户端保司选项、生成快照和渲染过程统一使用后台脱敏名
266 lines
8.9 KiB
Python
266 lines
8.9 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 public_company_option(company: dict) -> dict:
|
||
"""生成用户端保司选项，避免把原始名称和别名发送到浏览器。"""
|
||
item = apply_company_mask(dict(company or {}), bool((company or {}).get("maskingEnabled")))
|
||
return {
|
||
"id": item.get("id", ""),
|
||
"displayName": item.get("displayName", ""),
|
||
}
|
||
|
||
|
||
def public_product_option(product: dict) -> dict:
|
||
"""生成用户端产品选项，仅暴露选择所需字段。"""
|
||
item = apply_product_mask(dict(product or {}), bool((product or {}).get("maskingEnabled")))
|
||
return {
|
||
"id": item.get("id", ""),
|
||
"companyId": item.get("companyId", ""),
|
||
"planType": item.get("planType", ""),
|
||
"displayName": item.get("displayName", ""),
|
||
}
|
||
|
||
|
||
def build_brand_policy(company: dict | list[dict] | None, products: list[dict] | None) -> dict:
|
||
"""从后台配置生成不可由用户覆盖的品牌策略快照。"""
|
||
companies = company if isinstance(company, list) else ([company] if company else [])
|
||
primary_company = companies[0] if companies else {}
|
||
products = products or []
|
||
company_policy_by_id = {}
|
||
for item in companies:
|
||
company_id = str(item.get("id") or "")
|
||
if not company_id:
|
||
continue
|
||
masking_enabled = bool(item.get("maskingEnabled"))
|
||
display_name = item.get("displayName", "")
|
||
if masking_enabled:
|
||
display_name = item.get("maskedDisplayName") or fallback_mask_name(display_name)
|
||
company_policy_by_id[company_id] = {
|
||
"maskingEnabled": masking_enabled,
|
||
"displayName": display_name,
|
||
"logoEnabled": bool(item.get("logoEnabled", True)),
|
||
}
|
||
return {
|
||
# 保留旧字段，兼容历史任务和海报链路。
|
||
"companyMaskingEnabled": bool(primary_company.get("maskingEnabled")),
|
||
"productMaskingById": {
|
||
str(product.get("id")): bool(product.get("maskingEnabled"))
|
||
for product in products
|
||
if product.get("id")
|
||
},
|
||
"logoEnabled": bool(primary_company.get("logoEnabled", True)),
|
||
"companyPolicyById": company_policy_by_id,
|
||
"policyVersion": 2,
|
||
}
|
||
|
||
|
||
def apply_brand_policy(
|
||
company: dict | None,
|
||
product: dict | None,
|
||
policy: dict | None = None,
|
||
) -> tuple[dict, dict]:
|
||
"""把品牌策略应用到保司和单个产品字典副本。"""
|
||
company = dict(company or {})
|
||
product = dict(product or {})
|
||
policy = policy or build_brand_policy(company, [product])
|
||
|
||
product_id = str(product.get("id") or "")
|
||
product_masking = (policy.get("productMaskingById") or {}).get(
|
||
product_id,
|
||
bool(product.get("maskingEnabled")),
|
||
)
|
||
apply_product_mask(product, bool(product_masking))
|
||
company_id = str(company.get("id") or "")
|
||
company_policy = (policy.get("companyPolicyById") or {}).get(company_id) or {}
|
||
company_masking = company_policy.get(
|
||
"maskingEnabled",
|
||
policy.get("companyMaskingEnabled", company.get("maskingEnabled")),
|
||
)
|
||
if company_policy.get("displayName"):
|
||
company["maskedDisplayName"] = company_policy["displayName"]
|
||
apply_company_mask(company, bool(company_masking))
|
||
logo_enabled = company_policy.get(
|
||
"logoEnabled",
|
||
policy.get("logoEnabled", company.get("logoEnabled", True)),
|
||
)
|
||
if not bool(logo_enabled):
|
||
company["logoUrl"] = ""
|
||
return company, product
|
||
|
||
|
||
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
|