import contextlib import datetime import json import random import time from typing import Any, Dict import requests import xbmc from lib.api.trakt.base_cache import BASE_DELETE, connect_database from lib.api.trakt.lists_cache import lists_cache, lists_cache_object from lib.api.trakt.main_cache import cache_object from lib.api.trakt.trakt_cache import ( cache_trakt_object, clear_trakt_collection_watchlist_data, clear_trakt_favorites, clear_trakt_list_contents_data, clear_trakt_list_data, trakt_watched_cache, ) from lib.api.trakt.trakt_utils import clean_ids, sort_for_article, sort_list from lib.gui.qr_progress_dialog import QRProgressDialog from lib.jacktook.utils import ADDON_PATH from lib.utils.debrid.qrcode_utils import make_qrcode from lib.utils.general.utils import set_pluging_category from lib.utils.kodi.settings import ( EMPTY_USER, trakt_client, trakt_lists_sort_order, trakt_secret, ) from lib.utils.kodi.utils import ( copy2clip, get_datetime, get_property, kodilog, notification, set_property, set_setting, sleep, translation, ) class TraktBase: def __init__(self): self.api_endpoint = "https://api.trakt.tv/%s" self.timeout = 20 self.empty_setting_check = (None, "empty_setting", "") self.standby_date = "2050-01-01T01:00:00.000Z" self.trakt_user = EMPTY_USER def ensure_token_valid(self): expires = get_property("trakt_expires") refresh_token = get_property("trakt_refresh") if expires and refresh_token: try: expires = float(expires) # Refresh if less than 1 hour left if expires - time.time() < 3600: self.trakt_refresh = refresh_token self.trakt_refresh_token() except Exception as e: kodilog(f"Error checking token expiry: {e}") def no_client_key(self): notification(translation(90382)) return None def no_secret_key(self): notification(translation(90383)) return None def _handle_unauthorized(self): kodilog("Trakt unauthorized - Revoking authentication") set_property("trakt_token", "") set_property("trakt_refresh", "") set_property("trakt_expires", "") set_setting("trakt_user", EMPTY_USER) set_setting("is_trakt_auth", "false") notification(translation(90408), time=5000) with contextlib.suppress(BaseException): TraktCache().clear_all_trakt_cache_data() def call_trakt( self, path, params=None, data=None, is_delete=False, with_auth=True, method=None, pagination=False, page_no=1, ): params = params or {} headers = { "Content-Type": "application/json", "trakt-api-version": "2", "trakt-api-key": trakt_client(), } if headers["trakt-api-key"] in self.empty_setting_check: return self.no_client_key() if with_auth: self.ensure_token_valid() token = get_property("trakt_token") if token: headers["Authorization"] = f"Bearer {token}" if pagination: params["page"] = page_no response = self._send_request(path, params, data, headers, is_delete, method) # Rate Limiting Handling if response.status_code == 429: retry_after = int(response.headers.get("Retry-After", 1)) kodilog(f"Trakt Rate Limit Exceeded. Retrying in {retry_after} seconds.") sleep(retry_after * 1000) response = self._send_request(path, params, data, headers, is_delete, method) kodilog(f"Trakt response status code: {response.status_code}") try: response.raise_for_status() except requests.HTTPError as error: status_code = response.status_code if status_code == 401: self._handle_unauthorized() return [] status_code = response.status_code error_messages = { 400: "Bad Request", 401: "Unauthorized", 403: "Forbidden", 404: "Not Found", 422: "Unprocessable Entity", 429: "Rate Limit Exceeded", 500: "Internal Server Error", 503: "Service Unavailable", 504: "Gateway Timeout", } error_message = error_messages.get(status_code, f"HTTP Error: {status_code}") kodilog(f"Trakt API error: {response.text}") raise ProviderException(f"Trakt API error: {error_message}") from error except requests.RequestException as error: error_message = f"Trakt API error: {error}" raise ProviderException(error_message) from error return self._process_response(response, method, pagination) def _send_request(self, path, params, data, headers, is_delete, method): url = self.api_endpoint % path kodilog(f"Trakt URL: {url}") if method == "post": return requests.post(url, json=data, headers=headers, timeout=self.timeout) elif method == "delete": return requests.delete(url, json=data, headers=headers, timeout=self.timeout) elif method == "sort_by_headers": return requests.get(url, params=params, headers=headers, timeout=self.timeout) elif data is not None: return requests.post(url, json=data, headers=headers, timeout=self.timeout) elif is_delete: return requests.delete(url, headers=headers, timeout=self.timeout) else: return requests.get(url, params=params, headers=headers, timeout=self.timeout) def _process_response(self, response, method, pagination): response.encoding = "utf-8" try: result = response.json() except ValueError: return None if method == "sort_by_headers": headers = response.headers if "X-Sort-By" in headers and "X-Sort-How" in headers: with contextlib.suppress(Exception): result = sort_list(headers["X-Sort-By"], headers["X-Sort-How"], result) if pagination: return result, response.headers.get("X-Pagination-Page-Count") return result def get_trakt(self, params): try: path_insert = params.get("path_insert") if path_insert is not None: if not isinstance(path_insert, (tuple)): path_insert = (path_insert,) formatted_path = params["path"] % path_insert else: formatted_path = params["path"] result = self.call_trakt( formatted_path, params=params.get("params", {}), data=params.get("data"), is_delete=params.get("is_delete", False), with_auth=params.get("with_auth", False), method=params.get("method"), pagination=params.get("pagination", True), page_no=params.get("page_no"), ) if result is not None: if params.get("pagination", True): if isinstance(result, (list, tuple)) and len(result) > 0: return result[0] return [] return result except KeyError as e: kodilog(f"KeyError in get_trakt: {e}") raise except TypeError as e: kodilog(f"TypeError in get_trakt: {e}") raise def trakt_refresh_token(self): CLIENT_ID = trakt_client() if CLIENT_ID in self.empty_setting_check: return self.no_client_key() CLIENT_SECRET = trakt_secret() if CLIENT_SECRET in self.empty_setting_check: return self.no_secret_key() data = { "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, "redirect_uri": "urn:ietf:wg:oauth:2.0:oob", "grant_type": "refresh_token", "refresh_token": self.trakt_refresh, } response = self.call_trakt("oauth/token", data=data, with_auth=False) if response and isinstance(response, dict): set_property("trakt_token", response["access_token"]) set_property("trakt_refresh", response["refresh_token"]) set_property("trakt_expires", str(time.time() + 82800)) # 23 hours def get_trakt_id_by_tmdb(self, tmdb_id, media_type="movie"): trakt_object = self.get_trakt_object_by_tmdb(tmdb_id, media_type) if trakt_object: try: return trakt_object["ids"]["trakt"] except (KeyError, TypeError): return None return None def get_trakt_object_by_tmdb(self, tmdb_id, media_type="movie"): params = { "path": "search/tmdb/%s", "path_insert": tmdb_id, "params": {"type": media_type}, "with_auth": False, "pagination": False, } results = self.get_trakt(params) if results and isinstance(results, list) and len(results) > 0: try: return results[0][media_type] except (KeyError, IndexError, TypeError): return None return None class TraktAuthentication(TraktBase): def __init__(self): super().__init__() def get_user_settings(self): params = { "path": "users/settings", "path_insert": (), "with_auth": True, "pagination": False, } return self.get_trakt(params) def get_account_stats(self): params = { "path": "users/me/stats", "with_auth": True, "pagination": False, } return self.get_trakt(params) def trakt_get_device_code(self): CLIENT_ID = trakt_client() if CLIENT_ID in self.empty_setting_check: return self.no_client_key() data = {"client_id": CLIENT_ID} return self.call_trakt("oauth/device/code", data=data, with_auth=False) def trakt_get_device_token(self, device_codes): CLIENT_ID = trakt_client() if CLIENT_ID in self.empty_setting_check: return self.no_client_key() CLIENT_SECRET = trakt_secret() if CLIENT_SECRET in self.empty_setting_check: return self.no_secret_key() result = None headers = { "Content-Type": "application/json", "trakt-api-version": "2", "trakt-api-key": CLIENT_ID, } data = { "code": device_codes["device_code"], "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, } start = time.time() expires_in = device_codes["expires_in"] sleep_interval = device_codes["interval"] user_code = str(device_codes["user_code"]) verification_url = str(device_codes["verification_url"]) # Generate QR code for the verification URL qr_code = make_qrcode(verification_url) with contextlib.suppress(BaseException): copy2clip(user_code) # Create and display QR code dialog progressDialog = QRProgressDialog("qr_dialog.xml", ADDON_PATH) progressDialog.setup( translation(90563), qr_code, verification_url, user_code, "", is_debrid=False, ) progressDialog.show_dialog() try: time_passed = 0 while not progressDialog.iscanceled and time_passed < expires_in: sleep(max(sleep_interval, 1) * 1000) response = requests.post( self.api_endpoint % "oauth/device/token", data=json.dumps(data), headers=headers, timeout=self.timeout, ) status_code = response.status_code if status_code == 200: result = response.json() progressDialog.update_progress(100, translation(90545)) break elif status_code == 400: time_passed = time.time() - start progress = int(100 * time_passed / expires_in) progressDialog.update_progress(progress) else: break except Exception: pass with contextlib.suppress(BaseException): progressDialog.close_dialog() return result def trakt_authenticate(self): code = self.trakt_get_device_code() token = self.trakt_get_device_token(code) if not token: kodilog("Trakt authentication failed, no token received") notification(translation(90384), time=3000) return False set_property("trakt_token", str(token["access_token"])) set_property("trakt_refresh", str(token["refresh_token"])) set_property("trakt_expires", str(time.time() + 82800)) # 23 hours try: user = self.call_trakt("users/me") if user and isinstance(user, dict): set_setting("trakt_user", str(user["username"])) set_setting("is_trakt_auth", "true") notification(translation(90385), time=3000) return True except Exception: kodilog("Trakt user not found, setting to empty user") set_setting("is_trakt_auth", "false") notification(translation(90384), time=3000) return False def trakt_revoke_authentication(self): kodilog("Revoking Trakt authentication") TraktCache().clear_all_trakt_cache_data() CLIENT_ID = trakt_client() if CLIENT_ID in self.empty_setting_check: return self.no_client_key() CLIENT_SECRET = trakt_secret() if CLIENT_SECRET in self.empty_setting_check: return self.no_secret_key() data = { "token": get_property("trakt_token"), "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, } set_property("trakt_token", "") set_property("trakt_refresh", "") set_property("trakt_expires", "") set_setting("trakt_user", EMPTY_USER) set_setting("is_trakt_auth", "false") self.call_trakt("oauth/revoke", data=data, with_auth=False) notification(translation(90386), time=3000) class TraktMovies(TraktBase): def trakt_movies_trending(self, page_no): set_pluging_category(translation(90028)) string = f"trakt_movies_trending_{page_no}" params = { "path": "movies/trending", "params": {"limit": 20}, "page_no": page_no, } return lists_cache_object(self.get_trakt, string, params) def trakt_movies_trending_recent(self, page_no): set_pluging_category(translation(90024)) current_year = get_datetime().year years = f"{current_year - 1!s}-{current_year!s}" string = f"trakt_movies_trending_recent_{page_no}" params = { "path": "movies/trending", "params": {"limit": 20, "years": years}, "page_no": page_no, } return lists_cache_object(self.get_trakt, string, params) def trakt_movies_top10_boxoffice(self): set_pluging_category(translation(90036)) string = "trakt_movies_top10_boxoffice" params = {"path": "movies/boxoffice", "pagination": False} return lists_cache_object(self.get_trakt, string, params) def trakt_movies_most_watched(self, page_no): set_pluging_category(translation(90029)) string = f"trakt_movies_most_watched_{page_no}" params = { "path": "movies/watched/daily", "params": {"limit": 20}, "page_no": page_no, } return lists_cache_object(self.get_trakt, string, params) def trakt_movies_most_favorited(self, page_no): set_pluging_category(translation(90030)) string = f"trakt_movies_most_favorited{page_no}" params = { "path": "movies/favorited/daily", "params": {"limit": 20}, "page_no": page_no, } return lists_cache_object(self.get_trakt, string, params) def trakt_recommendations(self, media_type): set_pluging_category(translation(90033)) string = f"trakt_recommendations_{media_type}" params = { "path": "/recommendations/%s", "path_insert": media_type, "with_auth": True, "params": { "limit": 50, "ignore_collected": "true", "ignore_watchlisted": "true", }, "pagination": False, } return lists_cache_object(self.get_trakt, string, params) def get_watched_movies(self): params = { "path": "sync/watched/movies", "with_auth": True, "pagination": False, "params": {"extended": "full"}, } return self.get_trakt(params) def trakt_movies_progress(self): set_pluging_category(translation(90200)) try: return TraktScrobble().trakt_get_playback_progress("movies") except Exception: return [] def trakt_collection(self, page_no): set_pluging_category(translation(90294)) params = { "path": "sync/collection/movies", "params": {"limit": 20, "extended": "full"}, "page_no": page_no, "with_auth": True, } return self.get_trakt(params) class TraktTV(TraktBase): def trakt_tv_trending(self, page_no): set_pluging_category(translation(90028)) string = f"trakt_tv_trending_{page_no}" params = { "path": "shows/trending", "params": {"limit": 20}, "page_no": page_no, } return lists_cache_object(self.get_trakt, string, params) def trakt_tv_trending_recent(self, page_no): set_pluging_category(translation(90024)) current_year = get_datetime().year years = f"{current_year - 1!s}-{current_year!s}" string = f"trakt_tv_trending_recent_{page_no}" params = { "path": "shows/trending", "params": {"limit": 20, "years": years}, "page_no": page_no, } return lists_cache_object(self.get_trakt, string, params) def trakt_tv_most_watched(self, page_no): set_pluging_category(translation(90029)) string = f"trakt_tv_most_watched_{page_no}" params = { "path": "shows/watched/daily", "params": {"limit": 20}, "page_no": page_no, } return lists_cache_object(self.get_trakt, string, params) def trakt_tv_most_favorited(self, page_no): set_pluging_category(translation(90030)) string = f"trakt_tv_most_favorited_{page_no}" params = { "path": "shows/favorited/daily", "params": {"limit": 20}, "page_no": page_no, } return lists_cache_object(self.get_trakt, string, params) def trakt_recommendations(self, media_type): set_pluging_category(translation(90033)) string = f"trakt_recommendations_{media_type}" params = { "path": "/recommendations/%s", "path_insert": media_type, "with_auth": True, "params": { "limit": 50, "ignore_collected": "true", "ignore_watchlisted": "true", }, "pagination": False, } return lists_cache_object(self.get_trakt, string, params) def get_watched_shows(self): params = { "path": "sync/watched/shows", "with_auth": True, "pagination": False, "params": {"extended": "full"}, } return self.get_trakt(params) @staticmethod def _parse_iso_date(value): if not value: return None try: return datetime.datetime.fromisoformat(value.replace("Z", "+00:00")) except Exception: return None @staticmethod def _is_episode_aired(air_date, now_dt): if not air_date: return True now_date = now_dt.date() if hasattr(now_dt, "date") else now_dt try: return datetime.datetime.strptime(air_date, "%Y-%m-%d").date() <= now_date except ValueError: return True @staticmethod def _last_watched_episode(show_item): last_episode = None last_timestamp = None for season in show_item.get("seasons", []): season_number = season.get("number") if not season_number or season_number == 0: continue for episode in season.get("episodes", []): episode_number = episode.get("number") if not episode_number: continue episode_key = (int(season_number), int(episode_number)) watched_at = episode.get("last_watched_at") or show_item.get("last_watched_at") watched_dt = TraktTV._parse_iso_date(watched_at) if watched_dt: if last_timestamp is None or watched_dt > last_timestamp: last_timestamp = watched_dt last_episode = episode_key elif last_episode is None or episode_key > last_episode: last_episode = episode_key return last_episode, last_timestamp @staticmethod def _find_next_episode(fetcher, tmdb_id, starting_point, now_dt): if not tmdb_id or not starting_point: return None show_details = fetcher("tv_details", tmdb_id) total_seasons = getattr(show_details, "number_of_seasons", 0) or 0 start_season, start_episode = starting_point for season_number in range(max(1, int(start_season)), int(total_seasons) + 1): season_details = fetcher("season_details", {"id": tmdb_id, "season": season_number}) episodes = getattr(season_details, "episodes", []) or [] for episode in episodes: episode_number = getattr(episode, "episode_number", 0) or 0 if season_number == int(start_season) and episode_number <= int(start_episode): continue air_date = getattr(episode, "air_date", None) if not TraktTV._is_episode_aired(air_date, now_dt): continue return { "season": season_number, "number": episode_number, "title": getattr(episode, "name", ""), "first_aired": air_date, } return None @classmethod def _build_up_next_entries( cls, watched_shows, progress_items, fetcher=None, now_dt=None, hidden_items=None ): if fetcher is None: from lib.clients.tmdb.utils.utils import tmdb_get as fetcher now_dt = now_dt or get_datetime(string=False) entries = [] progress_by_show = {} hidden_items = hidden_items or [] for item in progress_items or []: show = item.get("show", {}) ids = show.get("ids", {}) tmdb_id = ids.get("tmdb") episode = item.get("episode", {}) if not tmdb_id or int(tmdb_id) in hidden_items or not episode: continue season_number = episode.get("season") episode_number = episode.get("number") if not season_number or int(season_number) == 0 or not episode_number: continue if not cls._is_episode_aired(episode.get("first_aired"), now_dt): continue progress_entry = { "type": "resume", "show": show, "episode": { "season": season_number, "number": episode_number, "title": episode.get("title", ""), "first_aired": episode.get("first_aired"), }, "progress": item.get("progress", 0), "activity_at": item.get("paused_at") or "", } existing_entry = progress_by_show.get(int(tmdb_id)) if existing_entry: existing_activity = existing_entry.get("activity_at", "") new_activity = progress_entry.get("activity_at", "") if (new_activity, progress_entry.get("progress", 0)) <= ( existing_activity, existing_entry.get("progress", 0), ): continue progress_by_show[int(tmdb_id)] = progress_entry entries.extend(progress_by_show.values()) for show_item in watched_shows or []: show = show_item.get("show", {}) ids = show.get("ids", {}) tmdb_id = ids.get("tmdb") if not tmdb_id: continue tmdb_id = int(tmdb_id) if tmdb_id in progress_by_show or tmdb_id in hidden_items: continue last_episode, last_timestamp = cls._last_watched_episode(show_item) if not last_episode: continue next_episode = cls._find_next_episode(fetcher, tmdb_id, last_episode, now_dt) if not next_episode: continue entries.append( { "type": "next", "show": show, "episode": next_episode, "progress": 0, "activity_at": last_timestamp.isoformat() if last_timestamp else "", } ) return sorted( entries, key=lambda item: ( 1 if item.get("type") == "resume" else 0, item.get("activity_at", ""), item.get("progress", 0), ), reverse=True, ) def trakt_get_hidden_items(self, section="progress_watched"): def _process(params): result = self.get_trakt(params) hidden = [] if result: for item in result: if item.get("type") == "show": tmdb_id = item.get("show", {}).get("ids", {}).get("tmdb") if tmdb_id: hidden.append(int(tmdb_id)) return hidden string = f"trakt_hidden_items_{section}" params = { "path": f"users/hidden/{section}", "params": {"limit": 1000}, "with_auth": True, "pagination": False, } return cache_trakt_object(_process, string, params) def trakt_up_next(self): watched_shows = self.get_watched_shows() progress_items = TraktScrobble().trakt_get_playback_progress("episodes") hidden_items = self.trakt_get_hidden_items("progress_watched") return self._build_up_next_entries(watched_shows, progress_items, hidden_items=hidden_items) def trakt_collection(self, page_no): set_pluging_category(translation(90294)) params = { "path": "sync/collection/shows", "params": {"limit": 20, "extended": "full"}, "page_no": page_no, "with_auth": True, } return self.get_trakt(params) class TraktAnime(TraktBase): def _get_path_prefix(self, mode): return "movies" if mode == "movies" else "shows" def trakt_anime_trending(self, page_no, mode="tv"): path_prefix = self._get_path_prefix(mode) string = f"trakt_anime_trending_{mode}_{page_no}" params = { "path": f"{path_prefix}/trending", "params": {"genres": "anime", "limit": 20}, "page_no": page_no, } return lists_cache_object(self.get_trakt, string, params) def trakt_anime_trending_recent(self, page_no, mode="tv"): path_prefix = self._get_path_prefix(mode) current_year = get_datetime().year years = f"{current_year - 1!s}-{current_year!s}" string = f"trakt_anime_trending_recent_{mode}_{page_no}" params = { "path": f"{path_prefix}/trending", "params": {"genres": "anime", "limit": 20, "years": years}, "page_no": page_no, } return lists_cache_object(self.get_trakt, string, params) def trakt_anime_most_watched(self, page_no, mode="tv"): path_prefix = self._get_path_prefix(mode) string = f"trakt_anime_most_watched_{mode}_{page_no}" params = { "path": f"{path_prefix}/watched/daily", "params": {"genres": "anime", "limit": 20}, "page_no": page_no, } return lists_cache_object(self.get_trakt, string, params) def trakt_anime_most_favorited(self, page_no, mode="tv"): path_prefix = self._get_path_prefix(mode) string = f"trakt_anime_most_favorited_{mode}_{page_no}" params = { "path": f"{path_prefix}/favorited/daily", "params": {"genres": "anime", "limit": 20}, "page_no": page_no, } return lists_cache_object(self.get_trakt, string, params) class TraktLists(TraktBase): @staticmethod def _history_payload(media_type, season, episode, ids): if media_type in ("movie", "movies"): media_type_key = "movies" payload = {media_type_key: [{"ids": {"tmdb": int(ids["tmdb"])}}]} else: media_type_key = "shows" show_item: Dict[str, Any] = {"ids": {"tmdb": int(ids["tmdb"])}} if season: show_item["seasons"] = [{"number": int(season)}] if season and episode: show_item["seasons"] = [ {"number": int(season), "episodes": [{"number": int(episode)}]} ] payload = {media_type_key: [show_item]} return payload @staticmethod def _ids_match(candidate_ids, target_ids): for key in ("trakt", "tmdb", "tvdb", "imdb"): candidate = candidate_ids.get(key) target = target_ids.get(key) or target_ids.get(f"{key}_id") if candidate and target and str(candidate) == str(target): return True return False def get_collection_items(self, media_type): if media_type in ("movie", "movies"): path = "sync/collection/movies" else: path = "sync/collection/shows" return self.get_trakt( { "path": path, "with_auth": True, "pagination": False, "params": {"extended": "full"}, } ) def build_collection_remove_payload(self, media_type, ids): if media_type in ("movie", "movies"): return self._list_item_payload(media_type, ids) items = self.get_collection_items(media_type) if not items: return None for item in items: show = item.get("show") or item show_ids = show.get("ids", {}) if not self._ids_match(show_ids, ids): continue payload = {"shows": [{"ids": clean_ids(show_ids)}]} if show.get("title"): payload["shows"][0]["title"] = show.get("title") if show.get("year"): payload["shows"][0]["year"] = show.get("year") seasons = item.get("seasons", []) or [] if seasons: payload["shows"][0]["seasons"] = [] for season in seasons: season_number = season.get("number") if not season_number: continue season_payload = {"number": int(season_number)} episodes = season.get("episodes", []) or [] if episodes: season_payload["episodes"] = [] for episode in episodes: episode_number = episode.get("number") if episode_number: season_payload["episodes"].append({"number": int(episode_number)}) payload["shows"][0]["seasons"].append(season_payload) return payload return None def trakt_watchlist(self, media_type): kodilog("Fetching trakt watchlist") result = self.trakt_fetch_sorted_list("watchlist", media_type) kodilog(f"Watchlist result: {result}", level=xbmc.LOGDEBUG) return result def add_to_watchlist(self, media_type, ids): original_media_type = media_type if media_type in ("movie", "movies"): media_type = "movies" else: media_type = "shows" payload = self._list_item_payload(media_type, ids) result = self.call_trakt( "sync/watchlist", data=payload, with_auth=True, pagination=False, ) lists_cache.delete_prefix( "trakt_%s_watchlist_" % ("movies" if media_type == "movies" else "tv") ) clear_trakt_collection_watchlist_data("watchlist", original_media_type) return result def remove_from_watchlist(self, media_type, ids): original_media_type = media_type if media_type in ("movie", "movies"): media_type = "movies" else: media_type = "shows" payload = self._list_item_payload(media_type, ids) result = self.call_trakt( "sync/watchlist/remove", data=payload, with_auth=True, pagination=False, ) lists_cache.delete_prefix( "trakt_%s_watchlist_" % ("movies" if media_type == "movies" else "tv") ) clear_trakt_collection_watchlist_data("watchlist", original_media_type) return result def trakt_comments(self, media_type, tmdb_id, sort_type="likes", page_no=1): if media_type in ("movie", "movies"): media_type = "movies" else: media_type = "shows" trakt_id = self.get_trakt_id_by_tmdb(tmdb_id, media_type=media_type[:-1]) if not trakt_id: return [] params = { "path": f"{media_type}/%s/comments/%s", "path_insert": (trakt_id, sort_type), "with_auth": False, "params": {"limit": 20}, "pagination": True, "page_no": page_no, } return self.get_trakt(params) def add_to_collection(self, media_type, ids, payload=None): original_media_type = media_type if media_type in ("movie", "movies"): media_type = "movies" else: media_type = "shows" payload = payload or self._list_item_payload(media_type, ids) result = self.call_trakt( "sync/collection", data=payload, with_auth=True, pagination=False, ) lists_cache.delete_prefix( "trakt_%s_collection_" % ("movies" if media_type == "movies" else "tv") ) clear_trakt_collection_watchlist_data("collection", original_media_type) return result def remove_from_collection(self, media_type, ids, payload=None): original_media_type = media_type if media_type in ("movie", "movies"): media_type = "movies" else: media_type = "shows" payload = payload or self._list_item_payload(media_type, ids) result = self.call_trakt( "sync/collection/remove", data=payload, with_auth=True, pagination=False, ) lists_cache.delete_prefix( "trakt_%s_collection_" % ("movies" if media_type == "movies" else "tv") ) clear_trakt_collection_watchlist_data("collection", original_media_type) return result def trakt_watched_history(self, media_type, page_no, sort_type="recent"): def _process(params, media_type): response = self.get_trakt(params) history = [] if not response or not isinstance(response, (list, tuple)): return history if media_type == "movies": for item in response: if item["type"] == "movie": history.append( { "media_ids": item["movie"]["ids"], "title": item["movie"]["title"], "type": "movie", "watched_at": item.get("watched_at"), } ) elif media_type == "shows": for item in response: if item["type"] == "episode": history.append( { "media_ids": item["show"]["ids"], "title": f"{item['show']['title']} - S{item['episode']['season']}E{item['episode']['number']} - {item['episode']['title']}", "type": "show", "watched_at": item.get("watched_at"), "show_title": item["show"]["title"], "ep_title": item["episode"]["title"], "season": item["episode"]["season"], "episode": item["episode"]["number"], } ) # --- Sorting logic similar to trakt_fetch_sorted_list --- if sort_type == "recent": history.sort(key=lambda k: k.get("watched_at") or "", reverse=True) elif sort_type == "random": random.shuffle(history) elif sort_type is None: # Default: sort by watched_at descending history.sort(key=lambda k: k.get("watched_at") or "", reverse=True) return history if media_type in ("movie", "movies"): media_type = "movies" else: media_type = "shows" params = { "path": "sync/history/%s", "path_insert": media_type, "page_no": page_no, "with_auth": True, } return _process(params, media_type) def mark_as_watched(self, media_type, season, episode, ids): payload = self._history_payload(media_type, season, episode, ids) kodilog(f"Payload: {payload}") response = self.call_trakt( "sync/history", data=payload, with_auth=True, pagination=False, ) if ( response and isinstance(response, dict) and "added" in response and response["added"].get("movies", 0) == 0 and response["added"].get("episodes", 0) == 0 ): notification(translation(90387), time=3000) else: notification(translation(90388), time=3000) return response def mark_as_unwatched(self, media_type, season, episode, ids): payload = self._history_payload(media_type, season, episode, ids) return self.call_trakt( "sync/history/remove", data=payload, with_auth=True, pagination=False, ) def trakt_fetch_sorted_list(self, list_type, media_type, sort_type=None, limit=None): data = self.trakt_fetch_watchlist(list_type, media_type) if sort_type == "recent": data.sort(key=lambda k: k["collected_at"], reverse=True) elif sort_type == "random": random.shuffle(data) elif sort_type is None and list_type == "watchlist": sort_order = trakt_lists_sort_order("watchlist") if sort_order == 0: data = sort_for_article(data, "title") elif sort_order == 1: data.sort(key=lambda k: k["collected_at"], reverse=True) else: data.sort(key=lambda k: k["released"], reverse=True) return data[:limit] if limit else data def trakt_fetch_watchlist(self, list_type, media_type): def _process(params): raw_data = self.get_trakt(params) if not raw_data: return [] return [ { "media_ids": { "tmdb": item[media_key]["ids"].get("tmdb", ""), "imdb": item[media_key]["ids"].get("imdb", ""), "tvdb": item[media_key]["ids"].get("tvdb", ""), }, "title": item[media_key]["title"], "collected_at": item.get(collected_at_key), "released": item[media_key].get(release_date_key, default_release_date), } for item in raw_data ] if media_type in ("movie", "movies"): media_key = "movie" release_date_key = "released" collected_at_key = "listed_at" default_release_date = "2050-01-01" else: media_key = "show" release_date_key = "first_aired" collected_at_key = "listed_at" default_release_date = self.standby_date api_path = "sync/%s/%s?extended=full" params = { "path": api_path, "path_insert": (list_type, media_key), "with_auth": True, "pagination": False, } return _process(params) def trakt_search_lists(self, search_title, page_no): def _process(dummy_arg): result = self.call_trakt( "search", params={ "type": "list", "fields": "name, description", "query": search_title, "limit": 50, }, pagination=True, page_no=page_no, ) if isinstance(result, tuple): result = result[0] if not result: return [] normalized = [] for item in result: list_data = item.get("list", {}) user_data = list_data.get("user", {}) or item.get("user", {}) or {} ids = list_data.get("ids", {}) user_ids = user_data.get("ids", {}) user_slug = ( user_ids.get("slug") or user_data.get("username") or list_data.get("username") or "" ) normalized.append( { "list_type": "search_lists", "name": list_data.get("name", ""), "description": list_data.get("description", ""), "slug": ids.get("slug", ""), "trakt_id": ids.get("trakt"), "user_slug": user_slug, "username": user_data.get("username", ""), "item_count": list_data.get("item_count", 0), "privacy": list_data.get("privacy", "public"), "with_auth": False, "can_like": True, } ) return normalized string = f"trakt_search_lists_{search_title}_{page_no}" return cache_object(_process, string, "dummy_arg", False, 4) def trakt_favorites(self, media_type): def _process(params): result = self.get_trakt(params) if not result: return [] return [ { "media_ids": { "tmdb": i[i["type"]]["ids"].get("tmdb", ""), "imdb": i[i["type"]]["ids"].get("imdb", ""), "tvdb": i[i["type"]]["ids"].get("tvdb", ""), }, "title": i[i["type"]].get("title", ""), "type": i["type"], } for i in result ] media_type = "movies" if media_type in ("movie", "movies") else "shows" string = f"trakt_favorites_{media_type}" params = { "path": "users/me/favorites/%s/%s", "path_insert": (media_type, "title"), "with_auth": True, "pagination": False, } return cache_trakt_object(_process, string, params) def get_trakt_list_contents(self, list_type, user, slug, with_auth, trakt_id=None): def _process(params): result = self.get_trakt(params) if not result: return [] return [ { "media_ids": i[i["type"]]["ids"], "title": i[i["type"]]["title"], "type": i["type"], "order": c, } for c, i in enumerate(result) if i["type"] in ("movie", "show") ] string = f"trakt_list_contents_{list_type}_{user}_{slug}_{trakt_id}" if list_type == "my_lists" and trakt_id: params = { "path": "users/me/lists/%s/items", "path_insert": trakt_id, "params": {"extended": "full"}, "with_auth": True, "method": "sort_by_headers", } elif trakt_id: params = { "path": "lists/%s/items", "path_insert": trakt_id, "params": {"extended": "full"}, "with_auth": with_auth, "method": "sort_by_headers", } else: params = { "path": "users/%s/lists/%s/items", "path_insert": (user, slug), "params": {"extended": "full"}, "with_auth": with_auth, "method": "sort_by_headers", } return cache_trakt_object(_process, string, params) def trakt_trending_popular_lists(self, list_type, page_no): set_pluging_category(translation(90072)) string = f"trakt_{list_type}_user_lists_{page_no}" params = { "path": "lists/%s", "path_insert": list_type, "params": {"limit": 50}, "page_no": page_no, } return cache_object(self.get_trakt, string, params, False) def trakt_get_lists(self, list_type): def _process(params): result = self.get_trakt(params) if not result: return [] normalized = [] for item in result: list_data = item.get("list", item) user_data = list_data.get("user", {}) ids = list_data.get("ids", {}) user_ids = user_data.get("ids", {}) user_slug = user_ids.get("slug") or user_data.get("username") or "" normalized.append( { "list_type": list_type, "name": list_data.get("name", ""), "description": list_data.get("description", ""), "slug": ids.get("slug", ""), "trakt_id": ids.get("trakt"), "user_slug": user_slug, "username": user_data.get("username", ""), "item_count": list_data.get("item_count", 0), "privacy": list_data.get("privacy", "public"), "with_auth": True, "can_delete": list_type == "my_lists", "can_unlike": list_type == "liked_lists", } ) return normalized if list_type == "my_lists": string = "trakt_my_lists" path = "users/me/lists" elif list_type == "liked_lists": string = "trakt_liked_lists" path = "users/likes/lists" else: return [] params = { "path": path, "params": {"limit": 1000}, "pagination": False, "with_auth": True, } return cache_trakt_object(_process, string, params) def add_to_favorites(self, media_type, ids): response = self.call_trakt( "sync/favorites", data=self._list_item_payload(media_type, ids), with_auth=True, pagination=False, ) clear_trakt_favorites() return response def remove_from_favorites(self, media_type, ids): response = self.call_trakt( "sync/favorites/remove", data=self._list_item_payload(media_type, ids), with_auth=True, pagination=False, ) clear_trakt_favorites() return response def create_list(self, name, description="", privacy="private"): response = self.call_trakt( "users/me/lists", data={ "name": name, "description": description, "privacy": privacy, "display_numbers": True, "allow_comments": True, }, with_auth=True, pagination=False, ) clear_trakt_list_data("my_lists") clear_trakt_list_contents_data("my_lists") return response def delete_list(self, trakt_id): response = self.call_trakt( f"users/me/lists/{trakt_id}", method="delete", with_auth=True, pagination=False, ) clear_trakt_list_data("my_lists") clear_trakt_list_contents_data("my_lists") return response def like_list(self, user_slug, trakt_id): path = ( f"users/{user_slug}/lists/{trakt_id}/like" if user_slug else f"lists/{trakt_id}/like" ) response = self.call_trakt( path, data={}, with_auth=True, pagination=False, ) clear_trakt_list_data("liked_lists") return response def unlike_list(self, user_slug, trakt_id): path = ( f"users/{user_slug}/lists/{trakt_id}/like" if user_slug else f"lists/{trakt_id}/like" ) response = self.call_trakt( path, method="delete", with_auth=True, pagination=False, ) clear_trakt_list_data("liked_lists") clear_trakt_list_contents_data("liked_lists") return response def _list_item_payload(self, media_type, ids): media_key = "movies" if media_type in ("movie", "movies") else "shows" normalized_ids = clean_ids( { "tmdb": ids.get("tmdb") or ids.get("tmdb_id"), "tvdb": ids.get("tvdb") or ids.get("tvdb_id"), "imdb": ids.get("imdb") or ids.get("imdb_id"), } ) for key in ("tmdb", "tvdb"): if key in normalized_ids: with contextlib.suppress(TypeError, ValueError): normalized_ids[key] = int(normalized_ids[key]) payload = {media_key: [{"ids": normalized_ids}]} return payload def add_item_to_list(self, trakt_id, media_type, ids): response = self.call_trakt( f"users/me/lists/{trakt_id}/items", data=self._list_item_payload(media_type, ids), with_auth=True, pagination=False, ) clear_trakt_list_data("my_lists") clear_trakt_list_contents_data("my_lists") return response def remove_item_from_list(self, trakt_id, media_type, ids): response = self.call_trakt( f"users/me/lists/{trakt_id}/items/remove", data=self._list_item_payload(media_type, ids), with_auth=True, pagination=False, ) clear_trakt_list_data("my_lists") clear_trakt_list_contents_data("my_lists") return response def make_trakt_slug(self, name): import re name = name.strip() name = name.lower() name = re.sub("[^a-z0-9_]", "-", name) name = re.sub("--+", "-", name) return name class TraktScrobble(TraktBase): @staticmethod def _scrobble_payload(data): from datetime import datetime from lib.utils.kodi.utils import ADDON_VERSION progress = data.get("progress", 0) if isinstance(progress, str): try: progress = float(progress) except ValueError: progress = 0.0 payload: Dict[str, Any] = { "progress": progress, "app_version": ADDON_VERSION, "app_date": datetime.now().strftime("%Y-%m-%d"), } if data["mode"] == "movies": payload["movie"] = {"ids": {"tmdb": data["ids"]["tmdb_id"]}} elif data["mode"] == "tv": if data.get("tv_data") is None: return None payload["show"] = {"ids": {"tmdb": data["ids"]["tmdb_id"]}} payload["episode"] = { "season": data.get("tv_data").get("season"), "number": data.get("tv_data").get("episode"), } return payload def trakt_start_scrobble(self, data): payload = self._scrobble_payload(data) if payload is None: return self.call_trakt("scrobble/start", data=payload, with_auth=True) def trakt_pause_scrobble(self, data): payload = self._scrobble_payload(data) if payload is None or payload.get("progress", 0) < 1.0: return self.call_trakt("scrobble/pause", data=payload, with_auth=True) def trakt_stop_scrobble(self, data): kodilog("Stopping scrobble") payload = self._scrobble_payload(data) if payload is None or payload.get("progress", 0) < 1.0: return self.call_trakt("scrobble/stop", data=payload, with_auth=True) def trakt_get_last_tracked_position(self, data): try: media_type = "movies" if data["mode"] == "movies" else "episodes" season = data.get("tv_data", {}).get("season") episode = data.get("tv_data", {}).get("episode") tmdb_id = data.get("ids", {}).get("tmdb_id") # Check local cache first if data["mode"] == "movies": cached_progress = trakt_watched_cache.get_progress("movie", str(tmdb_id)) else: cached_progress = trakt_watched_cache.get_progress( "episode", str(tmdb_id), season, episode ) if cached_progress > 0: kodilog(f"Trakt: Found cached resume point: {cached_progress}%") return cached_progress path = f"sync/playback/{media_type}" response = self.call_trakt(path, with_auth=True) if response: for item in response: try: if item["type"] == "movie": trakt_tmdb = int(item["movie"]["ids"]["tmdb"]) local_tmdb = int(tmdb_id) if trakt_tmdb == local_tmdb: progress = item.get("progress", 0) kodilog(f"Trakt: Found cloud resume point for movie: {progress}%") return progress elif item["type"] == "episode": trakt_tmdb = int(item["show"]["ids"]["tmdb"]) local_tmdb = int(tmdb_id) # TV Show comparison logic if ( trakt_tmdb == local_tmdb and item["episode"]["season"] == int(season) and item["episode"]["number"] == int(episode) ): progress = item.get("progress", 0) kodilog(f"Trakt: Found cloud resume point for episode: {progress}%") return progress except Exception: continue except Exception as e: kodilog(f"Trakt: Error fetching last tracked position: {e}") return 0.0 return 0.0 def trakt_get_playback_progress(self, media_type): # media_type: "movies" or "episodes" try: path = f"sync/playback/{media_type}" params = { "path": path, "params": {"limit": 1000}, # Safety limit "with_auth": True, "pagination": False, } return self.get_trakt(params) except Exception as e: kodilog(f"Error fetching playback progress: {e}") return [] return 0 class TraktCache(TraktBase): def clear_all_trakt_cache_data(self): try: dbcon = connect_database("trakt_db") for table in ("trakt_data", "progress", "watched", "watched_status"): dbcon.execute(BASE_DELETE % table) dbcon.execute("VACUUM") return True except Exception: return False def clear_cache(self, cache_type): success = True if cache_type == "trakt": success = self.clear_all_trakt_cache_data() elif cache_type == "list": success = lists_cache.delete_all_lists() if success: return success class TraktCalendar(TraktBase): def trakt_my_calendar(self, start_date, days): params = { "path": "calendars/my/shows/%s/%s", "path_insert": (start_date, days), "with_auth": True, "pagination": False, "params": {"extended": "full"}, } return self.get_trakt(params) def trakt_all_shows_calendar(self, start_date, days): params = { "path": "calendars/all/shows/%s/%s", "path_insert": (start_date, days), "with_auth": True, "pagination": False, "params": {"extended": "full"}, } return self.get_trakt(params) class TraktSync(TraktBase): def get_last_activities(self): return self.call_trakt( "sync/last_activities", with_auth=True, pagination=False, ) class TraktAPI: def __init__(self): self.auth = TraktAuthentication() self.movies = TraktMovies() self.tv = TraktTV() self.anime = TraktAnime() self.lists = TraktLists() self.scrobble = TraktScrobble() self.sync = TraktSync() self.cache = TraktCache() self.calendar = TraktCalendar() class ProviderException(Exception): def __init__(self, message): self.message = message super().__init__(self.message) notification(self.message)