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

CUSTOM_EXT_TYPES = {"str", "num", "date"}
DIM_ATTRIBUTES = {"how", "what", "why"}


class IngestPayloadValidationError(ValueError):
    """Raised when a generated payload cannot satisfy the ingest API contract."""


def _text(value: Any) -> str:
    return value if isinstance(value, str) else ""


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 knowledge.get("explanation") 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 validate_ingest_payload(payload: dict[str, Any]) -> None:
    """Validate the minimal external ingest API contract before storing/sending."""

    errors: list[str] = []
    source = payload.get("source")
    if not isinstance(source, dict):
        source = {}
        errors.append("source 必须是对象")
    source_id = _text(source.get("id")).strip()
    if not source_id:
        errors.append("source.id 不能为空")
    elif len(source_id) > 64:
        errors.append("source.id 最长 64 字符")
    source_type = _text(source.get("source_type"))
    if len(source_type) > 32:
        errors.append("source.source_type 最长 32 字符")
    source_title = _text(source.get("title"))
    if len(source_title) > 512:
        errors.append("source.title 最长 512 字符")
    source_author = _text(source.get("author"))
    if len(source_author) > 128:
        errors.append("source.author 最长 128 字符")
    title = payload.get("title")
    if title is not None and len(_text(title)) > 512:
        errors.append("title 最长 512 字符")
    content = payload.get("content")
    if not isinstance(content, str) or not content.strip():
        errors.append("content 必须是非空字符串")
    dim_attributes = payload.get("dim_attributes")
    if not isinstance(dim_attributes, list) or len(dim_attributes) != 1:
        errors.append("dim_attributes 必须恰好包含一个值")
    elif dim_attributes[0] not in DIM_ATTRIBUTES:
        errors.append("dim_attributes 只能是 how/what/why")
    for idx, scope in enumerate(payload.get("scopes") or [], 1):
        if not isinstance(scope, dict):
            errors.append(f"scopes[{idx}] 必须是对象")
            continue
        value = _text(scope.get("value"))
        if len(value) > 128:
            errors.append(f"scopes[{idx}].value 最长 128 字符")
    for idx, ext in enumerate(payload.get("custom_ext") or [], 1):
        if not isinstance(ext, dict):
            errors.append(f"custom_ext[{idx}] 必须是对象")
            continue
        ext_type = ext.get("type")
        if ext_type not in CUSTOM_EXT_TYPES:
            errors.append(f"custom_ext[{idx}].type 必须是 str/num/date")
    if errors:
        title_hint = _text(payload.get("title")) or source_id or "unknown"
        raise IngestPayloadValidationError(f"{title_hint}: {'; '.join(errors)}")


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 _source_metadata(post: Any) -> dict[str, Any]:
    metadata = {
        "platform": _post_value(post, "platform"),
        "url": _post_value(post, "url"),
        "unique_key": _post_value(post, "unique_key"),
        "platform_item_id": _post_value(post, "content_id"),
        "content_type": _post_value(post, "content_type"),
        "content_mode": _post_value(post, "content_mode"),
        "media_count": _post_value(post, "media_count", 0),
        "cover_url": _post_value(post, "cover_url"),
    }
    return {
        key: value
        for key, value in metadata.items()
        if value is not None and value != ""
    }


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")
    if t not in TYPE2ATTR:
        raise IngestPayloadValidationError(f"{knowledge.get('title') or 'unknown'}: unknown knowledge type {t}")
    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')}步",
            }
        )
    payload = {
        "source": {
            "id": _post_value(post, "id"),
            "source_type": "post",
            "title": _post_value(post, "title") or "",
            "author": _post_value(post, "author_name") or "",
            "source_metadata": _source_metadata(post),
        },
        "title": knowledge.get("title"),
        "content": _payload_content(knowledge),
        "dim_creations": ["创作"],
        "dim_attributes": [TYPE2ATTR[t]],
        "scopes": collect_payload_scopes(knowledge),
        "custom_ext": ext,
    }
    validate_ingest_payload(payload)
    return payload


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]