From e8cc6a6920462f154aa38b7930a2de7ab1cf1feb Mon Sep 17 00:00:00 2001 From: meisnate12 Date: Fri, 3 Feb 2023 10:42:27 -0500 Subject: [PATCH] [57] update overlay removal/reset process --- VERSION | 2 +- defaults/overlays/runtimes.yml | 4 -- modules/library.py | 37 +++++++++++++++- modules/overlays.py | 77 +++++++--------------------------- modules/plex.py | 43 +++++++++++++++++++ 5 files changed, 95 insertions(+), 68 deletions(-) diff --git a/VERSION b/VERSION index 29ed74f7..7b2e4a33 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.18.3-develop56 +1.18.3-develop57 diff --git a/defaults/overlays/runtimes.yml b/defaults/overlays/runtimes.yml index 3a507447..7ad92dfe 100644 --- a/defaults/overlays/runtimes.yml +++ b/defaults/overlays/runtimes.yml @@ -13,10 +13,6 @@ external_templates: horizontal_align: right vertical_align: bottom conditionals: - overlay_level: - conditions: - - library_type: show - value: episode vertical_offset: default: 15 conditions: diff --git a/modules/library.py b/modules/library.py index 8b7e6b8e..cceccf8c 100644 --- a/modules/library.py +++ b/modules/library.py @@ -1,9 +1,10 @@ -import os +import os, time from abc import ABC, abstractmethod from modules import util, operations from modules.meta import MetadataFile, OverlayFile from modules.operations import Operations from modules.util import Failed, NotScheduled, YAML +from PIL import Image, ImageFilter logger = util.logger @@ -249,6 +250,40 @@ class Library(ABC): def item_labels(self, item): pass + @abstractmethod + def find_poster_url(self, item): + pass + + def check_image_for_overlay(self, image_url, image_path, remove=False): + image_response = self.config.get(image_url) + if image_response.status_code >= 400: + raise Failed("Image Download Failed") + if image_response.headers["Content-Type"] not in ["image/png", "image/jpeg", "image/webp"]: + raise Failed("Image Not PNG, JPG, or WEBP") + if image_response.headers["Content-Type"] == "image/jpeg": + image_path += ".jpg" + elif image_response.headers["Content-Type"] == "image/webp": + image_path += ".webp" + else: + image_path += ".png" + with open(image_path, "wb") as handler: + handler.write(image_response.content) + while util.is_locked(image_path): + time.sleep(1) + with Image.open(image_path) as image: + exif_tags = image.getexif() + if 0x04bc in exif_tags and exif_tags[0x04bc] == "overlay": + os.remove(image_path) + raise Failed("Poster already has an Overlay") + if remove: + os.remove(image_path) + else: + return image_path + + @abstractmethod + def item_posters(self, item): + pass + @abstractmethod def get_all(self, builder_level=None, load=False): pass diff --git a/modules/overlays.py b/modules/overlays.py index 9a428ec6..b2d53865 100644 --- a/modules/overlays.py +++ b/modules/overlays.py @@ -1,4 +1,4 @@ -import os, re, time +import os, re from datetime import datetime from modules import plex, util, overlay from modules.builder import CollectionBuilder @@ -170,53 +170,19 @@ class Overlays: reset_list = ["plex", "tmdb"] else: reset_list = [] - reset_attempted = False - for reset in reset_list: - if reset == "plex": - reset_attempted = True - temp_poster = next((p for p in item.posters()), None) - if temp_poster: - new_backup = f"{self.library.url}{temp_poster.key}&X-Plex-Token={self.library.token}" - break - else: - logger.trace("Plex Error: Plex Poster Download Failed") - if reset == "tmdb": - reset_attempted = True - try: - new_backup = self.find_poster_url(item) - break - except Failed as e: - logger.trace(e) - if reset_attempted and not new_backup: - logger.error("Overlay Error: Reset Failed") + try: + new_backup = self.library.item_posters(item) + except Failed as e: + if any(r in reset_list for r in ["plex", "tmdb"]): + logger.error(e) else: new_backup = item.posterUrl if new_backup: changed_image = True - image_response = self.config.get(new_backup) - if image_response.status_code >= 400: - raise Failed(f"{item_title[:60]:<60} | Overlay Error: Image Download Failed") - if image_response.headers["Content-Type"] not in ["image/png", "image/jpeg", "image/webp"]: - raise Failed(f"{item_title[:60]:<60} | Overlay Error: Image Not PNG, JPG, or WEBP") - if image_response.headers["Content-Type"] == "image/jpeg": - i_ext = "jpg" - elif image_response.headers["Content-Type"] == "image/webp": - i_ext = "webp" - else: - i_ext = "png" - backup_image_path = os.path.join(self.library.overlay_backup, f"{item.ratingKey}.{i_ext}") - with open(backup_image_path, "wb") as handler: - handler.write(image_response.content) - while util.is_locked(backup_image_path): - time.sleep(1) - backup_poster = Image.open(backup_image_path) - exif_tags = backup_poster.getexif() - if 0x04bc in exif_tags and exif_tags[0x04bc] == "overlay": - logger.error(f"{item_title[:60]:<60} | Overlay Backup Error: Poster already has an Overlay") - os.remove(backup_image_path) - else: - has_original = backup_image_path - + try: + has_original = self.library.check_image_for_overlay(new_backup, os.path.join(self.library.overlay_backup, f"{item.ratingKey}")) + except Failed as e: + raise Failed(f"{item_title[:60]:<60} | Overlay Error: {e}") poster_compare = None if poster is None and has_original is None: logger.error(f"{item_title[:60]:<60} | Overlay Error: No poster found") @@ -387,6 +353,8 @@ class Overlays: except Failed as e: logger.error(f"{e}\nOverlays Attempted on {item_title}: {', '.join(over_names)}") except Exception as e: + logger.info(e) + logger.info(type(e)) logger.stacktrace(e) logger.error("") logger.error(f"Overlays Attempted on {item_title}: {', '.join(over_names)}") @@ -509,21 +477,6 @@ class Overlays: key_to_overlays[over_key][1].remove(v) return key_to_overlays, properties - def find_poster_url(self, item): - if isinstance(item, Movie): - if item.ratingKey in self.library.movie_rating_key_map: - return self.config.TMDb.get_movie(self.library.movie_rating_key_map[item.ratingKey]).poster_url - elif isinstance(item, (Show, Season, Episode)): - check_key = item.ratingKey if isinstance(item, Show) else item.show().ratingKey - if check_key in self.library.show_rating_key_map: - tmdb_id = self.config.Convert.tvdb_to_tmdb(self.library.show_rating_key_map[check_key]) - if isinstance(item, Show) and item.ratingKey in self.library.show_rating_key_map: - return self.config.TMDb.get_show(tmdb_id).poster_url - elif isinstance(item, Season): - return self.config.TMDb.get_season(tmdb_id, item.seasonNumber).poster_url - elif isinstance(item, Episode): - return self.config.TMDb.get_episode(tmdb_id, item.seasonNumber, item.episodeNumber).still_url - def get_overlay_items(self, label="Overlay", libtype=None, ignore=None): items = self.library.search(label=label, libtype=libtype) return items if not ignore else [o for o in items if o.ratingKey not in ignore] @@ -542,9 +495,9 @@ class Overlays: if not poster_location: is_url = True try: - poster_location = self.find_poster_url(item) - except Failed as e: - logger.error(e) + poster_location = self.library.item_posters(item) + except Failed: + pass if poster_location: self.library.upload_poster(item, poster_location, url=is_url) self.library.edit_tags("label", item, remove_tags=[label], do_print=False) diff --git a/modules/plex.py b/modules/plex.py index d52b5ae4..a23aa0d2 100644 --- a/modules/plex.py +++ b/modules/plex.py @@ -595,6 +595,49 @@ class Plex(Library): except BadRequest: raise Failed(f"Item: {item.title} Labels failed to load") + def find_poster_url(self, item): + if isinstance(item, Movie): + if item.ratingKey in self.movie_rating_key_map: + return self.config.TMDb.get_movie(self.movie_rating_key_map[item.ratingKey]).poster_url + elif isinstance(item, (Show, Season, Episode)): + check_key = item.ratingKey if isinstance(item, Show) else item.show().ratingKey + if check_key in self.show_rating_key_map: + tmdb_id = self.config.Convert.tvdb_to_tmdb(self.show_rating_key_map[check_key]) + if isinstance(item, Show) and item.ratingKey in self.show_rating_key_map: + return self.config.TMDb.get_show(tmdb_id).poster_url + elif isinstance(item, Season): + return self.config.TMDb.get_season(tmdb_id, item.seasonNumber).poster_url + elif isinstance(item, Episode): + return self.config.TMDb.get_episode(tmdb_id, item.seasonNumber, item.episodeNumber).still_url + + def item_posters(self, item, providers=None): + if not providers: + providers = ["plex", "tmdb"] + image_url = None + for provider in providers: + if provider == "plex": + for poster in item.posters(): + if poster.key.startswith("/"): + image_url = f"{self.url}{poster.key}&X-Plex-Token={self.token}" + if poster.ratingKey.startswith("upload"): + try: + self.check_image_for_overlay(image_url, os.path.join(self.overlay_backup, "temp"), remove=True) + except Failed as e: + logger.trace(f"Plex Error: {e}") + continue + break + if provider == "tmdb": + try: + image_url = self.find_poster_url(item) + except Failed as e: + logger.trace(e) + continue + if image_url: + break + if not image_url: + raise Failed("Overlay Error: No Poster found to reset") + return image_url + @retry(stop_max_attempt_number=6, wait_fixed=10000, retry_on_exception=util.retry_if_not_plex) def reload(self, item, force=False): is_full = False