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()