"""发送前合规终审(工作台本地独立校验,不依赖 Agent 返回值)。 PRD §4.5.2 / §4.8:发送环节必须再次独立执行全套校验(敏感词 + 适当性 + 免责声明), 不可完全依赖 Agent 保存阶段的结果。三检全过才放行,任一不过抛异常拦截(fail-closed)。 """ from __future__ import annotations from common_const import DISCLAIMER_TEXT from service.advisor.suitability import check_suitability from utils.exceptions import ForbiddenError, NotSuitableError def check_disclaimer(content: str | None) -> bool: """免责声明完整性:正文必须「逐字包含」完整原文(精确子串匹配)。""" return bool(content) and DISCLAIMER_TEXT in content def check_sensitive_words(content: str, words: list[str]) -> list[str]: """敏感词扫描:返回命中的词列表(空列表=通过)。""" text = content or "" return [w for w in words if w and w in text] def review_send( *, content: str | None, title: str | None, customer_risk: str | None, product_risks: list[str | None], sensitive_words: list[str], ) -> None: """发送终审主入口:三项校验,任一不过抛异常。 - 敏感词:命中即 ForbiddenError(403); - 适当性:任一建议产品风险高于客户风险即 NotSuitableError(1005); - 免责声明:正文未逐字包含完整原文即 ForbiddenError(403)。 """ # 1) 敏感词(标题 + 正文合并扫描) hit = check_sensitive_words(f"{title or ''}\n{content or ''}", sensitive_words) if hit: raise ForbiddenError(f"报告包含敏感词:{'、'.join(hit[:5])}") # 2) 适当性:对建议清单中每个产品风险等级逐一校验 for pr in product_risks: result = check_suitability(customer_risk, pr) if not result["ok"]: raise NotSuitableError(result["reason"]) # 3) 免责声明完整性(精确匹配;编辑改动导致不匹配即拦截) if not check_disclaimer(content): raise ForbiddenError("报告缺少免责声明,禁止发送")