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

from __future__ import annotations

from uuid import uuid4

from acquisition.domain import CandidateItem, MediaAsset
from core.models import Card, CardExtract, ExtractedContent, Post
from decode_content.readers.service import (
    UnsupportedPostModeError,
    post_from_candidate_item,
    read_item,
    read_post,
)


def test_post_from_candidate_item_prefers_cdn_then_oss_then_source():
    item_id = uuid4()
    item = CandidateItem(
        id=item_id,
        platform="xiaohongshu",
        platform_item_id="abc",
        unique_key="xhs:abc",
        canonical_url="https://xhs/item/abc",
        content_mode="image_post",
        title="脚本结构",
        author_name="sam",
        raw_summary="正文",
        source_payload={"topic_list": ["脚本"]},
    )
    media = [
        MediaAsset(item_id=item_id, media_type="image", source_url="source1", cdn_url="cdn1", position=2),
        MediaAsset(item_id=item_id, media_type="image", source_url="source0", oss_url="oss0", position=1),
    ]

    post = post_from_candidate_item(item, media)

    assert post.id == "xiaohongshu_abc"
    assert post.content_id == "abc"
    assert post.unique_key == "xhs:abc"
    assert post.content_mode == "image_post"
    assert post.image_urls == ["oss0", "cdn1"]
    assert post.media_count == 2
    assert post.cover_url == "oss0"
    assert [card.index for card in post.cards] == [1, 2]
    assert post.body_text == "正文"
    assert post.topic_list == ["脚本"]


def test_post_from_candidate_item_prefers_full_body_text_over_summary():
    item_id = uuid4()
    item = CandidateItem(
        id=item_id,
        platform="weixin",
        platform_item_id="wx1",
        content_mode="article",
        title="长文",
        body_text="完整正文",
        raw_summary="摘要",
    )

    post = post_from_candidate_item(item, [])

    assert post.body_text == "完整正文"


def test_post_from_candidate_item_reads_legacy_nested_body_before_summary():
    item_id = uuid4()
    item = CandidateItem(
        id=item_id,
        platform="xiaohongshu",
        platform_item_id="x1",
        content_mode="image_post",
        raw_summary="摘要",
        source_payload={"detail": {"data": {"data": {"body_text": "旧 raw 正文"}}}},
    )

    post = post_from_candidate_item(item, [])

    assert post.body_text == "旧 raw 正文"


def test_read_post_imgtext_merges_text_cards_and_media():
    post = Post(
        id="xhs_1",
        platform="xiaohongshu",
        url="u",
        content_id="1",
        image_urls=["img1", "img2"],
        cards=[Card(index=1, kind="image", url="img1"), Card(index=2, kind="image", url="img2")],
    )

    def image_reader(p: Post) -> ExtractedContent:
        assert p.id == "xhs_1"
        return ExtractedContent(
            text="主体",
            from_image="图片知识",
            cards=[CardExtract(index=1, content="卡1"), CardExtract(index=2, content="卡2")],
        )

    result = read_post(post, image_reader=image_reader)

    assert "图片知识" in result.text
    assert result.cards[0].content == "卡1"
    assert result.media == {"type": "image", "video_url": None, "images": ["img1", "img2"]}


def test_read_item_routes_video_to_video_reader():
    item_id = uuid4()
    item = CandidateItem(
        id=item_id,
        platform="douyin",
        platform_item_id="v1",
        content_mode="video_post",
        title="视频",
    )
    media = [MediaAsset(item_id=item_id, media_type="video", cdn_url="https://cdn/video.mp4", position=1)]

    def video_reader(post: Post) -> ExtractedContent:
        assert post.video_urls == ["https://cdn/video.mp4"]
        post.cards = [Card(index=1, kind="segment", url=post.video_urls[0], start=0, end=12)]
        return ExtractedContent(text="视频理解", cards=[CardExtract(index=1, content="片段知识")])

    result = read_item(item, media, video_reader=video_reader)

    assert result.media["type"] == "video"
    assert result.cards[0].kind == "segment"
    assert result.cards[0].content == "片段知识"


def test_video_source_url_without_oss_or_cdn_is_not_decodable():
    import pytest

    item_id = uuid4()
    item = CandidateItem(
        id=item_id,
        platform="douyin",
        platform_item_id="v1",
        content_mode="video_post",
        title="视频",
    )
    media = [
        MediaAsset(
            item_id=item_id,
            media_type="video",
            source_url="https://video.test/temporary.mp4",
            position=1,
        )
    ]

    post = post_from_candidate_item(item, media)

    assert post.video_urls == []
    with pytest.raises(UnsupportedPostModeError, match="video_url_missing"):
        read_item(item, media)


def test_read_post_rejects_unsupported_and_missing_video():
    unsupported = Post(
        id="dy_x",
        platform="douyin",
        url="u",
        content_id="x",
        content_mode="unsupported",
    )
    missing_video = Post(
        id="dy_v",
        platform="douyin",
        url="u",
        content_id="v",
        content_mode="video_post",
    )

    import pytest

    with pytest.raises(UnsupportedPostModeError, match="unsupported_content_mode"):
        read_post(unsupported)
    with pytest.raises(UnsupportedPostModeError, match="video_url_missing"):
        read_post(missing_video)