"""wall.json → CompiledWall 编译器(Task 4 词表数据化)。 seed_profiles//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")))