Create-knowledge-find-decode/scripts/build_creation_demo.py

"""创作知识 query 正交 demo：多种正交组合 → LLM 评 valid/relevant(query_filter.txt) → 存 JSON。

不真实搜，只产 query 供前端看。轴严格取分类树的"创作支"：
  实质 = 实质树·理念支(排除表象)   形式 = 形式树·架构支(排除呈现)
  目的池 = 作用树 + 感受树 + 意图树 全部合一(随机取)
  业务阶段 = 灵感/选题/脚本(只这三个裸阶段，不展开)——脊柱，每族必带
  模态 = 视频/图文   知识类型 = 怎么做/有哪些/为什么
新尾缀：模态 × 业务阶段 × 知识类型；老尾缀对照：阶段 × 动作 × [作用] × 知识类型。
实质 / 形式 / 目的等轴从同一张 MASTER 主表投影，方便控制变量横向对比。
每条原串过 query_filter.txt(valid + relevant)，存 keep+valid+reason 供前端展示。
用法：PYTHONPATH=. CK_ENV_FILE=.env python scripts/build_creation_demo.py
"""
from __future__ import annotations

import json
import random
import sys
from pathlib import Path

from acquisition.query import ACTIONS, _nonleaf_d4   # 老正交动作轴 + 4级非叶子取作用节点
from acquisition.query import STAGES as OLD_STAGES   # 老的裸阶段 灵感/选题/脚本
from acquisition.query_filter import filter_queries   # 共享筛选器(query_filter.txt)
from core.config import Settings

ROOT = Path(__file__).resolve().parent.parent
TREES = ROOT / "scope_trees" / "trees_index.json"
OUT = ROOT / "data" / "queries" / "creation_demo.json"
PER = 30
BATCH_N = 30        # 全 demo 统一抽这么多个「实质 / 形式」，各族共用同一批；PER=BATCH_N 保证整批都用上、左右一一对应
KTYPE = ["怎么做", "有哪些", "为什么"]
MODALITY = ["视频", "图文"]   # 被创作内容的形态(与教学帖本身格式无关),正交进所有家族
# 业务阶段(脊柱)：只留 3 个裸阶段，不再展开成 query构造.md 的衍生词(找素材/开头/钩子/标题…全去掉)
INTENT = ["灵感", "选题", "脚本"]


def _segs(p):
    return [x for x in (p or "").split("/") if x]


def _leaves(idx, source_type, under=None):
    """某树某支下的叶子节点名(没有更深子节点的=元素层)。under 限定分支。"""
    paths = [(_segs(n["path"]), n.get("name")) for n in idx if n.get("source_type") == source_type]
    if under:
        paths = [(s, nm) for s, nm in paths if under in s]
    allp = {"/".join(s) for s, _ in paths}
    out, seen = [], set()
    for s, nm in paths:
        if len(s) < 2:
            continue
        full = "/".join(s)
        is_leaf = not any(o != full and o.startswith(full + "/") for o in allp)
        name = nm or s[-1]
        if is_leaf and name and name not in seen:
            seen.add(name)
            out.append(name)
    return out


def _nonleaf(idx, source_type, depths=(3, 4), under=None):
    """某树某支下、指定层级的【非叶子"类目"节点】(底下还有元素,不取元素本身)。
    对齐制作侧取法:实质/形式 取 depth 3-4 的类目层,而非最深的元素层。"""
    paths = [(_segs(n["path"]), n.get("name")) for n in idx if n.get("source_type") == source_type]
    if under:
        paths = [(s, nm) for s, nm in paths if under in s]
    allp = {"/".join(s) for s, _ in paths}
    out, seen = [], set()
    for s, nm in paths:
        if len(s) not in depths:
            continue
        full = "/".join(s)
        is_nonleaf = any(o != full and o.startswith(full + "/") for o in allp)
        name = nm or (s[-1] if s else "")
        if is_nonleaf and name and name not in seen:
            seen.add(name)
            out.append(name)
    return out


def main():
    settings = Settings.from_env()
    rng = random.Random(7)
    DRY = "--dry" in sys.argv          # 干跑：跳过 LLM 筛选(全 keep)，只验证生成结构/控制变量
    idx = json.loads(TREES.read_text("utf-8"))
    SHI = _nonleaf(idx, "实质", depths=(3, 4), under="理念")   # 类目层,非元素
    XING = _nonleaf(idx, "形式", depths=(3, 4), under="架构")  # 类目层,非元素
    POOL = _leaves(idx, "作用") + _leaves(idx, "感受") + _leaves(idx, "意图")
    print(f"实质 {len(SHI)} / 形式 {len(XING)} / 目的池 {len(POOL)} / 业务阶段 {len(INTENT)}")

    # 统一批：30 实质 / 30 形式 / 30 目的(作用感受意图)，全 demo 共用
    SHI_BATCH = rng.sample(SHI, min(BATCH_N, len(SHI)))
    XING_BATCH = rng.sample(XING, min(BATCH_N, len(XING)))
    POOL_BATCH = rng.sample(POOL, min(BATCH_N, len(POOL)))
    F6_ZY = _nonleaf_d4("作用", 10)                                       # 老正交的"作用"(4级非叶子)
    F6_STAGE_ACT = [(s, a) for s in OLD_STAGES for a in ACTIONS] + [("", "")]   # 阶段×动作 + 无动作变体
    print(f"统一批: 实质×{len(SHI_BATCH)} 形式×{len(XING_BATCH)} 目的×{len(POOL_BATCH)}")

    # 主表：第 i 行把【所有轴】取值一次定死；各组合方式只是从这行挑自己用到的轴(投影)，严格控制变量
    # ——同一个实质(在第 i 行)无论出现在哪种组合里，配的模态/业务阶段/知识类型都相同。
    MASTER = []
    for i in range(PER):
        st, ac = F6_STAGE_ACT[i % len(F6_STAGE_ACT)]
        MASTER.append({
            "实质": SHI_BATCH[i % len(SHI_BATCH)],
            "形式": XING_BATCH[i % len(XING_BATCH)],
            "目的": POOL_BATCH[i % len(POOL_BATCH)],
            "模态": MODALITY[i % len(MODALITY)],
            "业务阶段": INTENT[i % len(INTENT)],
            "知识类型": KTYPE[(i // 3) % len(KTYPE)],     # 与业务阶段解耦，纯尾缀族也能多出几种
            "阶段": st or "/", "动作": ac or "/", "_st": st, "_ac": ac,
            "作用": F6_ZY[i % len(F6_ZY)],
        })

    def proj(i, keys):       # 新脊柱族：从主表第 i 行取这些 parts 键(query 由 order 拼)
        row = MASTER[i]
        return {"parts": {k: row[k] for k in keys}}

    def gen_old(i, *, shi=False, xing=False, purpose=False, zy=True):    # 老正交族：[实质] [形式] [目的] (阶段+动作) [作用] 知识类型
        r = MASTER[i]
        seg = (r["_st"] + r["_ac"]) if r["_ac"] else ""               # 脚本撰写 / 空
        head = ([r["实质"]] if shi else []) + ([r["形式"]] if xing else []) + ([r["目的"]] if purpose else [])
        tail = ([r["作用"]] if zy else []) + [r["知识类型"]]
        q = " ".join(head + ([seg] if seg else []) + tail)
        parts = {}
        if shi:
            parts["实质"] = r["实质"]
        if xing:
            parts["形式"] = r["形式"]
        if purpose:
            parts["目的"] = r["目的"]
        parts["阶段"], parts["动作"] = r["阶段"], r["动作"]
        if zy:
            parts["作用"] = r["作用"]
        parts["知识类型"] = r["知识类型"]
        return {"parts": parts, "query": q}

    # 每种组合方式：生成器 + 用到的轴。name = axes 用「×」连接，直接反映正交结构。
    # axes 顺序即前端列顺序；前 5 种使用新尾缀，后 6 种用于对照老尾缀。
    families = [
        {"key": "f1", "axes": ["实质", "模态", "业务阶段", "知识类型"],
         "gen": lambda i: proj(i, ["实质", "模态", "业务阶段", "知识类型"])},
        {"key": "f2", "axes": ["形式", "模态", "业务阶段", "知识类型"],
         "gen": lambda i: proj(i, ["形式", "模态", "业务阶段", "知识类型"])},
        {"key": "f4", "axes": ["作用/感受/意图", "模态", "业务阶段", "知识类型"],
         "gen": lambda i: proj(i, ["目的", "模态", "业务阶段", "知识类型"])},
        {"key": "f3", "axes": ["实质", "形式", "模态", "业务阶段", "知识类型"],
         "gen": lambda i: proj(i, ["实质", "形式", "模态", "业务阶段", "知识类型"])},
        {"key": "f5", "axes": ["模态", "业务阶段", "知识类型"],
         "gen": lambda i: proj(i, ["模态", "业务阶段", "知识类型"])},
        # 老正交·尾缀A：阶段 × 动作 × 作用 × 知识类型(五种正交)
        {"key": "a_shi", "axes": ["实质", "阶段", "动作", "作用", "知识类型"],
         "gen": lambda i: gen_old(i, shi=True, zy=True)},
        {"key": "a_xing", "axes": ["形式", "阶段", "动作", "作用", "知识类型"],
         "gen": lambda i: gen_old(i, xing=True, zy=True)},
        {"key": "a_both", "axes": ["实质", "形式", "阶段", "动作", "作用", "知识类型"],
         "gen": lambda i: gen_old(i, shi=True, xing=True, zy=True)},
        {"key": "a_purpose", "axes": ["作用/感受/意图", "阶段", "动作", "作用", "知识类型"],
         "gen": lambda i: gen_old(i, purpose=True, zy=True)},
        {"key": "a_tail", "axes": ["阶段", "动作", "作用", "知识类型"],
         "gen": lambda i: gen_old(i, zy=True)},
        # 老正交·尾缀B：阶段 × 动作 × 知识类型(抽掉作用，五种正交)
        {"key": "b_shi", "axes": ["实质", "阶段", "动作", "知识类型"],
         "gen": lambda i: gen_old(i, shi=True, zy=False)},
        {"key": "b_xing", "axes": ["形式", "阶段", "动作", "知识类型"],
         "gen": lambda i: gen_old(i, xing=True, zy=False)},
        {"key": "b_both", "axes": ["实质", "形式", "阶段", "动作", "知识类型"],
         "gen": lambda i: gen_old(i, shi=True, xing=True, zy=False)},
        {"key": "b_purpose", "axes": ["作用/感受/意图", "阶段", "动作", "知识类型"],
         "gen": lambda i: gen_old(i, purpose=True, zy=False)},
        {"key": "b_tail", "axes": ["阶段", "动作", "知识类型"],
         "gen": lambda i: gen_old(i, zy=False)},
    ]
    # 各部件按固定顺序拼成原串（内容维度在前，模态贴题材后，业务阶段+知识类型收尾）
    order = ["实质", "形式", "目的", "模态", "业务阶段", "知识类型"]

    # 业务阶段只剩 灵感/选题/脚本 三个，扁平数组(前端按池子平铺，不再分组缩进)
    out = {"axis_values": {"实质": SHI, "形式": XING, "目的池": POOL, "业务阶段": INTENT,
                           "模态": MODALITY, "知识类型": KTYPE,
                           "阶段": OLD_STAGES, "动作": ACTIONS, "作用": F6_ZY},   # 老尾缀对照轴
           "families": []}
    for fam in families:
        name = " × ".join(fam["axes"])   # 家族名直接反映正交结构
        seen, items = set(), []
        for i in range(PER):                          # 单趟过主表、去重(纯尾缀族会自然少于 PER)
            g = fam["gen"](i)
            parts = g["parts"]
            q = g.get("query") or " ".join(parts[k] for k in order if k in parts)   # f6/f7 自带 query 串
            if q in seen:
                continue
            seen.add(q)
            items.append({"query": q, "parts": parts})
        verdicts = ([{"keep": True, "valid": None, "relevant": True, "reason": "(dry:未过筛)"} for _ in items]
                    if DRY else filter_queries([it["query"] for it in items], settings))
        for it, v in zip(items, verdicts):
            it.update(v)
        kept = sum(1 for it in items if it["keep"])
        print(f"[{name}] 生成 {len(items)} 条, 筛后保留 {kept}")
        out["families"].append({"key": fam["key"], "name": name, "axes": fam["axes"], "items": items})

    OUT.parent.mkdir(parents=True, exist_ok=True)
    OUT.write_text(json.dumps(out, ensure_ascii=False, indent=1), encoding="utf-8")
    print(f"→ {OUT}")


if __name__ == "__main__":
    main()