Create-knowledge-find-decode/decode_content/readers/service.py

"""Unified post reader for image-text and video decode inputs."""
from __future__ import annotations

from typing import Protocol

from acquisition.domain import CandidateItem, MediaAsset
from core.config import Settings
from core.models import Card, ExtractedContent, Post
from decode_content.models import ReadCard, ReadResult
from decode_content.readers.imgtext import BailianExtractor, read_imgtext
from decode_content.readers.video import read_video


class ImageReader(Protocol):
    def __call__(self, post: Post) -> ExtractedContent:
        ...


class VideoReader(Protocol):
    def __call__(self, post: Post) -> ExtractedContent:
        ...


def _media_url(asset: MediaAsset) -> str | None:
    return asset.cdn_url or asset.oss_url or asset.source_url


def post_from_candidate_item(item: CandidateItem, media_assets: list[MediaAsset]) -> Post:
    """Build the reader input from formal acquisition DB objects."""

    ordered = sorted(media_assets, key=lambda asset: asset.position)
    image_urls: list[str] = []
    video_urls: list[str] = []
    cards: list[Card] = []
    for asset in ordered:
        url = _media_url(asset)
        if not url:
            continue
        media_type = asset.media_type.lower()
        if media_type == "video":
            video_urls.append(url)
            continue
        if media_type in {"image", "cover", "frame"}:
            image_urls.append(url)
            cards.append(
                Card(
                    index=len(cards) + 1,
                    kind="frame" if media_type == "frame" else "image",
                    url=url,
                    timestamp=asset.metadata.get("timestamp"),
                )
            )
    source_payload = item.source_payload or {}
    content_id = item.platform_item_id or (str(item.id) if item.id else "")
    return Post(
        id=f"{item.platform}_{content_id}" if content_id and not content_id.startswith(f"{item.platform}_") else content_id,
        platform=item.platform,
        url=item.canonical_url or "",
        content_id=content_id,
        title=item.title or "",
        content_type=item.content_type or ("video" if video_urls else "图文"),
        body_text=(
            item.raw_summary
            or source_payload.get("body_text")
            or source_payload.get("desc")
            or source_payload.get("content")
            or ""
        ),
        topic_list=source_payload.get("topic_list") or source_payload.get("topics") or [],
        image_urls=image_urls,
        video_urls=video_urls,
        cards=[] if video_urls else cards,
        author_name=item.author_name,
        raw=source_payload,
    )


def read_result_from_extracted(post: Post, extracted: ExtractedContent) -> ReadResult:
    cmap = {card.index: card.content for card in extracted.cards}
    cards: list[ReadCard] = []
    source_cards = post.cards
    if not source_cards and post.image_urls:
        source_cards = [
            {"index": idx, "kind": "image", "url": url}
            for idx, url in enumerate(post.image_urls, start=1)
        ]
    for raw in source_cards:
        if isinstance(raw, dict):
            index = int(raw.get("index") or 0)
            kind = str(raw.get("kind") or "image")
            url = raw.get("url")
            timestamp = raw.get("timestamp")
            start = raw.get("start")
            end = raw.get("end")
        else:
            index = raw.index
            kind = raw.kind
            url = raw.url
            timestamp = raw.timestamp
            start = raw.start
            end = raw.end
        cards.append(
            ReadCard(
                index=index,
                kind=kind,
                url=url,
                timestamp=timestamp,
                start=start,
                end=end,
                content=cmap.get(index, ""),
            )
        )
    parts = [extracted.text]
    if extracted.from_image:
        parts.append("【图片要点】\n" + extracted.from_image)
    if extracted.from_video:
        parts.append("【视频要点】\n" + extracted.from_video)
    parts.extend(f"【卡片{card.index}】{card.content}" for card in extracted.cards if card.content)
    text = "\n\n".join(p for p in parts if p)
    media = (
        {"type": "video", "video_url": post.video_urls[0] if post.video_urls else None, "images": []}
        if post.video_urls
        else {"type": "image", "video_url": None, "images": list(post.image_urls)}
    )
    return ReadResult(text=text, cards=cards, media=media, is_empty=bool(extracted.is_empty))


def read_post(
    post: Post,
    *,
    settings: Settings | None = None,
    extractor: BailianExtractor | None = None,
    image_reader: ImageReader | None = None,
    video_reader: VideoReader | None = None,
) -> ReadResult:
    if post.video_urls:
        if video_reader is not None:
            extracted = video_reader(post)
        else:
            if settings is None:
                raise ValueError("settings is required when reading video posts")
            extracted = read_video(post, settings=settings)
    else:
        extracted = image_reader(post) if image_reader is not None else read_imgtext(post, extractor=extractor)
    return read_result_from_extracted(post, extracted)


def read_item(
    item: CandidateItem,
    media_assets: list[MediaAsset],
    *,
    settings: Settings | None = None,
    extractor: BailianExtractor | None = None,
    image_reader: ImageReader | None = None,
    video_reader: VideoReader | None = None,
) -> ReadResult:
    return read_post(
        post_from_candidate_item(item, media_assets),
        settings=settings,
        extractor=extractor,
        image_reader=image_reader,
        video_reader=video_reader,
    )


__all__ = [
    "ImageReader",
    "VideoReader",
    "post_from_candidate_item",
    "read_item",
    "read_post",
    "read_result_from_extracted",
]