Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 17 additions & 4 deletions cli_battery/app/direct_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -68,16 +68,29 @@ def _ensure_worker():
logger.warning(f"Could not start refresh worker: {e}")


def _tmdb_only_available() -> bool:
"""True when shows must come from TMDB: no TVDB key, no Trakt tokens.

tvdb_client serves shows entirely from TMDB in that case, which keeps TV
working on setups that have neither a TVDB key nor a Trakt account. Trakt
wins when configured - TMDB has no per-episode IMDb ids, no absolute
numbering, and date-only air times.
"""
return tvdb_client.tmdb_only_mode()


def _get_metadata_client():
"""Return tvdb_client if TVDB API key is set, else trakt_client."""
if tvdb_client.is_available():
"""Return tvdb_client if TVDB or TMDB is configured, else trakt_client."""
if tvdb_client.is_available() or _tmdb_only_available():
return tvdb_client
return trakt_client


def _get_metadata_source_name() -> str:
"""Return 'tvdb' or 'trakt' depending on which client is active."""
return 'tvdb' if tvdb_client.is_available() else 'trakt'
"""Return 'tvdb', 'tmdb' or 'trakt' depending on which source is active."""
if tvdb_client.is_available():
return 'tvdb'
return 'tmdb' if _tmdb_only_available() else 'trakt'


def _get_local_tz():
Expand Down
130 changes: 127 additions & 3 deletions cli_battery/app/tvdb_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -118,6 +118,32 @@ def is_available() -> bool:
return False


def _trakt_configured() -> bool:
"""True when Trakt has stored tokens.

Deliberately does not call trakt_auth.is_authenticated(), which can fire a
network token refresh - this runs on every metadata client selection.
"""
try:
from utilities.settings import get_setting
return bool((get_setting('Trakt', 'access_token', default='') or '').strip()
or (get_setting('Trakt', 'refresh_token', default='') or '').strip())
except Exception:
return False


def tmdb_only_mode() -> bool:
"""True when shows can only be served from TMDB.

Requires a TMDB key and the absence of both richer sources. Trakt is
preferred over TMDB when configured: TMDB gives no per-episode IMDb ids, no
absolute numbering, and date-only air times.
"""
if is_available() or _trakt_configured():
return False
return bool(_get_tmdb_api_key())


def _ensure_token() -> bool:
"""Authenticate with TVDB and cache the bearer token. Returns True on success."""
global _token
Expand Down Expand Up @@ -604,6 +630,17 @@ def _get_trakt_status(imdb_id: str) -> Optional[str]:

def get_show_data(imdb_id: str) -> Optional[dict]:
"""Get full show metadata + aliases + seasons/episodes."""
# No usable TVDB key: don't resolve a TVDB id we cannot then fetch with.
# Resolving one via TMDB and continuing would send us into the TVDB request
# below, which needs the key we don't have.
if not is_available():
if tmdb_only_mode():
return _fetch_tmdb_show_data(imdb_id, _get_tmdb_api_key())
if _trakt_configured():
from . import trakt_client
return trakt_client.get_show_data(imdb_id)
return None

tvdb_id = _resolve_tvdb_id(imdb_id, media_type='show')
if not tvdb_id:
# TVDB hasn't linked this IMDb ID yet — try resolving via TMDB ID
Expand Down Expand Up @@ -991,6 +1028,32 @@ def _extract_seasons_from_extended(raw: dict) -> Optional[dict]:

def get_show_seasons_and_episodes(imdb_id: str, include_specials: bool = False) -> Tuple[Optional[dict], Optional[str]]:
"""Fetch seasons and episodes for a show."""
# No usable TVDB key: prefer Trakt, else build seasons from TMDB.
# See get_show_data above.
if not is_available():
if not tmdb_only_mode():
if _trakt_configured():
from . import trakt_client
return trakt_client.get_show_seasons_and_episodes(
imdb_id, include_specials=include_specials)
return None, None
show = _fetch_tmdb_show_data(imdb_id, _get_tmdb_api_key())
if not show:
return None, None
seasons = {}
for sn_str, sdata in (show.get('seasons') or {}).items():
try:
sn = int(sn_str)
except (TypeError, ValueError):
continue
seasons[sn] = {
'episode_count': sdata.get('episode_count', 0),
'episodes': sdata.get('episodes', {}),
}
if seasons and not include_specials:
seasons.pop(0, None)
return (seasons, 'tmdb') if seasons else (None, None)

tvdb_id = _resolve_tvdb_id(imdb_id, media_type='show')
if not tvdb_id:
logger.warning(f"TVDB: could not resolve IMDb {imdb_id} to TVDB ID for episodes, trying Trakt fallback")
Expand Down Expand Up @@ -1531,6 +1594,59 @@ def _fetch_tmdb_movie_data(imdb_id: str, api_key: str) -> Optional[dict]:
return None


def _fetch_tmdb_episodes(tmdb_id: int, api_key: str,
season_numbers: List[Optional[int]]) -> Optional[dict]:
"""Fetch episodes per season from TMDB, shaped like _fetch_episodes_paginated.

Returns {season_number: {episode_number: {title, overview, runtime,
first_aired, imdb_id, absolute}}}.

TMDB gives air_date as a plain YYYY-MM-DD with no air time or network
timezone, so first_aired is formatted date-only. Downstream treats a
date-only value the same way it does for TVDB shows lacking airsTime.
"""
if not tmdb_id:
return None

by_season: dict = {}
for sn in season_numbers:
if sn is None:
continue
try:
resp = requests.get(
f"https://api.themoviedb.org/3/tv/{tmdb_id}/season/{sn}",
params={'api_key': api_key},
timeout=REQUEST_TIMEOUT,
)
if resp.status_code != 200:
logger.debug(f"TMDB season {sn} fetch failed for tmdb_id={tmdb_id}: {resp.status_code}")
continue

ep_dict: dict = {}
for ep in (resp.json().get('episodes') or []):
ep_num = ep.get('episode_number')
if ep_num is None:
continue
ep_dict[ep_num] = {
'title': ep.get('name', '') or '',
'overview': ep.get('overview', '') or '',
'runtime': ep.get('runtime') or 0,
'first_aired': _format_air_date(ep.get('air_date')),
'imdb_id': None,
'absolute': None,
}
if ep_dict:
by_season[sn] = ep_dict
except Exception as e:
logger.debug(f"TMDB season {sn} error for tmdb_id={tmdb_id}: {e}")

if by_season:
total = sum(len(v) for v in by_season.values())
logger.info(f"TMDB: fetched {total} episode(s) across {len(by_season)} season(s) "
f"for tmdb_id={tmdb_id}")
return by_season or None


def _fetch_tmdb_show_data(imdb_id: str, api_key: str) -> Optional[dict]:
"""Fetch show metadata from TMDB as fallback when TVDB cannot resolve the ID."""
tmdb_id = _resolve_tmdb_id_from_imdb(imdb_id, api_key, media_type='show')
Expand Down Expand Up @@ -1570,15 +1686,23 @@ def _fetch_tmdb_show_data(imdb_id: str, api_key: str) -> Optional[dict]:

genres = [g.get('name', '') for g in (raw.get('genres') or []) if isinstance(g, dict)]

# Build seasons dict
# Build seasons dict, populating episodes from TMDB. Without episodes
# the show cannot expand into wanted items, so a show-level-only result
# is no more useful than none at all.
episodes_by_season = _fetch_tmdb_episodes(
tmdb_id, api_key,
[s.get('season_number') for s in (raw.get('seasons') or [])],
) or {}

seasons = {}
for s in (raw.get('seasons') or []):
sn = s.get('season_number')
if sn is not None:
fetched = episodes_by_season.get(sn) or {}
seasons[str(sn)] = {
'number': sn,
'episode_count': s.get('episode_count', 0),
'episodes': {},
'episode_count': len(fetched) or s.get('episode_count', 0),
'episodes': fetched,
}

data = {
Expand Down