import json import logging import sys import time from abc import ABC, abstractmethod import trakt.core from tinydb import Query from tinydb.table import Document from database import syncedEpisodesTable, syncedMoviesTable from searcher import TVShowSearcher, MovieSearcher, TraktTVShow, TraktMovie, TraktItem, TVTimeItem, TVTimeTVShow, \ TVTimeMovie class Processor(ABC): @abstractmethod def _get_synced_items(self, tv_time_item: TVTimeItem) -> list[Document]: pass @abstractmethod def _log_already_imported(self, tv_time_item: TVTimeItem, progress: str) -> None: pass @abstractmethod def _should_continue(self, tv_time_item: TVTimeItem) -> bool: pass @abstractmethod def _search(self, tv_time_item: TVTimeItem) -> TraktItem: pass @abstractmethod def _process(self, tv_time_item: TVTimeItem, trakt_item: TraktItem, progress: str) -> None: pass def process_item(self, tv_time_item: TVTimeItem, progress: str, delay: int = 1) -> None: # Query the local database for previous entries indicating that # the item has already been imported in the past. Which will # ease pressure on Trakt's API server during a retry of the import # process, and just save time overall without needing to create network requests. synced_episodes = self._get_synced_items(tv_time_item) if len(synced_episodes) != 0: self._log_already_imported(tv_time_item, progress) return # If the query returned no results, then continue to import it into Trakt # Create a repeating loop, which will break on success, but repeats on failures error_streak = 0 while True: # If more than 10 errors occurred in one streak, whilst trying to import the item # then give up, and move onto the next item, but warn the user. if error_streak > 10: logging.warning("An error occurred 10 times in a row... skipping episode...") break if not self._should_continue(tv_time_item): break try: # Sleep for a second between each process, before going onto the next watched item. # This is required to remain within the API rate limit, and use the API server fairly. # Other developers share the service, for free - so be considerate of your usage. time.sleep(delay) trakt_item = self._search(tv_time_item) if trakt_item is None: break self._process(tv_time_item, trakt_item, progress) error_streak = 0 break # Catch errors which occur because of an incorrect array index. This occurs when # an incorrect Trakt show has been selected, with season/episodes which don't match TV Time. # It can also occur due to a bug in Trakt Py, whereby some seasons contain an empty array of episodes. except IndexError: self._handle_index_error(tv_time_item, trakt_item, progress) break except trakt.core.errors.NotFoundException: self._handle_not_found_exception(tv_time_item, progress) break except trakt.core.errors.RateLimitException: logging.warning( "The program is running too quickly and has hit Trakt's API rate limit!" " Please increase the delay between" " movies via the variable 'DELAY_BETWEEN_EPISODES_IN_SECONDS'." " The program will now wait 60 seconds before" " trying again." ) time.sleep(60) error_streak += 1 # Catch a JSON decode error - this can be raised when the API server is down and produces an HTML page, # instead of JSON except json.decoder.JSONDecodeError: logging.warning( f"({progress}) - A JSON decode error occurred whilst processing {tv_time_item.name}" " This might occur when the server is down and has produced" " a HTML document instead of JSON. The script will wait 60 seconds before trying again." ) time.sleep(60) error_streak += 1 # Catch a CTRL + C keyboard input, and exits the program except KeyboardInterrupt: sys.exit("Cancel requested...") except Exception as e: logging.error( f"Got unknown error {e}," f" while processing {tv_time_item.name}" ) error_streak += 1 @abstractmethod def _handle_index_error(self, tv_time_item: TVTimeItem, trakt_item: TraktItem, progress: str) -> None: pass @abstractmethod def _handle_not_found_exception(self, tv_time_item: TVTimeItem, progress: str) -> None: pass class TVShowProcessor(Processor): def __init__(self): super().__init__() def _get_synced_items(self, tv_time_show: TVTimeTVShow) -> list[Document]: episode_completed_query = Query() return syncedEpisodesTable.search(episode_completed_query.episodeId == tv_time_show.episode_id) def _log_already_imported(self, tv_time_show: TVTimeTVShow, progress: str) -> None: logging.info( f"({progress}) - Already imported," f" skipping \'{tv_time_show.name}\' Season {tv_time_show.season_number} /" f" Episode {tv_time_show.episode_number}." ) def _should_continue(self, tv_time_show: TVTimeTVShow) -> bool: return True def _search(self, tv_time_show: TVTimeTVShow) -> TraktTVShow: return TVShowSearcher(tv_time_show).search(tv_time_show.title) def _process(self, tv_time_show: TVTimeTVShow, trakt_show: TraktItem, progress: str) -> None: logging.info( f"({progress}) - Processing '{tv_time_show.name}'" f" Season {tv_time_show.season_number} /" f" Episode {tv_time_show.episode_number}" ) season = trakt_show.seasons[tv_time_show.parse_season_number(trakt_show)] episode = season.episodes[int(tv_time_show.episode_number) - 1] episode.mark_as_seen(tv_time_show.date_watched) # Add the episode to the local database as imported, so it can be skipped, # if the process is repeated syncedEpisodesTable.insert({"episodeId": tv_time_show.episode_id}) logging.info( f"'{tv_time_show.name} Season {tv_time_show.season_number}," f" Episode {tv_time_show.episode_number}' marked as seen" ) def _handle_index_error(self, tv_time_show: TVTimeTVShow, trakt_show: TraktTVShow, progress: str) -> None: tv_show_slug = trakt_show.to_json()["shows"][0]["ids"]["ids"]["slug"] logging.warning( f"({progress}) - {tv_time_show.name} Season {tv_time_show.season_number}," f" Episode {tv_time_show.episode_number} does not exist in Trakt!" f" (https://trakt.tv/shows/{tv_show_slug}/seasons/{tv_time_show.season_number}/episodes/{tv_time_show.episode_number})" ) def _handle_not_found_exception(self, tv_time_show: TVTimeTVShow, progress: str) -> None: logging.warning( f"({progress}) - {tv_time_show.name} Season {tv_time_show.season_number}," f" Episode {tv_time_show.episode_number} does not exist (search) in Trakt!" ) class MovieProcessor(Processor): def __init__(self, watched_list: list): super().__init__() self._watched_list = watched_list def _get_synced_items(self, tv_time_movie: TVTimeMovie) -> list[Document]: movie_query = Query() return syncedMoviesTable.search( (movie_query.movie_name == tv_time_movie.name) & (movie_query.type == "watched") ) def _log_already_imported(self, tv_time_movie: TVTimeMovie, progress: str) -> None: logging.info(f"({progress}) - Already imported, skipping '{tv_time_movie.name}'.") def _should_continue(self, tv_time_movie: TVTimeMovie) -> bool: # If movie is watched but this is an entry for watchlist, then skip if tv_time_movie.name in self._watched_list and tv_time_movie.activity_type != "watch": logging.info(f"Skipping '{tv_time_movie.name}' to avoid redundant watchlist entry.") return False return True def _search(self, tv_time_movie: TVTimeMovie) -> TraktMovie: return MovieSearcher().search(tv_time_movie.title) def _process(self, tv_time_movie: TVTimeMovie, trakt_movie: TraktMovie, progress: str) -> None: logging.info(f"({progress}) - Processing '{tv_time_movie.name}'") watchlist_query = Query() movies_in_watchlist = syncedMoviesTable.search( (watchlist_query.movie_name == tv_time_movie.name) & (watchlist_query.type == "watchlist") ) if tv_time_movie.activity_type == "watch": trakt_movie.mark_as_seen(tv_time_movie.date_watched) # Add the episode to the local database as imported, so it can be skipped, # if the process is repeated syncedMoviesTable.insert( {"movie_name": tv_time_movie.name, "type": "watched"} ) logging.info(f"'{tv_time_movie.name}' marked as seen") elif len(movies_in_watchlist) == 0: trakt_movie.add_to_watchlist() # Add the episode to the local database as imported, so it can be skipped, # if the process is repeated syncedMoviesTable.insert( {"movie_name": tv_time_movie.name, "type": "watchlist"} ) logging.info(f"'{tv_time_movie.name}' added to watchlist") else: logging.warning(f"{tv_time_movie.name} already in watchlist") def _handle_index_error(self, tv_time_movie: TVTimeMovie, trakt_movie: TraktMovie, progress: str) -> None: movie_slug = trakt_movie.to_json()["movies"][0]["ids"]["ids"]["slug"] logging.warning( f"({progress}) - {tv_time_movie.name}" f" does not exist in Trakt! (https://trakt.tv/movies/{movie_slug}/)" ) def _handle_not_found_exception(self, tv_time_movie: TVTimeMovie, progress: str) -> None: logging.warning(f"({progress}) - {tv_time_movie.name} does not exist (search) in Trakt!")