manage/backend/src/media_library_viewer_api/utils.py

"""Formatting and ffprobe summarization helpers.

These helpers are intentionally UI-framework independent. Streamlit renders the
returned dictionaries/dataframes, but another frontend can reuse the same
summaries.
"""

from __future__ import annotations

from datetime import datetime
from pathlib import PurePosixPath
from typing import Any

VIDEO_FILE_EXTENSIONS = {
    ".3g2",
    ".3gp",
    ".avi",
    ".divx",
    ".flv",
    ".m2ts",
    ".m4v",
    ".mkv",
    ".mov",
    ".mp4",
    ".mpeg",
    ".mpg",
    ".mts",
    ".ogm",
    ".ogv",
    ".rmvb",
    ".ts",
    ".vob",
    ".webm",
    ".wmv",
}


def ticks_to_minutes(ticks: int | None) -> int | None:
    """Convert Jellyfin/Emby 100-nanosecond ticks to rounded minutes."""
    if not ticks:
        return None
    return round(ticks / 10_000_000 / 60)


def human_size(num: int | float | None) -> str:
    """Format a byte count as B/KB/MB/GB/etc."""
    if num is None:
        return ""
    value = float(num)
    for unit in ["B", "KB", "MB", "GB", "TB", "PB"]:
        if value < 1024 or unit == "PB":
            return f"{value:.1f} {unit}" if unit != "B" else f"{int(value)} B"
        value /= 1024
    return f"{value:.1f} PB"


def timestamp_to_local(ts: float | None) -> str:
    if ts is None:
        return ""
    return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S")


def is_known_video_file(path: str | None) -> bool:
    """Return True when a path extension is one we should ffprobe automatically."""
    if not path:
        return False
    return PurePosixPath(path).suffix.lower() in VIDEO_FILE_EXTENSIONS


def format_duration(seconds: str | int | float | None) -> str:
    if seconds in (None, ""):
        return ""
    try:
        total = float(seconds)
    except (TypeError, ValueError):
        return str(seconds)
    hours = int(total // 3600)
    minutes = int((total % 3600) // 60)
    secs = int(total % 60)
    return f"{hours:02d}:{minutes:02d}:{secs:02d}"


def format_bitrate(bit_rate: str | int | float | None) -> str:
    if bit_rate in (None, ""):
        return ""
    try:
        value = float(bit_rate)
    except (TypeError, ValueError):
        return str(bit_rate)
    if value >= 1_000_000:
        return f"{value / 1_000_000:.2f} Mbps"
    if value >= 1_000:
        return f"{value / 1_000:.0f} kbps"
    return f"{value:.0f} bps"


def _tags(stream: dict[str, Any]) -> dict[str, Any]:
    return stream.get("tags") or {}


def _disposition(stream: dict[str, Any], key: str) -> str:
    value = (stream.get("disposition") or {}).get(key)
    return "yes" if value == 1 else ""


def _side_data_types(stream: dict[str, Any]) -> str:
    values = []
    for item in stream.get("side_data_list") or []:
        if item.get("side_data_type"):
            values.append(item["side_data_type"])
    return ", ".join(values)


def ffprobe_format_summary(ffprobe: dict[str, Any]) -> dict[str, str]:
    """Summarize ffprobe container/format-level metadata."""
    fmt = ffprobe.get("format") or {}
    return {
        "filename": fmt.get("filename", ""),
        "format": fmt.get("format_name", ""),
        "format_long": fmt.get("format_long_name", ""),
        "duration": format_duration(fmt.get("duration")),
        "size": human_size(float(fmt["size"])) if fmt.get("size") else "",
        "bit_rate": format_bitrate(fmt.get("bit_rate")),
        "stream_count": str(fmt.get("nb_streams", "")),
    }


def summarize_video_streams(ffprobe: dict[str, Any]) -> list[dict[str, Any]]:
    """Return video-only stream rows with video/HDR-related fields."""
    rows = []
    for stream in ffprobe.get("streams", []):
        if stream.get("codec_type") != "video":
            continue
        tags = _tags(stream)
        rows.append(
            {
                "index": stream.get("index"),
                "codec": stream.get("codec_name"),
                "profile": stream.get("profile"),
                "resolution": f"{stream.get('width', '')}x{stream.get('height', '')}",
                "pix_fmt": stream.get("pix_fmt"),
                "bit_rate": format_bitrate(stream.get("bit_rate")),
                "avg_fps": stream.get("avg_frame_rate"),
                "color_range": stream.get("color_range"),
                "color_space": stream.get("color_space"),
                "color_transfer": stream.get("color_transfer"),
                "color_primaries": stream.get("color_primaries"),
                "side_data": _side_data_types(stream),
                "language": tags.get("language"),
                "title": tags.get("title"),
                "default": _disposition(stream, "default"),
            }
        )
    return rows


def summarize_audio_streams(ffprobe: dict[str, Any]) -> list[dict[str, Any]]:
    """Return audio-only stream rows with channel/language/default fields."""
    rows = []
    for stream in ffprobe.get("streams", []):
        if stream.get("codec_type") != "audio":
            continue
        tags = _tags(stream)
        rows.append(
            {
                "index": stream.get("index"),
                "codec": stream.get("codec_name"),
                "profile": stream.get("profile"),
                "channels": stream.get("channels"),
                "layout": stream.get("channel_layout"),
                "sample_rate": stream.get("sample_rate"),
                "bit_rate": format_bitrate(stream.get("bit_rate")),
                "language": tags.get("language"),
                "title": tags.get("title"),
                "default": _disposition(stream, "default"),
                "forced": _disposition(stream, "forced"),
            }
        )
    return rows


def summarize_subtitle_streams(ffprobe: dict[str, Any]) -> list[dict[str, Any]]:
    """Return subtitle-only stream rows with language/forced/default fields."""
    rows = []
    for stream in ffprobe.get("streams", []):
        if stream.get("codec_type") != "subtitle":
            continue
        tags = _tags(stream)
        rows.append(
            {
                "index": stream.get("index"),
                "codec": stream.get("codec_name"),
                "codec_long": stream.get("codec_long_name"),
                "language": tags.get("language"),
                "title": tags.get("title"),
                "default": _disposition(stream, "default"),
                "forced": _disposition(stream, "forced"),
                "hearing_impaired": _disposition(stream, "hearing_impaired"),
            }
        )
    return rows


def summarize_streams(ffprobe: dict[str, Any]) -> list[dict[str, Any]]:
    rows = []
    for stream in ffprobe.get("streams", []):
        rows.append(
            {
                "index": stream.get("index"),
                "type": stream.get("codec_type"),
                "codec": stream.get("codec_name"),
                "profile": stream.get("profile"),
                "width": stream.get("width"),
                "height": stream.get("height"),
                "pix_fmt": stream.get("pix_fmt"),
                "color_transfer": stream.get("color_transfer"),
                "color_primaries": stream.get("color_primaries"),
                "color_space": stream.get("color_space"),
                "bit_rate": format_bitrate(stream.get("bit_rate")),
                "channels": stream.get("channels"),
                "sample_rate": stream.get("sample_rate"),
                "language": stream.get("tags", {}).get("language"),
                "title": stream.get("tags", {}).get("title"),
            }
        )
    return rows