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 content(k: dict) -> str:
    t = k["type"]
    if t == "how":
        L = [f"目标：{k.get('purpose','')}"]
        for i, s in enumerate(k.get("steps", []), 1):
            L += [f"步骤{i}（目的：{s.get('intent','')}）",
                  f"  指引：{s.get('directive','')}", f"  产出：{s.get('output','')}"]
        return "\n".join(L)
    if t == "what":
        return "\n".join([f"界定：{k.get('界定','')}", "构成："]
                         + [f"- {c.get('要素','')}：{c.get('说明','')}" for c in k.get("构成", [])])
    return f"主张：{k.get('主张','')}\n依据：{k.get('依据','')}\n对创作的影响：{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 ("灵感", "选题", "脚本")]
            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 == "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()