import contextlib import json from datetime import datetime, timedelta from lib.api.trakt.trakt import TraktAPI, TraktLists, TraktMovies, TraktTV from lib.api.trakt.trakt_utils import ( add_trakt_collection_context_menu, add_trakt_custom_list_context_menu, add_trakt_favorites_context_menu, add_trakt_watched_context_menu, add_trakt_watchlist_context_menu, is_trakt_auth, ) from lib.clients.tmdb.utils.utils import build_play_trailer_context_menu_item, tmdb_get from lib.clients.trakt.utils import add_kodi_dir_item, extract_ids from lib.utils.general.utils import ( Anime, Enum, add_next_button, execute_thread_pool, set_content_type, set_media_infoTag, truncate_text, ) from lib.utils.kodi.utils import ( action_url_run, build_url, container_update, dialog_select, dialog_text, dialogyesno, end_of_directory, execute_builtin, get_datetime, get_setting, kodi_play_media, kodilog, make_list_item, notification, refresh, show_keyboard, translation, ) from .paginator import paginator_db def _normalize_user_slug(value): if value in (None, "", "None"): return "" return str(value) class Trakt(Enum): TRENDING = "trakt_trending" TRENDING_RECENT = "trakt_trending_recent" TOP10 = "trakt_top10" WATCHED = "trakt_watched" FAVORITED = "trakt_favorited" RECOMENDATIONS = "trakt_recommendations" TRENDING_LISTS = "trakt_trending_lists" POPULAR_LISTS = "trakt_popular_lists" WATCHED_HISTORY = "trakt_watched_history" WATCHLIST = "trakt_watchlist" CALENDAR = "trakt_calendar" UP_NEXT = "trakt_up_next" MOVIES_PROGRESS = "trakt_movies_progress" COLLECTION = "trakt_collection" FAVORITES = "trakt_favorites" MY_LISTS = "trakt_my_lists" LIKED_LISTS = "trakt_liked_lists" SEARCH_LISTS = "trakt_search_lists" ACCOUNT_INFO = "trakt_account_info" CREATE_LIST = "trakt_create_list" class BaseTraktClient: @staticmethod def _exclusive_context_flags(remove_action): flags = { "watchlist_add": False, "watchlist_remove": False, "watched_add": False, "watched_remove": False, "collection_add": False, "collection_remove": False, "favorites_add": False, "favorites_remove": False, "custom_list_add": False, "custom_list_remove": False, } if remove_action == "watchlist": flags["watchlist_remove"] = True elif remove_action == "watched": flags["watched_remove"] = True elif remove_action == "collection": flags["collection_remove"] = True elif remove_action == "favorites": flags["favorites_remove"] = True elif remove_action == "custom_list": flags["custom_list_remove"] = True return flags @staticmethod def _trakt_context_menu(mode, ids, context_flags=None): context_flags = context_flags or {} media_type = "movies" if mode == "movies" else "shows" return ( add_trakt_watchlist_context_menu( media_type, ids, include_add=context_flags.get("watchlist_add", True), include_remove=context_flags.get("watchlist_remove", True), ) + add_trakt_watched_context_menu( media_type, ids=ids, include_add=context_flags.get("watched_add", True), include_remove=context_flags.get("watched_remove", True), ) + add_trakt_collection_context_menu( media_type, ids, include_add=context_flags.get("collection_add", True), include_remove=context_flags.get("collection_remove", True), ) + add_trakt_favorites_context_menu( media_type, ids, include_add=context_flags.get("favorites_add", True), include_remove=context_flags.get("favorites_remove", True), ) + add_trakt_custom_list_context_menu( media_type, ids, include_add=context_flags.get("custom_list_add", True), include_remove=context_flags.get("custom_list_remove", True), ) ) @staticmethod def _add_media_directory_item(list_item, mode, title, ids, media_type=None, context_flags=None): trailer_item = build_play_trailer_context_menu_item( ids=ids, media_type="movie" if mode == "movies" else "tv", title=title, title_key="title", ) if mode == "movies": context_menu = [ ( translation(90049), kodi_play_media( name="search", mode=mode, query=title, ids=ids, rescrape=True, ), ), ] if trailer_item: context_menu.append(trailer_item) context_menu += ( BaseTraktClient._trakt_context_menu("movies", ids, context_flags=context_flags) if is_trakt_auth() else [] ) list_item.addContextMenuItems(context_menu) add_kodi_dir_item( list_item=list_item, url=build_url( "search", mode=mode, query=title, ids=ids, ), is_folder=False, is_playable=True, ) else: context_menu = [] if trailer_item: context_menu.append(trailer_item) if is_trakt_auth(): context_menu += BaseTraktClient._trakt_context_menu( "tv", ids, context_flags=context_flags ) if context_menu: list_item.addContextMenuItems(context_menu) add_kodi_dir_item( list_item=list_item, url=build_url( "show_seasons_details", ids=ids, mode=mode, media_type=media_type, ), is_folder=True, ) class TraktClient: @staticmethod def _refresh_after_list_action(params): if params.get("query") == Trakt.SEARCH_LISTS and params.get("search_term"): execute_builtin( container_update( "search_item", query=params.get("query"), category=params.get("category"), mode=params.get("mode"), submode=params.get("submode", ""), api=params.get("api", "trakt"), page=params.get("page", 1), search_term=params.get("search_term"), ) ) return refresh() @staticmethod def _trakt_sync_result_count(result, key, media_type): if not isinstance(result, dict): return 0 section = result.get(key, {}) if not isinstance(section, dict): return 0 media_keys = [] if media_type in ("movie", "movies"): media_keys = ["movies", "movie"] else: media_keys = ["shows", "show"] for media_key in media_keys: value = section.get(media_key) if isinstance(value, int): return value return 0 @staticmethod def _normalize_trakt_media_type(media_type, ids): if media_type in ("tv", "show", "shows", "tvshow", "tvshows", "anime"): return "shows" if media_type in ("movie", "movies") and (ids.get("tvdb") or ids.get("tvdb_id")): return "shows" if ids.get("tvdb") or ids.get("tvdb_id"): return "shows" tmdb_id = ids.get("tmdb") or ids.get("tmdb_id") if tmdb_id: tv_details = tmdb_get("tv_details", tmdb_id) tv_name = getattr(tv_details, "name", None) if tv_name is None and isinstance(tv_details, dict): tv_name = tv_details.get("name") if tv_name: return "shows" movie_details = tmdb_get("movie_details", tmdb_id) movie_title = getattr(movie_details, "title", None) if movie_title is None and isinstance(movie_details, dict): movie_title = movie_details.get("title") if movie_title: return "movies" imdb_id = ids.get("imdb") or ids.get("imdb_id") if imdb_id: found = tmdb_get("find_by_imdb_id", imdb_id) if getattr(found, "tv_results", []): return "shows" if getattr(found, "movie_results", []): return "movies" return "movies" @staticmethod def _build_show_collection_payload(ids): resolved_ids = TraktPresentation._resolve_media_ids("tv", ids) tmdb_id = resolved_ids.get("tmdb_id") show_details = tmdb_get("tv_details", tmdb_id) if tmdb_id else None trakt_object = ( TraktAPI().lists.get_trakt_object_by_tmdb(tmdb_id, media_type="show") if tmdb_id else None ) normalized_ids = {} if resolved_ids.get("tmdb_id"): normalized_ids["tmdb"] = int(resolved_ids["tmdb_id"]) trakt_id = trakt_object.get("ids", {}).get("trakt") if trakt_object else None if trakt_id: normalized_ids["trakt"] = trakt_id if resolved_ids.get("tvdb_id"): normalized_ids["tvdb"] = int(resolved_ids["tvdb_id"]) if resolved_ids.get("imdb_id"): normalized_ids["imdb"] = resolved_ids["imdb_id"] slug = trakt_object.get("ids", {}).get("slug") if trakt_object else None if slug: normalized_ids["slug"] = slug if not normalized_ids: return None show_item = {"ids": normalized_ids} title = trakt_object.get("title") if trakt_object else None if title is None: title = getattr(show_details, "name", None) if title is None and isinstance(show_details, dict): title = show_details.get("name") if title: show_item["title"] = title first_air_date = trakt_object.get("year") if trakt_object else None if isinstance(first_air_date, int): show_item["year"] = first_air_date return {"shows": [show_item]} first_air_date = getattr(show_details, "first_air_date", None) if first_air_date is None and isinstance(show_details, dict): first_air_date = show_details.get("first_air_date") if first_air_date: with contextlib.suppress(TypeError, ValueError): show_item["year"] = int(str(first_air_date)[:4]) return {"shows": [show_item]} @staticmethod def _build_show_collection_episode_payload(ids): resolved_ids = TraktPresentation._resolve_media_ids("tv", ids) tmdb_id = resolved_ids.get("tmdb_id") if not tmdb_id: return None show_details = tmdb_get("tv_details", tmdb_id) total_seasons = getattr(show_details, "number_of_seasons", 0) or show_details.get( "number_of_seasons", 0 ) today = datetime.utcnow().date() seasons_payload = [] for season_number in range(1, int(total_seasons) + 1): season_details = tmdb_get("season_details", {"id": tmdb_id, "season": season_number}) episodes = getattr(season_details, "episodes", None) if episodes is None and isinstance(season_details, dict): episodes = season_details.get("episodes", []) if not episodes: continue episode_entries = [] for episode in episodes: air_date = getattr(episode, "air_date", None) if air_date is None and isinstance(episode, dict): air_date = episode.get("air_date") if air_date: try: if datetime.strptime(air_date, "%Y-%m-%d").date() > today: continue except ValueError: pass episode_number = getattr(episode, "episode_number", None) if episode_number is None and isinstance(episode, dict): episode_number = episode.get("episode_number") if not episode_number: continue episode_entries.append({"number": int(episode_number)}) if episode_entries: seasons_payload.append({"number": int(season_number), "episodes": episode_entries}) if not seasons_payload: return None trakt_ids = {k: v for k, v in resolved_ids.items() if v} trakt_object = TraktAPI().lists.get_trakt_object_by_tmdb(tmdb_id, media_type="show") normalized_ids = {} if trakt_ids.get("tmdb_id"): normalized_ids["tmdb"] = int(trakt_ids["tmdb_id"]) trakt_id = trakt_object.get("ids", {}).get("trakt") if trakt_object else None if trakt_id: normalized_ids["trakt"] = trakt_id if trakt_ids.get("tvdb_id"): normalized_ids["tvdb"] = int(trakt_ids["tvdb_id"]) if trakt_ids.get("imdb_id"): normalized_ids["imdb"] = trakt_ids["imdb_id"] slug = trakt_object.get("ids", {}).get("slug") if trakt_object else None if slug: normalized_ids["slug"] = slug show_item = {"ids": normalized_ids, "seasons": seasons_payload} if trakt_object and trakt_object.get("title"): show_item["title"] = trakt_object.get("title") if trakt_object and trakt_object.get("year"): show_item["year"] = trakt_object.get("year") return {"shows": [show_item]} @staticmethod def _build_show_collection_season_payload(ids): episode_payload = TraktClient._build_show_collection_episode_payload(ids) if not episode_payload: return None show_item = episode_payload["shows"][0] seasons = show_item.get("seasons", []) if not seasons: return None season_payload = { "shows": [ { "ids": show_item.get("ids", {}), "seasons": [ {"number": season.get("number")} for season in seasons if season.get("number") ], } ] } if show_item.get("title"): season_payload["shows"][0]["title"] = show_item.get("title") if show_item.get("year"): season_payload["shows"][0]["year"] = show_item.get("year") return season_payload @staticmethod def _result_has_collection_change(result, action): if action == "add": relevant_keys = ("added", "updated", "existing") else: relevant_keys = ("deleted",) for key in relevant_keys: section = result.get(key, {}) if isinstance(result, dict) else {} if not isinstance(section, dict): continue if any(int(section.get(count_key, 0) or 0) > 0 for count_key in ("movies", "episodes")): return True return False @staticmethod def handle_trakt_query(query, category, mode, page, submode, api, params=None): set_content_type(mode) params = params or {} handlers = { "movies": lambda q, m, p: TraktClient.handle_trakt_movie_query(q, m, p, params), "tv": lambda q, m, p: TraktClient.handle_trakt_show_query(q, m, p, params), "anime": lambda q, m, p: TraktClient.handle_trakt_anime_query(category, p, submode), } handler = handlers.get(mode) if handler: return handler(query, mode, page) @staticmethod def handle_trakt_movie_query(query, mode, page, params=None): params = params or {} query_handlers = { Trakt.TRENDING: lambda: TraktMovies().trakt_movies_trending(page), Trakt.TOP10: lambda: TraktMovies().trakt_movies_top10_boxoffice(), Trakt.WATCHED: lambda: TraktMovies().trakt_movies_most_watched(page), Trakt.WATCHED_HISTORY: lambda: TraktLists().trakt_watched_history(mode, page), Trakt.FAVORITED: lambda: TraktMovies().trakt_movies_most_favorited(page), Trakt.RECOMENDATIONS: lambda: TraktMovies().trakt_recommendations("movies"), Trakt.FAVORITES: lambda: TraktLists().trakt_favorites(mode), Trakt.MY_LISTS: lambda: TraktLists().trakt_get_lists("my_lists"), Trakt.LIKED_LISTS: lambda: TraktLists().trakt_get_lists("liked_lists"), Trakt.SEARCH_LISTS: lambda: TraktClient._search_trakt_lists(params, page), Trakt.ACCOUNT_INFO: TraktClient.get_account_info, } list_handlers = { Trakt.TRENDING_LISTS: lambda: TraktLists().trakt_trending_popular_lists( list_type="trending", page_no=page ), Trakt.POPULAR_LISTS: lambda: TraktLists().trakt_trending_popular_lists( list_type="popular", page_no=page ), Trakt.WATCHLIST: lambda: TraktLists().trakt_watchlist(mode), Trakt.COLLECTION: lambda: TraktMovies().trakt_collection(page), Trakt.MOVIES_PROGRESS: lambda: TraktMovies().trakt_movies_progress(), } if query in query_handlers: return query_handlers[query]() elif query in list_handlers: return list_handlers[query]() @staticmethod def handle_trakt_show_query(query, mode, page, params=None): params = params or {} query_handlers = { Trakt.TRENDING: lambda: TraktTV().trakt_tv_trending(page), Trakt.WATCHED: lambda: TraktTV().trakt_tv_most_watched(page), Trakt.WATCHED_HISTORY: lambda: TraktLists().trakt_watched_history(mode, page), Trakt.FAVORITED: lambda: TraktTV().trakt_tv_most_favorited(page), Trakt.RECOMENDATIONS: lambda: TraktTV().trakt_recommendations("shows"), Trakt.FAVORITES: lambda: TraktLists().trakt_favorites(mode), Trakt.MY_LISTS: lambda: TraktLists().trakt_get_lists("my_lists"), Trakt.LIKED_LISTS: lambda: TraktLists().trakt_get_lists("liked_lists"), Trakt.SEARCH_LISTS: lambda: TraktClient._search_trakt_lists(params, page), Trakt.ACCOUNT_INFO: TraktClient.get_account_info, } list_handlers = { Trakt.TRENDING_LISTS: lambda: TraktLists().trakt_trending_popular_lists( list_type="trending", page_no=page ), Trakt.POPULAR_LISTS: lambda: TraktLists().trakt_trending_popular_lists( list_type="popular", page_no=page ), Trakt.WATCHLIST: lambda: TraktLists().trakt_watchlist(mode), Trakt.COLLECTION: lambda: TraktTV().trakt_collection(page), Trakt.CALENDAR: lambda: TraktClient.handle_calendar_request(), Trakt.UP_NEXT: lambda: TraktTV().trakt_up_next(), } if query in query_handlers: return query_handlers[query]() elif query in list_handlers: return list_handlers[query]() @staticmethod def handle_trakt_anime_query(query, page, submode="tv"): if query == Anime.TRENDING: return TraktAPI().anime.trakt_anime_trending(page, submode) elif query == Anime.TRENDING_RECENT: return TraktAPI().anime.trakt_anime_trending_recent(page, submode) elif query == Anime.MOST_WATCHED: return TraktAPI().anime.trakt_anime_most_watched(page, submode) elif query == Anime.FAVORITED: return TraktAPI().anime.trakt_anime_most_favorited(page, submode) @staticmethod def _search_trakt_lists(params, page): search_term = params.get("search_term", "") if not search_term: search_term = show_keyboard(id=30243, default="") if not search_term: params["_trakt_search_cancelled"] = True return [] params.pop("_trakt_search_cancelled", None) params["search_term"] = search_term return TraktLists().trakt_search_lists(search_term, page) @staticmethod def get_account_info(): auth_api = TraktAPI().auth settings = auth_api.get_user_settings() or {} stats = auth_api.get_account_stats() or {} return {"settings": settings, "stats": stats} @staticmethod def trakt_add_to_watchlist(params): ids = json.loads(params.get("ids", "{}")) media_type = TraktClient._normalize_trakt_media_type(params.get("media_type"), ids) try: TraktAPI().lists.add_to_watchlist(media_type, ids) notification(translation(90434), time=3000) refresh() except Exception as e: kodilog(f"Error adding to Trakt watchlist: {e}") notification(translation(90435), time=3000) @staticmethod def trakt_remove_from_watchlist(params): ids = json.loads(params.get("ids", "{}")) media_type = TraktClient._normalize_trakt_media_type(params.get("media_type"), ids) try: TraktAPI().lists.remove_from_watchlist(media_type, ids) notification(translation(90436), time=3000) refresh() except Exception as e: kodilog(f"Error removing from Trakt watchlist: {e}") notification(translation(90437), time=3000) @staticmethod def trakt_mark_as_watched(params): media_type = params.get("media_type") season = json.loads(params.get("season")) episode = json.loads(params.get("episode")) ids = json.loads(params.get("ids", "{}")) try: TraktAPI().lists.mark_as_watched(media_type, season, episode, ids) except Exception as e: kodilog(f"Error marking as watched on Trakt: {e}") notification(translation(90438), time=3000) @staticmethod def trakt_mark_as_unwatched(params): media_type = params.get("media_type") season = json.loads(params.get("season")) episode = json.loads(params.get("episode")) ids = json.loads(params.get("ids", "{}")) try: TraktAPI().lists.mark_as_unwatched(media_type, season, episode, ids) notification(translation(90439), time=3000) except Exception as e: kodilog(f"Error marking as unwatched on Trakt: {e}") notification(translation(90440), time=3000) @staticmethod def trakt_add_to_collection(params): ids = json.loads(params.get("ids", "{}")) media_type = TraktClient._normalize_trakt_media_type(params.get("media_type"), ids) try: payload = None if media_type not in ("movie", "movies"): payload = TraktClient._build_show_collection_payload(ids) if payload is None: notification(translation(90465), time=3500) return result = TraktAPI().lists.add_to_collection(media_type, ids, payload=payload) if media_type not in ( "movie", "movies", ) and not TraktClient._result_has_collection_change(result, "add"): payload = TraktClient._build_show_collection_season_payload(ids) if payload is not None: result = TraktAPI().lists.add_to_collection(media_type, ids, payload=payload) if media_type not in ( "movie", "movies", ) and not TraktClient._result_has_collection_change(result, "add"): payload = TraktClient._build_show_collection_episode_payload(ids) if payload is not None: result = TraktAPI().lists.add_to_collection(media_type, ids, payload=payload) added_count = TraktClient._trakt_sync_result_count(result, "added", media_type) existing_count = TraktClient._trakt_sync_result_count(result, "existing", media_type) not_found_count = TraktClient._trakt_sync_result_count(result, "not_found", media_type) if added_count > 0 or existing_count > 0: notification(translation(90441), time=3000) refresh() elif not_found_count > 0: notification(translation(90463), time=3500) else: notification(translation(90464), time=3500) except Exception as e: kodilog(f"Error adding to Trakt collection: {e}") notification(translation(90443), time=3000) @staticmethod def trakt_remove_from_collection(params): ids = json.loads(params.get("ids", "{}")) media_type = TraktClient._normalize_trakt_media_type(params.get("media_type"), ids) try: payload = None if media_type not in ("movie", "movies"): payload = TraktAPI().lists.build_collection_remove_payload(media_type, ids) if payload is None: payload = TraktClient._build_show_collection_payload(ids) if payload is None: notification(translation(90465), time=3500) return result = TraktAPI().lists.remove_from_collection(media_type, ids, payload=payload) if media_type not in ( "movie", "movies", ) and not TraktClient._result_has_collection_change(result, "remove"): payload = TraktClient._build_show_collection_season_payload(ids) if payload is not None: result = TraktAPI().lists.remove_from_collection( media_type, ids, payload=payload ) if media_type not in ( "movie", "movies", ) and not TraktClient._result_has_collection_change(result, "remove"): payload = TraktClient._build_show_collection_episode_payload(ids) if payload is not None: result = TraktAPI().lists.remove_from_collection( media_type, ids, payload=payload ) removed_count = TraktClient._trakt_sync_result_count(result, "deleted", media_type) not_found_count = TraktClient._trakt_sync_result_count(result, "not_found", media_type) if removed_count > 0: notification(translation(90442), time=3000) refresh() elif not_found_count > 0: notification(translation(90462), time=3500) else: notification(translation(90464), time=3500) except Exception as e: kodilog(f"Error removing from Trakt collection: {e}") notification(translation(90444), time=3000) @staticmethod def trakt_add_to_favorites(params): ids = json.loads(params.get("ids", "{}")) media_type = TraktClient._normalize_trakt_media_type(params.get("media_type"), ids) try: TraktAPI().lists.add_to_favorites(media_type, ids) notification(translation(90445), time=3000) refresh() except Exception as e: kodilog(f"Error adding to Trakt favorites: {e}") notification(translation(90447), time=3000) @staticmethod def trakt_remove_from_favorites(params): ids = json.loads(params.get("ids", "{}")) media_type = TraktClient._normalize_trakt_media_type(params.get("media_type"), ids) try: TraktAPI().lists.remove_from_favorites(media_type, ids) notification(translation(90446), time=3000) refresh() except Exception as e: kodilog(f"Error removing from Trakt favorites: {e}") notification(translation(90448), time=3000) @staticmethod def trakt_create_list(params): name = show_keyboard(id=30924, default="") if not name: return description = show_keyboard(id=30925, default="") or "" try: TraktAPI().lists.create_list(name=name, description=description) notification(translation(90449), time=3000) refresh() except Exception as e: kodilog(f"Error creating Trakt list: {e}") notification(translation(90450), time=3000) @staticmethod def trakt_delete_list(params): trakt_id = params.get("trakt_id") if not trakt_id: return if not dialogyesno(translation(90645), translation(90466)): return try: TraktAPI().lists.delete_list(trakt_id) notification(translation(90451), time=3000) refresh() except Exception as e: kodilog(f"Error deleting Trakt list: {e}") notification(translation(90452), time=3000) @staticmethod def trakt_like_list(params): trakt_id = params.get("trakt_id") user_slug = _normalize_user_slug(params.get("user")) if not trakt_id: return try: TraktAPI().lists.like_list(user_slug, trakt_id) notification(translation(90453), time=3000) TraktClient._refresh_after_list_action(params) except Exception as e: kodilog(f"Error liking Trakt list: {e}") notification(translation(90454), time=3000) @staticmethod def trakt_unlike_list(params): trakt_id = params.get("trakt_id") user_slug = _normalize_user_slug(params.get("user")) if not trakt_id: return try: TraktAPI().lists.unlike_list(user_slug, trakt_id) notification(translation(90455), time=3000) TraktClient._refresh_after_list_action(params) except Exception as e: kodilog(f"Error unliking Trakt list: {e}") notification(translation(90456), time=3000) @staticmethod def _select_trakt_list(): my_lists = TraktAPI().lists.trakt_get_lists("my_lists") or [] if not my_lists: notification(translation(90461), time=3000) return None options = [item.get("name", translation(90535)) for item in my_lists] choice = dialog_select(translation(30927), options) if choice < 0: return None return my_lists[choice] @staticmethod def trakt_add_item_to_list(params): media_type = params.get("media_type") ids = json.loads(params.get("ids", "{}")) selected_list = TraktClient._select_trakt_list() if not selected_list: return try: TraktAPI().lists.add_item_to_list(selected_list.get("trakt_id"), media_type, ids) notification(translation(90457), time=3000) except Exception as e: kodilog(f"Error adding item to Trakt list: {e}") notification(translation(90459), time=3000) @staticmethod def trakt_remove_item_from_list(params): media_type = params.get("media_type") ids = json.loads(params.get("ids", "{}")) selected_list = TraktClient._select_trakt_list() if not selected_list: return try: TraktAPI().lists.remove_item_from_list(selected_list.get("trakt_id"), media_type, ids) notification(translation(90536), time=3000) except Exception as e: kodilog(f"Error removing item from Trakt list: {e}") notification(translation(90537), time=3000) @staticmethod def handle_calendar_request(): previous_days = int(get_setting("trakt_calendar_previous_days", 0)) future_days = int(get_setting("trakt_calendar_future_days", 14)) # Calculate start date start_date_obj = get_datetime(string=False) - timedelta(days=previous_days) start_date = start_date_obj.strftime("%Y-%m-%d") # Calculate total days (past + future + today) days = previous_days + future_days + 1 # Log user identity logic try: user_settings = TraktAPI().auth.get_user_settings() if user_settings and "user" in user_settings: username = user_settings.get("user", {}).get("username") kodilog(f"Trakt Calendar Request for user: {username}") except Exception as e: kodilog(f"Error fetching Trakt user profile: {e}") if get_setting("trakt_calendar_show_all"): return TraktAPI().calendar.trakt_all_shows_calendar(start_date=start_date, days=days) return TraktAPI().calendar.trakt_my_calendar(start_date=start_date, days=days) @staticmethod def process_trakt_result(results, query, category, mode, submode, api, page, search_term=""): if query == Trakt.ACCOUNT_INFO: TraktPresentation.show_account_info(results) end_of_directory(cache=False) return if not results: if query == Trakt.SEARCH_LISTS and not search_term: end_of_directory(cache=False) return if query in (Trakt.CALENDAR, Trakt.UP_NEXT, Trakt.SEARCH_LISTS): notification("No results found", time=3000) query_handlers = { Trakt.TRENDING: lambda: execute_thread_pool( results, TraktPresentation.show_common_categories, mode ), Trakt.WATCHED: lambda: execute_thread_pool( results, TraktPresentation.show_common_categories, mode ), Trakt.FAVORITED: lambda: execute_thread_pool( results, TraktPresentation.show_common_categories, mode ), Trakt.TOP10: lambda: execute_thread_pool( results, TraktPresentation.show_common_categories, mode ), Trakt.RECOMENDATIONS: lambda: execute_thread_pool( results, TraktPresentation.show_recommendations, mode ), Trakt.TRENDING_LISTS: lambda: execute_thread_pool( results, TraktPresentation.show_trending_lists, mode ), Trakt.POPULAR_LISTS: lambda: execute_thread_pool( results, TraktPresentation.show_trending_lists, mode ), Trakt.WATCHLIST: lambda: execute_thread_pool( results, TraktPresentation.show_watchlist, mode ), Trakt.WATCHED_HISTORY: lambda: execute_thread_pool( results, TraktPresentation.show_watched_history ), Trakt.COLLECTION: lambda: execute_thread_pool( results, TraktPresentation.show_collection, mode ), Trakt.CALENDAR: lambda: execute_thread_pool( results, TraktPresentation.show_calendar_items ), Trakt.UP_NEXT: lambda: execute_thread_pool( results, TraktPresentation.show_up_next_items ), Trakt.MOVIES_PROGRESS: lambda: execute_thread_pool( results, TraktPresentation.show_movies_progress_items ), Trakt.FAVORITES: lambda: execute_thread_pool( results, TraktPresentation.show_favorites, mode ), Trakt.MY_LISTS: lambda: execute_thread_pool( results, TraktPresentation.show_user_lists, mode, query=query, page=page, search_term=search_term, api=api, category=category, submode=submode, ), Trakt.LIKED_LISTS: lambda: execute_thread_pool( results, TraktPresentation.show_user_lists, mode, query=query, page=page, search_term=search_term, api=api, category=category, submode=submode, ), Trakt.SEARCH_LISTS: lambda: execute_thread_pool( results, TraktPresentation.show_user_lists, mode, query=query, page=page, search_term=search_term, api=api, category=category, submode=submode, ), } anime_handlers = { Anime.TRENDING: lambda: execute_thread_pool( results, TraktPresentation.show_anime_common, submode ), Anime.TRENDING_RECENT: lambda: execute_thread_pool( results, TraktPresentation.show_anime_common, submode ), Anime.MOST_WATCHED: lambda: execute_thread_pool( results, TraktPresentation.show_anime_common, submode ), Anime.FAVORITED: lambda: execute_thread_pool( results, TraktPresentation.show_anime_common, submode ), } if query in query_handlers: query_handlers[query]() if query == Trakt.MY_LISTS: TraktPresentation.show_create_list_entry(mode) if category in anime_handlers: anime_handlers[category]() if TraktClient._should_add_next_button(query, category): next_kwargs = { "query": query, "category": category, "mode": mode, "submode": submode, "api": api, } if search_term: next_kwargs["search_term"] = search_term add_next_button("search_item", page=page, **next_kwargs) end_of_directory(cache=TraktClient._should_cache_directory(query)) @staticmethod def _should_add_next_button(query, category): paginated_queries = { Trakt.TRENDING, Trakt.WATCHED, Trakt.FAVORITED, Trakt.TRENDING_LISTS, Trakt.POPULAR_LISTS, Trakt.WATCHED_HISTORY, Trakt.COLLECTION, Trakt.SEARCH_LISTS, } paginated_categories = { Anime.TRENDING, Anime.TRENDING_RECENT, Anime.MOST_WATCHED, Anime.FAVORITED, } return query in paginated_queries or category in paginated_categories @staticmethod def _should_cache_directory(query): non_cached_queries = { Trakt.WATCHLIST, Trakt.WATCHED_HISTORY, Trakt.CALENDAR, Trakt.UP_NEXT, Trakt.COLLECTION, Trakt.FAVORITES, Trakt.MY_LISTS, Trakt.LIKED_LISTS, Trakt.ACCOUNT_INFO, } return query not in non_cached_queries @staticmethod def show_trakt_list_content(list_type, mode, user, slug, with_auth, page, trakt_id=None): data = TraktAPI().lists.get_trakt_list_contents(list_type, user, slug, with_auth, trakt_id) if not data: notification("No results found", time=3000) end_of_directory(cache=not bool(with_auth)) return paginator_db.initialize(data) items = paginator_db.get_page(page) execute_thread_pool(items, TraktPresentation.show_lists_content_items) add_next_button("list_trakt_page", page, mode=mode) end_of_directory(cache=not bool(with_auth)) @staticmethod def show_list_trakt_page(page, mode): items = paginator_db.get_page(page) execute_thread_pool(items, TraktPresentation.show_lists_content_items) add_next_button("list_trakt_page", page, mode=mode) end_of_directory() class TraktPresentation: @staticmethod def _format_bool_label(value): return "Yes" if value else "No" @staticmethod def _format_iso_date(value): if not value: return "" if "T" in str(value): return str(value).split("T")[0] return str(value) @staticmethod def _resolve_media_ids(mode, media_ids): tmdb_id = media_ids.get("tmdb") or media_ids.get("tmdb_id") imdb_id = media_ids.get("imdb") or media_ids.get("imdb_id", "") tvdb_id = media_ids.get("tvdb") or media_ids.get("tvdb_id", "") if mode == "tv": if not tmdb_id and imdb_id: found = tmdb_get("find_by_imdb_id", imdb_id) if getattr(found, "tv_results", []): tmdb_id = str(found.tv_results[0]["id"]) if not tmdb_id and tvdb_id: found = tmdb_get("find_by_tvdb", tvdb_id) if getattr(found, "tv_results", []): tmdb_id = str(found.tv_results[0]["id"]) else: if not tmdb_id and imdb_id: found = tmdb_get("find_by_imdb_id", imdb_id) if getattr(found, "movie_results", []): tmdb_id = str(found.movie_results[0]["id"]) return {"tmdb_id": tmdb_id, "tvdb_id": tvdb_id, "imdb_id": imdb_id} @staticmethod def show_create_list_entry(mode): list_item = make_list_item(label=f"[B]+ {translation(30926)}[/B]") add_kodi_dir_item( list_item=list_item, url=build_url("trakt_create_list", mode=mode), is_folder=False, ) @staticmethod def _format_account_info(results): settings = results.get("settings", {}) if isinstance(results, dict) else {} stats = results.get("stats", {}) if isinstance(results, dict) else {} user = settings.get("user", {}) account = settings.get("account", {}) connections = settings.get("connections", {}) lines = [str(translation(30916)), ""] lines.append(f"Username: {user.get('username', '')}") lines.append(f"Name: {user.get('name', '')}") lines.append(f"Joined: {TraktPresentation._format_iso_date(user.get('joined_at', ''))}") lines.append( f"Private: {TraktPresentation._format_bool_label(account.get('private', False))}" ) lines.append(f"VIP: {TraktPresentation._format_bool_label(user.get('vip', False))}") lines.append(f"Timezone: {account.get('timezone', '')}") lines.append(f"Locale: {account.get('locale', '')}") lines.append("") lines.append("Stats") lines.append(f"Movies Watched: {stats.get('movies', {}).get('watched', 0)}") lines.append(f"Shows Watched: {stats.get('shows', {}).get('watched', 0)}") lines.append(f"Episodes Watched: {stats.get('episodes', {}).get('watched', 0)}") lines.append(f"Movies Collected: {stats.get('movies', {}).get('collected', 0)}") lines.append(f"Shows Collected: {stats.get('shows', {}).get('collected', 0)}") lines.append(f"Lists: {stats.get('lists', {}).get('count', 0)}") lines.append(f"Ratings: {stats.get('ratings', {}).get('total', 0)}") lines.append("") lines.append("Connections") for key in ("facebook", "google", "twitter", "mastodon"): connection = connections.get(key, False) lines.append(f"{key.title()}: {TraktPresentation._format_bool_label(bool(connection))}") return "\n".join(lines) @staticmethod def show_account_info(results): content = TraktPresentation._format_account_info(results) dialog_text(translation(30916), content) @staticmethod def show_anime_common(res, mode): ids = extract_ids(res, mode) # Handle both show and movie responses if mode == "movies" and "movie" in res: title = res["movie"]["title"] else: title = res["show"]["title"] tmdb_id = ids["tmdb_id"] if mode == "tv": details = tmdb_get("tv_details", tmdb_id) else: details = tmdb_get("movie_details", tmdb_id) list_item = make_list_item(label=title) set_media_infoTag(list_item, data=details, mode=mode) BaseTraktClient._add_media_directory_item( list_item=list_item, mode=mode, title=title, ids=ids, media_type=res.get("media_type"), ) @staticmethod def show_common_categories(res, mode): if mode == "tv": title = res["show"]["title"] ids = extract_ids(res, mode) tmdb_id = ids["tmdb_id"] details = tmdb_get("tv_details", tmdb_id) else: title = res["movie"]["title"] ids = extract_ids(res, mode) tmdb_id = ids["tmdb_id"] details = tmdb_get("movie_details", tmdb_id) list_item = make_list_item(label=title) set_media_infoTag(list_item, data=details, mode=mode) BaseTraktClient._add_media_directory_item( list_item=list_item, mode=mode, title=title, ids=ids, media_type=res.get("media_type"), ) @staticmethod def show_watchlist(res, mode): title = res["title"] ids = TraktPresentation._resolve_media_ids(mode, res["media_ids"]) tmdb_id = ids["tmdb_id"] if not tmdb_id: return if mode == "tv": details = tmdb_get("tv_details", tmdb_id) else: details = tmdb_get("movie_details", tmdb_id) list_item = make_list_item(label=title) set_media_infoTag(list_item, data=details, mode=mode) BaseTraktClient._add_media_directory_item( list_item=list_item, mode=mode, title=title, ids=ids, media_type=res.get("media_type"), context_flags=BaseTraktClient._exclusive_context_flags("watchlist"), ) @staticmethod def show_favorites(res, mode): title = res["title"] ids = TraktPresentation._resolve_media_ids(mode, res["media_ids"]) tmdb_id = ids["tmdb_id"] if not tmdb_id: return if mode == "tv": details = tmdb_get("tv_details", tmdb_id) else: details = tmdb_get("movie_details", tmdb_id) list_item = make_list_item(label=title) set_media_infoTag(list_item, data=details, mode=mode) BaseTraktClient._add_media_directory_item( list_item=list_item, mode=mode, title=title, ids=ids, media_type=res.get("media_type"), context_flags=BaseTraktClient._exclusive_context_flags("favorites"), ) @staticmethod def show_collection(res, mode): if mode == "tv": title = res["show"]["title"] ids = TraktPresentation._resolve_media_ids(mode, res["show"].get("ids", {})) tmdb_id = ids["tmdb_id"] if not tmdb_id: return details = tmdb_get("tv_details", tmdb_id) else: title = res["movie"]["title"] ids = TraktPresentation._resolve_media_ids(mode, res["movie"].get("ids", {})) tmdb_id = ids["tmdb_id"] if not tmdb_id: return details = tmdb_get("movie_details", tmdb_id) list_item = make_list_item(label=title) set_media_infoTag(list_item, data=details, mode=mode) BaseTraktClient._add_media_directory_item( list_item=list_item, mode=mode, title=title, ids=ids, media_type=res.get("media_type"), context_flags=BaseTraktClient._exclusive_context_flags("collection"), ) @staticmethod def show_trending_lists(res, mode): list_title = res["list"]["name"] description = res["list"]["description"] user_data = res["list"].get("user", {}) user_ids = user_data.get("ids", {}) user_slug = _normalize_user_slug(user_ids.get("slug") or user_data.get("username")) url = build_url( "trakt_list_content", list_type=res["list"]["type"], mode=mode, user=user_slug, slug=res["list"]["ids"]["slug"], ) list_item = make_list_item(label=list_title) info_tag = list_item.getVideoInfoTag() info_tag.setTitle(list_title) info_tag.setPlot(truncate_text(description)) if user_slug: list_item.addContextMenuItems( [ ( "Like Trakt List", action_url_run( "trakt_like_list", trakt_id=res["list"]["ids"].get("trakt"), user=user_slug, ), ) ] ) add_kodi_dir_item( list_item, url=url, is_folder=True, ) @staticmethod def show_user_lists( res, mode, query=None, page=1, search_term="", api="trakt", category=None, submode="", ): list_title = res.get("name", "Untitled List") description = res.get("description", "") item_count = res.get("item_count") username = res.get("username") or res.get("user_slug", "") if item_count: label = f"{list_title} ({item_count})" else: label = list_title if username: label = f"{label} - [I]{username}[/I]" list_item = make_list_item(label=label) info_tag = list_item.getVideoInfoTag() info_tag.setTitle(list_title) info_tag.setPlot(truncate_text(description)) context_menu = [] if res.get("can_delete"): context_menu.append( ( "Delete Trakt List", action_url_run( "trakt_delete_list", trakt_id=res.get("trakt_id", ""), ), ) ) if res.get("can_unlike"): context_menu.append( ( "Unlike Trakt List", action_url_run( "trakt_unlike_list", trakt_id=res.get("trakt_id", ""), user=res.get("user_slug", ""), query=query, category=category, mode=mode, submode=submode, api=api, page=page, search_term=search_term, ), ) ) if res.get("can_like"): context_menu.append( ( "Like Trakt List", action_url_run( "trakt_like_list", trakt_id=res.get("trakt_id", ""), user=res.get("user_slug", ""), query=query, category=category, mode=mode, submode=submode, api=api, page=page, search_term=search_term, ), ) ) if context_menu: list_item.addContextMenuItems(context_menu) add_kodi_dir_item( list_item=list_item, url=build_url( "trakt_list_content", list_type=res.get("list_type", "user_lists"), mode=mode, user=res.get("user_slug", ""), slug=res.get("slug", ""), with_auth=res.get("with_auth", False), trakt_id=res.get("trakt_id", ""), ), is_folder=True, ) @staticmethod def show_recommendations(res, mode): title = res["title"] tmdb_id = res["ids"]["tmdb"] imdb_id = res["ids"]["imdb"] ids = {"tmdb_id": tmdb_id, "tvdb_id": "", "imdb_id": imdb_id} if mode == "tv": details = tmdb_get("tv_details", tmdb_id) else: details = tmdb_get("movie_details", tmdb_id) list_item = make_list_item(label=title) set_media_infoTag(list_item, data=details, mode=mode) BaseTraktClient._add_media_directory_item( list_item=list_item, mode=mode, title=title, ids=ids, media_type=res.get("media_type"), ) @staticmethod def show_watched_history(res): title = res["title"] if res["type"] == "show": mode = "tv" ids = TraktPresentation._resolve_media_ids(mode, res["media_ids"]) tmdb_id = ids["tmdb_id"] if not tmdb_id: return details = tmdb_get("tv_details", tmdb_id) url = build_url( "search", ids=ids, mode=mode, query=res["show_title"], tv_data={ "name": res["ep_title"], "episode": res["episode"], "season": res["season"], }, ) is_folder = True is_playable = False else: mode = "movies" ids = TraktPresentation._resolve_media_ids(mode, res["media_ids"]) tmdb_id = ids["tmdb_id"] if not tmdb_id: return details = tmdb_get("movie_details", tmdb_id) url = build_url( "search", query=title, mode=mode, ids=ids, ) is_folder = False is_playable = True list_item = make_list_item(label=title) set_media_infoTag(list_item, data=details, mode=mode) list_item.addContextMenuItems( BaseTraktClient._trakt_context_menu( mode, ids, context_flags=BaseTraktClient._exclusive_context_flags("watched"), ) ) add_kodi_dir_item( list_item, url=url, is_folder=is_folder, is_playable=is_playable, ) @staticmethod def show_lists_content_items(res): title = res["title"] if res["type"] == "show": mode = "tv" ids = TraktPresentation._resolve_media_ids(mode, res.get("media_ids", {})) tmdb_id = ids["tmdb_id"] if not tmdb_id: return details = tmdb_get("tv_details", tmdb_id) else: mode = "movies" ids = TraktPresentation._resolve_media_ids(mode, res.get("media_ids", {})) tmdb_id = ids["tmdb_id"] if not tmdb_id: return details = tmdb_get("movie_details", tmdb_id) list_item = make_list_item(label=title) set_media_infoTag(list_item, data=details, mode=mode) BaseTraktClient._add_media_directory_item( list_item=list_item, mode=mode, title=title, ids=ids, media_type=res.get("media_type"), context_flags=BaseTraktClient._exclusive_context_flags("custom_list"), ) @staticmethod def show_calendar_items(res): show_title = res.get("show", {}).get("title") ep_data = res.get("episode", {}) title = ep_data.get("title") season = ep_data.get("season") episode = ep_data.get("number") if not show_title or not title or season is None or episode is None: return first_aired = res.get("first_aired") or ep_data.get("first_aired") ids = TraktPresentation._resolve_media_ids("tv", res.get("show", {}).get("ids", {})) tmdb_id = ids["tmdb_id"] ids["imdb_id"] ids["tvdb_id"] if not tmdb_id: return display_title = f"{show_title} - S{int(season):02d}E{int(episode):02d} - {title}" if first_aired: date_part = first_aired.split("T")[0] display_title = f"{date_part} | {display_title}" details = tmdb_get("tv_details", tmdb_id) or {} list_item = make_list_item(label=display_title) set_media_infoTag(list_item, data=details, mode="tv") # Override title list_item.setLabel(display_title) info_tag = list_item.getVideoInfoTag() info_tag.setTitle(display_title) info_tag.setPlot(truncate_text(details.get("overview", ""))) url = build_url( "search", ids=ids, mode="tv", query=show_title, tv_data={ "name": title, "episode": episode, "season": season, }, ) add_kodi_dir_item( list_item=list_item, url=url, is_folder=True, ) @staticmethod def show_up_next_items(res): show_data = res.get("show", {}) episode = res.get("episode", {}) show_title = show_data.get("title", "") title = episode.get("title", "") season = episode.get("season") episode_number = episode.get("number") progress = res.get("progress", 0) tmdb_id = show_data.get("ids", {}).get("tmdb") imdb_id = show_data.get("ids", {}).get("imdb") tvdb_id = show_data.get("ids", {}).get("tvdb") if not tmdb_id or season is None or episode_number is None: return prefix = "[Resume]" if res.get("type") == "resume" else "[Next]" display_title = ( f"{prefix} {show_title} - S{int(season):02d}E{int(episode_number):02d} - {title}" ) if progress: display_title = f"{display_title} ({int(progress)}%)" ids = {"tmdb_id": tmdb_id, "imdb_id": imdb_id, "tvdb_id": tvdb_id} details = tmdb_get("tv_details", tmdb_id) or {} list_item = make_list_item(label=display_title) set_media_infoTag(list_item, data=details, mode="tv") info_tag = list_item.getVideoInfoTag() info_tag.setTitle(display_title) info_tag.setPlot(truncate_text(details.get("overview", ""))) if progress: list_item.setProperty("PercentPlayed", str(progress)) url = build_url( "search", ids=ids, mode="tv", query=show_title, tv_data={ "name": title, "episode": episode_number, "season": season, }, ) add_kodi_dir_item(list_item=list_item, url=url, is_folder=True) @staticmethod def show_movies_progress_items(res): movie_data = res.get("movie", {}) title = movie_data.get("title", "") progress = res.get("progress", 0) tmdb_id = movie_data.get("ids", {}).get("tmdb") imdb_id = movie_data.get("ids", {}).get("imdb") if not tmdb_id: return display_title = f"[Resume] {title}" if progress: display_title = f"{display_title} ({int(progress)}%)" ids = {"tmdb_id": tmdb_id, "imdb_id": imdb_id} details = tmdb_get("movie_details", tmdb_id) or {} list_item = make_list_item(label=display_title) set_media_infoTag(list_item, data=details, mode="movies") info_tag = list_item.getVideoInfoTag() info_tag.setTitle(display_title) info_tag.setPlot(truncate_text(details.get("overview", ""))) if progress: list_item.setProperty("PercentPlayed", str(progress)) url = build_url( "search", ids=ids, mode="movies", query=title, ) add_kodi_dir_item(list_item=list_item, url=url, is_folder=True)