"""seed_profiles 骨架提取脚本(Task 4 词表数据化,Step 0)。 从两个老项目确定性生成每个租户的 seed_profiles// 目录: - wall.json 墙词表/话术(boundary_filter.py + conversation.py 的 正则/话术常量,AST 解析源码取值,绝不 import 源模块—— conversation.py 的 import 链会拉起 ai_service,它 import 时读提示词文件,脚本会挂) - brand_values.json 品牌值草稿(chat.html / widget.js / chat.js 三处 硬编码 + .env NOTIFY_EMAIL 归一;Task 5 人工校对消费) - prompts/ 提示词 md 逐字节拷贝(Task 6 消费) 幂等: 相同源 → 逐字节相同产出(JSON 固定键序/缩进,无时间戳)。 用法: python scripts/extract_config.py [--jc DIR] [--auto DIR] [--out ROOT] """ from __future__ import annotations import argparse import ast import html as html_mod import json import re import shutil import sys from bisect import bisect_right from io import StringIO from pathlib import Path from tokenize import COMMENT, generate_tokens # 本仓库引擎编译器(仅 stdlib 依赖,无源项目 import 链,import 安全—— # "绝不 import 源模块"约束针对的是老项目 app.*,非本文件所在仓库)。 sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from app.engine.wall_schema import (TEMPLATE_PLACEHOLDERS, WallConfigError, compile_wall) # ---- 占位符映射(函数体内联话术的插值表达式 → wall 模板占位符) ---- # 源码里的 f"{shown}" / 拼接 NO_CONTACT_WEB 等,在 wall.json 里以 # {shown} / {no_contact_web} 占位,运行期由 conversation.py .format() 回填。 PLACEHOLDER_NAMES = { "shown": "shown", "intl": "intl", "NO_CONTACT_WEB": "no_contact_web", "NO_CONTACT_APP": "no_contact_app", "NO_CONTACT_ALREADY": "no_contact_already", } PLACEHOLDER_CALLS = {"_cc_script": "cc_script"} # 单一来源反查(审查 I-2): 引擎侧白名单 wall_schema.TEMPLATE_PLACEHOLDERS # 是唯一权威,本脚本产出的占位符必须 ⊆ 白名单——清单漂移(此处新增映射 # 而引擎未同步)在提取期立即显性失败,而非运行期 .format() KeyError。 if not (set(PLACEHOLDER_NAMES.values()) | set(PLACEHOLDER_CALLS.values()) <= TEMPLATE_PLACEHOLDERS): raise SystemExit( "提取失败: 占位符超出 wall_schema.TEMPLATE_PLACEHOLDERS 白名单: " + ", ".join(sorted((set(PLACEHOLDER_NAMES.values()) | set(PLACEHOLDER_CALLS.values())) - TEMPLATE_PLACEHOLDERS))) class NotExtractable(Exception): """表达式超出受限求值语法(非目标常量,跳过)。""" # -------------------------------------------------------------------------- # AST 受限求值 # -------------------------------------------------------------------------- def _flag_chars(node: ast.AST) -> str: """re.compile 的 flags 实参 → 标志字符串(仅支持 I/IGNORECASE)。""" names: set[str] = set() def walk(n: ast.AST) -> None: if isinstance(n, ast.BinOp): # re.I | re.M 形态 walk(n.left) walk(n.right) elif (isinstance(n, ast.Attribute) and isinstance(n.value, ast.Name) and n.value.id == "re"): names.add(n.attr) else: raise NotExtractable("不支持的 flags 形态") walk(node) chars = [] for name in sorted(names): if name in ("I", "IGNORECASE"): chars.append("i") else: raise NotExtractable(f"不支持的 re 标志: {name}") return "".join(chars) def eval_expr(node: ast.AST, constants: dict, mode: str = "bake"): """受限求值: Constant/隐式拼接/JoinedStr/BinOp(+)/List/Dict/Name/re.compile。 bake 模式(模块级常量): Name 从已收集常量表解析取值(静态组合烘焙成串)。 placeholder 模式(函数内联话术): Name/插值按 PLACEHOLDER_* 映射为占位符。 """ if isinstance(node, ast.Constant): return node.value if isinstance(node, ast.JoinedStr): out = [] for v in node.values: if isinstance(v, ast.Constant): out.append(str(v.value)) elif isinstance(v, ast.FormattedValue): fv = v.value if isinstance(fv, ast.Name) and fv.id in PLACEHOLDER_NAMES: out.append("{" + PLACEHOLDER_NAMES[fv.id] + "}") else: raise NotExtractable("插值非常量名") else: raise NotExtractable("JoinedStr 内未知节点") return "".join(out) if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add): left = eval_expr(node.left, constants, mode) right = eval_expr(node.right, constants, mode) if not isinstance(left, str) or not isinstance(right, str): raise NotExtractable("非字符串 +") return left + right if isinstance(node, ast.List): return [eval_expr(e, constants, mode) for e in node.elts] if isinstance(node, ast.Dict): return {eval_expr(k, constants, mode): eval_expr(v, constants, mode) for k, v in zip(node.keys, node.values)} if isinstance(node, ast.Name): if mode == "placeholder": if node.id in PLACEHOLDER_NAMES: return "{" + PLACEHOLDER_NAMES[node.id] + "}" raise NotExtractable("占位符模式未知 Name: " + node.id) if node.id in constants: return constants[node.id] raise NotExtractable("前向引用/未知常量: " + node.id) if isinstance(node, ast.Call): if (isinstance(node.func, ast.Attribute) and node.func.attr == "compile" and isinstance(node.func.value, ast.Name) and node.func.value.id == "re"): if not node.args: raise NotExtractable("re.compile 无参") pattern = eval_expr(node.args[0], constants, mode) if not isinstance(pattern, str): raise NotExtractable("re.compile 非字符串 pattern") flags_node = None if len(node.args) > 1: flags_node = node.args[1] else: flags_node = next((kw.value for kw in node.keywords if kw.arg == "flags"), None) return {"pattern": pattern, "flags": _flag_chars(flags_node) if flags_node is not None else ""} if (mode == "placeholder" and isinstance(node.func, ast.Name) and node.func.id in PLACEHOLDER_CALLS): return "{" + PLACEHOLDER_CALLS[node.func.id] + "}" raise NotExtractable("未知调用") raise NotExtractable("未支持节点: " + type(node).__name__) def extract_module_constants(tree: ast.Module) -> dict: """模块顶层赋值按源序受限求值(引用仅限更早常量)。""" constants: dict = {} for node in tree.body: if (isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name)): try: constants[node.targets[0].id] = eval_expr(node.value, constants) except NotExtractable: continue # 非墙数据形态(函数引用等),按名取用时才报错 return constants # -------------------------------------------------------------------------- # 函数体内联话术提取(指纹锚定: 内容取自源码,指纹只是定位器) # -------------------------------------------------------------------------- def _iter_stmts(body: list): for stmt in body: yield stmt for field in ("body", "orelse", "finalbody"): sub = getattr(stmt, field, None) if isinstance(sub, list): yield from _iter_stmts(sub) def extract_function_templates(tree: ast.Module, func_name: str) -> list[str]: """收集函数体内可求值的字符串模板(赋值/parts.append/return),按源序。 不可求值的语句(f"{key}: {val}" 等非目标插值)自然跳过——指纹匹配 只在可求值候选里找,找不到/多处命中都是提取错误(显性失败)。 """ # 目标是类方法(_build_context_reminder/_cc_confirm_note 在 Manager 类里) fn = next((n for n in ast.walk(tree) if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) and n.name == func_name), None) if fn is None: return [] out: list[str] = [] for stmt in _iter_stmts(fn.body): value = None if isinstance(stmt, ast.Assign) and len(stmt.targets) == 1: value = stmt.value elif isinstance(stmt, ast.Return): value = stmt.value elif (isinstance(stmt, ast.Expr) and isinstance(stmt.value, ast.Call) and isinstance(stmt.value.func, ast.Attribute) and stmt.value.func.attr == "append" and len(stmt.value.args) == 1): value = stmt.value.args[0] if value is None: continue try: v = eval_expr(value, {}, mode="placeholder") except NotExtractable: continue if isinstance(v, str) and v: out.append(v) return out def find_template(templates: list[str], needle: str, field: str, *, contains: bool = False, required: bool = True) -> str | None: """指纹定位唯一模板;required=False 时允许缺失(auto 上游无该特性)。""" if contains: hits = [t for t in templates if needle in t] else: hits = [t for t in templates if t.startswith(needle)] if len(hits) == 1: return hits[0] if not required and not hits: return None raise SystemExit( f"提取失败: {field} 锚点应唯一命中,实得 {len(hits)} 处: {needle!r}") # -------------------------------------------------------------------------- # price_patterns 行内注释保留(tokenize 关联到下一条目) # -------------------------------------------------------------------------- def attach_price_comments(src: str, assign: ast.Assign, entries: list[dict]) -> None: """PRICE_PATTERNS 列表内的 # 注释块挂到紧随其后的条目(comment 键)。 防误杀断言的设计说明随 pattern 一并数据化,编辑 wall.json 的人能看见。 边界注释有意丢弃(审查 M-2 备案): 首条目之前/末条目之后的注释没有 可挂靠的条目(bisect 落在 0/len(entries) 之外)——首条目前的通常是 分组标题注释,本就不属于任何单条 pattern 的设计说明,丢弃不损信息。 """ comments: list[tuple[int, str]] = [] for tok in generate_tokens(StringIO(src).readline): if tok.type == COMMENT and assign.lineno < tok.start[0] < assign.end_lineno: comments.append((tok.start[0], tok.string)) if not comments: return first_lines = [elt.lineno for elt in assign.value.elts] blocks: dict[int, list[str]] = {} for row, text in comments: idx = bisect_right(first_lines, row) # 注释属于行号大于它的第一条目 if 0 < idx < len(entries): blocks.setdefault(idx, []).append(text) for idx, lines in blocks.items(): entries[idx]["comment"] = "\n".join(lines) # -------------------------------------------------------------------------- # wall.json 组装 # -------------------------------------------------------------------------- def _regex_entries(specs: list[dict]) -> list: """re.compile 求值结果列表 → JSON 条目列表。 无标志 → 纯字符串(pattern 与源码逐字一致); 有标志 → {"pattern": ..., "flags": "i"}(显式承载源码的 re.I/IGNORECASE)。 """ return [s["pattern"] if not s["flags"] else {"pattern": s["pattern"], "flags": s["flags"]} for s in specs] def _lang_pools(constants: dict, base: str, suffix: str = "") -> dict: """DEFLECTIONS_EN / DEFLECTIONS_EN_CONTACT 形态按在场语言收集 (auto 无 ES 池,自然缺 es 键)。""" return {lang: constants[f"{base}_{lang.upper()}{suffix}"] for lang in ("en", "zh", "es") if f"{base}_{lang.upper()}{suffix}" in constants} def build_wall(project_root: Path, slug: str) -> dict: bf_src = (project_root / "app" / "services" / "boundary_filter.py").read_text(encoding="utf-8") cv_src = (project_root / "app" / "services" / "conversation.py").read_text(encoding="utf-8") bf_tree = ast.parse(bf_src) cv_tree = ast.parse(cv_src) bf, cv = extract_module_constants(bf_tree), extract_module_constants(cv_tree) def need(table: dict, name: str): if name not in table: raise SystemExit(f"提取失败: {slug} 源码缺常量 {name}") return table[name] # price_patterns: 统一先建 dict 以便挂行内注释,无 flags 且无 comment 再瘦回字符串 price_assign = next((n for n in bf_tree.body if isinstance(n, ast.Assign) and getattr(n.targets[0], "id", "") == "PRICE_PATTERNS"), None) if price_assign is None: # 防呆(审查 M-1): 源码重构挪走赋值节点时显性失败 raise SystemExit(f"提取失败: {slug} 源码缺 PRICE_PATTERNS 赋值节点") price_entries: list[dict] = [] for r in need(bf, "PRICE_PATTERNS"): e: dict = {"pattern": r["pattern"]} if r["flags"]: e["flags"] = r["flags"] price_entries.append(e) attach_price_comments(bf_src, price_assign, price_entries) price_entries_json = [e if (e.get("flags") or e.get("comment")) else e["pattern"] for e in price_entries] reminders_inline = extract_function_templates(cv_tree, "_build_context_reminder") cc_guide = extract_function_templates(cv_tree, "_cc_confirm_note") def fp(needle: str, field: str, **kw) -> str | None: return find_template(reminders_inline, needle, f"{slug} {field}", **kw) return { "price_patterns": price_entries_json, "deflections": _lang_pools(bf, "DEFLECTIONS"), "deflections_contact": _lang_pools(bf, "DEFLECTIONS", "_CONTACT"), "pivots": _lang_pools(bf, "PIVOTS"), "pivots_contact": _lang_pools(bf, "PIVOTS", "_CONTACT"), "no_contact": { "web": need(bf, "NO_CONTACT_WEB"), "app": need(bf, "NO_CONTACT_APP"), "already": need(bf, "NO_CONTACT_ALREADY"), }, "zero_number": { "long_number": _regex_entries([need(bf, "LONG_NUMBER_RE")])[0], "any_email": _regex_entries([need(bf, "ANY_EMAIL_RE")])[0], }, "reask_patterns": _regex_entries(need(bf, "REASK_PATTERNS")), "triggers": { "boundary": _regex_entries([need(cv, "BOUNDARY_TRIGGER_RE")])[0], "budget": _regex_entries([need(cv, "BUDGET_TRIGGER_RE")])[0], "contact_ask": _regex_entries([need(cv, "CONTACT_ASK_TRIGGER_RE")])[0], }, "reminders": { "boundary": { "budget": fp("⚠️ 职能边界提醒:客户报预算求推荐", "reminders.boundary.budget"), "normal": fp("⚠️ 职能边界提醒:客户正在询问价格", "reminders.boundary.normal"), }, "contact_ask": { "app_already": fp("⚠️ 客户正在索要我方联系方式,且客户已留过", "reminders.contact_ask.app_already"), "app": fp("⚠️ 客户正在索要我方联系方式(最高意向信号)。当前为 App 会话", "reminders.contact_ask.app"), "web": fp("⚠️ 客户正在索要我方联系方式(最高意向信号)。请逐字使用这句指路", "reminders.contact_ask.web"), }, # auto 上游无语言跟随特性 → 空键集(主流程以 jiancai 引擎标准回填) "language_follow": {k: v for k, v in { "zh": cv.get("LANG_NOTE_ZH"), "es": cv.get("LANG_NOTE_ES"), }.items() if v is not None}, "lead_given": _lead_given(slug, reminders_inline), "cc_guide": {k: v for k, v in { "echo": find_template(cc_guide, "系统已按 IP 推测补全为", f"{slug} reminders.cc_guide.echo", contains=True, required=False), "ask": find_template(cc_guide, "且系统无法按 IP 推测补全", f"{slug} reminders.cc_guide.ask", contains=True, required=False), }.items() if v is not None}, "email_light": {k: v for k, v in { "en": cv.get("_EMAIL_NOTE_EN"), "zh": cv.get("_EMAIL_NOTE_ZH"), }.items() if v is not None}, }, "cc_templates": {group: {lang: cv[f"_CC_{kind}_{lang.upper()}"] for lang in ("en", "zh", "es") if f"_CC_{kind}_{lang.upper()}" in cv} for group, kind in (("ask", "ASK"), ("wall", "WALL"), ("note", "NOTE"), ("echo", "ECHO"))}, } def _lead_given(slug: str, templates: list[str]) -> dict: """已留资禁令四件套。jc 为两条独立 append;auto 是单条合并串(按固定 标记"需要回应时可直接逐字使用"切分回两段,值不变)。""" field = f"{slug} reminders.lead_given" combined = find_template(templates, "⚠️ 客户已提供联系方式(", f"{field}.forbidden") marker = "需要回应时可直接逐字使用" if marker in combined: head, tail = combined.split(marker, 1) forbidden, respond = head, marker + tail else: forbidden = combined respond = find_template(templates, marker, f"{field}.respond_script") return { "forbidden": forbidden, "respond_script": respond, "boundary_tail": find_template( templates, "客户已留过联系方式——边界请求直接告知", f"{field}.boundary_tail"), "suspect": find_template( templates, "⚠️ 客户疑似已提供联系方式(", f"{field}.suspect"), } # -------------------------------------------------------------------------- # brand_values.json 草稿 / prompts 拷贝 # -------------------------------------------------------------------------- def build_brand_values(project_root: Path) -> dict: static = project_root / "static" chat = (static / "chat.html").read_text(encoding="utf-8") widget = (static / "widget.js").read_text(encoding="utf-8") chatjs = (static / "js" / "chat.js").read_text(encoding="utf-8") env: dict[str, str] = {} env_path = project_root / ".env" if env_path.exists(): for line in env_path.read_text(encoding="utf-8").splitlines(): if "=" in line and not line.lstrip().startswith("#"): k, _, v = line.partition("=") env[k.strip()] = v.strip() def find1(pattern: str, text: str) -> str | None: m = re.search(pattern, text) return m.group(1) if m else None def js_default(attr: str, text: str) -> str | None: # widget.js: getAttribute("data-x") || "…";chat.js 多一层括号: …")) || "…" return find1(r'data-%s"\)+\s*\|\|\s*"([^"]*)"' % attr, text) def leaf_values(d): for v in d.values(): if isinstance(v, str): yield v elif isinstance(v, list): yield from (x for x in v if isinstance(x, str)) elif isinstance(v, dict): yield from leaf_values(v) def with_conflict(field: dict) -> dict: vals = {v for v in leaf_values(field) if v} return {**field, "conflict": len(vals) > 1} title = { "chat_html_title": find1(r"(.*?)", chat), "chat_html_h1": find1(r"

(.*?)

", chat), "widget_js_default": js_default("title", widget), } whatsapp = { "chat_html_href": find1(r'href="(https://wa\.me/[^"]*)"', chat), "widget_js_default": js_default("whatsapp", widget), "chat_js_default": js_default("whatsapp", chatjs), } email = { "chat_html_href": find1(r'href="(mailto:[^"]*)"', chat), "widget_js_default": js_default("email", widget), "chat_js_default": js_default("email", chatjs), } return { "_doc": ("品牌值机器提取草稿(Task 5 消费): 值取自 chat.html/widget.js/" "chat.js 硬编码与 .env;conflict=true 处需人工归一定稿"), "title": with_conflict(title), "whatsapp": with_conflict(whatsapp), "email": with_conflict(email), "notify_email": {"env": env.get("NOTIFY_EMAIL", "")}, "help_description": { "chat_html": html_mod.unescape( find1(r'help-description">(.*?)', chat) or ""), }, "popular_topics": { "chat_html": [html_mod.unescape(t) for t in re.findall(r'data-topic="([^"]*)"', chat)], }, } def copy_prompts(project_root: Path, out_dir: Path) -> list[str]: copied = [] for md in sorted((project_root / "app" / "prompts").glob("*.md")): target = out_dir / "prompts" / md.name target.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(md, target) copied.append(md.name) return copied # -------------------------------------------------------------------------- # 主流程 # -------------------------------------------------------------------------- # auto 上游不存在的引擎特性(jc 后期 Task 的引擎能力)——对应字段沿用 # jiancai 出厂(引擎标准)话术,以 _meta 显性标注,Task 5+ 校对时可见。 FALLBACK_FIELDS = ["reminders.language_follow", "reminders.cc_guide", "reminders.email_light", "cc_templates"] # 租户自有词表(auto 用自己的值整体替换 jiancai 基座) OWN_FIELDS = ["price_patterns", "deflections", "deflections_contact", "pivots", "pivots_contact", "no_contact", "zero_number", "reask_patterns", "triggers"] OWN_REMINDER_CATS = ["boundary", "contact_ask", "lead_given"] def write_json(path: Path, data: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", newline="\n") def _self_check(wall: dict, slug: str) -> None: """提取自检(审查 I-4): 落盘前跑 compile_wall,把"提取静默漏项" (锚点漂移/结构缺失/坏 regex/非法占位符)收敛到提取期失败——带 slug 上下文显性报错,而非留到引擎加载或客户会话运行期才炸。""" try: compile_wall(wall) except WallConfigError as exc: raise SystemExit(f"提取自检失败({slug}): {exc}") from exc def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--jc", default=r"D:\refactor\ai-kefu-jc", help="jiancai 老项目根目录") parser.add_argument("--auto", default=r"D:\refactor\ai-kefu", help="auto 老项目根目录") parser.add_argument("--out", default=str(Path(__file__).resolve().parents[1]), help="seed_profiles 输出根目录(仓库根)") args = parser.parse_args() out_root = Path(args.out) jc_root, auto_root = Path(args.jc), Path(args.auto) for root in (jc_root, auto_root): if not (root / "app" / "services" / "boundary_filter.py").exists(): raise SystemExit(f"源项目不存在或结构不符: {root}") # jiancai(主线,完整提取;多语言墙 = 引擎出厂默认) jc_wall = build_wall(jc_root, "jiancai") _self_check(jc_wall, "jiancai") jc_dir = out_root / "seed_profiles" / "jiancai" write_json(jc_dir / "wall.json", jc_wall) write_json(jc_dir / "brand_values.json", build_brand_values(jc_root)) jc_prompts = copy_prompts(jc_root, jc_dir) # auto: 自有词表覆盖 jiancai 基座;引擎特性字段沿用基座(见 FALLBACK_FIELDS) auto_own = build_wall(auto_root, "auto") auto_wall = json.loads(json.dumps(jc_wall, ensure_ascii=False)) for key in OWN_FIELDS: auto_wall[key] = auto_own[key] for cat in OWN_REMINDER_CATS: auto_wall["reminders"][cat] = auto_own["reminders"][cat] auto_wall["_meta"] = { "doc": ("auto 上游无语言跟随/CC 回显确认/邮箱轻补引擎特性," "以下字段沿用 jiancai 出厂(引擎标准)话术"), "fallback_from": "jiancai", "fields": FALLBACK_FIELDS, } _self_check(auto_wall, "auto") auto_dir = out_root / "seed_profiles" / "auto" write_json(auto_dir / "wall.json", auto_wall) write_json(auto_dir / "brand_values.json", build_brand_values(auto_root)) auto_prompts = copy_prompts(auto_root, auto_dir) print(f"jiancai: price {len(jc_wall['price_patterns'])} 条 / " f"deflections {'+'.join(sorted(jc_wall['deflections']))} / " f"prompts {jc_prompts}") print(f"auto: price {len(auto_wall['price_patterns'])} 条 / " f"deflections {'+'.join(sorted(auto_wall['deflections']))} / " f"prompts {auto_prompts}") print(f"auto fallback(引擎标准)字段: {', '.join(FALLBACK_FIELDS)}") if __name__ == "__main__": main()