group_fqcd_jr/app/core/compliance_context.py

"""零容忍词匹配的**语境豁免**：区分"作出承诺"与"禁止承诺/事实性表述"。

## 为什么需要这个模块

治理层的零容忍规则是**朴素子串匹配**（`pattern in output`），它无法区分：

- **作出承诺**（必须拦）："本产品保证收益"、"稳赚不赔"
- **禁止承诺**（不该拦）："严禁承诺保本保收益"、"不得使用『保证收益』等违规表述"
- **事实性表述**（不该拦）："安全防范设施"、"七日年化收益率 1.85%"

实测复现（真机 `customer_service` Agent）：用户问「基金销售有哪些合规要求」，
检索到的政策片段里含"严禁承诺保本保收益""禁止使用『保证收益』等违规表述"，
模型**忠实复述**了它们（未新增任何词），却被治理层整条替换成
"该内容需要人工核实…"，**答复长度从 1097 字掉到 83 字** —— 合规问答反而答不出来。

## 判定口径（保守优先）

**只有在命中位置附近出现明确的否定/禁止线索时才豁免**；否则一律按命中处理（宁可误拦）。

线索分两类：

1. **前置禁止词**：紧邻命中词之前出现（如 `严禁承诺保本`、`禁止使用保证收益`）。
2. **后置/前置的"引用语境"**：命中词被引号包裹（如 `禁止使用"保证收益"`），
   或紧邻出现"表述/话术/字眼/字样/宣传"等元语言词（说明在谈论这个词本身，而非作承诺）。

豁免窗口**严格限定在命中词前后各 `CONTEXT_WINDOW` 个字符内**，不倒查全文 ——
否则"前面某处提到过禁止，后面真的作出承诺"会被错误放过。

## 与"库内规则"的关系

本模块**只作用于代码侧治理路径**（`review_output`）。库内 `agent_negative_word` 的原始语义
仍然是"命中即拦"，豁免是在匹配**之后**按语境判定。这样：
- 规则的**入库/审核**流程不变（合规人员照常录入）；
- 判定变严的只有"明显在禁止或谈论该表述"的场景，且**有测试固定**。
"""

from __future__ import annotations

#: 命中词前后各看多少字符来判定语境。
CONTEXT_WINDOW = 12

#: 前置否定/禁止线索：出现在命中词**之前**的窗口内即视为"在禁止或否定该表述"。
NEGATION_CUES: tuple[str, ...] = (
    "严禁", "禁止", "不得", "不可", "不允许", "不准", "杜绝", "严禁使用",
    "避免", "勿", "拒绝", "防止", "防范", "打击", "处罚", "违规",
    "不承诺", "不保证", "不作", "不能", "绝不", "从不",
    "且不", "并不", "未保证", "不予保证",
)

#: 元语言线索：说明"在谈论这个词本身"而不是在作承诺（如『保证收益』等违规表述）。
METALANGUAGE_CUES: tuple[str, ...] = (
    "表述", "话术", "字眼", "字样", "宣传", "用语", "措辞", "提法", "说法",
    "等违规", "等禁用", "等禁止", "等表述",
)

#: 询问语境：问“能否/是否/会不会保证收益”是在确认或质疑产品属性，不是作出承诺。
#: 风控回访话术中常见此类问句，若按承诺拦截，会把正常人工回访脚本误判为违规输出。
QUESTION_CUES: tuple[str, ...] = (
    "能否", "是否", "可否", "会不会", "能不能", "请问", "吗",
)

#: 引号字符：命中词被引号包裹时视为引用（在谈论该表述本身）。
QUOTE_CHARS: tuple[str, ...] = ('"', "'", "“", "”", "‘", "’", "「", "」", "『", "』")

#: 句子边界：豁免线索**不得跨句生效**。
#: 反例（实测发现）：`"严禁承诺保本。但这只基金保本"` —— 第二个「保本」的前置窗口
#: 会读到前一句的「严禁」，若允许跨句就会把**真实承诺**放过，成为合规漏洞。
#: 因此线索只在与命中词**同一句内**才算数。
SENTENCE_BOUNDARIES: tuple[str, ...] = ("。", "！", "？", "；", "\n", ";", "!", "?")


def _same_sentence_before(text: str, index: int) -> str:
    """命中词之前、**同一句内**的文本（遇到最近的句末标点即截断）。"""
    start = max(0, index - CONTEXT_WINDOW)
    window = text[start:index]
    cut = max(window.rfind(ch) for ch in SENTENCE_BOUNDARIES)
    return window[cut + 1:] if cut >= 0 else window


def _same_sentence_after(text: str, index: int, length: int) -> str:
    """命中词之后、**同一句内**的文本（遇到最近的句末标点即截断）。"""
    end = min(len(text), index + length + CONTEXT_WINDOW)
    window = text[index + length:end]
    positions = [window.find(ch) for ch in SENTENCE_BOUNDARIES]
    positions = [p for p in positions if p >= 0]
    return window[:min(positions)] if positions else window


def _window(text: str, index: int, length: int) -> tuple[str, str]:
    """返回命中处的**同句**前置/后置窗口（各自不超过 `CONTEXT_WINDOW` 字符）。"""
    return _same_sentence_before(text, index), _same_sentence_after(text, index, length)


def _is_quoted(text: str, index: int, length: int) -> bool:
    """命中词是否被引号包裹（紧邻前一个字符与紧邻后一个字符都是引号）。"""
    if index == 0 or index + length >= len(text):
        return False
    return text[index - 1] in QUOTE_CHARS and text[index + length] in QUOTE_CHARS


def is_prohibited_context(text: str, index: int, pattern: str) -> bool:
    """命中位置是否处于"禁止/否定/引用"语境（是则豁免该次命中）。

    保守口径：任一明确线索命中即豁免；没有线索一律**不**豁免。
    """
    before, after = _window(text, index, len(pattern))
    if _is_quoted(text, index, len(pattern)):
        return True
    if any(cue in before for cue in NEGATION_CUES):
        return True
    if any(cue in after for cue in NEGATION_CUES):
        return True
    if any(cue in before or cue in after for cue in QUESTION_CUES):
        return True
    # 元语言线索：前后任一侧出现"表述/话术/字样"等，说明在谈论该词本身。
    return any(cue in before or cue in after for cue in METALANGUAGE_CUES)


def match_with_exemptions(text: str, pattern: str) -> bool:
    """`pattern` 在 `text` 中是否存在**未被豁免**的命中（True = 应判为违规）。

    所有命中处都被语境豁免时才返回 False（即"没有实质违规"）。
    """
    if not pattern:
        return False
    start = 0
    while True:
        index = text.find(pattern, start)
        if index < 0:
            return False
        if not is_prohibited_context(text, index, pattern):
            return True
        start = index + 1


def first_violation(text: str, patterns: tuple[str, ...]) -> str | None:
    """返回第一个**未被豁免**的命中词；无实质违规时返回 `None`。供测试与排障使用。"""
    for pattern in patterns:
        if match_with_exemptions(text, pattern):
            return pattern
    return None