Create-knowledge-find-decode/tests/test_decode_payloads.py

from __future__ import annotations

import json

import pytest

from core.models import Post
from decode_content.payloads import IngestPayloadValidationError, build_payload, build_payloads, validate_ingest_payload


def _post() -> Post:
    return Post(
        id="xhs_1",
        platform="xiaohongshu",
        url="https://xhs/1",
        content_id="1",
        unique_key="xhs:1",
        title="脚本教程",
        content_type="note",
        content_mode="image_post",
        media_count=2,
        cover_url="https://cdn.test/cover.webp",
        author_name="作者",
    )


def test_build_how_payload_uses_formal_dim_attribute_and_scope_union():
    knowledge = {
        "id": "k1",
        "type": "how",
        "title": "先定受众再写脚本",
        "purpose": "写出稳定脚本",
        "业务阶段": ["脚本"],
        "steps": [
            {
                "input": "选题",
                "directive": "收窄受众",
                "output": "受众画像",
                "创作阶段": "定向",
                "动作": "收窄",
                "作用域": [
                    {"scope_type": "intent", "value": "吸引注意", "link": "复用"},
                    {"scope_type": "intent", "value": "吸引注意", "link": "复用"},
                ],
            }
        ],
    }

    payload = build_payload(_post(), knowledge)

    assert payload["dim_attributes"] == ["how"]
    assert payload["source"]["source_metadata"] == {
        "platform": "xiaohongshu",
        "url": "https://xhs/1",
        "unique_key": "xhs:1",
        "platform_item_id": "1",
        "content_type": "note",
        "content_mode": "image_post",
        "media_count": 2,
        "cover_url": "https://cdn.test/cover.webp",
    }
    assert "输入：选题" in payload["content"]
    assert "指引：收窄受众" in payload["content"]
    assert payload["scopes"] == [{"scope_type": "intent", "value": "吸引注意"}]
    assert {"key": "创作阶段", "type": "str", "value": "定向"} in payload["custom_ext"]


def test_build_what_and_why_payload_content_is_string_contract():
    what = {
        "id": "w1",
        "type": "what",
        "title": "脚本三要素",
        "kind": "子集",
        "概要": "脚本由开头中段结尾构成",
        "维度拆分规则": "按表达顺序",
        "body": [{"item_name": "开头", "item_desc": "抓注意"}],
        "业务阶段": ["脚本"],
        "作用域": [{"scope_type": "form", "value": "脚本结构", "top": []}],
    }
    why = {
        "id": "y1",
        "type": "why",
        "title": "冲突带来停留",
        "阐述": "冲突能制造期待，所以提升停留。",
        "业务阶段": ["选题"],
        "作用域": [{"scope_type": "effect", "value": "提升停留", "score": 0.95}],
    }

    what_payload, why_payload = build_payloads(_post(), [what, why])

    assert what_payload["dim_attributes"] == ["what"]
    assert json.loads(what_payload["content"])["body"][0]["item_name"] == "开头"
    assert {"key": "概要", "type": "str", "value": "脚本由开头中段结尾构成"} in what_payload["custom_ext"]
    assert why_payload["dim_attributes"] == ["why"]
    assert why_payload["content"] == "冲突能制造期待，所以提升停留。"
    assert "score" not in why_payload["scopes"][0]


def test_build_why_payload_accepts_explanation_alias():
    payload = build_payload(
        _post(),
        {
            "id": "y2",
            "type": "why",
            "title": "动作变化带来画面节奏",
            "explanation": "动作变化能制造连续观看的期待。",
            "业务阶段": ["脚本"],
            "作用域": [{"scope_type": "effect", "value": "提升节奏感"}],
        },
    )

    assert payload["content"] == "动作变化能制造连续观看的期待。"


def test_validate_ingest_payload_blocks_external_api_contract_violations():
    payload = build_payload(
        _post(),
        {
            "id": "h1",
            "type": "how",
            "title": "先定受众再写脚本",
            "purpose": "写出稳定脚本",
            "业务阶段": ["脚本"],
            "steps": [
                {
                    "input": "选题",
                    "directive": "收窄受众",
                    "output": "受众画像",
                    "作用域": [{"scope_type": "intent", "value": "吸引注意"}],
                }
            ],
        },
    )

    bad = {**payload, "content": ""}
    with pytest.raises(IngestPayloadValidationError, match="content"):
        validate_ingest_payload(bad)

    bad = {**payload, "source": {**payload["source"], "id": "x" * 65}}
    with pytest.raises(IngestPayloadValidationError, match="source.id"):
        validate_ingest_payload(bad)

    bad = {**payload, "title": "x" * 513}
    with pytest.raises(IngestPayloadValidationError, match="title"):
        validate_ingest_payload(bad)

    bad = {**payload, "source": {**payload["source"], "author": "x" * 129}}
    with pytest.raises(IngestPayloadValidationError, match="author"):
        validate_ingest_payload(bad)

    bad = {**payload, "scopes": [{"scope_type": "intent", "value": "x" * 129}]}
    with pytest.raises(IngestPayloadValidationError, match="scopes"):
        validate_ingest_payload(bad)

    bad = {**payload, "custom_ext": [{"key": "推荐指数", "type": "bad", "value": "5"}]}
    with pytest.raises(IngestPayloadValidationError, match="custom_ext"):
        validate_ingest_payload(bad)


def test_validate_ingest_payload_rejects_legacy_dim_attributes():
    payload = build_payload(
        _post(),
        {
            "id": "h1",
            "type": "how",
            "title": "先定受众再写脚本",
            "purpose": "写出稳定脚本",
            "业务阶段": ["脚本"],
            "steps": [
                {
                    "input": "选题",
                    "directive": "收窄受众",
                    "output": "受众画像",
                    "作用域": [{"scope_type": "intent", "value": "吸引注意"}],
                }
            ],
        },
    )
    legacy_values = [
        "how" + "\u5de5\u5e8f",
        "what" + "\u6784\u6210",
        "why" + "\u539f\u7406",
    ]

    for legacy_value in legacy_values:
        bad = {**payload, "dim_attributes": [legacy_value]}
        with pytest.raises(IngestPayloadValidationError, match="dim_attributes"):
            validate_ingest_payload(bad)