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` 的原始语义
仍然是"命中即拦"，豁免是在匹配**之后**按语境判定。这样：
- 规则的**入库/审核**流程不变（合规人员照常录入）；
- 判定变严的只有"明显在禁止或谈论该表述"的场景，且**有测试固定**。

## C-04 补充的两类豁免（**放宽方向**，与上面「宁可误拦」相反）

1. **短否定式**（`SHORT_NEGATION_CUES`）：`不保本` / `非保本` 这类线索**跨越命中词本身**，
   窗口判定读不到（命中的「保本」之前只剩一个「不」）。改为**与命中位置重叠**判定，
   只有压住命中词的那一次才生效 —— 避免「前文否定 → 后文真实承诺被放过」。
2. **指标名 / 概念词**（`FACTUAL_TERMS`，目前为 `安全` / `年化收益率`）：它们本身是合规的
   业务名词（「资金安全」「七日年化收益率」天天出现在正式披露里），命中不等于违规。
   判据与输入侧句式判定同构：**同句内出现承诺性线索（`CLAIM_CUES`）才算违规**。
   刻意**不含 `预期收益率`** —— 该词是明令禁用的营销表述、输入侧也未放行它，保持命中即拦。

⚠️ 这两条是**放宽**方向，因此：覆盖它们的测试必须同时给出**反例**（真实承诺仍被拦）；
   `CLAIM_CUES` 的失效方向是偏严（漏收线索只会更容易判违规），可以按需增长。
"""

from __future__ import annotations

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

#: 前置否定/禁止线索：出现在命中词**之前**的窗口内即视为"在禁止或否定该表述"。
NEGATION_CUES: tuple[str, ...] = (
    "严禁", "禁止", "不得", "不可", "不允许", "不准", "杜绝", "严禁使用",
    "避免", "勿", "拒绝", "防止", "防范", "打击", "处罚", "违规",
    "不承诺", "不保证", "不作", "不能", "绝不", "从不",
    "且不", "并不", "未保证", "不予保证",
    # C-04 补充：否定**存在**的短式。`预期收益率` 不是指标名（理由见 `FACTUAL_TERMS` 说明），
    # 仍按「命中即拦」处理，靠这条线索豁免「净值型产品是没有固定预期收益率…」这类定义性表述。
    # 刻意**不收**单字或两字的「没有」—— 它会在同一句里把后面的真实承诺一并豁免
    # （实测反例：「这款产品没有风险，保本保收益」）。
    "没有固定",
)

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

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

#: 短否定式线索（`C-04` DoD ①）：这类线索**跨越命中词本身**，例如 `不保本` 里的 `保本`。
#: 窗口判定读不到它 —— 命中词之前的窗口里只剩一个「不」，线索被命中词截断。
#: 因此改按**与命中位置重叠**判定（见 `_short_negation_overlaps`），且只收明确的否定短语、
#: 不收单字「不」（单字会把「不可保本？」这类边界写法也放过）。
SHORT_NEGATION_CUES: tuple[str, ...] = (
    "不保本", "不保收益", "非保本", "并非保本", "不确保", "不担保",
)

#: 指标名 / 概念词（`C-04` DoD ②）：它们**本身是合规的业务名词**，命中不等于违规。
#: 判据与输入侧句式判定同构：**同句内出现承诺性线索（`CLAIM_CUES`）才算违规**，
#: 否则视为事实性提及（「资金安全」「七日年化收益率」天天出现在正式披露里）。
#:
#: 为什么必须成对处理：输入侧已放行「什么是年化收益率」「你们的平台安全吗」这类概念题，
#: 若输出侧仍按裸词拦，检索回来的**正确答案**会被整条替换成转人工 —— 输入侧的收益被输出侧抹掉。
#: 所以口径是「**输入侧放行了哪个概念词，输出侧就必须同口径放行**」。
#:
#: 刻意**不含 `预期收益率`**：该词是明令禁用的营销表述，且输入侧并未放行它
#: （它不在含义题句式里），故输出侧**保持命中即拦** —— 收紧方向不失守。
FACTUAL_TERMS: tuple[str, ...] = ("安全", "年化收益率")

#: 承诺性线索：与输入侧的承诺词表同一套语义，另加**收益可实现性断言**与**推销/比较口吻**。
#: ⚠️ 失效方向是**偏严**（漏收线索 = 更容易判违规），与「白名单漏词 = 更容易放行」正好相反，
#: 所以这份表可以按需增长而不牺牲安全。
CLAIM_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 _short_negation_overlaps(text: str, index: int, length: int) -> bool:
    """短否定式是否**与命中位置重叠**（如 `不保本` 中的 `保本`）。

    窗口判定做不到这件事：命中词之前的窗口里只剩「不」，`不保本` 这个线索被命中词截断。
    判据是「线索与命中词**相接或重叠**」：`不保本` 压住命中词，`不确保` / `不担保` 紧贴在它
    之前（中间不得有标点或别的字）。这样既读得到跨词线索，又不会因为「前文出现过否定」就放过
    后文的真实承诺 —— 实测反例「不保本的产品也很多，这只基金保本」仍然判违规。
    """
    for cue in SHORT_NEGATION_CUES:
        start = 0
        while True:
            position = text.find(cue, start)
            if position < 0:
                break
            if position <= index + length and index <= position + len(cue):
                return True
            start = position + 1
    return False


def _is_factual_term_mention(text: str, index: int, pattern: str) -> bool:
    """指标名 / 概念词在**同句内没有承诺性线索**时，视为事实性提及（豁免）。"""
    if pattern not in FACTUAL_TERMS:
        return False
    before, after = _window(text, index, len(pattern))
    local = before + pattern + after
    return not any(cue in local for cue in CLAIM_CUES)


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

    判定顺序：短否定式（重叠）→ 引号 → 否定线索 → 询问语境 → 元语言线索 → 指标名句式判定。
    **只有最后一条是放宽方向**（`C-04` DoD ②），前面几条都是既有的收紧项；顺序如此是为了让
    「不得使用『安全』等误导性表述」这类**引用语境**优先成立，不让指标名判定抢先给结论。
    保守口径不变：没有明确线索一律**不**豁免。
    """
    if _short_negation_overlaps(text, index, len(pattern)):
        return True
    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
    # 元语言线索：前后任一侧出现"表述/话术/字样"等，说明在谈论该词本身。
    if any(cue in before or cue in after for cue in METALANGUAGE_CUES):
        return True
    return _is_factual_term_mention(text, index, pattern)


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