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

"""从 web/frameworks.json 重建 web/payloads.json（不调 LLM）——确定性。

顺带两个守卫修正：
  · 组件颗的 parent.how_id 必须指向同帖一个 how 颗，否则降级为 orphan 主颗（role=主, parent=null）。
  · 组件颗 payload 的「出自」用所属 how 的 title（而非裸 id）。
用法：python3 scripts/rebuild_payloads.py
"""
from __future__ import annotations

import json
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
FW = ROOT / "web/frameworks.json"
PL = ROOT / "web/payloads.json"
TYPE2ATTR = {"how": "how工序", "what": "what构成", "why": "why原理"}


def _sections(blocks) -> list:
    out = []
    for b in blocks or []:
        head = b.get("小标题") or ""
        form = b.get("形式")
        out.append(f"【{head}】" + (f"（{form}）" if form else ""))
        if b.get("内容"):
            out.append(f"  {b['内容']}")
        for it in b.get("条目") or []:
            word = it.get("词") or it.get("要素") or ""
            cue = it.get("选择线索")
            line = f"  - {word}：{it.get('说明','')}" if word else f"  - {it.get('说明','')}"
            if cue:
                line += f"（选用：{cue}）"
            out.append(line)
    return out


def content(k: dict):
    """How/Why → 拍平字符串；What → 结构化实体对象 {name,kind,维度拆分规则,body[]}。"""
    t = k["type"]
    if t == "how":
        L = [f"目标：{k.get('purpose','')}"]
        for i, s in enumerate(k.get("steps", []), 1):
            L += [f"步骤{i}", f"  输入：{s.get('input','')}",
                  f"  方法：{s.get('directive','')}", f"  产出：{s.get('output','')}"]
        return "\n".join(L)
    if t == "what":  # 结构化实体；概要走 custom_ext
        return {"name": k.get("title") or "", "kind": k.get("kind"),
                "维度拆分规则": k.get("维度拆分规则") if k.get("kind") == "子集" else None,
                "body": [{"item_name": it.get("item_name", ""), "item_desc": it.get("item_desc", ""),
                          "作用域": it.get("作用域") or []} for it in (k.get("body") or [])]}
    return "\n".join([f"主张：{k.get('主张','')}"] + _sections(k.get("支撑")))


def main() -> None:
    d = json.loads(FW.read_text(encoding="utf-8"))
    payloads = []
    demoted = 0
    for p in d["posts"]:
        ks = p["knowledges"]
        how_ids = {k["id"] for k in ks if k["type"] == "how"}
        how_titles = {k["id"]: k.get("title") for k in ks if k["type"] == "how"}
        for k in ks:
            k["业务阶段"] = [b for b in (k.get("业务阶段") or []) if b in ("灵感", "选题", "脚本")]
            for s in k.get("steps", []):
                if s.get("创作阶段") not in ("定向", "构思", "结构", "成文", "打磨"):
                    s["创作阶段"] = None
            if k.get("role") == "组件" and (k.get("parent") or {}).get("how_id") not in how_ids:
                k["role"] = "主"
                k["parent"] = None
                demoted += 1
        for k in ks:
            t = k["type"]
            scopes, seen = [], set()
            def add(lst):
                for sc in lst:
                    key = (sc["scope_type"], sc["value"])
                    if key not in seen:
                        seen.add(key)
                        scopes.append({"scope_type": sc["scope_type"], "value": sc["value"]})
            if t == "how":
                for s in k.get("steps", []):
                    add(s.get("作用域", []))
            else:
                add(k.get("作用域", []))
            ext = [{"key": "业务阶段", "type": "str", "value": v} for v in (k.get("业务阶段") or [])]
            if t == "what" and k.get("概要"):
                ext.append({"key": "概要", "type": "str", "value": k["概要"]})
            if t == "how":
                cs, cseen = [], set()
                for s in k.get("steps", []):
                    c = s.get("创作阶段")
                    if c and c not in cseen:
                        cseen.add(c)
                        cs.append(c)
                ext += [{"key": "创作阶段", "type": "str", "value": v} for v in cs]
                ext += [{"key": "动作", "type": "str", "value": s["动作"]} for s in k.get("steps", []) if s.get("动作")]
            if k.get("role") == "组件" and k.get("parent"):
                par = k["parent"]
                ext.append({"key": "出自", "type": "str",
                            "value": f"{how_titles.get(par.get('how_id'), par.get('how_id'))} 第{par.get('step')}步"})
            payloads.append({
                "source": {"id": p.get("source_id") or f"xhs_{p['post_id']}", "source_type": "post", "title": p.get("title", ""),
                           "author": "", "source_metadata": {"platform": p.get("platform"), "url": p.get("url")}},
                "title": k.get("title"), "content": content(k), "dim_creations": ["创作"],
                "dim_attributes": [TYPE2ATTR.get(t, "how工序")], "scopes": scopes, "custom_ext": ext})
    FW.write_text(json.dumps(d, ensure_ascii=False, indent=1), encoding="utf-8")
    PL.write_text(json.dumps(payloads, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"rebuilt {len(payloads)} payloads; 降级组件→orphan: {demoted}")


if __name__ == "__main__":
    main()