group_fqcd_jr/tools/seed_compliance_baseline.py

r"""合规层种子数据：7 个零容忍负面词（+4 变体 + C-05 三类词 14 条）与 6 类固定话术。

C-05（`D2.1` §1.4 D-a）：在既有 11 行**之外追加**三类词 —— 收益比较 / 稀缺性 / 费率误导。
三条硬约束：① `applicable_agents` 非空且含客服 Agent（空数组 = 失败关闭跳过，规则永不生效）；
② **不得改写既有 11 行的 `word_pattern`**（追加项单独放在 `EXTRA_NEGATIVE_RULES`，既有元组
一字未动）；③ 按**承诺性短语**收录而非形容词。

⚠️ 为什么落点是**本文件**而不是另存一份数据（甲案，用户 2026-09-18 已拍板）：管理端
`negative-word-rules` 通道只能建 `draft`（须审核才 `active`），而本脚本是唯一可复现、可重跑
的装载器；另存第二份种子会让新环境**漏跑即静默缺失** —— 正是本项目反复踩过的坑。
`C-05` 的 DoD 写「不改任何 `.py`」，此处登记为「不改 `app/**` 与底座 `.py`」：本文件是
**数据装载器**，与 `D-b`（两条话术热线改定值）同源，追加的是数据行而不是拦截逻辑。

背景：`docs/02` §10.2/§10.3 要求这些数据，但迁移只建表、没有 seed。而
`app/service/agent/governance.py` L62-65 的取数条件是
`status='active' AND reviewer_id IS NOT NULL AND reviewed_at IS NOT NULL`，
两张表 0 行就意味着一**条规则都查不到**：客服 Agent 说"这只基金稳赚"不会被拦，
免责声明也不会出现在任何回复里。合规过滤是静默失效的——不报错，只是不生效。

取值来源（不臆造，均以真实 DDL 为准）：
- `agent_negative_word.match_type` CHECK：`exact` / `contains` / `regex`，
  这里只用 `contains`；`regex` 被 governance.py L75-77 显式拒绝
  （Python `re` 无执行超时，底座不跑未经约束的管理员正则）。
- `agent_negative_word.severity` CHECK：`warn` / `regenerate` / `block`，
  零容忍词一律 `block`。
- `agent_negative_word.category` **无 CHECK 约束**，按语义取值（《理财销售办法》
  的口径：误导性 / 绝对化 / 承诺 / 收益）。
- `agent_reply_template.scene` CHECK：`disclaimer` / `low_confidence` /
  `compliance_block` / `transfer` / `model_failure` / `system_busy` / `clarification`。
- `agent_reply_template.active_key` 是 **生成列**（`status='active'` 时等于
  `template_code:locale`，否则 NULL）：不能写、也不该写，由库自己算。
  它上面有唯一键 `uk_reply_template_active_one`，所以同一 template_code 只能有一条 active。

幂等：按 `rule_code`（唯一键 `uk_negative_rule_code`）与
`template_code`+`version`（唯一键 `uk_reply_template_version`）upsert，可重复执行；
重复执行不会新增行，只会把已存在的行重新置为 active 并刷新审核信息。

执行：`.\.venv\Scripts\python.exe tools\seed_compliance_baseline.py`（可重复执行）
"""

from __future__ import annotations

import asyncio
import json
import sys
from datetime import UTC, datetime
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from sqlalchemy import text  # noqa: E402

from app.infrastructure.db import SessionFactory  # noqa: E402

#: 审核人与创建人：`sys_user.id=9003`（T-ADMIN，由 tools/seed_test_rbac.py 种入）。
#: `created_by`/`reviewer_id` 都有指向 `sys_user` 的外键，写不存在的用户会直接失败。
ADMIN_ID = 9003

#: 只对客服 Agent 生效，避免影响其他 Agent 的表达自由度。
APPLICABLE_AGENTS = '["customer_service"]'

#: 命中后统一的合规拒答话术（就是本脚本要种的 6 条之一，所以话术先插）。
SAFE_REPLY_TEMPLATE_CODE = "TPL_COMPLIANCE_REFUSAL"

#: `docs/02` §10.2 逐字要求的 7 个零容忍词。前 7 条规则必须恰好覆盖它们。
REQUIRED_ZERO_TOLERANCE_WORDS = (
    "保本", "稳赚", "无风险", "保证收益", "预期收益率", "年化收益率", "安全",
)

#: (rule_code, word_pattern, category, reason)
#: 前 7 条是 `docs/02` §10.2 逐字要求的零容忍词，一个都不能少；
#: 后 4 条是专项设计 §6.5 评审建议的变体（换个说法绕过关键词的常见写法）。
NEGATIVE_RULES: tuple[tuple[str, str, str, str], ...] = (
    ("NEG-001", "保本", "guarantee", "资管新规后非保本"),
    ("NEG-002", "稳赚", "misleading", "误导性表述"),
    ("NEG-003", "无风险", "absolute", "绝对化表述"),
    ("NEG-004", "保证收益", "guarantee", "禁止刚兑承诺"),
    ("NEG-005", "预期收益率", "yield_claim", "《理财销售办法》明文禁止"),
    ("NEG-006", "年化收益率", "yield_claim", "监管处罚点名措辞"),
    ("NEG-007", "安全", "absolute", "绝对化表述"),
    ("NEG-008", "零风险", "absolute", "绝对化表述变体"),
    ("NEG-009", "稳赚不赔", "misleading", "误导性表述变体"),
    ("NEG-010", "躺着赚", "misleading", "误导性表述变体"),
    ("NEG-011", "坐享收益", "misleading", "误导性表述变体"),
)

#: C-05 追加的三类词（`D2.1` §1.4 D-a 硬约束 ③：**承诺性短语**，不按形容词收录）。
#:
#: 入库前做过**误伤体检**：对 `knowledge/_chunks.jsonl` 逐词计数，下列 14 个词**全部 0 命中**，
#: 且与 6 条话术零自绊（`verify()` 会强制复查）。两个刻意的排除：
#:   · **剔除 `X 折优惠`** —— 语料里「1 折优惠」「免申购费」是**公司自己的合法费率事实**
#:     （FAQ「申购和赎回的费率是多少」的答案正文就在写它），收进来会把费率问答整条拦成
#:     转人工，与「放宽答不上来时的去向」正好相反。
#:   · **不收「高收益」「稳健」「低风险」这类形容词** —— 它们是风险提示里的正常用词
#:     （「高收益伴随高风险」会被一并拦下）。
EXTRA_NEGATIVE_RULES: tuple[tuple[str, str, str, str], ...] = (
    ("NEG-012", "承诺高收益", "yield_compare", "收益比较：以承诺口吻作横向比较"),
    ("NEG-013", "保证高收益", "yield_compare", "收益比较：以保证口吻作横向比较"),
    ("NEG-014", "比同类收益高", "yield_compare", "收益比较：与同类产品比较"),
    ("NEG-015", "收益高于同类", "yield_compare", "收益比较：与同类产品比较"),
    ("NEG-016", "跑赢同类", "yield_compare", "收益比较：比较性表述"),
    ("NEG-017", "稀缺额度", "scarcity", "稀缺性：以额度稀缺促单"),
    ("NEG-018", "额度有限", "scarcity", "稀缺性：以额度有限促单"),
    ("NEG-019", "仅剩", "scarcity", "稀缺性：以示数量促单"),
    ("NEG-020", "优先认购权", "scarcity", "稀缺性：暗示优先资格"),
    ("NEG-021", "优先配置权", "scarcity", "稀缺性：暗示优先资格"),
    ("NEG-022", "零费率", "fee_misleading", "费率误导：宣称零费率"),
    ("NEG-023", "费率最低", "fee_misleading", "费率误导：最低化宣称"),
    ("NEG-024", "手续费全免", "fee_misleading", "费率误导：宣称全免"),
    ("NEG-025", "全网最低", "fee_misleading", "费率误导：全市场最低宣称"),
)

#: C-05 三类词必须齐备的类目（`verify()` 逐类检查至少 1 条 active）。
C05_CATEGORIES: tuple[str, ...] = ("yield_compare", "scarcity", "fee_misleading")

#: 真正落库的全部规则 = 既有 11 条 + C-05 追加 14 条。
ALL_NEGATIVE_RULES = NEGATIVE_RULES + EXTRA_NEGATIVE_RULES

#: 规则必须挂到的 Agent（D-a 硬约束 ①：空数组 = 规则永不生效，且是「假成功」）。
REQUIRED_APPLICABLE_AGENT = "customer_service"

#: (template_code, scene, title, content_text)
#: scene 取自 `chk_template_scene` 的 7 个合法值里的 6 个，覆盖 6 类固定话术。
#: TPL_AI_NOTICE 用 `clarification`（AI 生成标识属主动澄清披露）；
#: TPL_COMPLIANCE_REFUSAL 用 `compliance_block`（合规拦截后替换的安全话术）。
#:
#: ⚠️ **话术文本本身不得命中 NEGATIVE_RULES 里的任何词**（`verify()` 会强制校验）。
#: 治理层（governance.py L127-128）对 `contains` 是**朴素子串匹配**，没有否定式豁免：
#: 旧文案里的「非保本」会命中「保本」规则，形成"合规话术被自己的规则拦截"的自绊，
#: Task 2 接入话术后就会变成"替换文本又被过滤"的循环。因此合规语义必须用
#: **不含禁用字面**的说法表达（"不承诺本金不受损失" 而不是 "非保本"）。
REPLY_TEMPLATES: tuple[tuple[str, str, str, str], ...] = (
    ("TPL_COMPLIANCE_REFUSAL", "compliance_block", "合规拒答",
     "根据监管要求，我不能对收益做出任何承诺，也不承诺本金不受损失。"
     "本产品收益可能为负，请以产品说明书为准。"),
    ("TPL_DISCLAIMER", "disclaimer", "固定免责声明",
     "本内容仅为投资分析参考，不构成任何直接投资建议，不构成对任何产品的收益承诺，"
     "据此操作风险自负，请谨慎对待。"),
    # ✅ `W27` / `DEC-W27-10`：**轻量提示**。门禁 `F5` 分档后，非业务答复
    # （闲聊 / 联系方式 / 引导登录 / 澄清）挂这一句，业务答复仍挂 `TPL_DISCLAIMER`。
    # 与代码兜底 `app/service/agent/governance.FALLBACK_LIGHT_NOTICE` **必须同值**：
    # 一个走库、一个走代码，值不一致时"哪句生效"取决于数据库可不可用（已踩过一次）。
    ("TPL_DISCLAIMER_LIGHT", "disclaimer", "轻量提示（非业务答复）",
     "以上为一般性说明，仅供参考，具体以产品说明书与官方披露为准。"),
    ("TPL_AI_NOTICE", "clarification", "AI 生成标识",
     "本回答由 AI 生成，仅供参考。"),
    ("TPL_LOW_CONFIDENCE", "low_confidence", "低置信兜底",
     "抱歉，我暂时无法准确回答您的问题，建议您转接人工客服获取更准确的帮助。"),
    ("TPL_TRANSFER_HUMAN", "transfer", "转人工提示",
     "已为您转接人工客服，服务时间为每日 7:00—22:00，客服电话 400-889-8899。"),
    ("TPL_SYSTEM_BUSY", "system_busy", "系统繁忙/模型故障",
     "系统繁忙，暂时无法回答，请稍后重试或联系人工客服 400-889-8899。"),
)

#: template_code -> content_text，供 `verify()` 做自绊检查（须定义在 REPLY_TEMPLATES 之后）。
CONTENT_BY_TEMPLATE_CODE = {code: content for code, _scene, _title, content in REPLY_TEMPLATES}

INSERT_NEGATIVE_RULE = text(
    "INSERT INTO agent_negative_word"
    " (rule_code, word_pattern, match_type, category, severity, applicable_agents,"
    "  safe_reply_template_code, status, version, created_by, reviewer_id, reviewed_at,"
    "  created_at, updated_at)"
    " VALUES (:rule_code, :word, 'contains', :category, 'block', :agents,"
    "  :safe_code, 'active', 1, :admin, :admin, :now, :now, :now)"
    " ON DUPLICATE KEY UPDATE word_pattern=:word, match_type='contains', category=:category,"
    "  severity='block', applicable_agents=:agents, safe_reply_template_code=:safe_code,"
    "  status='active', reviewer_id=:admin, reviewed_at=:now, updated_at=:now"
)

INSERT_REPLY_TEMPLATE = text(
    "INSERT INTO agent_reply_template"
    " (template_code, scene, title, content_text, variables, locale, version, status,"
    "  created_by, reviewer_id, reviewed_at, created_at, updated_at)"
    " VALUES (:code, :scene, :title, :content, '{}', 'zh-CN', 1, 'active',"
    "  :admin, :admin, :now, :now, :now)"
    " ON DUPLICATE KEY UPDATE scene=:scene, title=:title, content_text=:content,"
    "  status='active', reviewer_id=:admin, reviewed_at=:now, updated_at=:now"
)


async def seed() -> None:
    """幂等落库。话术先于负面词插入：负面词会引用话术 code。"""
    now = datetime.now(UTC).replace(tzinfo=None)
    async with SessionFactory() as session, session.begin():
        for code, scene, title, content in REPLY_TEMPLATES:
            await session.execute(INSERT_REPLY_TEMPLATE, {
                "code": code, "scene": scene, "title": title, "content": content,
                "admin": ADMIN_ID, "now": now,
            })
        for rule_code, word, category, _reason in ALL_NEGATIVE_RULES:
            await session.execute(INSERT_NEGATIVE_RULE, {
                "rule_code": rule_code, "word": word, "category": category,
                "agents": APPLICABLE_AGENTS, "safe_code": SAFE_REPLY_TEMPLATE_CODE,
                "admin": ADMIN_ID, "now": now,
            })


async def verify() -> None:
    """按 governance.py 的**同一口径**复查，确认真的能被查到。"""
    async with SessionFactory() as session:
        rules = (await session.execute(text(
            "SELECT rule_code, word_pattern, category, severity, match_type, applicable_agents"
            " FROM agent_negative_word"
            " WHERE status='active' AND reviewer_id IS NOT NULL AND reviewed_at IS NOT NULL"
            " ORDER BY rule_code"
        ))).mappings().all()
        templates = (await session.execute(text(
            "SELECT template_code, scene, status FROM agent_reply_template"
            " WHERE status='active' AND reviewer_id IS NOT NULL AND reviewed_at IS NOT NULL"
            " ORDER BY template_code"
        ))).all()
    for row in rules:
        print(f"  rule {row['rule_code']} {row['word_pattern']!r}"
              f" category={row['category']} severity={row['severity']}"
              f" match_type={row['match_type']} agents={row['applicable_agents']}")
    for code, scene, status in templates:
        print(f"  tpl  {code} scene={scene} status={status}")
    # 种子层的自绊检查：话术不得命中自己这套规则。治理层是朴素子串匹配，
    # 没有否定式豁免——「非保本」这类写法会命中「保本」规则。
    # 在种子阶段就拦住，比等 Task 2 接入话术后再发现在替换结果里循环排查便宜得多。
    patterns = {row["word_pattern"] for row in rules}
    self_trips = [
        (code, word)
        for code, content in CONTENT_BY_TEMPLATE_CODE.items()
        for word in patterns
        if word in content
    ]
    if self_trips:
        raise SystemExit(f"话术命中了自己的禁用词（自绊）：{self_trips}")
    missing = {
        word for word in REQUIRED_ZERO_TOLERANCE_WORDS
        if not any(word in pattern for pattern in patterns)
    }
    if missing:
        raise SystemExit(f"零容忍词未生效：{sorted(missing)}")
    scene_count = len({scene for _, scene, _ in templates})
    if scene_count < 6:
        raise SystemExit(f"生效话术场景不足 6 类：{scene_count}")
    # D-a 硬约束 ①：`applicable_agents` 为空 = 治理层**失败关闭跳过**（规则永不生效）。
    # 这条最容易「看着种进去了、其实一条都没生效」，所以在种子阶段正面断言。
    ungated = [
        row["rule_code"] for row in rules
        if REQUIRED_APPLICABLE_AGENT not in {
            str(agent) for agent in (
                json.loads(row["applicable_agents"])
                if isinstance(row["applicable_agents"], str)
                else (row["applicable_agents"] or [])
            )
        }
    ]
    if ungated:
        raise SystemExit(f"规则未挂到客服 Agent（等于永不生效）：{ungated}")
    # C-05：三类词必须齐备 —— 否则「补了三类词」只是文档里的说法。
    active_categories = {row["category"] for row in rules}
    missing_categories = [c for c in C05_CATEGORIES if c not in active_categories]
    if missing_categories:
        raise SystemExit(f"C-05 三类词未齐备：{missing_categories}")
    print(f"verified: {len(rules)} active rules (含 7 个零容忍词), "
          f"{len(templates)} active templates covering {scene_count} scenes")


async def main() -> None:
    await seed()
    print(f"seeded {len(ALL_NEGATIVE_RULES)} negative rules"
          f"（既有 {len(NEGATIVE_RULES)} + C-05 {len(EXTRA_NEGATIVE_RULES)}），"
          f"{len(REPLY_TEMPLATES)} reply templates")
    await verify()


if __name__ == "__main__":
    asyncio.run(main())