浅念
238 lines
10 KiB
Python
238 lines
10 KiB
Python
"""wall.json → CompiledWall 编译器(Task 4 词表数据化)。
|
|
|
|
seed_profiles/<slug>/wall.json(由 scripts/extract_config.py 从老项目源码
|
|
AST 提取生成)是租户墙词表/话术的唯一数据源。compile_wall 把 JSON 编译为
|
|
运行时对象: pattern 字符串 → re.Pattern,模板话术(含 {shown}/{intl} 等
|
|
占位符)原样承载、由 conversation.py 运行期 .format() 回填;占位符仅允许
|
|
TEMPLATE_PLACEHOLDERS 白名单键,编译期干跑校验(字面花括号/未知占位符
|
|
显性失败,不留到运行期炸会话流)。
|
|
|
|
regex 条目两种形态(pattern 文本与源码逐字一致):
|
|
- 纯字符串: 无标志位,re.compile(s)
|
|
- {"pattern": "...", "flags": "i"}: flags 字符映射 re.IGNORECASE
|
|
(提取脚本只产出 "i";comment 键为设计说明注释,编译时忽略;dict 条目
|
|
出白名单 {pattern, flags, comment} 之外的键 → 显性拒绝,防键拼错静默放宽)
|
|
|
|
结构/regex 无效 → WallConfigError,信息含字段路径(坏 regex 含 "regex"
|
|
字样与 re.error 原因)——配置错误在编译期显性失败,不流入运行期。
|
|
顶层 "_"-前缀键(如 auto 的 _meta 数据来源标注)为元信息,不参与编译。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
_FLAG_MAP = {"": 0, "i": re.IGNORECASE}
|
|
|
|
|
|
class WallConfigError(Exception):
|
|
"""wall.json 结构/regex 无效。"""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CompiledWall:
|
|
price_patterns: tuple[re.Pattern, ...]
|
|
deflections: dict[str, tuple[str, ...]]
|
|
deflections_contact: dict[str, tuple[str, ...]]
|
|
pivots: dict[str, tuple[str, ...]]
|
|
pivots_contact: dict[str, tuple[str, ...]]
|
|
no_contact: dict[str, str]
|
|
zero_number: dict[str, re.Pattern]
|
|
reask_patterns: tuple[re.Pattern, ...]
|
|
triggers: dict[str, re.Pattern]
|
|
reminders: dict[str, dict[str, str]] # 六类轮中提醒话术(_build_context_reminder 数据源)
|
|
cc_templates: dict[str, dict[str, str]]
|
|
|
|
|
|
# 每字段必填的变体键(引擎代码按名引用,缺键=运行期 KeyError,前移到编译期)
|
|
_REQUIRED_POOLS = {
|
|
"deflections": ("en", "zh"),
|
|
"deflections_contact": ("en", "zh"),
|
|
"pivots": ("en", "zh"),
|
|
"pivots_contact": ("en", "zh"),
|
|
}
|
|
_REMINDER_VARIANTS = {
|
|
"boundary": ("budget", "normal"),
|
|
"contact_ask": ("app_already", "app", "web"),
|
|
"language_follow": ("zh", "es"),
|
|
"lead_given": ("forbidden", "respond_script", "boundary_tail", "suspect"),
|
|
"cc_guide": ("echo", "ask"),
|
|
"email_light": ("en", "zh"),
|
|
}
|
|
_CC_TEMPLATE_VARIANTS = {
|
|
"ask": ("en", "zh", "es"),
|
|
"wall": ("en", "zh", "es"),
|
|
"note": ("en", "zh"),
|
|
"echo": ("en", "zh", "es"),
|
|
}
|
|
_KNOWN_FIELDS = frozenset(
|
|
list(_REQUIRED_POOLS) + ["price_patterns", "no_contact", "zero_number",
|
|
"reask_patterns", "triggers", "reminders",
|
|
"cc_templates"])
|
|
|
|
# regex 条目 dict 的合法键(审查 I-3): 提取脚本只产出 {pattern, flags,
|
|
# comment};键拼错(如 "flagss")不再静默放宽标志位,而是编译期显性拒绝。
|
|
_LEGAL_PATTERN_KEYS = frozenset({"pattern", "flags", "comment"})
|
|
|
|
# 模板占位符白名单(审查 I-2): conversation.py 运行期 .format() 回填的全部
|
|
# 键名(grep 实证清单)。scripts/extract_config.py 反向引用此常量(单一来源),
|
|
# 保证提取产出的占位符与引擎回填键名不漂移。字面花括号(如把中文"格式"
|
|
# 写成 {格式})或未知占位符在编译期即拒绝——否则运行期 KeyError 直接炸
|
|
# 客户会话流。
|
|
TEMPLATE_PLACEHOLDERS = frozenset({
|
|
"shown", "intl", "cc_script",
|
|
"no_contact_web", "no_contact_app", "no_contact_already",
|
|
})
|
|
|
|
|
|
def _err(msg: str) -> WallConfigError:
|
|
return WallConfigError(msg)
|
|
|
|
|
|
def _compile_pattern(entry, field: str) -> re.Pattern:
|
|
"""str | {pattern, flags[, comment]} → re.Pattern;失败含字段路径与原因。"""
|
|
if isinstance(entry, str):
|
|
pattern, flags = entry, 0
|
|
elif isinstance(entry, dict):
|
|
unknown = set(entry) - _LEGAL_PATTERN_KEYS
|
|
if unknown:
|
|
# 审查 I-3: 键拼错(如 "flagss")曾静默忽略→墙悄悄放宽,显性拒绝
|
|
raise _err(f"{field}: 条目含未知键 {', '.join(sorted(unknown))}"
|
|
f"(合法: {', '.join(sorted(_LEGAL_PATTERN_KEYS))}): {entry!r}")
|
|
pattern = entry.get("pattern")
|
|
flag_chars = str(entry.get("flags", ""))
|
|
if not isinstance(pattern, str) or not pattern:
|
|
raise _err(f"{field}: 条目缺非空 pattern 字符串: {entry!r}")
|
|
if flag_chars not in _FLAG_MAP:
|
|
raise _err(f"{field}: 不支持的 flags {flag_chars!r}(仅支持 'i')")
|
|
flags = _FLAG_MAP[flag_chars]
|
|
else:
|
|
raise _err(f"{field}: 条目必须是字符串或 {{pattern, flags}}: {entry!r}")
|
|
try:
|
|
return re.compile(pattern, flags)
|
|
except re.error as exc:
|
|
raise _err(f"{field}: regex 编译失败: {pattern!r}: {exc}") from exc
|
|
|
|
|
|
def _compile_patterns(entries, field: str) -> tuple[re.Pattern, ...]:
|
|
if not isinstance(entries, list) or not entries:
|
|
raise _err(f"{field}: 必须为非空列表")
|
|
return tuple(_compile_pattern(e, f"{field}[{i}]") for i, e in enumerate(entries))
|
|
|
|
|
|
def _compile_pool(cfg: dict, field: str) -> dict[str, tuple[str, ...]]:
|
|
pool_cfg = cfg.get(field)
|
|
if not isinstance(pool_cfg, dict) or not pool_cfg:
|
|
raise _err(f"{field}: 必须为非空语言→列表映射")
|
|
out: dict[str, tuple[str, ...]] = {}
|
|
for lang, items in pool_cfg.items():
|
|
if (not isinstance(items, list) or not items
|
|
or not all(isinstance(t, str) and t for t in items)):
|
|
raise _err(f"{field}.{lang}: 必须为非空字符串列表")
|
|
out[lang] = tuple(items)
|
|
for lang in _REQUIRED_POOLS[field]:
|
|
if lang not in out:
|
|
raise _err(f"{field}: 缺语言 {lang}")
|
|
return out
|
|
|
|
|
|
def _validate_template(template: str, label: str) -> None:
|
|
"""编译期干跑 .format()(审查 I-2): 全白名单占位符代入空串,字面花括号/
|
|
未知占位符/位置参数问题(原属运行期 KeyError/IndexError,炸客户会话流)
|
|
前移为编译期 WallConfigError。纯文本模板(无占位符)天然通过。"""
|
|
try:
|
|
template.format(**{name: "" for name in TEMPLATE_PLACEHOLDERS})
|
|
except (KeyError, IndexError, ValueError, AttributeError,
|
|
TypeError) as exc:
|
|
raise _err(f"{label}: 模板占位符非法(仅支持 "
|
|
f"{', '.join(sorted(TEMPLATE_PLACEHOLDERS))}): "
|
|
f"{template!r}: {exc}") from exc
|
|
|
|
|
|
def _str_mapping(sub, label: str, required: tuple[str, ...],
|
|
*, template: bool = False) -> dict[str, str]:
|
|
"""一层 str 映射校验(no_contact 各键/reminders 各类变体/cc_templates 各组)。
|
|
|
|
template=True 时额外干跑 .format() 校验占位符(conversation.py 运行期
|
|
回填的字段才需要;no_contact 为纯文本,不校验)。"""
|
|
if not isinstance(sub, dict):
|
|
raise _err(f"{label}: 必须为映射(缺类别或类型错误)")
|
|
out: dict[str, str] = {}
|
|
for key in required:
|
|
val = sub.get(key)
|
|
if not isinstance(val, str) or not val:
|
|
raise _err(f"{label}.{key}: 缺非空字符串")
|
|
if template:
|
|
_validate_template(val, f"{label}.{key}")
|
|
out[key] = val
|
|
for key in sub:
|
|
if key not in required:
|
|
raise _err(f"{label}.{key}: 未知变体键(合法: {', '.join(required)})")
|
|
return out
|
|
|
|
|
|
def compile_wall(cfg: dict) -> CompiledWall:
|
|
"""wall.json 反序列化后的 dict → CompiledWall。任何结构/regex 问题显性失败。"""
|
|
if not isinstance(cfg, dict):
|
|
raise _err("wall 配置必须为 JSON 对象")
|
|
unknown = [k for k in cfg if k not in _KNOWN_FIELDS and not k.startswith("_")]
|
|
if unknown:
|
|
raise _err(f"wall 未知字段: {', '.join(sorted(unknown))}")
|
|
missing = [k for k in sorted(_KNOWN_FIELDS) if k not in cfg]
|
|
if missing:
|
|
raise _err(f"wall 缺字段: {', '.join(missing)}")
|
|
|
|
zero_cfg = cfg["zero_number"]
|
|
if not isinstance(zero_cfg, dict):
|
|
raise _err("zero_number: 必须为映射")
|
|
zero = {key: _compile_pattern(zero_cfg.get(key), f"zero_number.{key}")
|
|
for key in ("long_number", "any_email")}
|
|
|
|
triggers_cfg = cfg["triggers"]
|
|
if not isinstance(triggers_cfg, dict):
|
|
raise _err("triggers: 必须为映射")
|
|
triggers = {key: _compile_pattern(triggers_cfg.get(key), f"triggers.{key}")
|
|
for key in ("boundary", "budget", "contact_ask")}
|
|
|
|
if not isinstance(cfg["reminders"], dict):
|
|
raise _err("reminders: 必须为映射")
|
|
reminders = {cat: _str_mapping(cfg["reminders"].get(cat), f"reminders.{cat}",
|
|
variants, template=True)
|
|
for cat, variants in _REMINDER_VARIANTS.items()}
|
|
extra = set(cfg["reminders"]) - set(_REMINDER_VARIANTS)
|
|
if extra:
|
|
raise _err(f"reminders: 未知类别 {', '.join(sorted(extra))}")
|
|
|
|
if not isinstance(cfg["cc_templates"], dict):
|
|
raise _err("cc_templates: 必须为映射")
|
|
cc_templates = {group: _str_mapping(cfg["cc_templates"].get(group),
|
|
f"cc_templates.{group}", variants,
|
|
template=True)
|
|
for group, variants in _CC_TEMPLATE_VARIANTS.items()}
|
|
extra = set(cfg["cc_templates"]) - set(_CC_TEMPLATE_VARIANTS)
|
|
if extra:
|
|
raise _err(f"cc_templates: 未知组 {', '.join(sorted(extra))}")
|
|
|
|
return CompiledWall(
|
|
price_patterns=_compile_patterns(cfg["price_patterns"], "price_patterns"),
|
|
deflections=_compile_pool(cfg, "deflections"),
|
|
deflections_contact=_compile_pool(cfg, "deflections_contact"),
|
|
pivots=_compile_pool(cfg, "pivots"),
|
|
pivots_contact=_compile_pool(cfg, "pivots_contact"),
|
|
no_contact=_str_mapping(cfg.get("no_contact"), "no_contact",
|
|
("web", "app", "already")),
|
|
zero_number=zero,
|
|
reask_patterns=_compile_patterns(cfg["reask_patterns"], "reask_patterns"),
|
|
triggers=triggers,
|
|
reminders=reminders,
|
|
cc_templates=cc_templates,
|
|
)
|
|
|
|
|
|
def load_wall_file(path: str | Path) -> CompiledWall:
|
|
"""从 wall.json 文件编译(出厂默认/测试共用;调用方负责路径锚定)。"""
|
|
return compile_wall(
|
|
json.loads(Path(path).read_text(encoding="utf-8")))
|