feat: 词表数据化——AST 提取脚本 + wall.json + 编译器 + boundary/conversation 数据源切换(等价回归全绿)
- scripts/extract_config.py: AST 解析两老项目源码取常量(不 import 源模块),
生成 seed_profiles/<slug>/{wall.json, brand_values.json, prompts/};幂等可重跑
- app/engine/wall_schema.py: compile_wall(json)→CompiledWall,坏结构/坏 regex
编译期显性失败(WallConfigError 含字段路径)
- boundary_filter/conversation: 墙词表/话术数据源切换为 DEFAULT_WALL
(jiancai 出厂 seed),函数逻辑不动、常量名保留为 wall 视图(既有测试零修改);
wall 数据化新增 pivots/pivots_contact/zero_number/reask_patterns 字段
- 脆锚点迁移: test_inline_prohibitions 断言对象从 conversation.py 源码改为
wall.json 数据文件,语义等价(唯一允许的测试修改)
- 回归: 322 既有全绿 + 10 新增 wall_schema 测试 = 332;提取脚本两遍无 diff
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
@@ -1,5 +1,7 @@
|
||||
"""批次1 集成测试:管道接线/邮件触发/结束守卫/卡单标记。"""
|
||||
import asyncio
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy import select
|
||||
|
||||
@@ -1004,13 +1006,19 @@ def test_reminder_cc_due_turn_email_note_blocked():
|
||||
|
||||
|
||||
def test_inline_prohibitions_all_carry_reask_exception():
|
||||
"""三轮修复防第四次漏: conversation.py 三处内联禁令运行时文本(app分支/
|
||||
confirmed块/边界分支)都必须带回显确认例外——号码确认是对"缺国家代码"的
|
||||
补全动作,只豁免缺失渠道的禁令会压制它(生产验收两轮复发的根因);
|
||||
"""三轮修复防第四次漏: 三处内联禁令运行时文本(app分支/confirmed块/
|
||||
边界分支)都必须带回显确认例外——号码确认是对"缺国家代码"的补全动作,
|
||||
只豁免缺失渠道的禁令会压制它(生产验收两轮复发的根因);
|
||||
Task 4(spec §4b-3)例外从"可请客户重发完整号一次"收敛为回显语义
|
||||
(防同场竞争降低首过命中率)。"""
|
||||
with open("app/engine/services/conversation.py", encoding="utf-8") as f:
|
||||
src = f.read()
|
||||
(防同场竞争降低首过命中率)。
|
||||
锚点迁移(Task 4 词表数据化): 三处禁令已从 conversation.py 源码搬入
|
||||
seed_profiles/jiancai/wall.json 的 reminders 字段——断言对象改为该
|
||||
数据文件,断言语义保持等价(同样的禁令字符串与回显例外必须存在于
|
||||
wall 数据中,且例外与禁令同串就近在场)。"""
|
||||
wall_path = (Path(__file__).resolve().parents[1]
|
||||
/ "seed_profiles" / "jiancai" / "wall.json")
|
||||
src = json.dumps(
|
||||
json.loads(wall_path.read_text(encoding="utf-8")), ensure_ascii=False)
|
||||
assert src.count("回显补全后的完整号码") >= 3
|
||||
anchors = [
|
||||
"也绝对禁止再索要客户已提供过", # contact_note app+已留资分支
|
||||
|
||||
@@ -0,0 +1,123 @@
|
||||
"""wall.json → CompiledWall 编译器契约(Task 4 词表数据化)。
|
||||
|
||||
seed_profiles/<slug>/wall.json 是租户墙词表/话术的唯一数据源;
|
||||
compile_wall 把 JSON 编译为运行时对象,坏配置必须显性失败(信息含字段名)。
|
||||
"""
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from app.engine.wall_schema import CompiledWall, WallConfigError, compile_wall
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
def _load(slug: str) -> dict:
|
||||
return json.loads(
|
||||
(REPO_ROOT / "seed_profiles" / slug / "wall.json").read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def test_compile_produces_regex_objects():
|
||||
wall = compile_wall(_load("jiancai"))
|
||||
assert isinstance(wall, CompiledWall)
|
||||
assert wall.price_patterns
|
||||
assert all(isinstance(p, re.Pattern) for p in wall.price_patterns)
|
||||
assert wall.deflections["es"]
|
||||
assert all(isinstance(p, re.Pattern) for p in wall.reask_patterns)
|
||||
assert all(isinstance(p, re.Pattern) for p in wall.triggers.values())
|
||||
assert isinstance(wall.zero_number["long_number"], re.Pattern)
|
||||
|
||||
|
||||
def test_compile_rejects_bad_regex():
|
||||
cfg = _load("jiancai")
|
||||
cfg["price_patterns"] = ["([unclosed"]
|
||||
with pytest.raises(WallConfigError) as ei:
|
||||
compile_wall(cfg)
|
||||
assert "price_patterns" in str(ei.value)
|
||||
assert "regex" in str(ei.value)
|
||||
|
||||
|
||||
def test_compile_rejects_bad_regex_in_nested_field():
|
||||
cfg = _load("jiancai")
|
||||
cfg["triggers"]["boundary"] = {"pattern": "(?P<n>budget", "flags": "i"}
|
||||
with pytest.raises(WallConfigError) as ei:
|
||||
compile_wall(cfg)
|
||||
assert "triggers.boundary" in str(ei.value)
|
||||
|
||||
|
||||
def test_compile_rejects_unknown_flag():
|
||||
cfg = _load("jiancai")
|
||||
cfg["reask_patterns"] = [{"pattern": "ok", "flags": "x"}]
|
||||
with pytest.raises(WallConfigError) as ei:
|
||||
compile_wall(cfg)
|
||||
assert "reask_patterns" in str(ei.value)
|
||||
|
||||
|
||||
def test_missing_and_unknown_top_level_fields():
|
||||
cfg = _load("jiancai")
|
||||
del cfg["no_contact"]
|
||||
with pytest.raises(WallConfigError) as ei:
|
||||
compile_wall(cfg)
|
||||
assert "no_contact" in str(ei.value)
|
||||
|
||||
cfg = _load("jiancai")
|
||||
cfg["deflectionz"] = [] # 拼写错误必须在编译期显性失败
|
||||
with pytest.raises(WallConfigError) as ei:
|
||||
compile_wall(cfg)
|
||||
assert "deflectionz" in str(ei.value)
|
||||
|
||||
|
||||
def test_meta_key_is_ignored():
|
||||
cfg = _load("jiancai")
|
||||
cfg["_meta"] = {"doc": "任何 _ 前缀键为数据来源标注,不参与编译"}
|
||||
assert compile_wall(cfg).no_contact["web"]
|
||||
|
||||
|
||||
def test_flags_entry_restores_ignorecase():
|
||||
wall = compile_wall({"price_patterns": [{"pattern": "BUDGET|预算", "flags": "i"}],
|
||||
**{k: v for k, v in _load("jiancai").items()
|
||||
if k != "price_patterns"}})
|
||||
assert any(p.search("my BUDGET plan") for p in wall.price_patterns)
|
||||
|
||||
|
||||
def test_both_seed_walls_compile():
|
||||
for slug in ("jiancai", "auto"):
|
||||
wall = compile_wall(_load(slug))
|
||||
assert wall.triggers["boundary"].search("how much is it?")
|
||||
assert wall.no_contact["web"]
|
||||
assert wall.reminders["boundary"]["normal"].startswith("⚠️ 职能边界提醒")
|
||||
# 租户差异钉死: jiancai 多语言墙/auto 中英+$ 且无 es 池
|
||||
if slug == "jiancai":
|
||||
assert "es" in wall.deflections
|
||||
assert len(wall.price_patterns) > 10
|
||||
else:
|
||||
assert "es" not in wall.deflections
|
||||
assert "_meta" in _load(slug) # 引擎特性字段沿用 jiancai(显性标注)
|
||||
|
||||
|
||||
def test_default_wall_matches_jiancai_seed():
|
||||
"""boundary_filter 的出厂默认墙 = jiancai seed 逐字段等价(等价回归的锚)。"""
|
||||
from app.engine.services import boundary_filter as bf
|
||||
wall = compile_wall(_load("jiancai"))
|
||||
assert [p.pattern for p in bf.DEFAULT_WALL.price_patterns] == \
|
||||
[p.pattern for p in wall.price_patterns]
|
||||
assert [p.flags for p in bf.DEFAULT_WALL.price_patterns] == \
|
||||
[p.flags for p in wall.price_patterns]
|
||||
assert bf.DEFLECTIONS_ZH == list(wall.deflections["zh"])
|
||||
assert bf.NO_CONTACT_WEB == wall.no_contact["web"]
|
||||
assert bf.LONG_NUMBER_RE.pattern == wall.zero_number["long_number"].pattern
|
||||
assert [p.pattern for p in bf.REASK_PATTERNS] == \
|
||||
[p.pattern for p in wall.reask_patterns]
|
||||
|
||||
|
||||
def test_conversation_triggers_match_seed():
|
||||
"""conversation 三组触发正则 = wall.triggers 编译产物(pattern+flags 双等价)。"""
|
||||
from app.engine.services import conversation as conv
|
||||
wall = compile_wall(_load("jiancai"))
|
||||
for name, module_re in (("boundary", conv.BOUNDARY_TRIGGER_RE),
|
||||
("budget", conv.BUDGET_TRIGGER_RE),
|
||||
("contact_ask", conv.CONTACT_ASK_TRIGGER_RE)):
|
||||
assert module_re.pattern == wall.triggers[name].pattern, name
|
||||
assert module_re.flags == wall.triggers[name].flags, name
|
||||
Reference in New Issue
Block a user