Create-knowledge-find-decode/decode_content/framing.py

"""Knowledge particle framing and type-shape cleanup."""
from __future__ import annotations

import json
from typing import Any

from core.llm import chat_json as default_chat_json
from core.models import Post
from decode_content.contracts import BUSINESS_STAGES, CREATION_STAGES, WHAT_KINDS, SkillContract, load_contract
from decode_content.gates import ChatJsonFn, how_gate, why_refute_gate


CSTAGE = set(CREATION_STAGES)
KINDS = set(WHAT_KINDS)
KIND_FIX = {"多维度关系": "多维关系", "多维": "多维关系", "集合": "子集", "序列型": "序列"}


def frame_knowledges(
    post: Post,
    read_text: str,
    *,
    contract: SkillContract | None = None,
    chat_json_fn: ChatJsonFn = default_chat_json,
) -> list[dict[str, Any]]:
    contract = contract or load_contract()
    user = (
        f"原帖标题：{post.title or '(无)'}\n\n读懂后的完整内容：\n{read_text}\n\n"
        "按上面规则拆颗+判类型+成形+轻标签。作用域字段一律留空 []。"
        "只输出 JSON：{\"knowledges\":[ ... 见模板 ... ]}"
    )
    return chat_json_fn(contract.phase1, user, timeout=120).get("knowledges") or []


def reshape_nonhow(
    knowledge: dict[str, Any],
    *,
    contract: SkillContract | None = None,
    chat_json_fn: ChatJsonFn = default_chat_json,
) -> list[dict[str, Any]]:
    contract = contract or load_contract()
    body = f"目标：{knowledge.get('purpose', '')}\n" + "\n".join(
        f"- 输入：{step.get('input', '')}｜方法：{step.get('directive', '')}｜产出：{step.get('output', '')}"
        for step in knowledge.get("steps", [])
    )
    user = (
        "【下面这块原被误判为 how 工序，实为「离散构成 / 原理」，请只拆成 What/Why 主颗——"
        "每个'是什么/分几类'的构成块拆一颗 What，背后的原理/标准拆一颗 Why；"
        "不要 how、不要组件颗，parent 一律 null。作用域字段留空 []。】\n\n"
        f"原标题：{knowledge.get('title', '')}\n{body}\n\n"
        "只输出 JSON：{\"knowledges\":[ ... 仅 what/why，见模板 ... ]}"
    )
    try:
        out = chat_json_fn(contract.phase1, user, timeout=120).get("knowledges") or []
    except Exception:
        out = []
    reshaped: list[dict[str, Any]] = []
    for idx, item in enumerate(out, 1):
        if item.get("type") == "how":
            continue
        item["id"] = f"{knowledge.get('id', 'k')}r{idx}"
        item["role"], item["parent"] = "主", None
        reshaped.append(item)
    return reshaped or [knowledge]


def fix_fake_hows(
    knowledges: list[dict[str, Any]],
    *,
    contract: SkillContract | None = None,
    chat_json_fn: ChatJsonFn = default_chat_json,
) -> list[dict[str, Any]]:
    out: list[dict[str, Any]] = []
    for knowledge in knowledges:
        if knowledge.get("type") == "how" and len(knowledge.get("steps", [])) >= 2:
            gate = how_gate(knowledge, chat_json_fn=chat_json_fn)
            if not gate.passed:
                reshaped = reshape_nonhow(knowledge, contract=contract, chat_json_fn=chat_json_fn)
                for item in reshaped:
                    item.setdefault("metadata", {})["how_gate_reason"] = gate.reason
                out.extend(reshaped)
                continue
        out.append(knowledge)
    return out


def drop_fake_whys(
    knowledges: list[dict[str, Any]],
    *,
    chat_json_fn: ChatJsonFn = default_chat_json,
) -> list[dict[str, Any]]:
    out: list[dict[str, Any]] = []
    for knowledge in knowledges:
        if knowledge.get("type") == "why":
            gate = why_refute_gate(knowledge, chat_json_fn=chat_json_fn)
            if not gate.passed:
                verdict = gate.details.get("verdict")
                if verdict == "废话":
                    continue
                if verdict in {"实为what", "实为how"}:
                    knowledge["inferred"] = True
                    knowledge["inferred_reason"] = f"why闸疑似{verdict}：{gate.reason}"
        out.append(knowledge)
    return out


def normalize_knowledge_shapes(knowledges: list[dict[str, Any]]) -> list[dict[str, Any]]:
    how_ids = {k.get("id") for k in knowledges if k.get("type") == "how"}
    for knowledge in knowledges:
        knowledge["业务阶段"] = [
            stage for stage in (knowledge.get("业务阶段") or []) if stage in BUSINESS_STAGES
        ]
        if knowledge.get("type") == "what" and knowledge.get("kind") not in KINDS:
            knowledge["kind"] = KIND_FIX.get((knowledge.get("kind") or "").strip(), knowledge.get("kind"))
        for step in knowledge.get("steps", []):
            if step.get("创作阶段") not in CSTAGE:
                step["创作阶段"] = None
        if knowledge.get("role") == "组件" and (knowledge.get("parent") or {}).get("how_id") not in how_ids:
            knowledge["role"] = "主"
            knowledge["parent"] = None
    return knowledges


def frame_and_clean(
    post: Post,
    read_text: str,
    *,
    contract: SkillContract | None = None,
    chat_json_fn: ChatJsonFn = default_chat_json,
) -> list[dict[str, Any]]:
    contract = contract or load_contract()
    knowledges = frame_knowledges(post, read_text, contract=contract, chat_json_fn=chat_json_fn)
    knowledges = fix_fake_hows(knowledges, contract=contract, chat_json_fn=chat_json_fn)
    knowledges = drop_fake_whys(knowledges, chat_json_fn=chat_json_fn)
    return normalize_knowledge_shapes(knowledges)


def knowledge_debug_json(knowledges: list[dict[str, Any]]) -> str:
    return json.dumps(knowledges, ensure_ascii=False)