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

"""创作知识解构引擎 v2：一帖 → N 颗（how/what/why，含组件颗）→ frameworks.json + payloads.json。

编排全流程，把 skill 的 phase 文档当 prompt 喂给 LLM（skill 是唯一真源）：
  ① 读懂：图文帖→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
import os
import re
import sys
import time
from pathlib import Path

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 creation_knowledge.integrations.oss import to_oss, upload_stream
from creation_knowledge.integrations.search import search_keyword
from creation_knowledge.prompts import load_prompt
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")
GATE_ADMIT = load_prompt("gate_admit")        # ①.5 创作判定闸：判"是创作"
GATE_REFUTE = load_prompt("gate_refute")      # ①.5：挑刺"是制作/越界"
GATE_TIEBREAK = load_prompt("gate_tiebreak")  # ①.5：分歧裁决（边界倾向排除）
NORMALIZE = load_prompt("normalize_scope")    # ③前：作用域值名词化（③LLM）
GATE_HOW_ADMIT = load_prompt("gate_how_admit")        # ②.5 有序性闸：判"真流水线"
GATE_HOW_REFUTE = load_prompt("gate_how_refute")      # ②.5：挑"假how（离散构成硬串）"
GATE_HOW_TIEBREAK = load_prompt("gate_how_tiebreak")  # ②.5：分歧裁决（边界倾向假how）
GATE_WHY_REFUTE = load_prompt("gate_why_refute")      # ②.7 why 轻闸：单票 refute（废话/实为what/实为how）

# ① 窄表确定性兜底：只放无歧义裸动作动词（绝不放 营造/引导/表达 等名词语素）
_STRIP_VERBS = ("寻找", "定位", "推导", "核验", "提取", "挖掘", "捕捉",
                "识别", "梳理", "归纳", "判断", "验证", "确认", "复盘")
_PROTECT = ("营造", "引导", "表达", "塑造", "叙述", "呈现", "刻画",
            "升华", "控制", "推进", "转化", "传达")  # 名词语素，永不砍

# 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": "6901e15a0000000003019fa3", "platform": "xiaohongshu", "from": "live"},  # 新增1
    {"cid": "69ff58670000000023016fa2", "platform": "xiaohongshu", "from": "live"},  # 新增2
    {"cid": "7589257893544165455", "platform": "douyin", "from": "live"},        # 抖音视频（倒数第2）
    {"cid": "6a33655e000000000f0055af", "platform": "xiaohongshu", "from": "live"},  # 无知识 Voodoo（最后）
]
# 入口二：按关键词搜索召回（与 SOURCES 并存）。跑法：python scripts/decompose.py queries
# 每条 {query, platform, content_type, limit}；limit 缺省取 settings.search_default_limit。
# 抖音 /keyword 实测 code:10000（需鉴权），本迭代只上小红书。
QUERIES = [
    {"query": "分镜脚本", "platform": "xiaohongshu", "content_type": "图文", "limit": 5},
]
SRC2CN = {"substance": "实质", "form": "形式", "feeling": "感受", "effect": "作用", "intent": "意图"}
TYPE2ATTR = {"how": "how", "what": "what", "why": "why"}
CSTAGE = {"定向", "构思", "结构", "成文", "打磨"}  # 创作阶段受控 5 值
KINDS = {"子集", "多维关系", "序列"}               # what.kind 受控 3 值
KIND_FIX = {"多维度关系": "多维关系", "多维": "多维关系", "集合": "子集", "序列型": "序列"}  # LLM 偶发变体归一
REUSE_THRESHOLD = 0.90


# ---------- 取数 ----------
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/{src['cid']}"
    return post


# ---------- ① 读懂（图文/视频分流） ----------
def read_one(src: dict, post, settings: Settings, extractor: GeminiExtractor):
    live = src["from"] == "live"
    if post.video_urls:  # 视频帖
        oss_url = ""
        if live:
            try:  # live：转存视频到 OSS 拿公网直链直喂（已验证 Gemini 可直读）；失败→显式回退
                oss_url = upload_stream(post.video_urls[0], src_type="video", settings=settings)
            except Exception as exc:
                print(f"  ⚠ 视频 OSS 转存失败，回退下载+base64：{exc}")
        if oss_url:  # 转存成功：OSS 直链直喂，跳过下载+base64
            ec = video_extract.extract_video(post, settings=settings, oss_video_url=oss_url, public_url=oss_url)
            pub = oss_url
        else:        # fixture / 转存失败：下载 mp4 + base64 兜底（落盘按平台分目录，post.id 已含平台前缀）
            save = DATA / post.platform / post.id / "video.mp4"
            pub = f"/data/demo/{post.platform}/{post.id}/video.mp4"
            ec = video_extract.extract_video(post, settings=settings, save_path=save, public_url=pub)
        media = {"type": "video", "video_url": pub, "images": []}
        cmap = {c.index: c.content for c in ec.cards}
        cards = [{"index": c.index, "content": cmap.get(c.index, ""), "video_url": pub,
                  "start": c.start, "end": c.end} for c in post.cards]  # 段卡：时间戳 + 读到的内容
    else:                # 图文帖：读图
        if live:  # live 帖：每张 OSS 转存绕防盗链，cdn_url 写回 post 再喂 extractor（直读 card.url）
            imgs = [to_oss(u, "image", settings=settings) for u in post.image_urls]
            post.image_urls = imgs
            for c in post.cards:
                if c.kind == "image" and 1 <= c.index <= len(imgs):
                    c.url = imgs[c.index - 1]
            ec = extractor.extract(post)
        else:     # fixture：读图后用预下载的本地 /data 路径展示
            ec = extractor.extract(post)
            imgs = [f"/data/demo/xiaohongshu/{post.id}/image_{n}.webp" for n in range(1, len(post.image_urls) + 1)]
        media = {"type": "image", "video_url": None, "images": imgs}
        cmap = {c.index: c.content for c in ec.cards}
        cards = [{"index": n, "content": cmap.get(n, ""), "image_url": imgs[n - 1] if n <= len(imgs) else None}
                 for n in range(1, len(imgs) + 1)]  # 每图一张卡：图 url + 读到的内容
    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 "\n\n".join(p for p in parts if p), bool(ec.is_empty), media, cards


# ---------- ② 判颗+成形+轻标签 ----------
def shape(post, read: str) -> list[dict]:
    user = (f"原帖标题：{post.title or '(无)'}\n\n读懂后的完整内容：\n{read}\n\n"
            "按上面规则拆颗+判类型+成形+轻标签。作用域字段一律留空 []。"
            "只输出 JSON：{\"knowledges\":[ ... 见模板 ... ]}")
    return chat_json(PHASE1, user, timeout=120).get("knowledges") or []


# ---------- ①.5 创作判定闸（voting：admit + refute，分歧上 tiebreak；边界倾向排除）----------
def _vote(system: str, read: str, key: str, on_fail: bool) -> bool:
    """跑一次判定，取 key 字段为 bool；出错按 on_fail 兜底（避免 API 抖动误杀）。"""
    try:
        return bool(chat_json(system, read, timeout=90).get(key))
    except Exception:
        return on_fail


def creation_gate(read: str) -> tuple[bool, str]:
    """判这帖是不是【图文/视频内容创作知识】。返回 (in_scope, 说明)。
    甲方案：admit + refute 两票；一致即定；分歧→tiebreak 裁决（边界倾向排除）。
    判定只看 ① 读懂后的内容，不看标题、不数关键词。"""
    v_admit = _vote(GATE_ADMIT, read, "in_scope", on_fail=True)            # 判"是创作"
    v_refute = not _vote(GATE_REFUTE, read, "out_of_scope", on_fail=False)  # 挑刺"是制作/越界"→归一成 in_scope
    if v_admit == v_refute:
        return v_admit, f"admit={v_admit}/refute={v_refute} 一致"
    v_tie = _vote(GATE_TIEBREAK, read, "in_scope", on_fail=False)           # 分歧裁决，失败也倾向排除
    return v_tie, f"admit={v_admit}/refute={v_refute} 分歧→裁决={v_tie}"


# ---------- ②.5 假how根治：有序性审查（层2信号+层3对抗投票）+ 假how重拆为 what/why ----------
def _chain_signals(steps: list[dict]) -> list[str]:
    """层2 确定性信号：并列输入（input 不指向前步产出）/ 产出近义。喂给层3当证据。"""
    sigs, outs = [], [(s.get("output") or "") for s in steps]
    indep = 0
    for i, s in enumerate(steps):
        if i == 0:
            continue
        inp = s.get("input") or ""
        if not (("←" in inp) or any(o and o[:4] in inp for o in outs[:i])):
            indep += 1
    if indep >= 2:
        sigs.append(f"{indep} 个后步的 input 未指向前步产出（各自起头）")
    for i in range(len(outs)):
        for j in range(i + 1, len(outs)):
            a, b = outs[i], outs[j]
            if a and b and (a in b or b in a):
                sigs.append(f"步骤{i+1}与{j+1}产出近义（{a} / {b}）")
                break
    return sigs


def how_gate(k: dict) -> tuple[bool, str]:
    """层3a 有序性闸（对抗投票：admit 判真链 / refute 挑假链 / 分歧裁决，边界倾向假how）。"""
    payload = json.dumps({
        "purpose": k.get("purpose"),
        "steps": [{"input": s.get("input"), "方法": (s.get("directive") or "")[:300], "产出": s.get("output")}
                  for s in k.get("steps", [])],
        "代码信号": _chain_signals(k.get("steps", [])),
    }, ensure_ascii=False)
    v_admit = _vote(GATE_HOW_ADMIT, payload, "is_real_how", on_fail=True)
    v_refute = not _vote(GATE_HOW_REFUTE, payload, "is_fake", on_fail=False)  # 不假 → 真
    if v_admit == v_refute:
        return v_admit, f"admit={v_admit}/refute={v_refute} 一致"
    v_tie = _vote(GATE_HOW_TIEBREAK, payload, "is_real_how", on_fail=False)
    return v_tie, f"admit={v_admit}/refute={v_refute} 分歧→裁决={v_tie}"


def reshape_nonhow(k: dict) -> list[dict]:
    """层3b 假how重拆：只拆成 What/Why 主颗（不要 how/组件）。失败则保留原颗，不丢内容。"""
    body = f"目标：{k.get('purpose','')}\n" + "\n".join(
        f"- 输入：{s.get('input','')}｜方法：{s.get('directive','')}｜产出：{s.get('output','')}"
        for s in k.get("steps", []))
    user = ("【下面这块原被误判为 how 工序，实为「离散构成 / 原理」，请只拆成 What/Why 主颗——"
            "每个'是什么/分几类'的构成块拆一颗 What，背后的原理/标准拆一颗 Why；"
            "不要 how、不要组件颗，parent 一律 null。作用域字段留空 []。】\n\n"
            f"原标题：{k.get('title','')}\n{body}\n\n"
            "只输出 JSON：{\"knowledges\":[ ... 仅 what/why，见模板 ... ]}")
    try:
        out = chat_json(PHASE1, user, timeout=120).get("knowledges") or []
    except Exception:
        out = []
    res = []
    for i, nk in enumerate(out, 1):
        if nk.get("type") == "how":   # 保险：拒绝又冒出来的 how
            continue
        nk["id"] = f"{k.get('id','k')}r{i}"
        nk["role"], nk["parent"] = "主", None
        res.append(nk)
    return res or [k]   # 兜底：没拆出来就保留原颗


def fix_fake_hows(knowledges: list[dict]) -> list[dict]:
    """逐颗 how 审查；假how → 重拆为 what/why（替换原颗）。"""
    out = []
    for k in knowledges:
        if k.get("type") == "how" and len(k.get("steps", [])) >= 2:
            real, why = how_gate(k)
            if not real:
                new = reshape_nonhow(k)
                print(f"  ②.5 假how「{k.get('title')}」（{why}）→ 重拆 {len(new)} 颗")
                out.extend(new)
                continue
        out.append(k)
    return out


# ---------- ②.7 why 轻闸：单票 refute（废话→drop；实为what/how→标 inferred 保留，不丢内容）----------
def drop_fake_whys(knowledges: list[dict]) -> list[dict]:
    out = []
    for k in knowledges:
        if k.get("type") == "why":
            payload = json.dumps({"阐述": k.get("阐述")}, ensure_ascii=False)
            try:
                r = chat_json(GATE_WHY_REFUTE, payload, timeout=90)
                not_why, verdict, reason = bool(r.get("not_why")), r.get("verdict", ""), r.get("reason", "")
            except Exception:
                not_why, verdict, reason = False, "", "API错误→保留"   # 兜底：保留，避免误杀
            if not_why:
                if verdict == "废话":   # 正确的废话：无复用价值，直接 drop
                    print(f"  ②.7 why闸：drop 废话「{k.get('title')}」（{reason}）")
                    continue
                if verdict in ("实为what", "实为how"):  # 误分类：保留但标 inferred，待人工看，不丢内容
                    k["inferred"] = True
                    k["inferred_reason"] = f"why闸疑似{verdict}：{reason}"
                    print(f"  ②.7 why闸：标记「{k.get('title')}」（{verdict}：{reason}）")
                # 其它（not_why 与 verdict 自相矛盾/空）→ 保守保留、不扣帽子
        out.append(k)
    return out


# ---------- ③ 作用域候选 + 定位 ----------
def _slim(knowledges: list[dict]) -> list[dict]:
    slim = []
    for k in knowledges:
        e = {"id": k.get("id"), "type": k.get("type"), "title": k.get("title")}
        if k.get("type") == "how":
            e["steps"] = [{"id": s.get("id"), "input": s.get("input"),
                           "directive": (s.get("directive") or "")[:500], "output": s.get("output")}
                          for s in k.get("steps", [])]
        else:
            e["内容"] = {x: k.get(x) for x in ("概要", "维度拆分规则", "body", "阐述") if k.get(x)}
        slim.append(e)
    return slim


def scope_candidates(knowledges: list[dict]) -> list:
    user = ("给下面每颗知识标作用域候选（how 逐步：每个 step 一组；what/why 颗级：整颗一组）。\n"
            "每类平铺列出所有相关值（同类可多条、全部对等、不选主次）；先脑内摊正交、收敛近义；意图通常 1 个。\n"
            "只输出 JSON：{\"scopes\":[{\"knowledge_id\":\"k1\",\"step_id\":\"s1\",\"items\":["
            "{\"scope_type\":\"substance\",\"value\":\"实质值1\"},{\"scope_type\":\"substance\",\"value\":\"实质值2\"},"
            "{\"scope_type\":\"form\",\"value\":\"形式值1\"},{\"scope_type\":\"intent\",\"value\":\"意图值\"}]},"
            "{\"knowledge_id\":\"k2\",\"step_id\":null,\"items\":[...]}]}\n\n"
            + json.dumps(_slim(knowledges), ensure_ascii=False))
    return chat_json(PHASE2, user, timeout=120).get("scopes") or []


def strip_verb_tail(v: str) -> str:
    """① 窄表确定性兜底：砍掉值开头/结尾的无歧义裸动词；砍到 <2 字则回退原值。"""
    if not v or len(v) < 3:
        return v
    for verb in _STRIP_VERBS:                       # 开头裸动词
        if v.startswith(verb) and len(v) - len(verb) >= 2:
            v = v[len(verb):]
            break
    for verb in _STRIP_VERBS:                        # 结尾裸动词（_PROTECT 与之不相交，名词语素天然不在表里）
        if v.endswith(verb) and len(v) - len(verb) >= 2:
            v = v[:-len(verb)]
            break
    return v


def nounify_scopes(scopes: list) -> list:
    """作用域值名词化（仅 实质/形式/感受/作用）：③ 批量 LLM 名词化 → ① 窄表兜底。在定位前做。
    意图豁免：意图值就该是动词（对齐意图树），不名词化、不剥动词。"""
    vals = sorted({it["value"] for sc in scopes for it in (sc.get("items") or [])
                   if it.get("value") and it.get("scope_type") != "intent"})
    mp = {}
    if vals:
        try:
            mp = chat_json(NORMALIZE, json.dumps(vals, ensure_ascii=False), timeout=90).get("映射") or {}
        except Exception:
            mp = {}
    for sc in scopes:
        for it in sc.get("items") or []:
            v = it.get("value")
            if not v or it.get("scope_type") == "intent":   # 意图原样保留（动词）
                continue
            it["value"] = strip_verb_tail(mp.get(v) or v)   # ③ 映射优先，再 ① 兜底
    return scopes


def link_scope(linker: ScopeLinker, scope_type: str, value: str) -> dict:
    try:
        hits = linker.link(value, source_type=SRC2CN.get(scope_type, scope_type), top_k=3)
    except Exception:
        hits = []
    top = hits[0] if hits else {}
    score = float(top.get("score", 0.0))
    reuse = score >= REUSE_THRESHOLD and top.get("name")
    return {"scope_type": scope_type, "value": top["name"] if reuse else value,
            "candidate": value, "link": "复用" if reuse else "新建", "score": round(score, 4),
            "top": [{"name": h["name"], "score": h["score"], "path": h.get("path", "")} for h in hits]}


_SCOPE_CONN = re.compile(r"\s*[和与、，,/／&]\s*")  # 作用域值确定性拆分：含连接词必拆成多原子


def _split_scope_items(items: list) -> list:
    """连接词拆分守卫：把"社会共识与个体真实矛盾"这种拆成多原子，平铺。LLM 漏拆时兜底。"""
    out = []
    for it in (items or []):
        st, v = it.get("scope_type"), it.get("value")
        if not (st and v):
            continue
        parts = [p.strip() for p in _SCOPE_CONN.split(v) if p.strip()]
        for p in (parts if len(parts) > 1 else [v]):
            out.append({"scope_type": st, "value": p})
    return out


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"))
        if not k:
            continue
        linked = [link_scope(linker, it["scope_type"], it["value"])
                  for it in _split_scope_items(sc.get("items"))]
        if k.get("type") == "how" and sc.get("step_id"):
            for s in k.get("steps", []):
                if s.get("id") == sc["step_id"]:
                    s["作用域"] = linked
        else:
            k["作用域"] = (k.get("作用域") or []) + linked


# ---------- ⑤ 组装 ----------
def build_content(k: dict):
    """How → 拍平字符串；What → 结构化实体对象 {name,kind,维度拆分规则,body[]}；Why → 结构化对象 {name,desc}。"""
    t = k.get("type")
    if t == "how":
        lines = [f"目标：{k.get('purpose','')}"]
        for i, s in enumerate(k.get("steps", []), 1):
            lines += [f"步骤{i}",
                      f"  输入：{s.get('input','')}", f"  方法：{s.get('directive','')}", f"  产出：{s.get('output','')}"]
        return "\n".join(lines)
    if t == "what":  # 结构化实体（非字符串）；概要走 custom_ext，不进 content
        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 {"name": k.get("title") or "", "desc": k.get("阐述") or ""}


def _sections(blocks) -> list[str]:
    """把 why.支撑 的自由小节拼成文本行（What 已改结构化 body，不再走这里）。"""
    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 build_payload(post, k: dict, how_titles: dict | None = None) -> dict:
    how_titles = how_titles or {}
    t = k.get("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("概要"):  # What 的概要迁到 custom_ext（不再进 content）
        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"):
        p = k["parent"]
        ext.append({"key": "出自", "type": "str",
                    "value": f"{how_titles.get(p.get('how_id'), p.get('how_id'))} 第{p.get('step')}步"})
    c0 = build_content(k)  # how=字符串；what/why=结构化对象→json.dumps 成字符串（ingest 接口要求 content 为 string）
    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": c0 if isinstance(c0, str) else json.dumps(c0, ensure_ascii=False),
            "dim_creations": ["创作"], "dim_attributes": [TYPE2ATTR.get(t, "how")],
            "scopes": scopes, "custom_ext": ext}


def process_one(src: dict, settings: Settings, extractor: GeminiExtractor,
                linker: ScopeLinker) -> tuple[dict, list[dict]]:
    """跑通一帖：取数→读懂→闸→拆颗→作用域→组装。返回 (post_out_entry, payloads)。

    从 main 循环体原样抽出（零行为变更）；SOURCES 与 QUERIES 两条入口共用。"""
    cid = src["cid"]
    print(f"\n=== {src['platform']} {cid[:10]} ({src['from']}) ===")
    try:
        post = load_post(src, settings)
        read, is_empty, media, cards = read_one(src, post, settings, extractor)
    except Exception as exc:
        print(f"  ✗ 取数/读懂失败：{exc}")
        return ({"post_id": cid, "source_id": cid, "title": f"(取数失败 {cid})",
                 "platform": src["platform"], "url": "", "media": {"type": "image", "images": []},
                 "cards": [], "error": str(exc)[:200], "knowledges": []}, [])
    meta = {"post_id": cid, "source_id": post.id, "title": post.title or "",
            "platform": post.platform, "url": post.url, "media": media, "cards": cards}
    if is_empty:  # ① 总闸：纯展示/无可提取
        print(f"  ① 读懂 {len(read)} 字 → ① 判纯展示/无知识，跳过")
        return ({**meta, "no_knowledge": True, "knowledges": []}, [])
    in_scope, gate_why = creation_gate(read)  # ①.5 创作判定闸（创作 vs 制作 vs 越界）
    if not in_scope:
        print(f"  ① 读懂 {len(read)} 字 → ①.5 闸判【非创作】（{gate_why}）→ 整帖排除")
        return ({**meta, "no_knowledge": True, "knowledges": []}, [])
    print(f"  ① 读懂 {len(read)} 字（{media['type']}）· ①.5 闸：创作（{gate_why}）")
    knowledges = drop_fake_whys(fix_fake_hows(shape(post, read)))   # ② 拆颗 → ②.5 假how根治 → ②.7 why轻闸
    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:
        k["业务阶段"] = [b for b in (k.get("业务阶段") or []) if b in ("灵感", "选题", "脚本")]  # 守卫：只留合法业务阶段
        if k.get("type") == "what" and k.get("kind") not in KINDS:  # 守卫：what.kind 受控值，LLM 偶发变体归一
            k["kind"] = KIND_FIX.get((k.get("kind") or "").strip(), k.get("kind"))
        for s in k.get("steps", []):  # 守卫：创作阶段只留合法 5 值，非法（如"定稿/输出"）丢弃
            if s.get("创作阶段") not in CSTAGE:
                s["创作阶段"] = None
        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, nounify_scopes(scope_candidates(knowledges)), linker)  # ③名词化+①兜底 → 定位
    print("  ③⑤ 作用域定位 + 组装")
    return ({**meta, "knowledges": knowledges},
            [build_payload(post, k, how_titles) for k in knowledges])


# ---------- 入口二：query 搜索召回 + run 隔离产出 ----------
def expand_queries(queries: list[dict], settings: Settings) -> list[dict]:
    """逐 query 搜索 → content_id；跨 query 按 id 去重。搜索失败跳过该 query（不阻断整批）。"""
    seen: set[str] = set()
    out: list[dict] = []
    for q in queries:
        platform = q.get("platform", "xiaohongshu")
        try:
            ids = search_keyword(
                q["query"], platform=platform,
                content_type=q.get("content_type") or settings.search_content_type,
                sort_type=q.get("sort_type") or settings.search_sort_type,
                limit=q.get("limit") or settings.search_default_limit,
                settings=settings)
        except Exception as exc:
            print(f"  ✗ 搜索失败 [{q.get('query')}]：{exc}")
            continue
        new = 0
        for cid in ids:
            if cid in seen:
                continue
            seen.add(cid)
            out.append({"cid": cid, "platform": platform, "from": "live", "query": q["query"]})
            new += 1
        print(f"  🔎 [{q['query']}] 命中 {len(ids)} → 新增 {new}（去重后）")
    return out


def slugify(text: str) -> str:
    """run 文件名：保留中英文字与数字，其余压成 -。空则 'run'。"""
    s = re.sub(r"[^\w一-鿿]+", "-", text or "").strip("-").lower()
    return s or "run"


def write_run(posts_out: list[dict], payloads: list[dict], queries: list[dict]) -> str:
    """写 web/runs/<slug>.json + <slug>.payloads.json + 追加 web/runs/index.json 清单。
    同名 slug 已存在则加时间戳保留历史。返回最终 slug。"""
    runs_dir = ROOT / "web" / "runs"
    runs_dir.mkdir(parents=True, exist_ok=True)
    slug = slugify("-".join(q["query"] for q in queries))
    if (runs_dir / f"{slug}.json").exists():
        slug = f"{slug}-{int(time.time())}"
    (runs_dir / f"{slug}.json").write_text(
        json.dumps({"count": len(posts_out), "posts": posts_out, "queries": queries},
                   ensure_ascii=False, indent=1), encoding="utf-8")
    (runs_dir / f"{slug}.payloads.json").write_text(
        json.dumps(payloads, ensure_ascii=False, indent=2), encoding="utf-8")
    idx_path = runs_dir / "index.json"
    try:
        idx = json.loads(idx_path.read_text("utf-8")) if idx_path.exists() else []
    except Exception:
        idx = []
    idx = [r for r in idx if r.get("slug") != slug]  # 同 slug 去重
    idx.insert(0, {"slug": slug, "queries": [q["query"] for q in queries],
                   "count": len(posts_out), "created_at": int(time.time())})
    idx_path.write_text(json.dumps(idx, ensure_ascii=False, indent=1), encoding="utf-8")
    print(f"\nwrote {len(posts_out)} posts, {len(payloads)} payloads → web/runs/{slug}.json")
    return slug


def _run(srcs: list[dict], settings: Settings, extractor: GeminiExtractor,
         linker: ScopeLinker) -> tuple[list[dict], list[dict]]:
    posts_out, payloads = [], []
    for src in srcs:
        entry, pl = process_one(src, settings, extractor, linker)
        posts_out.append(entry)
        payloads += pl
    return posts_out, payloads


def main() -> None:
    settings = Settings.from_env()
    extractor = GeminiExtractor.from_env()
    linker = ScopeLinker()
    mode = sys.argv[1] if len(sys.argv) > 1 else "sources"
    if mode == "queries":  # 入口二：搜索召回 → run 隔离产出（不动 frameworks.json 样本）
        srcs = expand_queries(QUERIES, settings)
        if not srcs:
            print("没有召回到任何帖子，结束。")
            return
        posts_out, payloads = _run(srcs, settings, extractor, linker)
        write_run(posts_out, payloads, QUERIES)
    else:                  # 入口一（默认）：写死 SOURCES → web/frameworks.json
        posts_out, payloads = _run(SOURCES, settings, extractor, linker)
        suffix = os.environ.get("OUT", "")  # OUT=_v2 → 写 frameworks_v2.json，不覆盖原版
        (ROOT / f"web/frameworks{suffix}.json").write_text(
            json.dumps({"count": len(posts_out), "posts": posts_out}, ensure_ascii=False, indent=1), encoding="utf-8")
        (ROOT / f"web/payloads{suffix}.json").write_text(
            json.dumps(payloads, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"\nwrote {len(posts_out)} posts, {len(payloads)} payloads → web/frameworks{suffix}.json")


if __name__ == "__main__":
    main()