import os, re from datetime import datetime, timedelta, timezone from modules import plex, util, anidb from modules.util import Failed, LimitReached from plexapi.exceptions import NotFound from plexapi.video import Movie, Show logger = util.logger meta_operations = [ "mass_audience_rating_update", "mass_user_rating_update", "mass_critic_rating_update", "mass_episode_audience_rating_update", "mass_episode_user_rating_update", "mass_episode_critic_rating_update", "mass_genre_update", "mass_content_rating_update", "mass_originally_available_update", "mass_added_at_update", "mass_original_title_update", "mass_poster_update", "mass_background_update", "mass_studio_update" ] name_display = { "audienceRating": "Audience Rating", "rating": "Critic Rating", "userRating": "User Rating", "originallyAvailableAt": "Originally Available Date", "addedAt": "Added At Date", "contentRating": "Content Rating" } class Operations: def __init__(self, config, library): self.config = config self.library = library def run_operations(self): operation_start = datetime.now() logger.info("") logger.separator(f"{self.library.name} Library Operations") logger.info("") logger.debug(f"Assets For All: {self.library.assets_for_all}") logger.debug(f"Delete Collections: {self.library.delete_collections}") logger.debug(f"Show Unmanaged Collections: {self.library.show_unmanaged}") logger.debug(f"Show Unconfigured Collections: {self.library.show_unconfigured}") logger.debug(f"Mass Genre Update: {self.library.mass_genre_update}") logger.debug(f"Mass Audience Rating Update: {self.library.mass_audience_rating_update}") logger.debug(f"Mass Critic Rating Update: {self.library.mass_critic_rating_update}") logger.debug(f"Mass User Rating Update: {self.library.mass_user_rating_update}") logger.debug(f"Mass Episode Audience Rating Update: {self.library.mass_episode_audience_rating_update}") logger.debug(f"Mass Episode Critic Rating Update: {self.library.mass_episode_critic_rating_update}") logger.debug(f"Mass Episode User Rating Update: {self.library.mass_episode_user_rating_update}") logger.debug(f"Mass Content Rating Update: {self.library.mass_content_rating_update}") logger.debug(f"Mass Original Title Update: {self.library.mass_original_title_update}") logger.debug(f"Mass Originally Available Update: {self.library.mass_originally_available_update}") logger.debug(f"Mass Added At Update: {self.library.mass_added_at_update}") logger.debug(f"Mass IMDb Parental Labels: {self.library.mass_imdb_parental_labels}") logger.debug(f"Mass Poster Update: {self.library.mass_poster_update}") logger.debug(f"Mass Background Update: {self.library.mass_background_update}") logger.debug(f"Mass Collection Mode Update: {self.library.mass_collection_mode}") logger.debug(f"Split Duplicates: {self.library.split_duplicates}") logger.debug(f"Radarr Add All Existing: {self.library.radarr_add_all_existing}") logger.debug(f"Radarr Remove by Tag: {self.library.radarr_remove_by_tag}") logger.debug(f"Sonarr Add All Existing: {self.library.sonarr_add_all_existing}") logger.debug(f"Sonarr Remove by Tag: {self.library.sonarr_remove_by_tag}") logger.debug(f"Update Blank Track Titles: {self.library.update_blank_track_titles}") logger.debug(f"Update Remove Title Parentheses: {self.library.remove_title_parentheses}") logger.debug(f"Genre Mapper: {self.library.genre_mapper}") logger.debug(f"Content Rating Mapper: {self.library.content_rating_mapper}") logger.debug(f"Metadata Backup: {self.library.metadata_backup}") logger.debug(f"Item Operation: {self.library.items_library_operation}") logger.debug("") def should_be_deleted(col_in, labels_in, configured_in, managed_in, less_in): if all((x is None for x in [configured_in, managed_in, less_in])): return False less_check = True if less_in is not None: less_check = col_in.childCount < less_in logger.trace(f"{col_in.title} - collection size: {col_in.childCount} < less: {less_in}, DELETE: {less_check}") managed_check = True if managed_in is not None: is_managed = "PMM" in labels_in or "Kometa" in labels_in managed_check = managed_in == is_managed logger.trace(f"{col_in.title} - collection managed: {is_managed} vs managed: {managed_in}, DELETE: {managed_check}") configured_check = True if configured_in is not None: is_configured = col_in.title in self.library.collections configured_check = configured_in == is_configured logger.trace(f"{col_in.title} - collection configured: {is_configured} vs configured: {configured_in}, DELETE: {configured_check}") return all((less_check, managed_check, configured_check)) if self.library.split_duplicates: items = self.library.search(**{"duplicate": True}) for item in items: item.split() logger.info(f"{item.title[:25]:<25} | Splitting") if self.library.update_blank_track_titles: tracks = self.library.get_all(builder_level="track") num_edited = 0 for i, track in enumerate(tracks, 1): logger.ghost(f"Processing Track: {i}/{len(tracks)} {track.title}") if not track.title and track.titleSort: track.editTitle(track.titleSort) num_edited += 1 logger.info(f"Track: {track.titleSort} was updated with sort title") logger.info(f"{len(tracks)} Tracks Processed; {num_edited} Blank Track Titles Updated") if self.library.items_library_operation: items = self.library.get_all() radarr_adds = [] sonarr_adds = [] label_edits = {"add": {}, "remove": {}} rating_edits = {"audienceRating": {}, "rating": {}, "userRating": {}} genre_edits = {"add": {}, "remove": {}} content_edits = {} studio_edits = {} date_edits = {"originallyAvailableAt": {}, "addedAt": {}} remove_edits = {} reset_edits = {} lock_edits = {} unlock_edits = {} ep_rating_edits = {"audienceRating": {}, "rating": {}, "userRating": {}} ep_remove_edits = {} ep_reset_edits = {} ep_lock_edits = {} ep_unlock_edits = {} if self.library.assets_for_all and not self.library.asset_directory: logger.error("Asset Error: No Asset Directory for Assets For All") for i, item in enumerate(items, 1): logger.info("") logger.info(f"Processing: {i}/{len(items)} {item.title}") try: item = self.library.reload(item) except Failed as e: logger.error(e) continue current_labels = [la.tag for la in self.library.item_labels(item)] if self.library.label_operations else [] if self.library.assets_for_all and self.library.asset_directory: self.library.find_and_upload_assets(item, current_labels) locked_fields = [f.name for f in item.fields if f.locked] tmdb_id, tvdb_id, imdb_id = self.library.get_ids(item) item_edits = "" if self.library.remove_title_parentheses: if not any([f.name == "title" and f.locked for f in item.fields]) and item.title.endswith(")"): new_title = re.sub(" \\(\\w+\\)$", "", item.title) item.editTitle(new_title) item_edits += f"\nUpdated Title: {item.title[:25]:<25} | {new_title}" if self.library.mass_imdb_parental_labels: try: if self.library.mass_imdb_parental_labels == "remove": parental_labels = [] else: parental_guide = self.config.IMDb.parental_guide(imdb_id) parental_labels = [f"{k.capitalize()}:{v}" for k, v in parental_guide.items() if v not in util.parental_levels[self.library.mass_imdb_parental_labels]] add_labels = [la for la in parental_labels if la not in current_labels] remove_labels = [la for la in current_labels if la in util.parental_labels and la not in parental_labels] for label_list, edit_type in [(add_labels, "add"), (remove_labels, "remove")]: if label_list: for label in label_list: if label not in label_edits[edit_type]: label_edits[edit_type][label] = [] label_edits[edit_type][label].append(item.ratingKey) item_edits += f"\n{edit_type.capitalize()} IMDb Parental Labels (Batched) | {', '.join(label_list)}" except Failed: pass if item.locations: path = os.path.dirname(str(item.locations[0])) if self.library.is_movie else str(item.locations[0]) if self.library.Radarr and self.library.radarr_add_all_existing and tmdb_id: path = path.replace(self.library.Radarr.plex_path, self.library.Radarr.radarr_path) path = path[:-1] if path.endswith(('/', '\\')) else path radarr_adds.append((tmdb_id, path)) if self.library.Sonarr and self.library.sonarr_add_all_existing and tvdb_id: path = path.replace(self.library.Sonarr.plex_path, self.library.Sonarr.sonarr_path) path = path[:-1] if path.endswith(("/", "\\")) else path sonarr_adds.append((tvdb_id, path)) _trakt_ratings = None def trakt_ratings(): nonlocal _trakt_ratings if _trakt_ratings is None: _trakt_ratings = self.config.Trakt.user_ratings(self.library.is_movie) if not _trakt_ratings: raise Failed return _trakt_ratings _tmdb_obj = None def tmdb_obj(): nonlocal _tmdb_obj if _tmdb_obj is None: _tmdb_obj = False try: _item = self.config.TMDb.get_item(item, tmdb_id, tvdb_id, imdb_id, is_movie=self.library.is_movie) if _item: _tmdb_obj = _item except Failed as err: logger.error(str(err)) if not _tmdb_obj: raise Failed return _tmdb_obj _omdb_obj = None def omdb_obj(): nonlocal _omdb_obj if _omdb_obj is None: _omdb_obj = False if self.config.OMDb.limit is not False: logger.error("Daily OMDb Limit Reached") elif not imdb_id: logger.info(f"No IMDb ID for Guid: {item.guid}") else: try: _omdb_obj = self.config.OMDb.get_omdb(imdb_id) except Failed as err: logger.error(str(err)) except Exception: logger.error(f"IMDb ID: {imdb_id}") raise if not _omdb_obj: raise Failed return _omdb_obj _tvdb_obj = None def tvdb_obj(): nonlocal _tvdb_obj if _tvdb_obj is None: _tvdb_obj = False if tvdb_id: try: _tvdb_obj = self.config.TVDb.get_tvdb_obj(tvdb_id, is_movie=self.library.is_movie) except Failed as err: logger.error(str(err)) else: logger.info(f"No TVDb ID for Guid: {item.guid}") if not _tvdb_obj: raise Failed return _tvdb_obj _mdb_obj = None def mdb_obj(): nonlocal _mdb_obj if _mdb_obj is None: _mdb_obj = False if self.config.MDBList.limit is False: if self.library.is_show and tvdb_id: try: _mdb_obj = self.config.MDBList.get_series(tvdb_id) except LimitReached as err: logger.debug(err) except Failed as err: logger.error(str(err)) except Exception: logger.trace(f"TVDb ID: {tvdb_id}") raise if self.library.is_movie and tmdb_id: try: _mdb_obj = self.config.MDBList.get_movie(tmdb_id) except LimitReached as err: logger.debug(err) except Failed as err: logger.error(str(err)) except Exception: logger.trace(f"TMDb ID: {tmdb_id}") raise if imdb_id and not _mdb_obj: try: _mdb_obj = self.config.MDBList.get_imdb(imdb_id) except LimitReached as err: logger.debug(err) except Failed as err: logger.error(str(err)) except Exception: logger.trace(f"IMDb ID: {imdb_id}") raise if not _mdb_obj: logger.warning(f"No MdbItem for {item.title} (Guid: {item.guid})") if not _mdb_obj: raise Failed return _mdb_obj anidb_id = None def get_anidb_id(): temp_id = self.config.Convert.ids_to_anidb(self.library, item.ratingKey, tvdb_id, imdb_id, tmdb_id) return temp_id if temp_id else False _anidb_obj = None def anidb_obj(): nonlocal anidb_id, _anidb_obj if _anidb_obj is None: _anidb_obj = False if anidb_id is None: anidb_id = get_anidb_id() if anidb_id: try: _anidb_obj = self.config.AniDB.get_anime(anidb_id) except Failed as err: logger.error(str(err)) else: logger.warning(f"No AniDB ID for Guid: {item.guid}") if not _anidb_obj: raise Failed return _anidb_obj _mal_obj = None def mal_obj(): nonlocal anidb_id, _mal_obj if _mal_obj is None: _mal_obj = False if anidb_id is None: anidb_id = get_anidb_id() mal_id = None if item.ratingKey in self.library.reverse_mal: mal_id = self.library.reverse_mal[item.ratingKey] elif not anidb_id: logger.warning(f"Convert Warning: No AniDB ID to Convert to MyAnimeList ID for Guid: {item.guid}") else: try: mal_id = self.config.Convert.anidb_to_mal(anidb_id) except Failed as err: logger.warning(f"{err} of Guid: {item.guid}") if mal_id: try: _mal_obj = self.config.MyAnimeList.get_anime(mal_id) except Failed as err: logger.error(str(err)) if not _mal_obj: raise Failed return _mal_obj for attribute, item_attr in [ (self.library.mass_audience_rating_update, "audienceRating"), (self.library.mass_critic_rating_update, "rating"), (self.library.mass_user_rating_update, "userRating") ]: if attribute: current = getattr(item, item_attr) for option in attribute: if option in ["lock", "remove"]: if option == "remove" and current: if item_attr not in remove_edits: remove_edits[item_attr] = [] remove_edits[item_attr].append(item.ratingKey) item_edits += f"\nRemove {name_display[item_attr]} (Batched)" elif item_attr not in locked_fields: if item_attr not in lock_edits: lock_edits[item_attr] = [] lock_edits[item_attr].append(item.ratingKey) item_edits += f"\nLock {name_display[item_attr]} (Batched)" break elif option in ["unlock", "reset"]: if option == "reset" and current: if item_attr not in reset_edits: reset_edits[item_attr] = [] reset_edits[item_attr].append(item.ratingKey) item_edits += f"\nReset {name_display[item_attr]} (Batched)" elif item_attr in locked_fields: if item_attr not in unlock_edits: unlock_edits[item_attr] = [] unlock_edits[item_attr].append(item.ratingKey) item_edits += f"\nUnlock {name_display[item_attr]} (Batched)" break else: try: if option == "tmdb": found_rating = tmdb_obj().vote_average # noqa elif option == "imdb": found_rating = self.config.IMDb.get_rating(imdb_id) elif option == "omdb": found_rating = omdb_obj().imdb_rating # noqa elif option == "trakt_user": _ratings = trakt_ratings() _id = tmdb_id if self.library.is_movie else tvdb_id if _id in _ratings: found_rating = _ratings[_id] else: raise Failed elif str(option).startswith("mdb"): mdb_item = mdb_obj() if option == "mdb_average": found_rating = mdb_item.average / 10 if mdb_item.average else None # noqa elif option == "mdb_imdb": found_rating = mdb_item.imdb_rating if mdb_item.imdb_rating else None # noqa elif option == "mdb_metacritic": found_rating = mdb_item.metacritic_rating / 10 if mdb_item.metacritic_rating else None # noqa elif option == "mdb_metacriticuser": found_rating = mdb_item.metacriticuser_rating if mdb_item.metacriticuser_rating else None # noqa elif option == "mdb_trakt": found_rating = mdb_item.trakt_rating / 10 if mdb_item.trakt_rating else None # noqa elif option == "mdb_tomatoes": found_rating = mdb_item.tomatoes_rating / 10 if mdb_item.tomatoes_rating else None # noqa elif option == "mdb_tomatoesaudience": found_rating = mdb_item.tomatoesaudience_rating / 10 if mdb_item.tomatoesaudience_rating else None # noqa elif option == "mdb_tmdb": found_rating = mdb_item.tmdb_rating / 10 if mdb_item.tmdb_rating else None # noqa elif option == "mdb_letterboxd": found_rating = mdb_item.letterboxd_rating * 2 if mdb_item.letterboxd_rating else None # noqa elif option == "mdb_myanimelist": found_rating = mdb_item.myanimelist_rating if mdb_item.myanimelist_rating else None # noqa else: found_rating = mdb_item.score / 10 if mdb_item.score else None # noqa elif option == "anidb_rating": found_rating = anidb_obj().rating # noqa elif option == "anidb_average": found_rating = anidb_obj().average # noqa elif option == "anidb_score": found_rating = anidb_obj().score # noqa elif option == "mal": found_rating = mal_obj().score # noqa else: found_rating = option if not found_rating: logger.info(f"No {option} {name_display[item_attr]} Found") raise Failed found_rating = f"{float(found_rating):.1f}" if str(current) != found_rating: if found_rating not in rating_edits[item_attr]: rating_edits[item_attr][found_rating] = [] rating_edits[item_attr][found_rating].append(item.ratingKey) item_edits += f"\nUpdate {name_display[item_attr]} (Batched) | {found_rating}" break except Failed: continue if self.library.mass_genre_update or self.library.genre_mapper: new_genres = [] extra_option = None if self.library.mass_genre_update: for option in self.library.mass_genre_update: if option in ["lock", "unlock", "remove", "reset"]: extra_option = option break try: if option == "tmdb": new_genres = tmdb_obj().genres # noqa elif option == "imdb": new_genres = self.config.IMDb.get_genres(imdb_id) elif option == "omdb": new_genres = omdb_obj().genres # noqa elif option == "tvdb": new_genres = tvdb_obj().genres # noqa elif str(option) in anidb.weights: new_genres = [str(t).title() for t, w in anidb_obj().tags.items() if w >= anidb.weights[str(option)]] # noqa elif option == "mal": new_genres = mal_obj().genres # noqa else: new_genres = option if not new_genres: logger.info(f"No {option} Genres Found") raise Failed break except Failed: continue item_genres = [g.tag for g in item.genres] if not new_genres and extra_option not in ["remove", "reset"]: new_genres = item_genres if self.library.genre_mapper: mapped_genres = [] for genre in new_genres: if genre in self.library.genre_mapper: if self.library.genre_mapper[genre]: mapped_genres.append(self.library.genre_mapper[genre]) else: mapped_genres.append(genre) new_genres = mapped_genres _add = list(set(new_genres) - set(item_genres)) _remove = list(set(item_genres) - set(new_genres)) for genre_list, edit_type in [(_add, "add"), (_remove, "remove")]: if genre_list: for g in genre_list: if g not in genre_edits[edit_type]: genre_edits[edit_type][g] = [] genre_edits[edit_type][g].append(item.ratingKey) item_edits += f"\n{edit_type.capitalize()} Genres (Batched) | {', '.join(genre_list)}" if extra_option in ["unlock", "reset"] and ("genre" in locked_fields or _add or _remove): if "genre" not in unlock_edits: unlock_edits["genre"] = [] unlock_edits["genre"].append(item.ratingKey) item_edits += "\nUnlock Genre (Batched)" elif extra_option in ["lock", "remove"] and "genre" not in locked_fields and not _add and not _remove: if "genre" not in lock_edits: lock_edits["genre"] = [] lock_edits["genre"].append(item.ratingKey) item_edits += "\nLock Genre (Batched)" if self.library.mass_content_rating_update or self.library.content_rating_mapper: new_rating = None extra_option = None if self.library.mass_content_rating_update: for option in self.library.mass_content_rating_update: if option in ["lock", "unlock", "remove", "reset"]: extra_option = option break try: if option == "omdb": new_rating = omdb_obj().content_rating # noqa elif option == "mdb": _rating = mdb_obj().content_rating # noqa new_rating = _rating if _rating else None elif str(option).startswith("mdb_commonsense"): _rating = mdb_obj().commonsense # noqa if not _rating: new_rating = None elif option == "mdb_commonsense0": new_rating = str(_rating).rjust(2, "0") else: new_rating = _rating elif str(option).startswith("mdb_age_rating"): _rating = mdb_obj().age_rating # noqa if not _rating: new_rating = None elif option == "mdb_age_rating0": new_rating = str(_rating).rjust(2, "0") else: new_rating = _rating elif option == "mal": new_rating = mal_obj().rating # noqa else: new_rating = option if new_rating is None: logger.info(f"No {option} Content Rating Found") raise Failed else: new_rating = str(new_rating) break except Failed: continue is_none = False do_lock = False do_unlock = False current_rating = item.contentRating if not new_rating: new_rating = current_rating if self.library.content_rating_mapper: if new_rating in self.library.content_rating_mapper: new_rating = self.library.content_rating_mapper[new_rating] if not new_rating: is_none = True if extra_option == "reset": if current_rating: if "contentRating" not in reset_edits: reset_edits["contentRating"] = [] reset_edits["contentRating"].append(item.ratingKey) item_edits += "\nReset Content Rating (Batched)" elif "contentRating" in locked_fields: do_unlock = True elif extra_option == "remove" or is_none: if current_rating: if "contentRating" not in remove_edits: remove_edits["contentRating"] = [] remove_edits["contentRating"].append(item.ratingKey) item_edits += "\nRemove Content Rating (Batched)" elif "contentRating" not in locked_fields: do_lock = True elif new_rating and new_rating != current_rating: if new_rating not in content_edits: content_edits[new_rating] = [] content_edits[new_rating].append(item.ratingKey) item_edits += f"\nUpdate Content Rating (Batched) | {new_rating}" do_lock = False if extra_option == "lock" or do_lock: if "contentRating" not in lock_edits: lock_edits["contentRating"] = [] lock_edits["contentRating"].append(item.ratingKey) item_edits += "\nLock Content Rating (Batched)" elif extra_option == "unlock" or do_unlock: if "contentRating" not in unlock_edits: unlock_edits["contentRating"] = [] unlock_edits["contentRating"].append(item.ratingKey) item_edits += "\nUnlock Content Rating (Batched)" if self.library.mass_original_title_update: current_original = item.originalTitle for option in self.library.mass_original_title_update: if option in ["lock", "remove"]: if option == "remove" and current_original: if "originalTitle" not in remove_edits: remove_edits["originalTitle"] = [] remove_edits["originalTitle"].append(item.ratingKey) item_edits += "\nRemove Original Title (Batched)" elif "originalTitle" not in locked_fields: if "originalTitle" not in lock_edits: lock_edits["originalTitle"] = [] lock_edits["originalTitle"].append(item.ratingKey) item_edits += "\nLock Original Title (Batched)" break elif option in ["unlock", "reset"]: if option == "reset" and current_original: if "originalTitle" not in reset_edits: reset_edits["originalTitle"] = [] reset_edits["originalTitle"].append(item.ratingKey) item_edits += "\nReset Original Title (Batched)" elif "originalTitle" in locked_fields: if "originalTitle" not in unlock_edits: unlock_edits["originalTitle"] = [] unlock_edits["originalTitle"].append(item.ratingKey) item_edits += "\nUnlock Original Title (Batched)" break else: try: if option == "anidb": new_original_title = anidb_obj().main_title # noqa elif option == "anidb_official": new_original_title = anidb_obj().official_title # noqa elif option == "mal": new_original_title = mal_obj().title # noqa elif option == "mal_english": new_original_title = mal_obj().title_english # noqa elif option == "mal_japanese": new_original_title = mal_obj().title_japanese # noqa else: new_original_title = option if not new_original_title: logger.info(f"No {option} Original Title Found") raise Failed if str(current_original) != str(new_original_title): item.editOriginalTitle(new_original_title) item_edits += f"\nUpdated Original Title | {new_original_title}" break except Failed: continue if self.library.mass_studio_update: current_studio = item.studio for option in self.library.mass_studio_update: if option in ["lock", "remove"]: if option == "remove" and current_studio: if "studio" not in remove_edits: remove_edits["studio"] = [] remove_edits["studio"].append(item.ratingKey) item_edits += "\nRemove Studio (Batched)" elif "studio" not in locked_fields: if "studio" not in lock_edits: lock_edits["studio"] = [] lock_edits["studio"].append(item.ratingKey) item_edits += "\nLock Studio (Batched)" break elif option in ["unlock", "reset"]: if option == "reset" and current_studio: if "studio" not in reset_edits: reset_edits["studio"] = [] reset_edits["studio"].append(item.ratingKey) item_edits += "\nReset Studio (Batched)" elif "studio" in locked_fields: if "studio" not in unlock_edits: unlock_edits["studio"] = [] unlock_edits["studio"].append(item.ratingKey) item_edits += "\nUnlock Studio (Batched)" break else: try: if option == "tmdb": new_studio = tmdb_obj().studio # noqa elif option == "anidb": new_studio = anidb_obj().studio # noqa elif option == "mal": new_studio = mal_obj().studio # noqa else: new_studio = option if not new_studio: logger.info(f"No {option} Studio Found") raise Failed if str(current_studio) != str(new_studio): if new_studio not in studio_edits: studio_edits[new_studio] = [] studio_edits[new_studio].append(item.ratingKey) item_edits += f"\nUpdate Studio (Batched) | {new_studio}" break except Failed: continue for attribute, item_attr in [ (self.library.mass_originally_available_update, "originallyAvailableAt"), (self.library.mass_added_at_update, "addedAt") ]: if attribute: current = getattr(item, item_attr) if current: current = current.strftime("%Y-%m-%d") for option in attribute: if option in ["lock", "remove"]: if option == "remove" and current: if item_attr not in remove_edits: remove_edits[item_attr] = [] remove_edits[item_attr].append(item.ratingKey) item_edits += f"\nRemove {name_display[item_attr]} (Batched)" elif item_attr not in locked_fields: if item_attr not in lock_edits: lock_edits[item_attr] = [] lock_edits[item_attr].append(item.ratingKey) item_edits += f"\nLock {name_display[item_attr]} (Batched)" break elif option in ["unlock", "reset"]: if option == "reset" and current: if item_attr not in reset_edits: reset_edits[item_attr] = [] reset_edits[item_attr].append(item.ratingKey) item_edits += f"\nReset {name_display[item_attr]} (Batched)" elif item_attr in locked_fields: if item_attr not in unlock_edits: unlock_edits[item_attr] = [] unlock_edits[item_attr].append(item.ratingKey) item_edits += f"\nUnlock {name_display[item_attr]} (Batched)" break else: try: if option == "tmdb": new_date = tmdb_obj().release_date if self.library.is_movie else tmdb_obj().first_air_date # noqa elif option == "omdb": new_date = omdb_obj().released # noqa elif option == "tvdb": new_date = tvdb_obj().release_date # noqa elif option == "mdb": new_date = mdb_obj().released # noqa elif option == "mdb_digital": new_date = mdb_obj().released_digital # noqa elif option == "anidb": new_date = anidb_obj().released # noqa elif option == "mal": new_date = mal_obj().aired # noqa else: new_date = option if not new_date: logger.info(f"No {option} {name_display[item_attr]} Found") raise Failed new_date = new_date.strftime("%Y-%m-%d") if current != new_date: if new_date not in date_edits[item_attr]: date_edits[item_attr][new_date] = [] date_edits[item_attr][new_date].append(item.ratingKey) item_edits += f"\nUpdate {name_display[item_attr]} (Batched) | {new_date}" break except Failed: continue if len(item_edits) > 0: logger.info(f"Item Edits{item_edits}") else: logger.info("No Item Edits") if self.library.mass_poster_update or self.library.mass_background_update: try: new_poster, new_background, item_dir, name = self.library.find_item_assets(item) except Failed: item_dir = None name = None new_poster = None new_background = None try: tmdb_item = tmdb_obj() except Failed: tmdb_item = None if self.library.mass_poster_update: self.library.poster_update(item, new_poster, tmdb=tmdb_item.poster_url if tmdb_item else None, title=item.title) # noqa if self.library.mass_background_update: self.library.background_update(item, new_background, tmdb=tmdb_item.backdrop_url if tmdb_item else None, title=item.title) # noqa if self.library.is_show and ( (self.library.mass_poster_update and (self.library.mass_poster_update["seasons"] or self.library.mass_poster_update["episodes"])) or (self.library.mass_background_update and (self.library.mass_background_update["seasons"] or self.library.mass_background_update["episodes"])) ): real_show = None try: real_show = tmdb_item.load_show() if tmdb_item else None # noqa except Failed as e: logger.error(e) tmdb_seasons = {s.season_number: s for s in real_show.seasons} if real_show else {} for season in self.library.query(item.seasons): if (self.library.mass_poster_update and self.library.mass_poster_update["seasons"]) or \ (self.library.mass_background_update and self.library.mass_background_update["seasons"]): try: season_poster, season_background, _, _ = self.library.find_item_assets(season, item_asset_directory=item_dir, folder_name=name) except Failed: season_poster = None season_background = None season_title = f"S{season.seasonNumber} {season.title}" tmdb_poster = tmdb_seasons[season.seasonNumber].poster_url if season.seasonNumber in tmdb_seasons else None if self.library.mass_poster_update and self.library.mass_poster_update["seasons"]: self.library.poster_update(season, season_poster, tmdb=tmdb_poster, title=season_title if season else None) if self.library.mass_background_update and self.library.mass_background_update["seasons"]: self.library.background_update(season, season_background, title=season_title if season else None) if (self.library.mass_poster_update and self.library.mass_poster_update["episodes"]) or \ (self.library.mass_background_update and self.library.mass_background_update["episodes"]): tmdb_episodes = {} if season.seasonNumber in tmdb_seasons: for episode in tmdb_seasons[season.seasonNumber].episodes: episode._partial = False try: tmdb_episodes[episode.episode_number] = episode except NotFound: logger.error(f"TMDb Error: An Episode of Season {season.seasonNumber} was Not Found") for episode in self.library.query(season.episodes): try: episode = self.library.reload(episode) except Failed: logger.error(f"S{season.seasonNumber}E{episode.episodeNumber} {episode.title} Failed to Reload from Plex") continue try: episode_poster, episode_background, _, _ = self.library.find_item_assets(episode, item_asset_directory=item_dir, folder_name=name) except Failed: episode_poster = None episode_background = None episode_title = f"S{season.seasonNumber}E{episode.episodeNumber} {episode.title}" tmdb_poster = tmdb_episodes[episode.episodeNumber].still_url if episode.episodeNumber in tmdb_episodes else None if self.library.mass_poster_update and self.library.mass_poster_update["episodes"]: self.library.poster_update(episode, episode_poster, tmdb=tmdb_poster, title=episode_title if episode else None) if self.library.mass_background_update and self.library.mass_background_update["episodes"]: self.library.background_update(episode, episode_background, title=episode_title if episode else None) episode_ops = [ (self.library.mass_episode_audience_rating_update, "audienceRating"), (self.library.mass_episode_critic_rating_update, "rating"), (self.library.mass_episode_user_rating_update, "userRating") ] if any([x is not None for x, _ in episode_ops]): if any(["imdb" in x for x, _ in episode_ops if x]) and not imdb_id: logger.info(f"No IMDb ID for Guid: {item.guid}") for ep in item.episodes(): ep = self.library.reload(ep) item_title = self.library.get_item_sort_title(ep, atr="title") logger.info("") logger.info(f"Processing {item_title}") item_edits = "" for attribute, item_attr in episode_ops: if attribute: current = getattr(ep, item_attr) for option in attribute: if option in ["lock", "remove"]: if option == "remove" and current: if item_attr not in ep_remove_edits: ep_remove_edits[item_attr] = [] ep_remove_edits[item_attr].append(ep) item_edits += f"\nRemove {name_display[item_attr]} (Batched)" elif item_attr not in locked_fields: if item_attr not in ep_lock_edits: ep_lock_edits[item_attr] = [] ep_lock_edits[item_attr].append(ep) item_edits += f"\nLock {name_display[item_attr]} (Batched)" break elif option in ["unlock", "reset"]: if option == "reset" and current: if item_attr not in ep_reset_edits: ep_reset_edits[item_attr] = [] ep_reset_edits[item_attr].append(ep) item_edits += f"\nReset {name_display[item_attr]} (Batched)" elif item_attr in locked_fields: if item_attr not in ep_unlock_edits: ep_unlock_edits[item_attr] = [] ep_unlock_edits[item_attr].append(ep) item_edits += f"\nUnlock {name_display[item_attr]} (Batched)" break else: try: try: tmdb_item = tmdb_obj() except Failed: tmdb_item = None found_rating = None if tmdb_item and option == "tmdb": try: found_rating = self.config.TMDb.get_episode(tmdb_item.tmdb_id, ep.seasonNumber, ep.episodeNumber).vote_average # noqa except Failed as er: logger.error(er) elif imdb_id and option == "imdb": found_rating = self.config.IMDb.get_episode_rating(imdb_id, ep.seasonNumber, ep.episodeNumber) else: try: found_rating = float(option) except ValueError: pass if not found_rating: logger.info(f"No {option} {name_display[item_attr]} Found") raise Failed found_rating = f"{float(found_rating):.1f}" if str(current) != found_rating: if found_rating not in ep_rating_edits[item_attr]: ep_rating_edits[item_attr][found_rating] = [] ep_rating_edits[item_attr][found_rating].append(ep) item_edits += f"\nUpdate {name_display[item_attr]} (Batched) | {found_rating}" break except Failed: continue if len(item_edits) > 0: logger.info(f"Item Edits:{item_edits}") logger.info("") logger.separator("Batch Updates", space=False, border=False) logger.info("") def get_batch_info(placement, total, display_attr, total_count, display_value=None, is_episode=False, out_type=None, tag_type=None): return f"Batch {name_display[display_attr] if display_attr in name_display else display_attr.capitalize()} Update ({placement}/{total}): " \ f"{f'{out_type.capitalize()} ' if out_type else ''}" \ f"{f'Adding {display_value} to ' if tag_type == 'add' else f'Removing {display_value} from ' if tag_type == 'remove' else ''}" \ f"{total_count} {'Episode' if is_episode else 'Movie' if self.library.is_movie else 'Show'}" \ f"{'s' if total_count > 1 else ''}{'' if out_type or tag_type else f' updated to {display_value}'}" for tag_attribute, edit_dict in [("Label", label_edits), ("Genre", genre_edits)]: for edit_type, batch_edits in edit_dict.items(): _size = len(batch_edits.items()) for i, (tag_name, rating_keys) in enumerate(sorted(batch_edits.items()), 1): logger.info(get_batch_info(i, _size, tag_attribute, len(rating_keys), display_value=tag_name, tag_type=edit_type)) self.library.Plex.batchMultiEdits(self.library.load_list_from_cache(rating_keys)) getattr(self.library.Plex, f"{edit_type}{tag_attribute}")(tag_name) self.library.Plex.saveMultiEdits() for item_attr, _edits in rating_edits.items(): _size = len(rating_edits.items()) for i, (new_rating, rating_keys) in enumerate(sorted(_edits.items()), 1): logger.info(get_batch_info(i, _size, item_attr, len(rating_keys), display_value=new_rating)) self.library.Plex.batchMultiEdits(self.library.load_list_from_cache(rating_keys)) self.library.Plex.editField(item_attr, new_rating) self.library.Plex.saveMultiEdits() _size = len(content_edits.items()) for i, (new_rating, rating_keys) in enumerate(sorted(content_edits.items()), 1): logger.info(get_batch_info(i, _size, "contentRating", len(rating_keys), display_value=new_rating)) self.library.Plex.batchMultiEdits(self.library.load_list_from_cache(rating_keys)) self.library.Plex.editContentRating(new_rating) self.library.Plex.saveMultiEdits() _size = len(studio_edits.items()) for i, (new_studio, rating_keys) in enumerate(sorted(studio_edits.items()), 1): logger.info(get_batch_info(i, _size, "studio", len(rating_keys), display_value=new_studio)) self.library.Plex.batchMultiEdits(self.library.load_list_from_cache(rating_keys)) self.library.Plex.editStudio(new_studio) self.library.Plex.saveMultiEdits() _size = len(date_edits["originallyAvailableAt"].items()) for i, (new_date, rating_keys) in enumerate(sorted(date_edits["originallyAvailableAt"].items()), 1): logger.info(get_batch_info(i, _size, "originallyAvailableAt", len(rating_keys), display_value=new_date)) self.library.Plex.batchMultiEdits(self.library.load_list_from_cache(rating_keys)) self.library.Plex.editOriginallyAvailable(new_date) self.library.Plex.saveMultiEdits() epoch = datetime(1970, 1, 1) _size = len(date_edits["addedAt"].items()) for i, (new_date, rating_keys) in enumerate(sorted(date_edits["addedAt"].items()), 1): logger.info(get_batch_info(i, _size, "addedAt", len(rating_keys), display_value=new_date)) self.library.Plex.batchMultiEdits(self.library.load_list_from_cache(rating_keys)) new_date = datetime.strptime(new_date, "%Y-%m-%d") logger.trace(new_date) try: ts = int(round(new_date.timestamp())) except (TypeError, OSError): offset = int(datetime(2000, 1, 1, tzinfo=timezone.utc).timestamp() - datetime(2000, 1, 1).timestamp()) ts = int((new_date - epoch).total_seconds()) - offset logger.trace(epoch + timedelta(seconds=ts)) self.library.Plex.editAddedAt(ts) self.library.Plex.saveMultiEdits() _size = len(remove_edits.items()) for i, (field_attr, rating_keys) in enumerate(remove_edits.items(), 1): logger.info(get_batch_info(i, _size, field_attr, len(rating_keys), out_type="remov")) self.library.Plex.batchMultiEdits(self.library.load_list_from_cache(rating_keys)) self.library.Plex.editField(field_attr, None, locked=True) self.library.Plex.saveMultiEdits() _size = len(reset_edits.items()) for i, (field_attr, rating_keys) in enumerate(reset_edits.items(), 1): logger.info(get_batch_info(i, _size, field_attr, len(rating_keys), out_type="reset")) self.library.Plex.batchMultiEdits(self.library.load_list_from_cache(rating_keys)) self.library.Plex.editField(field_attr, None, locked=False) self.library.Plex.saveMultiEdits() _size = len(lock_edits.items()) for i, (field_attr, rating_keys) in enumerate(lock_edits.items(), 1): logger.info(get_batch_info(i, _size, field_attr, len(rating_keys), out_type="lock")) self.library.Plex.batchMultiEdits(self.library.load_list_from_cache(rating_keys)) self.library.Plex._edit(**{f"{field_attr}.locked": 1}) self.library.Plex.saveMultiEdits() _size = len(unlock_edits.items()) for i, (field_attr, rating_keys) in enumerate(unlock_edits.items(), 1): logger.info(get_batch_info(i, _size, field_attr, len(rating_keys), out_type="unlock")) self.library.Plex.batchMultiEdits(self.library.load_list_from_cache(rating_keys)) self.library.Plex._edit(**{f"{field_attr}.locked": 0}) self.library.Plex.saveMultiEdits() for item_attr, _edits in ep_rating_edits.items(): _size = len(_edits.items()) for i, (new_rating, rating_keys) in enumerate(sorted(_edits.items()), 1): logger.info(get_batch_info(i, _size, item_attr, len(rating_keys), display_value=new_rating, is_episode=True)) self.library.Plex.batchMultiEdits(rating_keys) self.library.Plex.editField(item_attr, new_rating) self.library.Plex.saveMultiEdits() _size = len(ep_remove_edits.items()) for i, (field_attr, rating_keys) in enumerate(ep_remove_edits.items(), 1): logger.info(get_batch_info(i, _size, field_attr, len(rating_keys), is_episode=True, out_type="remov")) self.library.Plex.batchMultiEdits(rating_keys) self.library.Plex.editField(field_attr, None, locked=True) self.library.Plex.saveMultiEdits() _size = len(ep_reset_edits.items()) for i, (field_attr, rating_keys) in enumerate(ep_reset_edits.items(), 1): logger.info(get_batch_info(i, _size, field_attr, len(rating_keys), is_episode=True, out_type="reset")) self.library.Plex.batchMultiEdits(rating_keys) self.library.Plex.editField(field_attr, None, locked=False) self.library.Plex.saveMultiEdits() _size = len(ep_lock_edits.items()) for i, (field_attr, rating_keys) in enumerate(ep_lock_edits.items(), 1): logger.info(get_batch_info(i, _size, field_attr, len(rating_keys), is_episode=True, out_type="lock")) self.library.Plex.batchMultiEdits(rating_keys) self.library.Plex._edit(**{f"{field_attr}.locked": 1}) self.library.Plex.saveMultiEdits() _size = len(ep_unlock_edits.items()) for i, (field_attr, rating_keys) in enumerate(ep_unlock_edits.items(), 1): logger.info(get_batch_info(i, _size, field_attr, len(rating_keys), is_episode=True, out_type="unlock")) self.library.Plex.batchMultiEdits(rating_keys) self.library.Plex._edit(**{f"{field_attr}.locked": 0}) self.library.Plex.saveMultiEdits() if self.library.Radarr and self.library.radarr_add_all_existing: logger.info("") logger.separator(f"Radarr Add All Existing: {len(radarr_adds)} Movies", space=False, border=False) logger.info("") try: self.library.Radarr.add_tmdb(radarr_adds) except Failed as e: logger.error(e) if self.library.Sonarr and self.library.sonarr_add_all_existing: logger.info("") logger.separator(f"Sonarr Add All Existing: {len(sonarr_adds)} Shows", space=False, border=False) logger.info("") try: self.library.Sonarr.add_tvdb(sonarr_adds) except Failed as e: logger.error(e) if self.library.radarr_remove_by_tag: logger.info("") logger.separator(f"Radarr Remove {len(self.library.radarr_remove_by_tag)} Movies with Tags: {', '.join(self.library.radarr_remove_by_tag)}", space=False, border=False) logger.info("") self.library.Radarr.remove_all_with_tags(self.library.radarr_remove_by_tag) if self.library.sonarr_remove_by_tag: logger.info("") logger.separator(f"Sonarr Remove {len(self.library.sonarr_remove_by_tag)} Shows with Tags: {', '.join(self.library.sonarr_remove_by_tag)}", space=False, border=False) logger.info("") self.library.Sonarr.remove_all_with_tags(self.library.sonarr_remove_by_tag) if self.library.delete_collections or self.library.show_unmanaged or self.library.show_unconfigured or self.library.assets_for_all or self.library.mass_collection_mode: logger.info("") logger.separator("Collection Operations", space=False, border=False) logger.info("") if self.library.delete_collections: logger.info("") logger.separator("Deleting Collections", space=False, border=False) logger.info("") less = self.library.delete_collections["less"] if self.library.delete_collections and self.library.delete_collections["less"] is not None else None managed = self.library.delete_collections["managed"] if self.library.delete_collections else None configured = self.library.delete_collections["configured"] if self.library.delete_collections else None unmanaged_collections = [] unconfigured_collections = [] all_collections = self.library.get_all_collections() for i, col in enumerate(all_collections, 1): logger.ghost(f"Reading Collection: {i}/{len(all_collections)} {col.title}") col = self.library.reload(col, force=True) labels = [la.tag for la in self.library.item_labels(col)] if should_be_deleted(col, labels, configured, managed, less): try: self.library.delete(col) logger.info(f"{col.title} Deleted") except Failed as e: logger.error(e) else: if "PMM" not in labels and "Kometa" not in labels: unmanaged_collections.append(col) if col.title not in self.library.collection_names: unconfigured_collections.append(col) if self.library.show_unmanaged and len(unmanaged_collections) > 0: logger.info("") logger.separator(f"Unmanaged Collections in {self.library.name} Library", space=False, border=False) logger.info("") for col in unmanaged_collections: logger.info(col.title) logger.info("") logger.info(f"{len(unmanaged_collections)} Unmanaged Collection{'s' if len(unmanaged_collections) > 1 else ''}") elif self.library.show_unmanaged: logger.info("") logger.separator(f"No Unmanaged Collections in {self.library.name} Library", space=False, border=False) logger.info("") if self.library.show_unconfigured and len(unconfigured_collections) > 0: logger.info("") logger.separator(f"Unconfigured Collections in {self.library.name} Library", space=False, border=False) logger.info("") for col in unconfigured_collections: logger.info(col.title) logger.info("") logger.info(f"{len(unconfigured_collections)} Unconfigured Collection{'s' if len(unconfigured_collections) > 1 else ''}") elif self.library.show_unconfigured: logger.info("") logger.separator(f"No Unconfigured Collections in {self.library.name} Library", space=False, border=False) logger.info("") if self.library.assets_for_all_collections and len(unconfigured_collections) > 0: logger.info("") logger.separator(f"Unconfigured Collection Assets Check for {self.library.name} Library", space=False, border=False) logger.info("") for col in unconfigured_collections: try: poster, background, item_dir, name = self.library.find_item_assets(col) if poster or background: self.library.upload_images(col, poster=poster, background=background) elif self.library.show_missing_assets: logger.warning(f"Asset Warning: No poster or background found in an assets folder for '{name}'") except Failed as e: logger.warning(e) if self.library.mass_collection_mode: logger.info("") logger.separator(f"Unconfigured Mass Collection Mode to {self.library.mass_collection_mode} for {self.library.name} Library", space=False, border=False) logger.info("") for col in unconfigured_collections: if int(col.collectionMode) not in plex.collection_mode_keys \ or plex.collection_mode_keys[int(col.collectionMode)] != self.library.mass_collection_mode: self.library.collection_mode_query(col, self.library.mass_collection_mode) logger.info(f"{col.title} Collection Mode Updated") if self.library.metadata_backup: logger.info("") logger.separator(f"Metadata Backup for {self.library.name} Library", space=False, border=False) logger.info("") logger.info(f"Metadata Backup Path: {self.library.metadata_backup['path']}") logger.info("") yaml = None if os.path.exists(self.library.metadata_backup["path"]): try: yaml = self.config.Requests.file_yaml(self.library.metadata_backup["path"]) except Failed as e: logger.error(e) filename, file_extension = os.path.splitext(self.library.metadata_backup["path"]) i = 1 while os.path.exists(f"{filename}{i}{file_extension}"): i += 1 os.rename(self.library.metadata_backup["path"], f"{filename}{i}{file_extension}") logger.error(f"Backup failed to load saving copy to {filename}{i}{file_extension}") if not yaml: yaml = self.config.Requests.file_yaml(self.library.metadata_backup["path"], create=True) if "metadata" not in yaml.data or not isinstance(yaml.data["metadata"], dict): yaml.data["metadata"] = {} special_names = {} for mk, mv in yaml.data["metadata"].items(): if mv and "title" in mv: special_names[mv["title"]] = mk if "year" in mv: special_names[f"{mv['title']} ({mv['year']})"] = mk items = self.library.get_all(load=True) titles = [] year_titles = [] for item in items: titles.append(item.title) if isinstance(item, (Movie, Show)): year_titles.append(f"{item.title} ({item.year})") for i, item in enumerate(items, 1): logger.ghost(f"Processing: {i}/{len(items)} {item.title}") map_key, attrs = self.library.get_locked_attributes(item, titles, year_titles) if map_key in special_names: map_key = special_names[map_key] og_dict = yaml.data["metadata"][map_key] if map_key in yaml.data["metadata"] and yaml.data["metadata"][map_key] and isinstance(yaml.data["metadata"][map_key], dict) else {} if attrs or (self.library.metadata_backup["add_blank_entries"] and not og_dict): def loop_dict(looping, dest_dict): if not looping: return None for lk, lv in looping.items(): if isinstance(lv, dict) and lk in dest_dict and dest_dict[lk] and isinstance(dest_dict[lk], dict): dest_dict[lk] = loop_dict(lv, dest_dict[lk]) else: dest_dict[lk] = lv return dest_dict yaml.data["metadata"][map_key] = loop_dict(attrs, og_dict) logger.exorcise() yaml.save() logger.info(f"{len(yaml.data['metadata'])} {self.library.type}{'s' if len(yaml.data['metadata']) > 1 else ''} Backed Up") operation_run_time = str(datetime.now() - operation_start).split('.')[0] logger.info("") logger.separator(f"Finished {self.library.name} Library Operations\nOperations Run Time: {operation_run_time}") return operation_run_time