解构引擎接入视频 + 实时取数 + 业务阶段守卫

- ① 读懂按媒体分流：视频帖→video_extract（下载 mp4+base64 喂 Gemini 原生理解），图文帖→extractor
- 数据源支持 fixture 或实时 crawler 取数（新帖按 URL/id 现爬）
- 每帖 try/except 隔离（一篇取数失败不拖垮整批）；meta 带 source_id + media(images/video_url)
- 守卫：业务阶段只留 灵感/选题/脚本（清掉 LLM 误塞的"结构/制作"）
- rebuild_payloads 同步 source_id + 业务阶段守卫

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

scripts/decompose.py

@@ -1,48 +1,79 @@
 """创作知识解构引擎 v2：一帖 → N 颗（how/what/why，含组件颗）→ frameworks.json + payloads.json。
 
 编排全流程，把 skill 的 phase 文档当 prompt 喂给 LLM（skill 是唯一真源）：
-  ① 读懂：extractor 读图 → 完整文字（vision, OpenRouter）
-  ② 判颗+类型闸+三lane成形+轻标签：system = phase1-frame.md
-  ③ 作用域：system = phase2-scope.md → 候选 → scope_link 回扣（火山）
-  ⑤ 组装：代码 → 每颗一个 ingest payload（按类型分拼）
-须在云端跑（OpenRouter vision + 火山 可达）。用法：PYTHONPATH=. python scripts/decompose.py
+  ① 读懂：图文帖→extractor 读图；视频帖→video_extract 下载 mp4+原生整段提炼（base64→Gemini）
+  ②  判颗+类型闸+三lane成形+轻标签：system = phase1-frame.md
+  ③  作用域：system = phase2-scope.md → 候选 → scope_link 回扣（火山）
+  ⑤  组装：代码 → 每颗一个 ingest payload（按类型分拼）
+① 判 is_empty=true（无可提取知识）→ 短路，跳过 ②③⑤。
+数据源：fixture（已有 5 帖）或实时 crawler 取数（新帖）。须在云端跑。用法：PYTHONPATH=. python scripts/decompose.py
 """
 from __future__ import annotations
 
 import json
 from pathlib import Path
 
-from creation_knowledge.integrations.crawler import parse_detail_response
+from creation_knowledge.config import Settings
+from creation_knowledge.integrations import video_extract
+from creation_knowledge.integrations.crawler import fetch_post_detail, parse_detail_response
 from creation_knowledge.integrations.extractor import GeminiExtractor
 from creation_knowledge.integrations.llm import chat_json
 from scripts.scope_link import ScopeLinker
 
 ROOT = Path(__file__).resolve().parent.parent
 FIX = ROOT / "tests" / "fixtures"
+DATA = ROOT / "data" / "demo"
 SKILL = ROOT / "创作知识提取-skill"
 PHASE1 = (SKILL / "extraction" / "phase1-frame.md").read_text(encoding="utf-8")
 PHASE2 = (SKILL / "extraction" / "phase2-scope.md").read_text(encoding="utf-8")
 
-CIDS = ["699308fa0000000016009697", "698481e1000000000a02a7c1",
-        "67e2e39b0000000003028ff0", "680659e8000000001a007a11",
-        "67e4bdf50000000006028a59"]
+# from: fixture（读 tests/fixtures）/ live（实时 crawler 取数）
+SOURCES = [
+    {"cid": "699308fa0000000016009697", "platform": "xiaohongshu", "from": "fixture"},
+    {"cid": "698481e1000000000a02a7c1", "platform": "xiaohongshu", "from": "fixture"},
+    {"cid": "67e2e39b0000000003028ff0", "platform": "xiaohongshu", "from": "fixture"},
+    {"cid": "680659e8000000001a007a11", "platform": "xiaohongshu", "from": "fixture"},
+    {"cid": "67e4bdf50000000006028a59", "platform": "xiaohongshu", "from": "fixture"},
+    {"cid": "7589257893544165455", "platform": "douyin", "from": "live"},        # 抖音视频
+    {"cid": "6a33655e000000000f0055af", "platform": "xiaohongshu", "from": "live"},  # 无知识
+]
 SRC2CN = {"substance": "实质", "form": "形式", "feeling": "感受", "effect": "作用", "intent": "意图"}
 TYPE2ATTR = {"how": "how工序", "what": "what构成", "why": "why原理"}
 REUSE_THRESHOLD = 0.90
 
 
-# ---------- ① 读懂 ----------
-def read_post(cid: str, extractor: GeminiExtractor):
-    resp = json.loads((FIX / f"xhs_case_{cid}.json").read_text("utf-8"))
-    post = parse_detail_response(resp, fallback_content_id=cid)
+# ---------- 取数 ----------
+def load_post(src: dict, settings: Settings):
+    if src["from"] == "fixture":
+        resp = json.loads((FIX / f"xhs_case_{src['cid']}.json").read_text("utf-8"))
+        post = parse_detail_response(resp, fallback_content_id=src["cid"])
+    else:
+        post = fetch_post_detail(src["cid"], settings=settings)
     if not post.url:
-        post.url = f"https://www.xiaohongshu.com/explore/{cid}"
-    ec = extractor.extract(post)
+        post.url = f"https://www.xiaohongshu.com/explore/{src['cid']}"
+    return post
+
+
+# ---------- ① 读懂（图文/视频分流） ----------
+def read_one(src: dict, post, settings: Settings, extractor: GeminiExtractor):
+    cid = src["cid"]
+    if post.video_urls:  # 视频帖：下载 mp4 + 原生整段提炼
+        save = DATA / "douyin" / f"dy_{cid}" / "video.mp4"
+        pub = f"/data/demo/douyin/dy_{cid}/video.mp4"
+        ec = video_extract.extract_video(post, settings=settings, save_path=save, public_url=pub)
+        media = {"type": "video", "video_url": pub, "images": []}
+    else:                # 图文帖：读图
+        ec = extractor.extract(post)
+        if src["from"] == "fixture":
+            imgs = [f"/data/demo/xiaohongshu/{post.id}/image_{n}.webp" for n in range(1, len(post.image_urls) + 1)]
+        else:
+            imgs = list(post.image_urls)  # 新帖未落盘 → 直接用 CDN url
+        media = {"type": "image", "video_url": None, "images": imgs}
     parts = [ec.text]
     if ec.from_image:
         parts.append("【图片要点】\n" + ec.from_image)
     parts += [f"【卡片{c.index}】{c.content}" for c in ec.cards if c.content]
-    return post, "\n\n".join(p for p in parts if p), bool(ec.is_empty)
+    return "\n\n".join(p for p in parts if p), bool(ec.is_empty), media
 
 
 # ---------- ② 判颗+成形+轻标签 ----------
@@ -50,8 +81,7 @@ def shape(post, read: str) -> list[dict]:
     user = (f"原帖标题：{post.title or '(无)'}\n\n读懂后的完整内容：\n{read}\n\n"
             "按上面规则拆颗+判类型+成形+轻标签。作用域字段一律留空 []。"
             "只输出 JSON：{\"knowledges\":[ ... 见模板 ... ]}")
-    out = chat_json(PHASE1, user, timeout=120)
-    return out.get("knowledges") or []
+    return chat_json(PHASE1, user, timeout=120).get("knowledges") or []
 
 
 # ---------- ③ 作用域候选 + 回扣 ----------
@@ -69,13 +99,12 @@ def _slim(knowledges: list[dict]) -> list[dict]:
     return slim
 
 
-def scope_candidates(knowledges: list[dict]) -> dict:
+def scope_candidates(knowledges: list[dict]) -> list:
     user = ("给下面每颗知识标作用域候选（how 逐步：每个 step 一组；what/why 颗级：整颗一组）。\n"
-            "只输出 JSON：{\"scopes\":[{\"knowledge_id\":\"k1\",\"step_id\":\"s1\",\"items\":[{\"scope_type\":\"substance\",\"value\":\"赛道共识\"}]},"
+            "只输出 JSON：{\"scopes\":[{\"knowledge_id\":\"k1\",\"step_id\":\"s1\",\"items\":[{\"scope_type\":\"substance\",\"value\":\"…\"}]},"
             "{\"knowledge_id\":\"k2\",\"step_id\":null,\"items\":[...]}]}\n\n"
             + json.dumps(_slim(knowledges), ensure_ascii=False))
-    out = chat_json(PHASE2, user, timeout=120)
-    return out.get("scopes") or []
+    return chat_json(PHASE2, user, timeout=120).get("scopes") or []
 
 
 def link_scope(linker: ScopeLinker, scope_type: str, value: str) -> dict:
@@ -91,7 +120,7 @@ def link_scope(linker: ScopeLinker, scope_type: str, value: str) -> dict:
             "top": [{"name": h["name"], "score": h["score"], "path": h.get("path", "")} for h in hits]}
 
 
-def apply_scopes(knowledges: list[dict], scopes: list[dict], linker: ScopeLinker) -> None:
+def apply_scopes(knowledges: list[dict], scopes: list, linker: ScopeLinker) -> None:
     by_k = {k.get("id"): k for k in knowledges}
     for sc in scopes:
         k = by_k.get(sc.get("knowledge_id"))
@@ -107,7 +136,7 @@ def apply_scopes(knowledges: list[dict], scopes: list[dict], linker: ScopeLinker
             k["作用域"] = (k.get("作用域") or []) + linked
 
 
-# ---------- ⑤ 组装 payload ----------
+# ---------- ⑤ 组装 ----------
 def build_content(k: dict) -> str:
     t = k.get("type")
     if t == "how":
@@ -117,9 +146,8 @@ def build_content(k: dict) -> str:
                       f"  指引：{s.get('directive','')}", f"  产出：{s.get('output','')}"]
         return "\n".join(lines)
     if t == "what":
-        lines = [f"界定：{k.get('界定','')}", "构成："]
-        lines += [f"- {c.get('要素','')}：{c.get('说明','')}" for c in k.get("构成", [])]
-        return "\n".join(lines)
+        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('对创作的影响','')}"
 
 
@@ -150,7 +178,7 @@ def build_payload(post, k: dict, how_titles: dict | None = None) -> dict:
         p = k["parent"]
         ext.append({"key": "出自", "type": "str",
                     "value": f"{how_titles.get(p.get('how_id'), p.get('how_id'))} 第{p.get('step')}步"})
-    return {"source": {"id": f"xhs_{post.content_id}", "source_type": "post", "title": post.title or "",
+    return {"source": {"id": post.id, "source_type": "post", "title": post.title or "",
                        "author": post.author_name or "", "source_metadata": {"platform": post.platform, "url": post.url}},
             "title": k.get("title"), "content": build_content(k),
             "dim_creations": ["创作"], "dim_attributes": [TYPE2ATTR.get(t, "how工序")],
@@ -158,28 +186,36 @@ def build_payload(post, k: dict, how_titles: dict | None = None) -> dict:
 
 
 def main() -> None:
+    settings = Settings.from_env()
     extractor = GeminiExtractor.from_env()
     linker = ScopeLinker()
     posts_out, payloads = [], []
-    for cid in CIDS:
-        print(f"\n=== {cid[:8]} ===")
-        post, read, is_empty = read_post(cid, extractor)
-        meta = {"post_id": cid, "title": post.title or "", "platform": post.platform,
-                "url": post.url, "img_base": f"/data/demo/xiaohongshu/xhs_{cid}",
-                "img_count": len(post.image_urls)}
-        if is_empty:  # ① 总闸：判无知识 → 不走 ②③⑤
+    for src in SOURCES:
+        cid = src["cid"]
+        print(f"\n=== {src['platform']} {cid[:10]} ({src['from']}) ===")
+        try:
+            post = load_post(src, settings)
+            read, is_empty, media = read_one(src, post, settings, extractor)
+        except Exception as exc:
+            print(f"  ✗ 取数/读懂失败：{exc}")
+            posts_out.append({"post_id": cid, "source_id": cid, "title": f"(取数失败 {cid})",
+                              "platform": src["platform"], "url": "", "media": {"type": "image", "images": []},
+                              "error": str(exc)[:200], "knowledges": []})
+            continue
+        meta = {"post_id": cid, "source_id": post.id, "title": post.title or "",
+                "platform": post.platform, "url": post.url, "media": media}
+        if is_empty:  # ① 总闸
             print(f"  ① 读懂 {len(read)} 字 → 判定无可提取的创作知识，跳过拆颗")
             posts_out.append({**meta, "no_knowledge": True, "knowledges": []})
             continue
-        print(f"  ① 读懂 {len(read)} 字")
+        print(f"  ① 读懂 {len(read)} 字（{media['type']}）")
         knowledges = shape(post, read)
-        # 守卫：组件颗 parent 必须指向同帖一个 how，否则降级为 orphan 主颗
         how_ids = {k.get("id") for k in knowledges if k.get("type") == "how"}
         how_titles = {k.get("id"): k.get("title") for k in knowledges if k.get("type") == "how"}
         for k in knowledges:
-            if k.get("role") == "组件" and (k.get("parent") or {}).get("how_id") not in how_ids:
-                k["role"] = "主"
-                k["parent"] = None
+            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:  # 守卫：组件 parent 必指向同帖 how
+                k["role"] = "主"; k["parent"] = None
         print(f"  ② {len(knowledges)} 颗：" + ", ".join(f"{k.get('type')}/{k.get('role')}" for k in knowledges))
         apply_scopes(knowledges, scope_candidates(knowledges), linker)
         print("  ③⑤ 作用域回扣 + 组装")

scripts/rebuild_payloads.py

@@ -39,6 +39,7 @@ def main() -> None:
         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
@@ -72,7 +73,7 @@ def main() -> None:
                 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": f"xhs_{p['post_id']}", "source_type": "post", "title": p.get("title", ""),
+                "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})