from time import monotonic import xbmc from lib.api.trakt.lists_cache import lists_cache from lib.api.trakt.trakt import TraktAPI from lib.api.trakt.trakt_cache import ( clear_trakt_calendar, clear_trakt_favorites, clear_trakt_hidden_data, clear_trakt_list_contents_data, clear_trakt_list_data, clear_trakt_watchlist, reset_activity, trakt_watched_cache, ) from lib.utils.kodi.utils import get_property_no_fallback, get_setting, kodilog PAUSE_SERVICES_PROP = "jacktook.pause_services" DEFAULT_SYNC_INTERVAL_MINUTES = 15 WAIT_STEP_SECONDS = 5 class TraktSyncService: def __init__(self, api=None, monitor=None): self.api = api or TraktAPI() self.monitor = monitor or xbmc.Monitor() self._cycle_context = None def _format_buckets(self, buckets): if not buckets: return "none" return ",".join(sorted(buckets)) def _get_cycle_context(self, force=False): if self._cycle_context: return self._cycle_context return "startup" if force else "periodic" def _log_sync_summary( self, cycle_context, force, outcome, detected_buckets, applied_buckets, forced_defaults, started_at, ): duration_ms = int((monotonic() - started_at) * 1000) kodilog( f"Trakt sync[{cycle_context}]: completed outcome={outcome} force={force} forced_defaults={forced_defaults} detected_buckets={self._format_buckets(detected_buckets)} applied_buckets={self._format_buckets(applied_buckets)} duration_ms={duration_ms}" ) def _log_sync_failure(self, cycle_context, force, error): kodilog( f"Trakt sync[{cycle_context}]: failed force={force} error={error}", level=xbmc.LOGERROR, ) def run(self): if not self._is_trakt_available(): kodilog("Trakt sync[startup]: skipped because Trakt is disabled or unauthenticated") return kodilog( "Starting Trakt Sync service (interval=%s minutes)" % (self._get_sync_interval_seconds() // 60) ) try: self._cycle_context = "startup" kodilog("Trakt sync[startup]: beginning initial sync (force=True)") try: self.sync_activities(force=True) except Exception as e: self._log_sync_failure("startup", True, e) return while not self.monitor.abortRequested(): if self._wait_for_next_cycle(): return if not self._is_trakt_available(): kodilog("Trakt sync[periodic]: skipped cycle because Trakt is unavailable") continue if self._services_paused(): kodilog("Trakt sync[periodic]: skipped cycle because services are paused") continue self._cycle_context = "periodic" kodilog("Trakt sync[periodic]: beginning cycle") try: self.sync_activities() except Exception as e: self._log_sync_failure("periodic", False, e) return except Exception as e: kodilog(f"Error during Trakt Sync: {e}", level=xbmc.LOGERROR) finally: self._cycle_context = None def _is_trakt_available(self): return get_setting("trakt_enabled") and get_setting("is_trakt_auth") def _services_paused(self): return get_property_no_fallback(PAUSE_SERVICES_PROP) == "true" def _get_sync_interval_seconds(self): try: interval = int(get_setting("trakt_sync_interval", DEFAULT_SYNC_INTERVAL_MINUTES)) except (TypeError, ValueError): interval = DEFAULT_SYNC_INTERVAL_MINUTES return max(interval, 1) * 60 def _wait_for_next_cycle(self): remaining = self._get_sync_interval_seconds() while remaining > 0: if self.monitor.waitForAbort(min(WAIT_STEP_SECONDS, remaining)): return True remaining -= WAIT_STEP_SECONDS return False def sync_activities(self, force=False): cycle_context = self._get_cycle_context(force=force) started_at = monotonic() forced_defaults = False detected_buckets = set() applied_buckets = set() try: latest_activities = self.api.sync.get_last_activities() if not latest_activities: self._log_sync_summary( cycle_context, force, "no_data", detected_buckets, applied_buckets, forced_defaults, started_at, ) return set() previous_activities = reset_activity(latest_activities) detected_buckets = self._get_changed_buckets(previous_activities, latest_activities) applied_buckets = set(detected_buckets) if force and not applied_buckets: applied_buckets = { "watched_movies", "watched_shows", "progress_movies", "progress_episodes", } forced_defaults = True if not applied_buckets: self._log_sync_summary( cycle_context, force, "no_changes", detected_buckets, applied_buckets, forced_defaults, started_at, ) return applied_buckets kodilog("Trakt sync: changed buckets = {}".format(", ".join(sorted(applied_buckets)))) self._apply_activity_changes(applied_buckets) self._log_sync_summary( cycle_context, force, "applied_changes", detected_buckets, applied_buckets, forced_defaults, started_at, ) return applied_buckets except Exception as e: self._log_sync_failure(cycle_context, force, e) raise def _activity_changed(self, previous, latest, *path): prev_value = previous latest_value = latest for key in path: if not isinstance(prev_value, dict) or not isinstance(latest_value, dict): return False prev_value = prev_value.get(key) latest_value = latest_value.get(key) return bool(latest_value) and prev_value != latest_value def _get_changed_buckets(self, previous, latest): changes = set() if self._activity_changed(previous, latest, "movies", "watched_at"): changes.add("watched_movies") if self._activity_changed(previous, latest, "episodes", "watched_at"): changes.add("watched_shows") if self._activity_changed(previous, latest, "movies", "paused_at"): changes.add("progress_movies") if self._activity_changed(previous, latest, "episodes", "paused_at"): changes.add("progress_episodes") if self._activity_changed(previous, latest, "movies", "collected_at"): changes.add("collection_movies") if self._activity_changed(previous, latest, "episodes", "collected_at"): changes.add("collection_shows") if self._activity_changed(previous, latest, "movies", "watchlisted_at"): changes.add("watchlist_movies") if self._activity_changed(previous, latest, "shows", "watchlisted_at"): changes.add("watchlist_shows") if self._activity_changed(previous, latest, "movies", "favorited_at"): changes.add("favorites_movies") if self._activity_changed(previous, latest, "shows", "favorited_at"): changes.add("favorites_shows") if self._activity_changed(previous, latest, "movies", "recommendations_at"): changes.add("recommendations_movies") if self._activity_changed(previous, latest, "shows", "recommendations_at"): changes.add("recommendations_shows") if self._activity_changed(previous, latest, "lists", "liked_at") or self._activity_changed( previous, latest, "lists", "updated_at" ): changes.add("lists") if self._activity_changed(previous, latest, "movies", "hidden_at"): changes.add("hidden_movies") if self._activity_changed(previous, latest, "shows", "hidden_at"): changes.add("hidden_shows") return changes def _apply_activity_changes(self, changes): if "watched_movies" in changes: self.sync_watched_movies() if "watched_shows" in changes: self.sync_watched_shows() if "progress_movies" in changes: self.sync_movie_progress() if "progress_episodes" in changes: self.sync_episode_progress() self.invalidate_cached_buckets(changes) def sync_watched_movies(self): movies = self.api.movies.get_watched_movies() if not movies: trakt_watched_cache.set_bulk_movie_watched([]) return insert_list = [] for item in movies: tmdb_id = item["movie"]["ids"].get("tmdb") if not tmdb_id: continue insert_list.append( ( "movie", str(tmdb_id), None, None, item["last_watched_at"], item["movie"]["title"], ) ) trakt_watched_cache.set_bulk_movie_watched(insert_list) kodilog(f"Synced {len(insert_list)} watched movies from Trakt") def sync_watched_shows(self): shows = self.api.tv.get_watched_shows() if not shows: trakt_watched_cache.set_bulk_tvshow_watched([]) return insert_list = [] for item in shows: show_tmdb = item["show"]["ids"].get("tmdb") if not show_tmdb: continue show_title = item["show"]["title"] for season in item.get("seasons", []): season_num = season["number"] for episode in season.get("episodes", []): insert_list.append( ( "episode", str(show_tmdb), season_num, episode["number"], episode["last_watched_at"], show_title, ) ) trakt_watched_cache.set_bulk_tvshow_watched(insert_list) kodilog(f"Synced {len(insert_list)} watched episodes from Trakt") def sync_movie_progress(self): paused_movies = self.api.scrobble.trakt_get_playback_progress("movies") if not paused_movies: trakt_watched_cache.set_bulk_movie_progress([]) return insert_list = [] for item in paused_movies: tmdb_id = item["movie"]["ids"].get("tmdb") if not tmdb_id: continue insert_list.append( ( "movie", str(tmdb_id), None, None, str(item.get("progress", 0)), "0", item.get("paused_at"), item.get("id"), item["movie"]["title"], ) ) trakt_watched_cache.set_bulk_movie_progress(insert_list) kodilog(f"Synced {len(insert_list)} movie progress items from Trakt") def sync_episode_progress(self): paused_episodes = self.api.scrobble.trakt_get_playback_progress("episodes") if not paused_episodes: trakt_watched_cache.set_bulk_tvshow_progress([]) return insert_list = [] for item in paused_episodes: show_tmdb = item["show"]["ids"].get("tmdb") if not show_tmdb: continue insert_list.append( ( "episode", str(show_tmdb), item["episode"]["season"], item["episode"]["number"], str(item.get("progress", 0)), "0", item.get("paused_at"), item.get("id"), f"{item['show']['title']} - {item['episode']['title']}", ) ) trakt_watched_cache.set_bulk_tvshow_progress(insert_list) kodilog(f"Synced {len(insert_list)} episode progress items from Trakt") def invalidate_cached_buckets(self, changes): if "collection_movies" in changes: lists_cache.delete_prefix("trakt_movies_collection_") if "collection_shows" in changes: lists_cache.delete_prefix("trakt_tv_collection_") if "watchlist_movies" in changes or "watchlist_shows" in changes: clear_trakt_watchlist() clear_trakt_calendar() if "favorites_movies" in changes or "favorites_shows" in changes: clear_trakt_favorites() if "recommendations_movies" in changes: lists_cache.delete_prefix("trakt_recommendations_movies") if "recommendations_shows" in changes: lists_cache.delete_prefix("trakt_recommendations_shows") if "lists" in changes: clear_trakt_list_data("my_lists") clear_trakt_list_data("liked_lists") clear_trakt_list_contents_data("my_lists") clear_trakt_list_contents_data("liked_lists") if "hidden_movies" in changes: clear_trakt_hidden_data("movies") if "hidden_shows" in changes: clear_trakt_hidden_data("shows")