baodan/api/insurance/ppt/masking.py

220 lines
6.9 KiB
Python
Raw Normal View History

2026-07-28 16:45:14 +08:00
"""脱敏工具模块。
提供名称脱敏功能用于 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
主要完成内容: 修复 PPT 异步任务无法生成的问题,包括任务变量引用错误、失败状态回写、心跳缺失任务恢复。 脱敏改为保司/产品后台统一配置,生成端不再让用户选择;任务创建时保存策略快照。 保司支持独立控制 PPT、海报 Logo 显示。 PPT 核验新增吸烟状态、币种及三个条件字段。 利益演示、退保提取调整为警告,不再阻止生成。 PPT 生成完成后可以直接返回数据核验页修改。 建立不同险种、单图/长图共六套海报字段画像。 PPT“生成场景”支持后台新增、启停和删除。 保司、产品、PPT 模板、文案模板均支持安全删除。 内置模板禁止删除,只允许停用;存在关联数据时拒绝危险删除。 补充策略变更及删除审计日志。 更新 API 文档、部署文档及修复计划实施记录。 关键交付文件: [数据库迁移 migrate_027.py](D:/work/code/python/coding/baodanagent/api/insurance/db/migrate_027.py) [海报字段画像 field_profiles.py](D:/work/code/python/coding/baodanagent/api/insurance/poster/field_profiles.py) [动态场景服务 scenarios.py](D:/work/code/python/coding/baodanagent/api/insurance/ppt/scenarios.py) [新增回归测试](D:/work/code/python/coding/baodanagent/tests/ppt_poster_optimization_test.py) [优化修复计划书](D:/work/code/python/coding/baodanagent/docs/保险智能客服系统_PPT与海报优化修复计划书_20260731.md) 验证结果: 核心链路测试:37 passed,1 skipped 扩展回归测试:140 passed PPT 渲染器测试:6 passed 前端生产构建:通过 Python 编译检查:通过 完整测试集:190 passed,1 failed 唯一失败为 tests/test_chat_save.py::test_chat_logs_query 未建立 Flask application context,与本次 PPT/海报链路无关。
2026-07-31 14:10:24 +08:00
def build_brand_policy(company: dict | None, products: list[dict] | None) -> dict:
"""从后台配置生成不可由用户覆盖的品牌策略快照。"""
company = company or {}
products = products or []
return {
"companyMaskingEnabled": bool(company.get("maskingEnabled")),
"productMaskingById": {
str(product.get("id")): bool(product.get("maskingEnabled"))
for product in products
if product.get("id")
},
"logoEnabled": bool(company.get("logoEnabled", True)),
"policyVersion": 1,
}
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))
apply_company_mask(company, bool(policy.get(
"companyMaskingEnabled",
company.get("maskingEnabled"),
)))
if not bool(policy.get("logoEnabled", company.get("logoEnabled", True))):
company["logoUrl"] = ""
return company, product
2026-07-28 16:45:14 +08:00
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