"""Unified post reader for image-text and video decode inputs.""" from __future__ import annotations from typing import Any, Protocol from acquisition.content_mode import infer_content_mode 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: ... class UnsupportedPostModeError(ValueError): pass def _nested_text(payload: dict, *path: str) -> str: current: Any = payload for key in path: if not isinstance(current, dict): return "" current = current.get(key) return current if isinstance(current, str) else "" def _body_text_from_payload(source_payload: dict) -> str: return ( _nested_text(source_payload, "body_text") or _nested_text(source_payload, "detail", "body_text") or _nested_text(source_payload, "detail", "data", "data", "body_text") or _nested_text(source_payload, "candidate", "body_text") or _nested_text(source_payload, "candidate", "raw", "body_text") or _nested_text(source_payload, "desc") or _nested_text(source_payload, "content") ) def _media_url(asset: MediaAsset) -> str | None: return asset.cdn_url or asset.oss_url or asset.source_url def _stable_media_url(asset: MediaAsset) -> str | None: return asset.cdn_url or asset.oss_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": stable_url = _stable_media_url(asset) if stable_url: video_urls.append(stable_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 {} body_text = item.body_text or _body_text_from_payload(source_payload) or item.raw_summary or "" content_id = item.platform_item_id or (str(item.id) if item.id else "") content_mode = item.content_mode or infer_content_mode( platform=item.platform, content_type=item.content_type or "", body_text=body_text, image_urls=image_urls, video_urls=video_urls, raw=source_payload if isinstance(source_payload, dict) 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 content_mode == "video_post" else "图文"), content_mode=content_mode, body_text=body_text, 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: mode = post.content_mode or ("video_post" if post.video_urls else "image_post") if mode == "unsupported": raise UnsupportedPostModeError("unsupported_content_mode") if mode == "video_post": if not post.video_urls: raise UnsupportedPostModeError("video_url_missing") 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, oss_video_url=post.video_urls[0], ) 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", "UnsupportedPostModeError", "VideoReader", "post_from_candidate_item", "read_item", "read_post", "read_result_from_extracted", ]