"""Business-level content mode helpers for acquisition.""" from __future__ import annotations from dataclasses import dataclass from typing import Any, Literal ContentMode = Literal["image_post", "video_post", "article", "unsupported"] IMAGE_POST: ContentMode = "image_post" VIDEO_POST: ContentMode = "video_post" ARTICLE: ContentMode = "article" UNSUPPORTED: ContentMode = "unsupported" @dataclass(frozen=True) class ContentModeGuard: can_process: bool reason: str = "" label: str = "" metadata: dict[str, Any] | None = None def _has_any(values: list[str] | None) -> bool: return any(bool(value) for value in values or []) def _raw_type(raw: dict[str, Any] | None) -> str: if not isinstance(raw, dict): return "" candidates = [ raw.get("type"), raw.get("content_type"), raw.get("note_type"), (raw.get("note_card") or {}).get("type") if isinstance(raw.get("note_card"), dict) else None, raw.get("aweme_type"), ] for value in candidates: if value is not None and str(value).strip(): return str(value).strip().lower() return "" def infer_content_mode( *, platform: str, content_type: str | None = None, body_text: str | None = None, image_urls: list[str] | None = None, video_urls: list[str] | None = None, raw: dict[str, Any] | None = None, ) -> ContentMode: """Infer the business mode without overwriting the platform raw type.""" platform_key = (platform or "").lower() raw_kind = _raw_type(raw) raw_content_type = (content_type or "").strip().lower() has_image = _has_any(image_urls) has_video = _has_any(video_urls) has_text = bool((body_text or "").strip()) if platform_key == "weixin": return ARTICLE if has_video: return VIDEO_POST if has_image or has_text: return IMAGE_POST if raw_kind in {"normal", "note", "image", "image_post", "图文"}: return IMAGE_POST if raw_content_type in {"normal", "note", "image", "image_post", "图文"}: return IMAGE_POST return UNSUPPORTED def guard_content_for_processing( *, content_mode: str | None, video_urls: list[str] | None = None, ) -> ContentModeGuard: """Return whether this item should enter media/classify/decode processing.""" if content_mode == UNSUPPORTED: return ContentModeGuard( can_process=False, reason="unsupported_content_mode", label="unsupported_content_mode", metadata={"skip_reason": "unsupported_content_mode"}, ) if content_mode == VIDEO_POST and not _has_any(video_urls): return ContentModeGuard( can_process=False, reason="video_url_missing", label="video_missing", metadata={"skip_reason": "video_url_missing", "video_url_missing": True}, ) return ContentModeGuard(can_process=True) __all__ = [ "ARTICLE", "ContentMode", "ContentModeGuard", "IMAGE_POST", "UNSUPPORTED", "VIDEO_POST", "guard_content_for_processing", "infer_content_mode", ]