ViewIT/addon/core/metadata.py

from __future__ import annotations
import asyncio
import os
import re
import threading
from datetime import datetime
from typing import Any, Callable, Dict, List, Optional, Tuple

from tmdb import (
    TmdbCastMember,
    fetch_tv_episode_credits,
    lookup_movie,
    lookup_tv_season,
    lookup_tv_season_summary,
    lookup_tv_show,
)

try:
    import xbmc
    import xbmcaddon
    import xbmcvfs
except ImportError:
    xbmc = None
    xbmcaddon = None
    xbmcvfs = None

# Caches
_TMDB_CACHE: dict[str, tuple[dict[str, str], dict[str, str]]] = {}
_TMDB_CAST_CACHE: dict[str, list[TmdbCastMember]] = {}
_TMDB_SEASON_CACHE: dict[tuple[int, int, str, str], dict[int, tuple[dict[str, str], dict[str, str]]]] = {}
_TMDB_SEASON_SUMMARY_CACHE: dict[tuple[int, int, str, str], tuple[dict[str, str], dict[str, str]]] = {}
_TMDB_EPISODE_CAST_CACHE: dict[tuple[int, int, int, str], list[TmdbCastMember]] = {}
_TMDB_ID_CACHE: dict[str, int] = {}
_TMDB_LOG_PATH: str | None = None
_TMDB_LOCK = threading.RLock()


# Dependency Injection variables
_initialized: bool = False
_get_setting_string: Callable[[str], str] = lambda k: ""
_get_setting_bool: Callable[[str, bool], bool] = lambda k, default=False: default
_get_setting_int: Callable[[str, int], int] = lambda k, default=0: default
_log: Callable[[str, int], None] = lambda msg, level=0: None
_run_async: Callable[[Any], Any] = lambda coro: None
_extract_first_int: Callable[[str], Optional[int]] = lambda val: None


def _require_init() -> None:
    """Gibt eine Warnung aus, wenn metadata.init() noch nicht aufgerufen wurde."""
    if not _initialized:
        import sys
        print("[ViewIT/metadata] WARNUNG: metadata.init() wurde nicht aufgerufen – Metadaten-Funktionen arbeiten mit Standardwerten!", file=sys.stderr)


def init(
    *,
    get_setting_string: Callable[[str], str],
    get_setting_bool: Callable[..., bool],
    get_setting_int: Callable[..., int],
    log_fn: Callable[[str, int], None],
    run_async_fn: Callable[[Any], Any],
    extract_first_int_fn: Callable[[str], Optional[int]],
) -> None:
    global _initialized, _get_setting_string, _get_setting_bool, _get_setting_int, _log, _run_async, _extract_first_int
    _get_setting_string = get_setting_string
    _get_setting_bool = get_setting_bool
    _get_setting_int = get_setting_int
    _log = log_fn
    _run_async = run_async_fn
    _extract_first_int = extract_first_int_fn
    _initialized = True


def _get_log_path(filename: str) -> str:
    if xbmcaddon and xbmcvfs:
        addon = xbmcaddon.Addon()
        profile = xbmcvfs.translatePath(addon.getAddonInfo("profile"))
        log_dir = os.path.join(profile, "logs")
        if not xbmcvfs.exists(log_dir):
            xbmcvfs.mkdirs(log_dir)
        return os.path.join(log_dir, filename)
    return os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), filename)


def tmdb_file_log(message: str) -> None:
    global _TMDB_LOG_PATH
    if _TMDB_LOG_PATH is None:
        _TMDB_LOG_PATH = _get_log_path("tmdb.log")
    timestamp = datetime.utcnow().isoformat(timespec="seconds") + "Z"
    line = f"{timestamp}\t{message}\n"
    try:
        with open(_TMDB_LOG_PATH, "a", encoding="utf-8") as handle:
            handle.write(line)
    except Exception:
        if xbmcvfs is None:
            return
        try:
            handle = xbmcvfs.File(_TMDB_LOG_PATH, "a") # type: ignore
            handle.write(line) # type: ignore
            handle.close() # type: ignore
        except Exception:
            return


def tmdb_cache_get(cache: dict, key, default=None):
    with _TMDB_LOCK:
        return cache.get(key, default)


def tmdb_cache_set(cache: dict, key, value) -> None:
    with _TMDB_LOCK:
        cache[key] = value


def tmdb_prefetch_concurrency() -> int:
    try:
        raw = _get_setting_string("tmdb_prefetch_concurrency").strip()
        value = int(raw) if raw else 6
    except Exception:
        value = 6
    return max(1, min(20, value))


def tmdb_enabled() -> bool:
    _require_init()
    return _get_setting_bool("tmdb_enabled", default=True)


def tmdb_list_enabled() -> bool:
    return tmdb_enabled() and _get_setting_bool("tmdb_genre_metadata", default=False)


def tmdb_labels_and_art(title: str) -> tuple[dict[str, str], dict[str, str], list[TmdbCastMember]]:
    if not tmdb_enabled():
        return {}, {}, []
    title_key = (title or "").strip().casefold()
    language = _get_setting_string("tmdb_language").strip() or "de-DE"
    show_plot = _get_setting_bool("tmdb_show_plot", default=True)
    show_art = _get_setting_bool("tmdb_show_art", default=True)
    show_fanart = _get_setting_bool("tmdb_show_fanart", default=True)
    show_rating = _get_setting_bool("tmdb_show_rating", default=True)
    show_votes = _get_setting_bool("tmdb_show_votes", default=False)
    show_cast = _get_setting_bool("tmdb_show_cast", default=False)
    flags = f"p{int(show_plot)}a{int(show_art)}f{int(show_fanart)}r{int(show_rating)}v{int(show_votes)}c{int(show_cast)}"
    cache_key = f"{language}|{flags}|{title_key}"
    cached = tmdb_cache_get(_TMDB_CACHE, cache_key)
    if cached is not None:
        info, art = cached
        cast_cached = tmdb_cache_get(_TMDB_CAST_CACHE, cache_key, [])
        return info, art, list(cast_cached)

    info_labels: dict[str, str] = {"title": title}
    art: dict[str, str] = {}
    cast: list[TmdbCastMember] = []
    query = (title or "").strip()
    api_key = _get_setting_string("tmdb_api_key").strip()
    log_requests = _get_setting_bool("tmdb_log_requests", default=False)
    log_responses = _get_setting_bool("tmdb_log_responses", default=False)
    if api_key:
        try:
            log_fn = tmdb_file_log if (log_requests or log_responses) else None
            candidates: list[str] = []
            if query:
                candidates.append(query)
                simplified = re.sub(r"\s*[-–]\s*der\s+film\s*$", "", query, flags=re.IGNORECASE).strip()
                if simplified and simplified not in candidates:
                    candidates.append(simplified)

            meta = None
            is_tv = False
            for candidate in candidates:
                meta = lookup_tv_show(
                    title=candidate,
                    api_key=api_key,
                    language=language,
                    log=log_fn,
                    log_responses=log_responses,
                    include_cast=show_cast,
                )
                if meta:
                    is_tv = True
                    break
            if not meta:
                for candidate in candidates:
                    movie = lookup_movie(
                        title=candidate,
                        api_key=api_key,
                        language=language,
                        log=log_fn,
                        log_responses=log_responses,
                        include_cast=show_cast,
                    )
                    if movie:
                        meta = movie
                        break
        except Exception as exc:
            try:
                tmdb_file_log(f"TMDB ERROR lookup_failed title={title!r} error={exc!r}")
            except Exception:
                pass
            _log(f"TMDB Meta fehlgeschlagen: {exc}", 1) # LOGWARNING/LOGDEBUG fallback
            meta = None
        if meta:
            if is_tv:
                tmdb_cache_set(_TMDB_ID_CACHE, title_key, int(getattr(meta, "tmdb_id", 0) or 0))
                info_labels.setdefault("mediatype", "tvshow")
            else:
                info_labels.setdefault("mediatype", "movie")
            if show_plot and getattr(meta, "plot", ""):
                info_labels["plot"] = getattr(meta, "plot", "")
            runtime_minutes = int(getattr(meta, "runtime_minutes", 0) or 0)
            if runtime_minutes > 0 and not is_tv:
                info_labels["duration"] = str(runtime_minutes * 60)
            rating = getattr(meta, "rating", 0.0) or 0.0
            votes = getattr(meta, "votes", 0) or 0
            if show_rating and rating:
                info_labels["rating"] = str(rating)
            if show_votes and votes:
                info_labels["votes"] = str(votes)
            if show_art and getattr(meta, "poster", ""):
                poster = getattr(meta, "poster", "")
                art.update({"thumb": poster, "poster": poster, "icon": poster})
            if show_fanart and getattr(meta, "fanart", ""):
                fanart = getattr(meta, "fanart", "")
                if fanart:
                    art.update({"fanart": fanart, "landscape": fanart})
            if show_cast:
                cast = list(getattr(meta, "cast", []) or [])
        elif log_requests or log_responses:
            tmdb_file_log(f"TMDB MISS title={title!r}")

    tmdb_cache_set(_TMDB_CACHE, cache_key, (info_labels, art))
    tmdb_cache_set(_TMDB_CAST_CACHE, cache_key, list(cast))
    return info_labels, art, list(cast)


async def _tmdb_labels_and_art_bulk_async(
    titles: list[str],
) -> dict[str, tuple[dict[str, str], dict[str, str], list[TmdbCastMember]]]:
    titles = [str(t).strip() for t in (titles or []) if t and str(t).strip()]
    if not titles:
        return {}

    unique_titles: list[str] = list(dict.fromkeys(titles))
    limit = tmdb_prefetch_concurrency()
    semaphore = asyncio.Semaphore(limit)

    async def fetch_one(title: str):
        async with semaphore:
            return title, await asyncio.to_thread(tmdb_labels_and_art, title)

    tasks = [fetch_one(title) for title in unique_titles]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    mapped: dict[str, tuple[dict[str, str], dict[str, str], list[TmdbCastMember]]] = {}
    for entry in results:
        if isinstance(entry, Exception):
            continue
        try:
            title, payload = entry
        except Exception:
            continue
        if isinstance(title, str) and isinstance(payload, tuple) and len(payload) == 3:
            mapped[title] = payload  # type: ignore[assignment]
    return mapped


def tmdb_labels_and_art_bulk(
    titles: list[str],
) -> dict[str, tuple[dict[str, str], dict[str, str], list[TmdbCastMember]]]:
    if not tmdb_enabled():
        return {}
    return _run_async(_tmdb_labels_and_art_bulk_async(titles))


def tmdb_episode_labels_and_art(*, title: str, season_label: str, episode_label: str) -> tuple[dict[str, str], dict[str, str]]:
    if not tmdb_enabled():
        return {"title": episode_label}, {}
    title_key = (title or "").strip().casefold()
    tmdb_id = tmdb_cache_get(_TMDB_ID_CACHE, title_key)
    if not tmdb_id:
        tmdb_labels_and_art(title)
        tmdb_id = tmdb_cache_get(_TMDB_ID_CACHE, title_key)
    if not tmdb_id:
        return {"title": episode_label}, {}

    season_number = _extract_first_int(season_label)
    episode_number = _extract_first_int(episode_label)
    if season_number is None or episode_number is None:
        return {"title": episode_label}, {}

    language = _get_setting_string("tmdb_language").strip() or "de-DE"
    show_plot = _get_setting_bool("tmdb_show_plot", default=True)
    show_art = _get_setting_bool("tmdb_show_art", default=True)
    flags = f"p{int(show_plot)}a{int(show_art)}"
    season_key = (tmdb_id, season_number, language, flags)
    cached_season = tmdb_cache_get(_TMDB_SEASON_CACHE, season_key)
    if cached_season is None:
        api_key = _get_setting_string("tmdb_api_key").strip()
        if not api_key:
            return {"title": episode_label}, {}
        log_requests = _get_setting_bool("tmdb_log_requests", default=False)
        log_responses = _get_setting_bool("tmdb_log_responses", default=False)
        log_fn = tmdb_file_log if (log_requests or log_responses) else None
        try:
            season_meta = lookup_tv_season(
                tmdb_id=tmdb_id,
                season_number=season_number,
                api_key=api_key,
                language=language,
                log=log_fn,
                log_responses=log_responses,
            )
        except Exception as exc:
            if log_fn:
                log_fn(f"TMDB ERROR season_lookup_failed tmdb_id={tmdb_id} season={season_number} error={exc!r}")
            season_meta = None
        mapped: dict[int, tuple[dict[str, str], dict[str, str]]] = {}
        if season_meta:
            for ep_no, ep in season_meta.items():
                info: dict[str, str] = {"title": f"Episode {ep_no}"}
                if show_plot and ep.plot:
                    info["plot"] = ep.plot
                if getattr(ep, "runtime_minutes", 0):
                    info["duration"] = str(int(getattr(ep, "runtime_minutes", 0)) * 60)
                art: dict[str, str] = {}
                if show_art and ep.thumb:
                    art = {"thumb": ep.thumb}
                mapped[ep_no] = (info, art)
        tmdb_cache_set(_TMDB_SEASON_CACHE, season_key, mapped)
        cached_season = mapped

    return cached_season.get(episode_number, ({"title": episode_label}, {}))


def tmdb_episode_cast(*, title: str, season_label: str, episode_label: str) -> list[TmdbCastMember]:
    if not tmdb_enabled():
        return []
    show_episode_cast = _get_setting_bool("tmdb_show_episode_cast", default=False)
    if not show_episode_cast:
        return []

    title_key = (title or "").strip().casefold()
    tmdb_id = tmdb_cache_get(_TMDB_ID_CACHE, title_key)
    if not tmdb_id:
        tmdb_labels_and_art(title)
        tmdb_id = tmdb_cache_get(_TMDB_ID_CACHE, title_key)
    if not tmdb_id:
        return []

    season_number = _extract_first_int(season_label)
    episode_number = _extract_first_int(episode_label)
    if season_number is None or episode_number is None:
        return []

    language = _get_setting_string("tmdb_language").strip() or "de-DE"
    cache_key = (tmdb_id, season_number, episode_number, language)
    cached = tmdb_cache_get(_TMDB_EPISODE_CAST_CACHE, cache_key)
    if cached is not None:
        return list(cached)

    api_key = _get_setting_string("tmdb_api_key").strip()
    if not api_key:
        tmdb_cache_set(_TMDB_EPISODE_CAST_CACHE, cache_key, [])
        return []

    log_requests = _get_setting_bool("tmdb_log_requests", default=False)
    log_responses = _get_setting_bool("tmdb_log_responses", default=False)
    log_fn = tmdb_file_log if (log_requests or log_responses) else None
    try:
        cast = fetch_tv_episode_credits(
            tmdb_id=tmdb_id,
            season_number=season_number,
            episode_number=episode_number,
            api_key=api_key,
            language=language,
            log=log_fn,
            log_responses=log_responses,
        )
    except Exception as exc:
        if log_fn:
            log_fn(
                f"TMDB ERROR episode_credits_failed tmdb_id={tmdb_id} season={season_number} episode={episode_number} error={exc!r}"
            )
        cast = []
    tmdb_cache_set(_TMDB_EPISODE_CAST_CACHE, cache_key, list(cast))
    return list(cast)


def tmdb_season_labels_and_art(
    *,
    title: str,
    season: str,
    title_info_labels: dict[str, str] | None = None,
) -> tuple[dict[str, str], dict[str, str]]:
    if not tmdb_enabled():
        return {"title": season}, {}
    language = _get_setting_string("tmdb_language").strip() or "de-DE"
    show_plot = _get_setting_bool("tmdb_show_plot", default=True)
    show_art = _get_setting_bool("tmdb_show_art", default=True)
    flags = f"p{int(show_plot)}a{int(show_art)}"
    api_key = _get_setting_string("tmdb_api_key").strip()
    log_requests = _get_setting_bool("tmdb_log_requests", default=False)
    log_responses = _get_setting_bool("tmdb_log_responses", default=False)
    log_fn = tmdb_file_log if (log_requests or log_responses) else None

    info_labels: dict[str, str] | None = None
    art: dict[str, str] | None = None
    season_number = _extract_first_int(season)

    if api_key and season_number is not None:
        title_key = (title or "").strip().casefold()
        tmdb_id = tmdb_cache_get(_TMDB_ID_CACHE, title_key) or 0
        cache_key = (tmdb_id, season_number, language, flags)
        cached = tmdb_cache_get(_TMDB_SEASON_SUMMARY_CACHE, cache_key)

        if cached is None and tmdb_id:
            try:
                meta = lookup_tv_season_summary(
                    tmdb_id=tmdb_id,
                    season_number=season_number,
                    api_key=api_key,
                    language=language,
                    log=log_fn,
                    log_responses=log_responses,
                )
            except Exception as exc:
                if log_fn:
                    log_fn(f"TMDB ERROR season_summary_failed tmdb_id={tmdb_id} season={season_number} error={exc!r}")
                meta = None
            labels = {"title": season}
            art_map: dict[str, str] = {}
            if meta:
                if show_plot and meta.plot:
                    labels["plot"] = meta.plot
                if show_art and meta.poster:
                    art_map = {"thumb": meta.poster, "poster": meta.poster}
            cached = (labels, art_map)
            tmdb_cache_set(_TMDB_SEASON_SUMMARY_CACHE, cache_key, cached)

        if cached is not None:
            info_labels, art = cached

    merged_labels = dict(info_labels or {})
    if title_info_labels:
        merged_labels = dict(title_info_labels)
        merged_labels.update(dict(info_labels or {}))

    return merged_labels, art or {}