baodan/api/insurance/ppt/masking.py
wsb1224 b8c4e8b672 主要完成内容:
修复 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

220 lines
6.9 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""脱敏工具模块。
提供名称脱敏功能,用于 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 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
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