From 24902c3c438035f0c32d91f1d82921a9343fcdb2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Wed, 14 Oct 2026 10:27:41 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20breach=5Frate=E9=9A=8F=E8=BF=81?= =?UTF-8?q?=EF=BC=88=E5=A4=9A=E7=A7=9F=E6=88=B7--tenant=E9=80=82=E9=85=8D?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 从 ai-kefu-jc/scripts/breach_rate.py 拷入并适配多租户: CLI 改为 --tenant 必填, 经 platform.db 门禁(租户须已 bootstrap 注册, 同 import_legacy/archive 纪律)后只读打开 tenants/.db; 新增 --platform-db/--tenants-dir 覆盖参数(默认 repo 根约定路径)。 其余逻辑与 jc 原版一致: PATTERNS 锚定式匹配(美元符/货币单位多语言/ 价格名词前置/交期承诺/号码回显≥10位/邮箱回显), 逐 assistant 轮次 首模式命中即计, 汇总破防率(>10% 启动方案B 输出过滤)。观察漏放侧 偏差——误杀侧不在本脚本口径(人工复核)。 冒烟: 临时库验证 happy path(5 轮 3 中 60%, user 行排除/NULL 内容 守卫/中文标签)与四类门禁拒绝(未注册/库缺文件/platform.db 缺/ schema 未就位), 临时文件已清。全量 pytest 630 passed 2 skipped。 Co-Authored-By: Claude Code --- scripts/breach_rate.py | 129 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 129 insertions(+) create mode 100644 scripts/breach_rate.py diff --git a/scripts/breach_rate.py b/scripts/breach_rate.py new file mode 100644 index 0000000..ffb2a4a --- /dev/null +++ b/scripts/breach_rate.py @@ -0,0 +1,129 @@ +"""职能边界破防率观测(只读): AI 回复中出现"价格形态数字"的轮次占比。 + +适配自 ai-kefu-jc/scripts/breach_rate.py(jc 全局库 → 多租户): 方案B +(输出过滤)的启动依据——>10% 即砌墙; 锚定式匹配, 年份/规格数字天然不 +命中。与 boundary_filter 试金石(wall/test, 判定侧)互补, 本脚本观察 +**漏放侧偏差**(线上 AI 回复实际漏出的价格/号码/邮箱形态); 误杀侧 +(客户正常消息被墙误拦)不在本脚本口径, 用后台会话详情人工复核。 + +多租户适配: --tenant 必填, 经 platform.db 门禁(租户须已 +bootstrap 注册, 同 import_legacy/archive 纪律——绝不摸未注册租户的库) +后只读打开 tenants/.db; 其余逻辑(模式表/统计口径)与 jc 原版 +一致——含 Minor-4 的号码回显(≥10位)与邮箱回显观测。 + +用法(Windows 须 PYTHONUTF8=1, 输出含中文): + PYTHONUTF8=1 python scripts/breach_rate.py --tenant \ + [--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}(? 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 路径(默认 /data/, 门禁查" + "租户是否已 bootstrap)") + parser.add_argument("--tenants-dir", + default=str(REPO_ROOT / "tenants"), + help="租户库目录(默认 /tenants, 只读打开" + ".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()