import os from abc import ABC, abstractmethod from modules import util from modules.meta import MetadataFile, OverlayFile from modules.operations import Operations from modules.util import Failed, ImageData from ruamel import yaml logger = util.logger class Library(ABC): def __init__(self, config, params): self.Radarr = None self.Sonarr = None self.Tautulli = None self.Webhooks = None self.Operations = Operations(config, self) self.Overlays = None self.Notifiarr = None self.collections = [] self.metadatas = [] self.overlays = [] self.metadata_files = [] self.overlay_files = [] self.missing = {} self.movie_map = {} self.show_map = {} self.imdb_map = {} self.anidb_map = {} self.mal_map = {} self.movie_rating_key_map = {} self.show_rating_key_map = {} self.cached_items = {} self.run_again = [] self.overlays_old = [] self.type = "" self.config = config self.name = params["name"] self.original_mapping_name = params["mapping_name"] self.metadata_path = params["metadata_path"] self.overlay_path = params["overlay_path"] self.skip_library = params["skip_library"] self.asset_depth = params["asset_depth"] self.asset_directory = params["asset_directory"] if params["asset_directory"] else [] self.default_dir = params["default_dir"] self.mapping_name, output = util.validate_filename(self.original_mapping_name) self.image_table_name = self.config.Cache.get_image_table_name(self.original_mapping_name) if self.config.Cache else None self.overlay_folder = os.path.join(self.config.default_dir, "overlays") self.overlay_backup = os.path.join(self.overlay_folder, f"{self.mapping_name} Original Posters") self.missing_path = params["missing_path"] if params["missing_path"] else os.path.join(self.default_dir, f"{self.mapping_name}_missing.yml") self.asset_folders = params["asset_folders"] self.create_asset_folders = params["create_asset_folders"] self.dimensional_asset_rename = params["dimensional_asset_rename"] self.download_url_assets = params["download_url_assets"] self.show_missing_season_assets = params["show_missing_season_assets"] self.show_missing_episode_assets = params["show_missing_episode_assets"] self.show_asset_not_needed = params["show_asset_not_needed"] self.sync_mode = params["sync_mode"] self.default_collection_order = params["default_collection_order"] self.minimum_items = params["minimum_items"] self.item_refresh_delay = params["item_refresh_delay"] self.delete_below_minimum = params["delete_below_minimum"] self.delete_not_scheduled = params["delete_not_scheduled"] self.missing_only_released = params["missing_only_released"] self.show_unmanaged = params["show_unmanaged"] self.show_filtered = params["show_filtered"] self.show_options = params["show_options"] self.show_missing = params["show_missing"] self.show_missing_assets = params["show_missing_assets"] self.save_missing = params["save_missing"] self.only_filter_missing = params["only_filter_missing"] self.ignore_ids = params["ignore_ids"] self.ignore_imdb_ids = params["ignore_imdb_ids"] self.assets_for_all = params["assets_for_all"] self.delete_unmanaged_collections = params["delete_unmanaged_collections"] self.delete_collections_with_less = params["delete_collections_with_less"] self.mass_genre_update = params["mass_genre_update"] self.mass_audience_rating_update = params["mass_audience_rating_update"] self.mass_critic_rating_update = params["mass_critic_rating_update"] self.mass_content_rating_update = params["mass_content_rating_update"] self.mass_originally_available_update = params["mass_originally_available_update"] self.mass_imdb_parental_labels = params["mass_imdb_parental_labels"] self.mass_trakt_rating_update = params["mass_trakt_rating_update"] self.radarr_add_all_existing = params["radarr_add_all_existing"] self.radarr_remove_by_tag = params["radarr_remove_by_tag"] self.sonarr_add_all_existing = params["sonarr_add_all_existing"] self.sonarr_remove_by_tag = params["sonarr_remove_by_tag"] self.update_blank_track_titles = params["update_blank_track_titles"] self.remove_title_parentheses = params["remove_title_parentheses"] self.remove_overlays = params["remove_overlays"] self.mass_collection_mode = params["mass_collection_mode"] self.metadata_backup = params["metadata_backup"] self.genre_mapper = params["genre_mapper"] self.content_rating_mapper = params["content_rating_mapper"] self.error_webhooks = params["error_webhooks"] self.changes_webhooks = params["changes_webhooks"] self.split_duplicates = params["split_duplicates"] # TODO: Here or just in Plex? self.clean_bundles = params["plex"]["clean_bundles"] # TODO: Here or just in Plex? self.empty_trash = params["plex"]["empty_trash"] # TODO: Here or just in Plex? self.optimize = params["plex"]["optimize"] # TODO: Here or just in Plex? self.stats = {"created": 0, "modified": 0, "deleted": 0, "added": 0, "unchanged": 0, "removed": 0, "radarr": 0, "sonarr": 0, "names": []} self.status = {} self.items_library_operation = True if self.assets_for_all or self.mass_genre_update or self.mass_audience_rating_update or self.remove_title_parentheses \ or self.mass_critic_rating_update or self.mass_content_rating_update or self.mass_originally_available_update or self.mass_imdb_parental_labels or self.mass_trakt_rating_update \ or self.genre_mapper or self.content_rating_mapper or self.radarr_add_all_existing or self.sonarr_add_all_existing else False self.library_operation = True if self.items_library_operation or self.delete_unmanaged_collections or self.delete_collections_with_less \ or self.radarr_remove_by_tag or self.sonarr_remove_by_tag or self.mass_collection_mode \ or self.show_unmanaged or self.metadata_backup or self.update_blank_track_titles else False self.meta_operations = [self.mass_genre_update, self.mass_audience_rating_update, self.mass_critic_rating_update, self.mass_content_rating_update, self.mass_originally_available_update] if self.asset_directory: logger.info("") for ad in self.asset_directory: logger.info(f"Using Asset Directory: {ad}") if output: logger.info("") logger.info(output) def scan_files(self): for file_type, metadata_file, temp_vars, asset_directory in self.metadata_path: try: meta_obj = MetadataFile(self.config, self, file_type, metadata_file, temp_vars, asset_directory) if meta_obj.collections: self.collections.extend([c for c in meta_obj.collections]) if meta_obj.metadata: self.metadatas.extend([c for c in meta_obj.metadata]) self.metadata_files.append(meta_obj) except Failed as e: logger.error(e) for file_type, overlay_file, temp_vars, asset_directory in self.overlay_path: try: over_obj = OverlayFile(self.config, self, file_type, overlay_file, temp_vars, asset_directory) self.overlays.extend([o.lower() for o in over_obj.overlays]) self.overlay_files.append(over_obj) except Failed as e: logger.error(e) def upload_images(self, item, poster=None, background=None): image = None image_compare = None poster_uploaded = False if self.config.Cache: image, image_compare, _ = self.config.Cache.query_image_map(item.ratingKey, self.image_table_name) if poster is not None: try: if image_compare and str(poster.compare) != str(image_compare): image = None if image is None or image != item.thumb: self._upload_image(item, poster) poster_uploaded = True logger.info(f"Detail: {poster.attribute} updated {poster.message}") elif self.show_asset_not_needed: logger.info(f"Detail: {poster.prefix}poster update not needed") except Failed: logger.stacktrace() logger.error(f"Detail: {poster.attribute} failed to update {poster.message}") background_uploaded = False if background is not None: try: image = None if self.config.Cache: image, image_compare, _ = self.config.Cache.query_image_map(item.ratingKey, f"{self.image_table_name}_backgrounds") if str(background.compare) != str(image_compare): image = None if image is None or image != item.art: self._upload_image(item, background) background_uploaded = True logger.info(f"Detail: {background.attribute} updated {background.message}") elif self.show_asset_not_needed: logger.info(f"Detail: {background.prefix}background update not needed") except Failed: logger.stacktrace() logger.error(f"Detail: {background.attribute} failed to update {background.message}") if self.config.Cache: if poster_uploaded: self.config.Cache.update_image_map(item.ratingKey, self.image_table_name, item.thumb, poster.compare if poster else "") if background_uploaded: self.config.Cache.update_image_map(item.ratingKey, f"{self.image_table_name}_backgrounds", item.art, background.compare) @abstractmethod def notify(self, text, collection=None, critical=True): pass @abstractmethod def _upload_image(self, item, image): pass @abstractmethod def upload_poster(self, item, image, url=False): pass @abstractmethod def reload(self, item, force=False): pass @abstractmethod def edit_tags(self, attr, obj, add_tags=None, remove_tags=None, sync_tags=None, do_print=True): pass @abstractmethod def get_all(self, collection_level=None, load=False): pass def find_assets(self, name="poster", folder_name=None, item_directory=None, prefix=""): poster = None background = None item_dir = None search_dir = item_directory if item_directory else None for ad in self.asset_directory: item_dir = None if not search_dir: search_dir = ad if folder_name: if os.path.isdir(os.path.join(ad, folder_name)): item_dir = os.path.join(ad, folder_name) else: for n in range(1, self.asset_depth + 1): new_path = ad for i in range(1, n + 1): new_path = os.path.join(new_path, "*") matches = util.glob_filter(os.path.join(new_path, folder_name)) if len(matches) > 0: item_dir = os.path.abspath(matches[0]) break if item_dir is None: continue search_dir = item_dir if item_directory: item_dir = item_directory file_name = name if item_dir else f"{folder_name}_{name}" poster_filter = os.path.join(search_dir, f"{file_name}.*") background_filter = os.path.join(search_dir, "background.*" if file_name == "poster" else f"{file_name}_background.*") poster_matches = util.glob_filter(poster_filter) if len(poster_matches) > 0: poster = ImageData("asset_directory", os.path.abspath(poster_matches[0]), prefix=prefix, is_url=False) background_matches = util.glob_filter(background_filter) if len(background_matches) > 0: background = ImageData("asset_directory", os.path.abspath(background_matches[0]), prefix=prefix, is_poster=False, is_url=False) break return poster, background, item_dir def add_missing(self, collection, items, is_movie): if collection not in self.missing: self.missing[collection] = {} section = "Movies Missing (TMDb IDs)" if is_movie else "Shows Missing (TVDb IDs)" if section not in self.missing[collection]: self.missing[collection][section] = {} for title, item_id in items: self.missing[collection][section][int(item_id)] = title with open(self.missing_path, "w"): pass try: yaml.round_trip_dump(self.missing, open(self.missing_path, "w", encoding="utf-8")) except yaml.scanner.ScannerError as e: logger.error(f"YAML Error: {util.tab_new_lines(e)}") def cache_items(self): logger.info("") logger.separator(f"Caching {self.name} Library Items", space=False, border=False) logger.info("") items = self.get_all() for item in items: self.cached_items[item.ratingKey] = (item, False) return items def map_guids(self, items): logger.separator(f"Mapping {self.type} Library: {self.name}", space=False, border=False) logger.info("") for i, item in enumerate(items, 1): logger.ghost(f"Processing: {i}/{len(items)} {item.title}") if item.ratingKey not in self.movie_rating_key_map and item.ratingKey not in self.show_rating_key_map: id_type, main_id, imdb_id = self.config.Convert.get_id(item, self) if main_id: if id_type == "movie": self.movie_rating_key_map[item.ratingKey] = main_id[0] util.add_dict_list(main_id, item.ratingKey, self.movie_map) elif id_type == "show": self.show_rating_key_map[item.ratingKey] = main_id[0] util.add_dict_list(main_id, item.ratingKey, self.show_map) if imdb_id: util.add_dict_list(imdb_id, item.ratingKey, self.imdb_map) logger.info("") logger.info(f"Processed {len(items)} {self.type}s")