Create-knowledge-find-decode/acquisition/content_mode.py

"""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",
]