manage/archive/src/media_library_viewer/domain/media.py

"""Media-domain normalization helpers.

Jellyfin item JSON is nested and inconsistent across item types. This module
flattens Jellyfin items into stable dictionaries suitable for storage in the
SQLite media index and display by any frontend.
"""

from __future__ import annotations

from typing import Any

import pandas as pd

from media_library_viewer.utils import human_size, ticks_to_minutes


def first_media_source(item: dict[str, Any]) -> dict[str, Any]:
    """Return the first Jellyfin media source, or an empty dict."""
    sources = item.get("MediaSources") or []
    return sources[0] if sources else {}


def media_streams(item: dict[str, Any], stream_type: str | None = None) -> list[dict[str, Any]]:
    """Return flattened media streams from all media sources.

    Jellyfin usually nests streams under MediaSources, while some endpoints may
    expose stream-like fields differently. This function gives callers one place
    to get streams and optionally filter by type.
    """
    streams = []
    for source in item.get("MediaSources") or []:
        streams.extend(source.get("MediaStreams") or [])
    if stream_type is None:
        return streams
    return [stream for stream in streams if str(stream.get("Type") or stream.get("codec_type") or "").lower() == stream_type.lower()]


def stream_value(stream: dict[str, Any], *keys: str) -> Any:
    for key in keys:
        if key in stream and stream[key] not in (None, ""):
            return stream[key]
    return None


def is_hdr_item(item: dict[str, Any]) -> bool:
    """Best-effort HDR detection from Jellyfin video stream metadata."""
    hdr_markers = {"hdr", "hdr10", "hdr10+", "dolbyvision", "dovi", "hlg", "pq", "smpte2084", "bt2020"}
    for stream in media_streams(item, "Video"):
        values = [
            stream_value(stream, "VideoRange", "video_range"),
            stream_value(stream, "VideoRangeType", "video_range_type"),
            stream_value(stream, "ColorTransfer", "color_transfer"),
            stream_value(stream, "ColorPrimaries", "color_primaries"),
            stream_value(stream, "ColorSpace", "color_space"),
            stream_value(stream, "DvVersionMajor", "dv_version_major"),
            stream_value(stream, "Hdr10PlusPresent", "hdr10_plus_present"),
        ]
        normalized = " ".join(str(value).lower() for value in values if value not in (None, "", False, 0))
        if any(marker in normalized for marker in hdr_markers):
            return True
    return False


def format_date_added(value: str | None) -> str:
    if not value:
        return ""
    try:
        return pd.to_datetime(value).strftime("%Y-%m-%d")
    except Exception:
        return str(value)


def timestamp_date_added(value: str | None) -> int | None:
    if not value:
        return None
    try:
        return int(pd.to_datetime(value).timestamp())
    except Exception:
        return None


def format_rate_bits_decimal(bits_per_second: float | int | str | None) -> str:
    if bits_per_second in (None, ""):
        return ""
    try:
        value = float(bits_per_second)
    except (TypeError, ValueError):
        return str(bits_per_second)
    for unit in ["bps", "Kbps", "Mbps", "Gbps", "Tbps"]:
        if value < 1000 or unit == "Tbps":
            return f"{value:.1f} {unit}"
        value /= 1000
    return f"{value:.1f} Tbps"


def normalize_media_item(item: dict[str, Any], library_id: str = "", library_name: str = "") -> dict[str, Any]:
    """Flatten one Jellyfin item into an indexable row.

    The returned row contains both display strings (``size``, ``bitrate``) and
    numeric sort fields (``size_bytes``, ``bitrate_bps``, ``date_added_ts``).
    """
    source = first_media_source(item)
    video_streams = media_streams(item, "Video")
    video = video_streams[0] if video_streams else {}
    size = source.get("Size") or source.get("size")
    bitrate = source.get("Bitrate") or source.get("bitrate") or item.get("Bitrate")
    width = stream_value(video, "Width", "width")
    height = stream_value(video, "Height", "height")
    season_number = item.get("ParentIndexNumber")
    episode_number = item.get("IndexNumber")

    hdr = is_hdr_item(item)
    return {
        "id": item.get("Id", ""),
        "title": item.get("Name", ""),
        "series": item.get("SeriesName", ""),
        "season": f"S{int(season_number):02d}" if season_number is not None else item.get("SeasonName", ""),
        "season_number": int(season_number) if season_number is not None else None,
        "episode": int(episode_number) if episode_number is not None else None,
        "type": item.get("Type", ""),
        "year": item.get("ProductionYear"),
        "runtime_ticks": item.get("RunTimeTicks"),
        "runtime_min": ticks_to_minutes(item.get("RunTimeTicks")),
        "size_bytes": int(size) if size not in (None, "") else None,
        "size": human_size(size),
        "bitrate_bps": int(bitrate) if bitrate not in (None, "") else None,
        "bitrate": format_rate_bits_decimal(bitrate),
        "hdr": 1 if hdr else 0,
        "hdr_label": "yes" if hdr else "",
        "video": video.get("Codec") or video.get("codec_name") or "",
        "width": int(width) if width not in (None, "") else None,
        "height": int(height) if height not in (None, "") else None,
        "resolution": f"{width}x{height}" if width and height else "",
        "date_added": format_date_added(item.get("DateCreated")),
        "date_added_ts": timestamp_date_added(item.get("DateCreated")),
        "path": item.get("Path") or source.get("Path") or "",
        "library_id": library_id,
        "library_name": library_name,
    }


def display_media_row(row: dict[str, Any]) -> dict[str, Any]:
    """Convert a SQLite row back into frontend display fields."""
    return {
        "title": row.get("title", ""),
        "series": row.get("series", ""),
        "season": row.get("season", ""),
        "episode": row.get("episode", ""),
        "type": row.get("type", ""),
        "year": row.get("year", ""),
        "runtime_min": row.get("runtime_min", ""),
        "size": row.get("size") or human_size(row.get("size_bytes")),
        "bitrate": row.get("bitrate") or format_rate_bits_decimal(row.get("bitrate_bps")),
        "hdr": "yes" if row.get("hdr") else "no",
        "video": row.get("video", ""),
        "resolution": row.get("resolution", ""),
        "date_added": row.get("date_added", ""),
        "library": row.get("library_name", ""),
        "path": row.get("path", ""),
        "id": row.get("id", ""),
    }