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 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",
        canonical_url="https://xhs/item/abc",
        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.image_urls == ["oss0", "cdn1"]
    assert [card.index for card in post.cards] == [1, 2]
    assert post.body_text == "正文"
    assert post.topic_list == ["脚本"]


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", 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 == "片段知识"