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

"""Deterministic payload assembly for decoded creation knowledge."""
from __future__ import annotations

import json
from typing import Any

from decode_content.contracts import DIM_ATTRIBUTE_BY_TYPE


TYPE2ATTR = DIM_ATTRIBUTE_BY_TYPE


def _post_value(post: Any, key: str, default: Any = "") -> Any:
    if isinstance(post, dict):
        return post.get(key, default)
    return getattr(post, key, default)


def build_content(knowledge: dict[str, Any]) -> str | dict[str, Any]:
    t = knowledge.get("type")
    if t == "how":
        lines = [f"目标：{knowledge.get('purpose', '')}"]
        for idx, step in enumerate(knowledge.get("steps", []), 1):
            intent = step.get("intent") or step.get("purpose") or ""
            label = f"步骤{idx}" + (f"（目的：{intent}）" if intent else "")
            lines.extend(
                [
                    label,
                    f"  输入：{step.get('input', '')}",
                    f"  指引：{step.get('directive', '')}",
                    f"  产出：{step.get('output', '')}",
                ]
            )
        return "\n".join(lines)
    if t == "what":
        return {
            "name": knowledge.get("title") or "",
            "kind": knowledge.get("kind"),
            "维度拆分规则": knowledge.get("维度拆分规则") if knowledge.get("kind") == "子集" else None,
            "body": [
                {
                    "item_name": item.get("item_name", ""),
                    "item_desc": item.get("item_desc", ""),
                    "作用域": item.get("作用域") or [],
                }
                for item in (knowledge.get("body") or [])
            ],
        }
    return knowledge.get("阐述") or knowledge.get("desc") or ""


def _payload_content(knowledge: dict[str, Any]) -> str:
    content = build_content(knowledge)
    return content if isinstance(content, str) else json.dumps(content, ensure_ascii=False)


def _add_scopes(scopes: list[dict[str, str]], seen: set[tuple[str, str]], items: list[dict[str, Any]]) -> None:
    for item in items:
        scope_type = item.get("scope_type")
        value = item.get("value")
        if not (scope_type and value):
            continue
        key = (scope_type, value)
        if key in seen:
            continue
        seen.add(key)
        scopes.append({"scope_type": scope_type, "value": value})


def collect_payload_scopes(knowledge: dict[str, Any]) -> list[dict[str, str]]:
    scopes: list[dict[str, str]] = []
    seen: set[tuple[str, str]] = set()
    if knowledge.get("type") == "how":
        for step in knowledge.get("steps", []):
            _add_scopes(scopes, seen, step.get("作用域", []))
    else:
        _add_scopes(scopes, seen, knowledge.get("作用域", []))
    return scopes


def build_payload(post: Any, knowledge: dict[str, Any], how_titles: dict[str, str] | None = None) -> dict[str, Any]:
    how_titles = how_titles or {}
    t = knowledge.get("type")
    ext = [
        {"key": "业务阶段", "type": "str", "value": stage}
        for stage in (knowledge.get("业务阶段") or [])
    ]
    if t == "what" and knowledge.get("概要"):
        ext.append({"key": "概要", "type": "str", "value": knowledge["概要"]})
    if t == "how":
        creation_stages: list[str] = []
        seen_stages: set[str] = set()
        for step in knowledge.get("steps", []):
            stage = step.get("创作阶段")
            if stage and stage not in seen_stages:
                seen_stages.add(stage)
                creation_stages.append(stage)
        ext.extend({"key": "创作阶段", "type": "str", "value": stage} for stage in creation_stages)
        ext.extend(
            {"key": "动作", "type": "str", "value": step["动作"]}
            for step in knowledge.get("steps", [])
            if step.get("动作")
        )
    if knowledge.get("role") == "组件" and knowledge.get("parent"):
        parent = knowledge["parent"]
        ext.append(
            {
                "key": "出自",
                "type": "str",
                "value": f"{how_titles.get(parent.get('how_id'), parent.get('how_id'))} 第{parent.get('step')}步",
            }
        )
    return {
        "source": {
            "id": _post_value(post, "id"),
            "source_type": "post",
            "title": _post_value(post, "title") or "",
            "author": _post_value(post, "author_name") or "",
            "source_metadata": {
                "platform": _post_value(post, "platform"),
                "url": _post_value(post, "url"),
            },
        },
        "title": knowledge.get("title"),
        "content": _payload_content(knowledge),
        "dim_creations": ["创作"],
        "dim_attributes": [TYPE2ATTR.get(t, "how工序")],
        "scopes": collect_payload_scopes(knowledge),
        "custom_ext": ext,
    }


def build_payloads(post: Any, knowledges: list[dict[str, Any]]) -> list[dict[str, Any]]:
    how_titles = {k.get("id"): k.get("title") for k in knowledges if k.get("type") == "how"}
    return [build_payload(post, knowledge, how_titles) for knowledge in knowledges]