"""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]