ai-kefu-admin/scripts/breach_rate.py

"""职能边界破防率观测(只读): AI 回复中出现"价格形态数字"的轮次占比。

适配自 ai-kefu-jc/scripts/breach_rate.py(jc 全局库 → 多租户): 方案B
(输出过滤)的启动依据——>10% 即砌墙; 锚定式匹配, 年份/规格数字天然不
命中。与 boundary_filter 试金石(wall/test, 判定侧)互补, 本脚本观察
**漏放侧偏差**(线上 AI 回复实际漏出的价格/号码/邮箱形态); 误杀侧
(客户正常消息被墙误拦)不在本脚本口径, 用后台会话详情人工复核。

多租户适配: --tenant <slug> 必填, 经 platform.db 门禁(租户须已
bootstrap 注册, 同 import_legacy/archive 纪律——绝不摸未注册租户的库)
后只读打开 tenants/<slug>.db; 其余逻辑(模式表/统计口径)与 jc 原版
一致——含 Minor-4 的号码回显(≥10位)与邮箱回显观测。

用法(Windows 须 PYTHONUTF8=1, 输出含中文):
  PYTHONUTF8=1 python scripts/breach_rate.py --tenant <slug> \
      [--platform-db <路径>] [--tenants-dir <目录>]
"""
from __future__ import annotations

import argparse
import re
import sqlite3
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[1]

PATTERNS = [
    (r"\$\s?\d", "美元符"),
    (r"[€£]\s?\d", "欧元/英镑符"),
    (r"\d[\d,.]*\s*(usd|美元|万块?|块钱)", "货币单位"),
    (r"\d[\d,.]*\s*(d[óo]lares?|euros?\b(?!\s*pallets?)|рублей|руб\.?|"
     r"доллар(?:ов|а|ы)?)", "货币单位(多语言)"),
    (r"\b(?:precio?s?|costo?s?|custo?s?|preço?s?|цен[аы]?|стоимость|"
     r"(?:ال)?سعر|(?:ال)?أسعار|(?:ال)?ثمن)\b"
     r"[^\d\n]{0,20}(?<![a-z])\d", "价格名词前置(多语言)"),
    (r"(price|cost|quote)[^.\d]{0,20}(?<![a-z])(?!(?:19|20)\d\d(?!\d))\d",
     "价格动词邻接(年份排除)"),
    (r"\d[\d,.]*\s*k\b(?!\w)", "25k形式"),
    (r"\d+\s*[- ]?(day|week|month|d[ií]as?|semanas?|дн(?:ей|я)|недел[ьи]|"
     r"أيام)s?\b", "交期/时间承诺"),
    # Minor-4(审查修复): 号码回显漏放观测——与零号码墙 LONG_NUMBER_RE 同形
    # ({8,17} 修复后口径),AI 回复中 ≥10 位数字串(含分隔/阿语数字形态)
    (r"\+?\d[\d\s\-().]{8,17}\d", "号码回显(≥10位)"),
    (r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}", "邮箱回显"),
]
COMPILED = [(re.compile(p, re.I), name) for p, name in PATTERNS]


def _tenant_bootstrapped(platform_db: Path, slug: str) -> None:
    """门禁(同 archive.py): 租户须已在 platform.db 注册; 只读打开
    (先看后动, 门禁绝不写 platform.db)。"""
    if not platform_db.is_file():
        raise SystemExit(
            f"platform.db 不存在: {platform_db}\n"
            f"租户须先 bootstrap: 先跑 scripts/bootstrap_platform.py")
    try:
        con = sqlite3.connect(
            platform_db.resolve().as_uri() + "?mode=ro", uri=True)
        try:
            row = con.execute(
                "SELECT 1 FROM tenants WHERE slug = ?", (slug,)).fetchone()
        finally:
            con.close()
    except sqlite3.Error as exc:
        raise SystemExit(
            f"platform.db 读取失败({exc}): 先跑 scripts/bootstrap_platform.py"
        ) from exc
    if row is None:
        raise SystemExit(
            f"租户 {slug} 未在 platform.db 注册: "
            f"先跑 scripts/bootstrap_platform.py 导入该租户档案")


def main() -> None:
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--tenant", required=True,
                        help="租户 slug(须已 bootstrap 注册)")
    parser.add_argument("--platform-db",
                        default=str(REPO_ROOT / "data" / "platform.db"),
                        help="platform.db 路径(默认 <repo>/data/, 门禁查"
                             "租户是否已 bootstrap)")
    parser.add_argument("--tenants-dir",
                        default=str(REPO_ROOT / "tenants"),
                        help="租户库目录(默认 <repo>/tenants, 只读打开"
                             "<slug>.db)")
    args = parser.parse_args()

    platform_db = Path(args.platform_db)
    tenants_dir = Path(args.tenants_dir)
    _tenant_bootstrapped(platform_db, args.tenant)
    db = tenants_dir / f"{args.tenant}.db"
    if not db.is_file():
        raise SystemExit(
            f"租户库不存在: {db}\n(租户已注册但库未建——应用首次服务该"
            f"租户时建库, 或经 scripts/import_legacy.py 导入老数据)")

    conn = sqlite3.connect(db.resolve().as_uri() + "?mode=ro", uri=True)
    try:
        rows = conn.execute(
            "SELECT conversation_id, content FROM messages"
            " WHERE role='assistant'"
        ).fetchall()
    except sqlite3.OperationalError as exc:
        raise SystemExit(
            f"租户库读取失败({exc}): {db}(schema 未就位, 先跑应用或"
            f" import_legacy)") from exc
    finally:
        conn.close()

    hits = []
    for cid, content in rows:
        text = content or ""
        for pattern, name in COMPILED:
            if pattern.search(text):
                hits.append((cid, name, text[:60]))
                break
    for cid, name, preview in hits:
        print(f"  [{name}] conv={cid[:8]} {preview}")
    total = len(rows)
    rate = len(hits) / total * 100 if total else 0
    print(f"租户 {args.tenant} | AI回复总轮次: {total} | "
          f"价格形态命中: {len(hits)} ({rate:.0f}%)")
    print("← 破防率。>10% 启动方案B(输出过滤)")


if __name__ == "__main__":
    main()