Create-knowledge-find-decode/acquisition/media/service.py

"""Media stabilization service for the formal acquisition pipeline."""
from __future__ import annotations

from dataclasses import dataclass
import time
from typing import Callable

from acquisition.media.oss_client import to_oss
from core.config import Settings


@dataclass(frozen=True)
class StabilizedMedia:
    media_type: str
    source_url: str
    cdn_url: str
    position: int
    status: str = "ready"
    error_message: str = ""


Uploader = Callable[[str, str], str]


def stabilize_media_urls(
    *,
    image_urls: list[str] | None = None,
    video_urls: list[str] | None = None,
    settings: Settings | None = None,
    uploader: Uploader | None = None,
    video_fallback_to_source: bool = True,
    video_retry_delays_seconds: tuple[float, ...] = (),
    sleep: Callable[[float], None] = time.sleep,
) -> list[StabilizedMedia]:
    """Transfer image/video URLs to OSS/CDN and return DB-ready media rows.

    Images keep the old soft-fallback behavior. Videos can opt into strict mode
    so expiring source URLs do not leak into classification/decode.
    """
    upload = uploader or (
        lambda url, media_type: to_oss(url, media_type, settings=settings)
    )
    rows: list[StabilizedMedia] = []
    position = 0
    for url in image_urls or []:
        if not url:
            continue
        position += 1
        try:
            cdn = upload(url, "image")
        except Exception:
            cdn = url
        rows.append(
            StabilizedMedia(
                media_type="image",
                source_url=url,
                cdn_url=cdn,
                position=position,
                status="ready" if cdn else "failed",
            )
        )
    for url in video_urls or []:
        if not url:
            continue
        position += 1
        cdn = ""
        error_message = ""
        attempts = len(video_retry_delays_seconds) + 1
        for attempt in range(attempts):
            try:
                cdn = upload(url, "video")
                if cdn and (video_fallback_to_source or cdn != url):
                    error_message = ""
                    break
                error_message = "oss_upload_returned_source_url"
                cdn = ""
            except Exception as exc:
                error_message = str(exc)
            if attempt < len(video_retry_delays_seconds):
                sleep(video_retry_delays_seconds[attempt])
        if not cdn and video_fallback_to_source:
            cdn = url
        rows.append(
            StabilizedMedia(
                media_type="video",
                source_url=url,
                cdn_url=cdn,
                position=position,
                status="ready" if cdn else "failed",
                error_message=error_message,
            )
        )
    return rows