decompose.py 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443
  1. """创作知识解构引擎 v2:一帖 → N 颗(how/what/why,含组件颗)→ frameworks.json + payloads.json。
  2. 编排全流程,把 skill 的 phase 文档当 prompt 喂给 LLM(skill 是唯一真源):
  3. ① 读懂:图文帖→extractor 读图;视频帖→video_extract 下载 mp4+原生整段提炼(base64→Gemini)
  4. ② 判颗+类型闸+三lane成形+轻标签:system = phase1-frame.md
  5. ③ 作用域:system = phase2-scope.md → 候选 → scope_link 定位(火山)
  6. ⑤ 组装:代码 → 每颗一个 ingest payload(按类型分拼)
  7. ① 判 is_empty=true(无可提取知识)→ 短路,跳过 ②③⑤。
  8. 数据源:fixture(已有 5 帖)或实时 crawler 取数(新帖)。须在云端跑。用法:PYTHONPATH=. python scripts/decompose.py
  9. """
  10. from __future__ import annotations
  11. import json
  12. import os
  13. from pathlib import Path
  14. from creation_knowledge.config import Settings
  15. from creation_knowledge.integrations import video_extract
  16. from creation_knowledge.integrations.crawler import fetch_post_detail, parse_detail_response
  17. from creation_knowledge.integrations.extractor import GeminiExtractor
  18. from creation_knowledge.integrations.llm import chat_json
  19. from creation_knowledge.prompts import load_prompt
  20. from scripts.scope_link import ScopeLinker
  21. ROOT = Path(__file__).resolve().parent.parent
  22. FIX = ROOT / "tests" / "fixtures"
  23. DATA = ROOT / "data" / "demo"
  24. SKILL = ROOT / "创作知识提取-skill"
  25. PHASE1 = (SKILL / "extraction" / "phase1-frame.md").read_text(encoding="utf-8")
  26. PHASE2 = (SKILL / "extraction" / "phase2-scope.md").read_text(encoding="utf-8")
  27. GATE_ADMIT = load_prompt("gate_admit") # ①.5 创作判定闸:判"是创作"
  28. GATE_REFUTE = load_prompt("gate_refute") # ①.5:挑刺"是制作/越界"
  29. GATE_TIEBREAK = load_prompt("gate_tiebreak") # ①.5:分歧裁决(边界倾向排除)
  30. NORMALIZE = load_prompt("normalize_scope") # ③前:作用域值名词化(③LLM)
  31. GATE_HOW_ADMIT = load_prompt("gate_how_admit") # ②.5 有序性闸:判"真流水线"
  32. GATE_HOW_REFUTE = load_prompt("gate_how_refute") # ②.5:挑"假how(离散构成硬串)"
  33. GATE_HOW_TIEBREAK = load_prompt("gate_how_tiebreak") # ②.5:分歧裁决(边界倾向假how)
  34. # ① 窄表确定性兜底:只放无歧义裸动作动词(绝不放 营造/引导/表达 等名词语素)
  35. _STRIP_VERBS = ("寻找", "定位", "推导", "核验", "提取", "挖掘", "捕捉",
  36. "识别", "梳理", "归纳", "判断", "验证", "确认", "复盘")
  37. _PROTECT = ("营造", "引导", "表达", "塑造", "叙述", "呈现", "刻画",
  38. "升华", "控制", "推进", "转化", "传达") # 名词语素,永不砍
  39. # from: fixture(读 tests/fixtures)/ live(实时 crawler 取数)
  40. SOURCES = [
  41. {"cid": "699308fa0000000016009697", "platform": "xiaohongshu", "from": "fixture"},
  42. {"cid": "698481e1000000000a02a7c1", "platform": "xiaohongshu", "from": "fixture"},
  43. {"cid": "67e2e39b0000000003028ff0", "platform": "xiaohongshu", "from": "fixture"},
  44. {"cid": "680659e8000000001a007a11", "platform": "xiaohongshu", "from": "fixture"},
  45. {"cid": "67e4bdf50000000006028a59", "platform": "xiaohongshu", "from": "fixture"},
  46. {"cid": "6901e15a0000000003019fa3", "platform": "xiaohongshu", "from": "live"}, # 新增1
  47. {"cid": "69ff58670000000023016fa2", "platform": "xiaohongshu", "from": "live"}, # 新增2
  48. {"cid": "7589257893544165455", "platform": "douyin", "from": "live"}, # 抖音视频(倒数第2)
  49. {"cid": "6a33655e000000000f0055af", "platform": "xiaohongshu", "from": "live"}, # 无知识 Voodoo(最后)
  50. ]
  51. SRC2CN = {"substance": "实质", "form": "形式", "feeling": "感受", "effect": "作用", "intent": "意图"}
  52. TYPE2ATTR = {"how": "how工序", "what": "what构成", "why": "why原理"}
  53. CSTAGE = {"定向", "构思", "结构", "成文", "打磨"} # 创作阶段受控 5 值
  54. REUSE_THRESHOLD = 0.90
  55. # ---------- 取数 ----------
  56. def load_post(src: dict, settings: Settings):
  57. if src["from"] == "fixture":
  58. resp = json.loads((FIX / f"xhs_case_{src['cid']}.json").read_text("utf-8"))
  59. post = parse_detail_response(resp, fallback_content_id=src["cid"])
  60. else:
  61. post = fetch_post_detail(src["cid"], settings=settings)
  62. if not post.url:
  63. post.url = f"https://www.xiaohongshu.com/explore/{src['cid']}"
  64. return post
  65. # ---------- ① 读懂(图文/视频分流) ----------
  66. def _download_images(post) -> list[str]:
  67. """live 图文帖:把 CDN 图下载到本地 /data(避开小红书防盗链),返回本地公开路径;
  68. 单张失败则退回 CDN 直链(至少有 url)。复用 video_extract 的下载器(iOS UA + 平台 Referer)。"""
  69. base = DATA / "xiaohongshu" / post.id
  70. base.mkdir(parents=True, exist_ok=True)
  71. out = []
  72. for n, url in enumerate(post.image_urls, 1):
  73. dst = base / f"image_{n}.webp"
  74. pub = f"/data/demo/xiaohongshu/{post.id}/image_{n}.webp"
  75. try:
  76. if not dst.exists():
  77. dst.write_bytes(video_extract._default_download(url, post.platform))
  78. out.append(pub)
  79. except Exception:
  80. out.append(url)
  81. return out
  82. def read_one(src: dict, post, settings: Settings, extractor: GeminiExtractor):
  83. cid = src["cid"]
  84. if post.video_urls: # 视频帖:下载 mp4 + 原生整段提炼
  85. save = DATA / "douyin" / f"dy_{cid}" / "video.mp4"
  86. pub = f"/data/demo/douyin/dy_{cid}/video.mp4"
  87. ec = video_extract.extract_video(post, settings=settings, save_path=save, public_url=pub)
  88. media = {"type": "video", "video_url": pub, "images": []}
  89. cmap = {c.index: c.content for c in ec.cards}
  90. cards = [{"index": c.index, "content": cmap.get(c.index, ""), "video_url": pub,
  91. "start": c.start, "end": c.end} for c in post.cards] # 段卡:时间戳 + 读到的内容
  92. else: # 图文帖:读图
  93. ec = extractor.extract(post)
  94. if src["from"] == "fixture":
  95. imgs = [f"/data/demo/xiaohongshu/{post.id}/image_{n}.webp" for n in range(1, len(post.image_urls) + 1)]
  96. else:
  97. imgs = _download_images(post) # live 帖:下载 CDN 图到本地 /data,避开防盗链
  98. media = {"type": "image", "video_url": None, "images": imgs}
  99. cmap = {c.index: c.content for c in ec.cards}
  100. cards = [{"index": n, "content": cmap.get(n, ""), "image_url": imgs[n - 1] if n <= len(imgs) else None}
  101. for n in range(1, len(imgs) + 1)] # 每图一张卡:图 url + 读到的内容
  102. parts = [ec.text]
  103. if ec.from_image:
  104. parts.append("【图片要点】\n" + ec.from_image)
  105. parts += [f"【卡片{c.index}】{c.content}" for c in ec.cards if c.content]
  106. return "\n\n".join(p for p in parts if p), bool(ec.is_empty), media, cards
  107. # ---------- ② 判颗+成形+轻标签 ----------
  108. def shape(post, read: str) -> list[dict]:
  109. user = (f"原帖标题:{post.title or '(无)'}\n\n读懂后的完整内容:\n{read}\n\n"
  110. "按上面规则拆颗+判类型+成形+轻标签。作用域字段一律留空 []。"
  111. "只输出 JSON:{\"knowledges\":[ ... 见模板 ... ]}")
  112. return chat_json(PHASE1, user, timeout=120).get("knowledges") or []
  113. # ---------- ①.5 创作判定闸(voting:admit + refute,分歧上 tiebreak;边界倾向排除)----------
  114. def _vote(system: str, read: str, key: str, on_fail: bool) -> bool:
  115. """跑一次判定,取 key 字段为 bool;出错按 on_fail 兜底(避免 API 抖动误杀)。"""
  116. try:
  117. return bool(chat_json(system, read, timeout=90).get(key))
  118. except Exception:
  119. return on_fail
  120. def creation_gate(read: str) -> tuple[bool, str]:
  121. """判这帖是不是【图文/视频内容创作知识】。返回 (in_scope, 说明)。
  122. 甲方案:admit + refute 两票;一致即定;分歧→tiebreak 裁决(边界倾向排除)。
  123. 判定只看 ① 读懂后的内容,不看标题、不数关键词。"""
  124. v_admit = _vote(GATE_ADMIT, read, "in_scope", on_fail=True) # 判"是创作"
  125. v_refute = not _vote(GATE_REFUTE, read, "out_of_scope", on_fail=False) # 挑刺"是制作/越界"→归一成 in_scope
  126. if v_admit == v_refute:
  127. return v_admit, f"admit={v_admit}/refute={v_refute} 一致"
  128. v_tie = _vote(GATE_TIEBREAK, read, "in_scope", on_fail=False) # 分歧裁决,失败也倾向排除
  129. return v_tie, f"admit={v_admit}/refute={v_refute} 分歧→裁决={v_tie}"
  130. # ---------- ②.5 假how根治:有序性审查(层2信号+层3对抗投票)+ 假how重拆为 what/why ----------
  131. def _chain_signals(steps: list[dict]) -> list[str]:
  132. """层2 确定性信号:并列输入(input 不指向前步产出)/ 产出近义。喂给层3当证据。"""
  133. sigs, outs = [], [(s.get("output") or "") for s in steps]
  134. indep = 0
  135. for i, s in enumerate(steps):
  136. if i == 0:
  137. continue
  138. inp = s.get("input") or ""
  139. if not (("←" in inp) or any(o and o[:4] in inp for o in outs[:i])):
  140. indep += 1
  141. if indep >= 2:
  142. sigs.append(f"{indep} 个后步的 input 未指向前步产出(各自起头)")
  143. for i in range(len(outs)):
  144. for j in range(i + 1, len(outs)):
  145. a, b = outs[i], outs[j]
  146. if a and b and (a in b or b in a):
  147. sigs.append(f"步骤{i+1}与{j+1}产出近义({a} / {b})")
  148. break
  149. return sigs
  150. def how_gate(k: dict) -> tuple[bool, str]:
  151. """层3a 有序性闸(对抗投票:admit 判真链 / refute 挑假链 / 分歧裁决,边界倾向假how)。"""
  152. payload = json.dumps({
  153. "purpose": k.get("purpose"),
  154. "steps": [{"input": s.get("input"), "方法": (s.get("directive") or "")[:300], "产出": s.get("output")}
  155. for s in k.get("steps", [])],
  156. "代码信号": _chain_signals(k.get("steps", [])),
  157. }, ensure_ascii=False)
  158. v_admit = _vote(GATE_HOW_ADMIT, payload, "is_real_how", on_fail=True)
  159. v_refute = not _vote(GATE_HOW_REFUTE, payload, "is_fake", on_fail=False) # 不假 → 真
  160. if v_admit == v_refute:
  161. return v_admit, f"admit={v_admit}/refute={v_refute} 一致"
  162. v_tie = _vote(GATE_HOW_TIEBREAK, payload, "is_real_how", on_fail=False)
  163. return v_tie, f"admit={v_admit}/refute={v_refute} 分歧→裁决={v_tie}"
  164. def reshape_nonhow(k: dict) -> list[dict]:
  165. """层3b 假how重拆:只拆成 What/Why 主颗(不要 how/组件)。失败则保留原颗,不丢内容。"""
  166. body = f"目标:{k.get('purpose','')}\n" + "\n".join(
  167. f"- 输入:{s.get('input','')}|方法:{s.get('directive','')}|产出:{s.get('output','')}"
  168. for s in k.get("steps", []))
  169. user = ("【下面这块原被误判为 how 工序,实为「离散构成 / 原理」,请只拆成 What/Why 主颗——"
  170. "每个'是什么/分几类'的构成块拆一颗 What,背后的原理/标准拆一颗 Why;"
  171. "不要 how、不要组件颗,parent 一律 null。作用域字段留空 []。】\n\n"
  172. f"原标题:{k.get('title','')}\n{body}\n\n"
  173. "只输出 JSON:{\"knowledges\":[ ... 仅 what/why,见模板 ... ]}")
  174. try:
  175. out = chat_json(PHASE1, user, timeout=120).get("knowledges") or []
  176. except Exception:
  177. out = []
  178. res = []
  179. for i, nk in enumerate(out, 1):
  180. if nk.get("type") == "how": # 保险:拒绝又冒出来的 how
  181. continue
  182. nk["id"] = f"{k.get('id','k')}r{i}"
  183. nk["role"], nk["parent"] = "主", None
  184. res.append(nk)
  185. return res or [k] # 兜底:没拆出来就保留原颗
  186. def fix_fake_hows(knowledges: list[dict]) -> list[dict]:
  187. """逐颗 how 审查;假how → 重拆为 what/why(替换原颗)。"""
  188. out = []
  189. for k in knowledges:
  190. if k.get("type") == "how" and len(k.get("steps", [])) >= 2:
  191. real, why = how_gate(k)
  192. if not real:
  193. new = reshape_nonhow(k)
  194. print(f" ②.5 假how「{k.get('title')}」({why})→ 重拆 {len(new)} 颗")
  195. out.extend(new)
  196. continue
  197. out.append(k)
  198. return out
  199. # ---------- ③ 作用域候选 + 定位 ----------
  200. def _slim(knowledges: list[dict]) -> list[dict]:
  201. slim = []
  202. for k in knowledges:
  203. e = {"id": k.get("id"), "type": k.get("type"), "title": k.get("title")}
  204. if k.get("type") == "how":
  205. e["steps"] = [{"id": s.get("id"), "input": s.get("input"),
  206. "directive": (s.get("directive") or "")[:500], "output": s.get("output")}
  207. for s in k.get("steps", [])]
  208. else:
  209. e["内容"] = {x: k.get(x) for x in ("概要", "维度拆分规则", "body", "主张", "支撑") if k.get(x)}
  210. slim.append(e)
  211. return slim
  212. def scope_candidates(knowledges: list[dict]) -> list:
  213. user = ("给下面每颗知识标作用域候选(how 逐步:每个 step 一组;what/why 颗级:整颗一组)。\n"
  214. "每类平铺列出所有相关值(同类可多条、全部对等、不选主次);先脑内摊正交、收敛近义;意图通常 1 个。\n"
  215. "只输出 JSON:{\"scopes\":[{\"knowledge_id\":\"k1\",\"step_id\":\"s1\",\"items\":["
  216. "{\"scope_type\":\"substance\",\"value\":\"实质值1\"},{\"scope_type\":\"substance\",\"value\":\"实质值2\"},"
  217. "{\"scope_type\":\"form\",\"value\":\"形式值1\"},{\"scope_type\":\"intent\",\"value\":\"意图值\"}]},"
  218. "{\"knowledge_id\":\"k2\",\"step_id\":null,\"items\":[...]}]}\n\n"
  219. + json.dumps(_slim(knowledges), ensure_ascii=False))
  220. return chat_json(PHASE2, user, timeout=120).get("scopes") or []
  221. def strip_verb_tail(v: str) -> str:
  222. """① 窄表确定性兜底:砍掉值开头/结尾的无歧义裸动词;砍到 <2 字则回退原值。"""
  223. if not v or len(v) < 3:
  224. return v
  225. for verb in _STRIP_VERBS: # 开头裸动词
  226. if v.startswith(verb) and len(v) - len(verb) >= 2:
  227. v = v[len(verb):]
  228. break
  229. for verb in _STRIP_VERBS: # 结尾裸动词(_PROTECT 与之不相交,名词语素天然不在表里)
  230. if v.endswith(verb) and len(v) - len(verb) >= 2:
  231. v = v[:-len(verb)]
  232. break
  233. return v
  234. def nounify_scopes(scopes: list) -> list:
  235. """作用域值名词化(仅 实质/形式/感受/作用):③ 批量 LLM 名词化 → ① 窄表兜底。在定位前做。
  236. 意图豁免:意图值就该是动词(对齐意图树),不名词化、不剥动词。"""
  237. vals = sorted({it["value"] for sc in scopes for it in (sc.get("items") or [])
  238. if it.get("value") and it.get("scope_type") != "intent"})
  239. mp = {}
  240. if vals:
  241. try:
  242. mp = chat_json(NORMALIZE, json.dumps(vals, ensure_ascii=False), timeout=90).get("映射") or {}
  243. except Exception:
  244. mp = {}
  245. for sc in scopes:
  246. for it in sc.get("items") or []:
  247. v = it.get("value")
  248. if not v or it.get("scope_type") == "intent": # 意图原样保留(动词)
  249. continue
  250. it["value"] = strip_verb_tail(mp.get(v) or v) # ③ 映射优先,再 ① 兜底
  251. return scopes
  252. def link_scope(linker: ScopeLinker, scope_type: str, value: str) -> dict:
  253. try:
  254. hits = linker.link(value, source_type=SRC2CN.get(scope_type, scope_type), top_k=3)
  255. except Exception:
  256. hits = []
  257. top = hits[0] if hits else {}
  258. score = float(top.get("score", 0.0))
  259. reuse = score >= REUSE_THRESHOLD and top.get("name")
  260. return {"scope_type": scope_type, "value": top["name"] if reuse else value,
  261. "candidate": value, "link": "复用" if reuse else "新建", "score": round(score, 4),
  262. "top": [{"name": h["name"], "score": h["score"], "path": h.get("path", "")} for h in hits]}
  263. def apply_scopes(knowledges: list[dict], scopes: list, linker: ScopeLinker) -> None:
  264. by_k = {k.get("id"): k for k in knowledges}
  265. for sc in scopes:
  266. k = by_k.get(sc.get("knowledge_id"))
  267. if not k:
  268. continue
  269. linked = [link_scope(linker, it["scope_type"], it["value"])
  270. for it in (sc.get("items") or []) if it.get("scope_type") and it.get("value")]
  271. if k.get("type") == "how" and sc.get("step_id"):
  272. for s in k.get("steps", []):
  273. if s.get("id") == sc["step_id"]:
  274. s["作用域"] = linked
  275. else:
  276. k["作用域"] = (k.get("作用域") or []) + linked
  277. # ---------- ⑤ 组装 ----------
  278. def build_content(k: dict):
  279. """How/Why → 拍平字符串;What → 结构化实体对象 {name,kind,维度拆分规则,body[]}。"""
  280. t = k.get("type")
  281. if t == "how":
  282. lines = [f"目标:{k.get('purpose','')}"]
  283. for i, s in enumerate(k.get("steps", []), 1):
  284. lines += [f"步骤{i}",
  285. f" 输入:{s.get('input','')}", f" 方法:{s.get('directive','')}", f" 产出:{s.get('output','')}"]
  286. return "\n".join(lines)
  287. if t == "what": # 结构化实体(非字符串);概要走 custom_ext,不进 content
  288. return {"name": k.get("title") or "", "kind": k.get("kind"),
  289. "维度拆分规则": k.get("维度拆分规则") if k.get("kind") == "子集" else None,
  290. "body": [{"item_name": it.get("item_name", ""), "item_desc": it.get("item_desc", ""),
  291. "作用域": it.get("作用域") or []} for it in (k.get("body") or [])]}
  292. return "\n".join([f"主张:{k.get('主张','')}"] + _sections(k.get("支撑")))
  293. def _sections(blocks) -> list[str]:
  294. """把 why.支撑 的自由小节拼成文本行(What 已改结构化 body,不再走这里)。"""
  295. out = []
  296. for b in blocks or []:
  297. head = b.get("小标题") or ""
  298. form = b.get("形式")
  299. out.append(f"【{head}】" + (f"({form})" if form else ""))
  300. if b.get("内容"):
  301. out.append(f" {b['内容']}")
  302. for it in b.get("条目") or []:
  303. word = it.get("词") or it.get("要素") or ""
  304. cue = it.get("选择线索")
  305. line = f" - {word}:{it.get('说明','')}" if word else f" - {it.get('说明','')}"
  306. if cue:
  307. line += f"(选用:{cue})"
  308. out.append(line)
  309. return out
  310. def build_payload(post, k: dict, how_titles: dict | None = None) -> dict:
  311. how_titles = how_titles or {}
  312. t = k.get("type")
  313. scopes, seen = [], set()
  314. def add(lst):
  315. for sc in lst:
  316. key = (sc["scope_type"], sc["value"])
  317. if key not in seen:
  318. seen.add(key); scopes.append({"scope_type": sc["scope_type"], "value": sc["value"]})
  319. if t == "how":
  320. for s in k.get("steps", []):
  321. add(s.get("作用域", []))
  322. else:
  323. add(k.get("作用域", []))
  324. ext = [{"key": "业务阶段", "type": "str", "value": v} for v in (k.get("业务阶段") or [])]
  325. if t == "what" and k.get("概要"): # What 的概要迁到 custom_ext(不再进 content)
  326. ext.append({"key": "概要", "type": "str", "value": k["概要"]})
  327. if t == "how":
  328. cs, cseen = [], set()
  329. for s in k.get("steps", []):
  330. c = s.get("创作阶段")
  331. if c and c not in cseen:
  332. cseen.add(c); cs.append(c)
  333. ext += [{"key": "创作阶段", "type": "str", "value": v} for v in cs]
  334. ext += [{"key": "动作", "type": "str", "value": s["动作"]} for s in k.get("steps", []) if s.get("动作")]
  335. if k.get("role") == "组件" and k.get("parent"):
  336. p = k["parent"]
  337. ext.append({"key": "出自", "type": "str",
  338. "value": f"{how_titles.get(p.get('how_id'), p.get('how_id'))} 第{p.get('step')}步"})
  339. return {"source": {"id": post.id, "source_type": "post", "title": post.title or "",
  340. "author": post.author_name or "", "source_metadata": {"platform": post.platform, "url": post.url}},
  341. "title": k.get("title"), "content": build_content(k),
  342. "dim_creations": ["创作"], "dim_attributes": [TYPE2ATTR.get(t, "how工序")],
  343. "scopes": scopes, "custom_ext": ext}
  344. def main() -> None:
  345. settings = Settings.from_env()
  346. extractor = GeminiExtractor.from_env()
  347. linker = ScopeLinker()
  348. posts_out, payloads = [], []
  349. for src in SOURCES:
  350. cid = src["cid"]
  351. print(f"\n=== {src['platform']} {cid[:10]} ({src['from']}) ===")
  352. try:
  353. post = load_post(src, settings)
  354. read, is_empty, media, cards = read_one(src, post, settings, extractor)
  355. except Exception as exc:
  356. print(f" ✗ 取数/读懂失败:{exc}")
  357. posts_out.append({"post_id": cid, "source_id": cid, "title": f"(取数失败 {cid})",
  358. "platform": src["platform"], "url": "", "media": {"type": "image", "images": []},
  359. "cards": [], "error": str(exc)[:200], "knowledges": []})
  360. continue
  361. meta = {"post_id": cid, "source_id": post.id, "title": post.title or "",
  362. "platform": post.platform, "url": post.url, "media": media, "cards": cards}
  363. if is_empty: # ① 总闸:纯展示/无可提取
  364. print(f" ① 读懂 {len(read)} 字 → ① 判纯展示/无知识,跳过")
  365. posts_out.append({**meta, "no_knowledge": True, "knowledges": []})
  366. continue
  367. in_scope, gate_why = creation_gate(read) # ①.5 创作判定闸(创作 vs 制作 vs 越界)
  368. if not in_scope:
  369. print(f" ① 读懂 {len(read)} 字 → ①.5 闸判【非创作】({gate_why})→ 整帖排除")
  370. posts_out.append({**meta, "no_knowledge": True, "knowledges": []})
  371. continue
  372. print(f" ① 读懂 {len(read)} 字({media['type']})· ①.5 闸:创作({gate_why})")
  373. knowledges = fix_fake_hows(shape(post, read)) # ② 拆颗 → ②.5 假how根治(假how重拆为 what/why)
  374. how_ids = {k.get("id") for k in knowledges if k.get("type") == "how"}
  375. how_titles = {k.get("id"): k.get("title") for k in knowledges if k.get("type") == "how"}
  376. for k in knowledges:
  377. k["业务阶段"] = [b for b in (k.get("业务阶段") or []) if b in ("灵感", "选题", "脚本")] # 守卫:只留合法业务阶段
  378. for s in k.get("steps", []): # 守卫:创作阶段只留合法 5 值,非法(如"定稿/输出")丢弃
  379. if s.get("创作阶段") not in CSTAGE:
  380. s["创作阶段"] = None
  381. if k.get("role") == "组件" and (k.get("parent") or {}).get("how_id") not in how_ids: # 守卫:组件 parent 必指向同帖 how
  382. k["role"] = "主"; k["parent"] = None
  383. print(f" ② {len(knowledges)} 颗:" + ", ".join(f"{k.get('type')}/{k.get('role')}" for k in knowledges))
  384. apply_scopes(knowledges, nounify_scopes(scope_candidates(knowledges)), linker) # ③名词化+①兜底 → 定位
  385. print(" ③⑤ 作用域定位 + 组装")
  386. posts_out.append({**meta, "knowledges": knowledges})
  387. payloads += [build_payload(post, k, how_titles) for k in knowledges]
  388. suffix = os.environ.get("OUT", "") # OUT=_v2 → 写 frameworks_v2.json,不覆盖原版
  389. (ROOT / f"web/frameworks{suffix}.json").write_text(
  390. json.dumps({"count": len(posts_out), "posts": posts_out}, ensure_ascii=False, indent=1), encoding="utf-8")
  391. (ROOT / f"web/payloads{suffix}.json").write_text(
  392. json.dumps(payloads, ensure_ascii=False, indent=2), encoding="utf-8")
  393. print(f"\nwrote {len(posts_out)} posts, {len(payloads)} payloads → web/frameworks{suffix}.json")
  394. if __name__ == "__main__":
  395. main()