130 lines
5.5 KiB
Python
130 lines
5.5 KiB
Python
|
"""职能边界破防率观测(只读): 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()
|