Mirrors/Plex-Meta-Manager
2
0
Fork 0
mirror of https://github.com/meisnate12/Plex-Meta-Manager synced 2024-11-27 15:01:00 +00:00
Code Issues Projects Releases Packages Wiki Activity
Plex-Meta-Manager/modules/builder.py
meisnate12 de95684532 None fix
2021-05-26 00:35:54 -04:00

1792 lines
121 KiB
Python
Raw Blame History

import logging, os, re
from datetime import datetime, timedelta
from modules import anidb, anilist, imdb, letterboxd, mal, plex, radarr, sonarr, tautulli, tmdb, trakttv, tvdb, util
from modules.util import Failed
from plexapi.exceptions import BadRequest, NotFound
from plexapi.video import Movie, Show
from urllib.parse import quote
logger = logging.getLogger("Plex Meta Manager")
string_filters = ["title", "episode_title", "studio"]
image_file_details = ["file_poster", "file_background", "asset_directory"]
advance_new_agent = ["item_metadata_language", "item_use_original_title"]
advance_show = ["item_episode_sorting", "item_keep_episodes", "item_delete_episodes", "item_season_display", "item_episode_sorting"]
method_alias = {
"actors": "actor", "role": "actor", "roles": "actor",
"collections": "collecion", "plex_collection": "collection",
"content_ratings": "content_rating", "contentRating": "content_rating", "contentRatings": "content_rating",
"countries": "country",
"decades": "decade",
"directors": "director",
"genres": "genre",
"labels": "label",
"rating": "critic_rating",
"studios": "studio",
"networks": "network",
"producers": "producer",
"writers": "writer",
"years": "year"
}
filter_translation = {
"actor": "actors",
"audience_rating": "audienceRating",
"collection": "collections",
"content_rating": "contentRating",
"country": "countries",
"critic_rating": "rating",
"director": "directors",
"genre": "genres",
"originally_available": "originallyAvailableAt",
"tmdb_vote_count": "vote_count",
"user_rating": "userRating",
"writer": "writers"
}
modifier_alias = {".greater": ".gt", ".less": ".lt"}
all_builders = anidb.builders + anilist.builders + imdb.builders + letterboxd.builders + mal.builders + plex.builders + tautulli.builders + tmdb.builders + trakttv.builders + tvdb.builders
dictionary_builders = [
"filters",
"anilist_genre",
"anilist_season",
"anilist_tag",
"mal_season",
"mal_userlist",
"plex_collectionless",
"plex_search",
"tautulli_popular",
"tautulli_watched",
"tmdb_discover"
]
show_only_builders = [
"tmdb_network",
"tmdb_show",
"tmdb_show_details",
"tvdb_show",
"tvdb_show_details"
]
movie_only_builders = [
"letterboxd_list",
"letterboxd_list_details",
"tmdb_collection",
"tmdb_collection_details",
"tmdb_movie",
"tmdb_movie_details",
"tmdb_now_playing",
"tvdb_movie",
"tvdb_movie_details"
]
numbered_builders = [
"anidb_popular",
"anilist_popular",
"anilist_top_rated",
"mal_all",
"mal_airing",
"mal_upcoming",
"mal_tv",
"mal_ova",
"mal_movie",
"mal_special",
"mal_popular",
"mal_favorite",
"mal_suggested",
"tmdb_popular",
"tmdb_top_rated",
"tmdb_now_playing",
"tmdb_trending_daily",
"tmdb_trending_weekly",
"trakt_trending",
"trakt_popular",
"trakt_recommended",
"trakt_watched",
"trakt_collected"
]
smart_collection_invalid = ["collection_mode", "collection_order"]
smart_url_collection_invalid = [
"run_again", "sync_mode", "show_filtered", "show_missing", "save_missing", "smart_label",
"radarr_add", "radarr_folder", "radarr_monitor", "radarr_availability",
"radarr_quality", "radarr_tag", "radarr_search",
"sonarr_add", "sonarr_folder", "sonarr_monitor", "sonarr_quality", "sonarr_language",
"sonarr_series", "sonarr_season", "sonarr_tag", "sonarr_search", "sonarr_cutoff_search",
"filters"
]
all_details = [
"sort_title", "content_rating", "collection_mode", "collection_order",
"summary", "tmdb_summary", "tmdb_description", "tmdb_biography", "tvdb_summary",
"tvdb_description", "trakt_description", "letterboxd_description",
"url_poster", "tmdb_poster", "tmdb_profile", "tvdb_poster", "file_poster",
"url_background", "tmdb_background", "tvdb_background", "file_background",
"name_mapping", "label", "show_filtered", "show_missing", "save_missing"
]
collectionless_details = [
"sort_title", "content_rating",
"summary", "tmdb_summary", "tmdb_description", "tmdb_biography",
"collection_order", "plex_collectionless",
"url_poster", "tmdb_poster", "tmdb_profile", "file_poster",
"url_background", "file_background",
"name_mapping", "label", "label_sync_mode", "test"
]
ignored_details = [
"smart_filter",
"smart_label",
"smart_url",
"run_again",
"schedule",
"sync_mode",
"template",
"test",
"tmdb_person"
]
boolean_details = [
"show_filtered",
"show_missing",
"save_missing"
]
all_filters = [
"actor", "actor.not",
"audio_language", "audio_language.not",
"audio_track_title", "audio_track_title.not",
"collection", "collection.not",
"content_rating", "content_rating.not",
"country", "country.not",
"director", "director.not",
"filepath", "filepath.not",
"genre", "genre.not",
"max_age",
"originally_available.gte", "originally_available.lte",
"tmdb_vote_count.gte", "tmdb_vote_count.lte",
"duration.gte", "duration.lte",
"original_language", "original_language.not",
"user_rating.gte", "user_rating.lte",
"audience_rating.gte", "audience_rating.lte",
"critic_rating.gte", "critic_rating.lte",
"studio", "studio.not",
"subtitle_language", "subtitle_language.not",
"video_resolution", "video_resolution.not",
"writer", "writer.not",
"year", "year.gte", "year.lte", "year.not"
]
movie_only_filters = [
"audio_language", "audio_language.not",
"audio_track_title", "audio_track_title.not",
"country", "country.not",
"director", "director.not",
"duration.gte", "duration.lte",
"original_language", "original_language.not",
"subtitle_language", "subtitle_language.not",
"video_resolution", "video_resolution.not",
"writer", "writer.not"
]
def _split(text):
attribute, modifier = os.path.splitext(str(text).lower())
attribute = method_alias[attribute] if attribute in method_alias else attribute
modifier = modifier_alias[modifier] if modifier in modifier_alias else modifier
final = f"{attribute}{modifier}"
if text != final:
logger.warning(f"Collection Warning: {text} plex search attribute will run as {final}")
return attribute, modifier, final
class CollectionBuilder:
def __init__(self, config, library, metadata, name, data):
self.config = config
self.library = library
self.metadata = metadata
self.name = name
self.data = data
self.details = {
"show_filtered": library.show_filtered,
"show_missing": library.show_missing,
"save_missing": library.save_missing
}
self.item_details = {}
self.radarr_options = {}
self.sonarr_options = {}
self.missing_movies = []
self.missing_shows = []
self.methods = []
self.filters = []
self.rating_keys = []
self.run_again_movies = []
self.run_again_shows = []
self.posters = {}
self.backgrounds = {}
self.summaries = {}
self.schedule = ""
self.add_to_radarr = None
self.add_to_sonarr = None
current_time = datetime.now()
current_year = current_time.year
methods = {m.lower(): m for m in self.data}
if "template" in methods:
logger.info("")
logger.info("Validating Method: template")
if not self.metadata.templates:
raise Failed("Collection Error: No templates found")
elif not self.data[methods["template"]]:
raise Failed("Collection Error: template attribute is blank")
else:
logger.debug(f"Value: {self.data[methods['template']]}")
for variables in util.get_list(self.data[methods["template"]], split=False):
if not isinstance(variables, dict):
raise Failed("Collection Error: template attribute is not a dictionary")
elif "name" not in variables:
raise Failed("Collection Error: template sub-attribute name is required")
elif not variables["name"]:
raise Failed("Collection Error: template sub-attribute name is blank")
elif variables["name"] not in self.metadata.templates:
raise Failed(f"Collection Error: template {variables['name']} not found")
elif not isinstance(self.metadata.templates[variables["name"]], dict):
raise Failed(f"Collection Error: template {variables['name']} is not a dictionary")
else:
for tm in variables:
if not variables[tm]:
raise Failed(f"Collection Error: template sub-attribute {tm} is blank")
if "collection_name" not in variables:
variables["collection_name"] = str(self.name)
template_name = variables["name"]
template = self.metadata.templates[template_name]
default = {}
if "default" in template:
if template["default"]:
if isinstance(template["default"], dict):
for dv in template["default"]:
if template["default"][dv]:
default[dv] = template["default"][dv]
else:
raise Failed(f"Collection Error: template default sub-attribute {dv} is blank")
else:
raise Failed("Collection Error: template sub-attribute default is not a dictionary")
else:
raise Failed("Collection Error: template sub-attribute default is blank")
optional = []
if "optional" in template:
if template["optional"]:
if isinstance(template["optional"], list):
for op in template["optional"]:
if op not in default:
optional.append(op)
else:
logger.warning(f"Template Warning: variable {op} cannot be optional if it has a default")
else:
optional.append(str(template["optional"]))
else:
raise Failed("Collection Error: template sub-attribute optional is blank")
def check_data(_data):
if isinstance(_data, dict):
final_data = {}
for sm, sd in _data.items():
try:
final_data[sm] = check_data(sd)
except Failed:
continue
elif isinstance(_data, list):
final_data = []
for li in _data:
try:
final_data.append(check_data(li))
except Failed:
continue
else:
txt = str(_data)
def scan_text(og_txt, var, var_value):
if og_txt == f"<<{var}>>":
return str(var_value)
elif f"<<{var}>>" in str(og_txt):
return str(og_txt).replace(f"<<{var}>>", str(var_value))
else:
return og_txt
for option in optional:
if option not in variables and f"<<{option}>>" in txt:
raise Failed
for variable, variable_data in variables.items():
if variable != "name":
txt = scan_text(txt, variable, variable_data)
for dm, dd in default.items():
txt = scan_text(txt, dm, dd)
if txt in ["true", "True"]:
final_data = True
elif txt in ["false", "False"]:
final_data = False
else:
try:
num_data = float(txt)
final_data = int(num_data) if num_data.is_integer() else num_data
except (ValueError, TypeError):
final_data = txt
return final_data
for method_name, attr_data in template.items():
if method_name not in self.data and method_name not in ["default", "optional"]:
if attr_data is None:
logger.error(f"Template Error: template attribute {method_name} is blank")
continue
try:
self.data[method_name] = check_data(attr_data)
methods[method_name.lower()] = method_name
except Failed:
continue
if "schedule" in methods:
logger.info("")
logger.info("Validating Method: schedule")
if not self.data[methods["schedule"]]:
raise Failed("Collection Error: schedule attribute is blank")
else:
logger.debug(f"Value: {self.data[methods['schedule']]}")
skip_collection = True
schedule_list = util.get_list(self.data[methods["schedule"]])
next_month = current_time.replace(day=28) + timedelta(days=4)
last_day = next_month - timedelta(days=next_month.day)
for schedule in schedule_list:
run_time = str(schedule).lower()
if run_time.startswith(("day", "daily")):
skip_collection = False
elif run_time.startswith(("hour", "week", "month", "year")):
match = re.search("\\(([^)]+)\\)", run_time)
if not match:
logger.error(f"Collection Error: failed to parse schedule: {schedule}")
continue
param = match.group(1)
if run_time.startswith("hour"):
try:
if 0 <= int(param) <= 23:
self.schedule += f"\nScheduled to run only on the {util.make_ordinal(param)} hour"
if config.run_hour == int(param):
skip_collection = False
else:
raise ValueError
except ValueError:
logger.error(f"Collection Error: hourly schedule attribute {schedule} invalid must be an integer between 0 and 23")
elif run_time.startswith("week"):
if param.lower() not in util.days_alias:
logger.error(f"Collection Error: weekly schedule attribute {schedule} invalid must be a day of the week i.e. weekly(Monday)")
continue
weekday = util.days_alias[param.lower()]
self.schedule += f"\nScheduled weekly on {util.pretty_days[weekday]}"
if weekday == current_time.weekday():
skip_collection = False
elif run_time.startswith("month"):
try:
if 1 <= int(param) <= 31:
self.schedule += f"\nScheduled monthly on the {util.make_ordinal(param)}"
if current_time.day == int(param) or (current_time.day == last_day.day and int(param) > last_day.day):
skip_collection = False
else:
raise ValueError
except ValueError:
logger.error(f"Collection Error: monthly schedule attribute {schedule} invalid must be an integer between 1 and 31")
elif run_time.startswith("year"):
match = re.match("^(1[0-2]|0?[1-9])/(3[01]|[12][0-9]|0?[1-9])$", param)
if not match:
logger.error(f"Collection Error: yearly schedule attribute {schedule} invalid must be in the MM/DD format i.e. yearly(11/22)")
continue
month = int(match.group(1))
day = int(match.group(2))
self.schedule += f"\nScheduled yearly on {util.pretty_months[month]} {util.make_ordinal(day)}"
if current_time.month == month and (current_time.day == day or (current_time.day == last_day.day and day > last_day.day)):
skip_collection = False
else:
logger.error(f"Collection Error: schedule attribute {schedule} invalid")
if len(self.schedule) == 0:
skip_collection = False
if skip_collection:
raise Failed(f"{self.schedule}\n\nCollection {self.name} not scheduled to run")
self.run_again = "run_again" in methods
self.collectionless = "plex_collectionless" in methods
self.run_again = False
if "run_again" in methods:
logger.info("")
logger.info("Validating Method: run_again")
if not self.data[methods["run_again"]]:
logger.warning(f"Collection Warning: run_again attribute is blank defaulting to false")
else:
logger.debug(f"Value: {self.data[methods['run_again']]}")
self.run_again = util.get_bool("run_again", self.data[methods["run_again"]])
self.sync = self.library.sync_mode == "sync"
if "sync_mode" in methods:
logger.info("")
logger.info("Validating Method: sync_mode")
if not self.data[methods["sync_mode"]]:
logger.warning(f"Collection Warning: sync_mode attribute is blank using general: {self.library.sync_mode}")
else:
logger.debug(f"Value: {self.data[methods['sync_mode']]}")
if self.data[methods["sync_mode"]].lower() not in ["append", "sync"]:
logger.warning(f"Collection Warning: {self.data[methods['sync_mode']]} sync_mode invalid using general: {self.library.sync_mode}")
else:
self.sync = self.data[methods["sync_mode"]].lower() == "sync"
self.build_collection = True
if "build_collection" in methods:
logger.info("")
logger.info("Validating Method: build_collection")
if not self.data[methods["build_collection"]]:
logger.warning(f"Collection Warning: build_collection attribute is blank defaulting to true")
else:
logger.debug(f"Value: {self.data[methods['build_collection']]}")
self.build_collection = util.get_bool("build_collection", self.data[methods["build_collection"]])
if "tmdb_person" in methods:
logger.info("")
logger.info("Validating Method: build_collection")
if not self.data[methods["tmdb_person"]]:
raise Failed("Collection Error: tmdb_person attribute is blank")
else:
logger.debug(f"Value: {self.data[methods['tmdb_person']]}")
valid_names = []
for tmdb_id in util.get_int_list(self.data[methods["tmdb_person"]], "TMDb Person ID"):
person = config.TMDb.get_person(tmdb_id)
valid_names.append(person.name)
if hasattr(person, "biography") and person.biography:
self.summaries["tmdb_person"] = person.biography
if hasattr(person, "profile_path") and person.profile_path:
self.posters["tmdb_person"] = f"{config.TMDb.image_url}{person.profile_path}"
if len(valid_names) > 0:
self.details["tmdb_person"] = valid_names
else:
raise Failed(f"Collection Error: No valid TMDb Person IDs in {self.data[methods['tmdb_person']]}")
self.smart_sort = "random"
self.smart_label_collection = False
if "smart_label" in methods:
logger.info("")
logger.info("Validating Method: smart_label")
self.smart_label_collection = True
if not self.data[methods["smart_label"]]:
logger.warning("Collection Error: smart_label attribute is blank defaulting to random")
else:
logger.debug(f"Value: {self.data[methods['smart_label']]}")
if (self.library.is_movie and str(self.data[methods["smart_label"]]).lower() in plex.movie_smart_sorts) \
or (self.library.is_show and str(self.data[methods["smart_label"]]).lower() in plex.show_smart_sorts):
self.smart_sort = str(self.data[methods["smart_label"]]).lower()
else:
logger.warning(f"Collection Error: smart_label attribute: {self.data[methods['smart_label']]} is invalid defaulting to random")
self.smart_url = None
self.smart_type_key = None
if "smart_url" in methods:
logger.info("")
logger.info("Validating Method: smart_url")
if not self.data[methods["smart_url"]]:
raise Failed("Collection Error: smart_url attribute is blank")
else:
logger.debug(f"Value: {self.data[methods['smart_url']]}")
try:
self.smart_url, self.smart_type_key = library.get_smart_filter_from_uri(self.data[methods["smart_url"]])
except ValueError:
raise Failed("Collection Error: smart_url is incorrectly formatted")
self.smart_filter_details = ""
if "smart_filter" in methods:
logger.info("")
logger.info("Validating Method: smart_filter")
filter_details = "\n"
smart_filter = self.data[methods["smart_filter"]]
if smart_filter is None:
raise Failed(f"Collection Error: smart_filter attribute is blank")
if not isinstance(smart_filter, dict):
raise Failed(f"Collection Error: smart_filter must be a dictionary: {smart_filter}")
logger.debug(f"Value: {self.data[methods['smart_filter']]}")
smart_methods = {m.lower(): m for m in smart_filter}
if "any" in smart_methods and "all" in smart_methods:
raise Failed(f"Collection Error: Cannot have more then one base")
if "any" not in smart_methods and "all" not in smart_methods:
raise Failed(f"Collection Error: Must have either any or all as a base for the filter")
if "type" in smart_methods and self.library.is_show:
if smart_filter[smart_methods["type"]] not in ["shows", "seasons", "episodes"]:
raise Failed(f"Collection Error: type: {smart_filter[smart_methods['type']]} is invalid, must be either shows, season, or episodes")
smart_type = smart_filter[smart_methods["type"]]
elif self.library.is_show:
smart_type = "shows"
else:
smart_type = "movies"
filter_details += f"Smart {smart_type.capitalize()[:-1]} Filter\n"
self.smart_type_key, smart_sorts = plex.smart_types[smart_type]
smart_sort = "random"
if "sort_by" in smart_methods:
if smart_filter[smart_methods["sort_by"]] is None:
raise Failed(f"Collection Error: sort_by attribute is blank")
if smart_filter[smart_methods["sort_by"]] not in smart_sorts:
raise Failed(f"Collection Error: sort_by: {smart_filter[smart_methods['sort_by']]} is invalid")
smart_sort = smart_filter[smart_methods["sort_by"]]
filter_details += f"Sort By: {smart_sort}\n"
limit = None
if "limit" in smart_methods:
if smart_filter[smart_methods["limit"]] is None:
raise Failed("Collection Error: limit attribute is blank")
if not isinstance(smart_filter[smart_methods["limit"]], int) or smart_filter[smart_methods["limit"]] < 1:
raise Failed("Collection Error: limit attribute must be an integer greater then 0")
limit = smart_filter[smart_methods["limit"]]
filter_details += f"Limit: {limit}\n"
validate = True
if "validate" in smart_methods:
if smart_filter[smart_methods["validate"]] is None:
raise Failed("Collection Error: validate attribute is blank")
if not isinstance(smart_filter[smart_methods["validate"]], bool):
raise Failed("Collection Error: validate attribute must be either true or false")
validate = smart_filter[smart_methods["validate"]]
filter_details += f"Validate: {validate}\n"
def _filter(filter_dict, fail, is_all=True, level=1):
output = ""
display = f"\n{' ' * level}Match {'all' if is_all else 'any'} of the following:"
level += 1
indent = f"\n{' ' * level}"
conjunction = f"{'and' if is_all else 'or'}=1&"
for smart_key, smart_data in filter_dict.items():
smart, smart_mod, smart_final = _split(smart_key)
def build_url_arg(arg, mod=None, arg_s=None, mod_s=None, param_s=None):
arg_key = plex.search_translation[smart] if smart in plex.search_translation else smart
if mod is None:
mod = plex.modifier_translation[smart_mod] if smart_mod in plex.modifier_translation else smart_mod
if arg_s is None:
arg_s = arg
if smart in string_filters and smart_mod in ["", ".not"]:
mod_s = "does not contain" if smart_mod == ".not" else "contains"
elif mod_s is None:
mod_s = plex.mod_displays[smart_mod]
if param_s is None:
param_s = smart.title().replace('_', ' ')
display_line = f"{indent}{param_s} {mod_s} {arg_s}"
return f"{arg_key}{mod}={arg}&", display_line
if smart_final in plex.movie_only_smart_searches and self.library.is_show:
raise Failed(f"Collection Error: {smart_final} smart filter attribute only works for movie libraries")
elif smart_final in plex.show_only_smart_searches and self.library.is_movie:
raise Failed(f"Collection Error: {smart_final} smart filter attribute only works for show libraries")
elif smart_final not in plex.smart_searches:
raise Failed(f"Collection Error: {smart_final} is not a valid smart filter attribute")
elif smart_data is None:
raise Failed(f"Collection Error: {smart_final} smart filter attribute is blank")
elif smart in ["all", "any"]:
dicts = util.get_list(smart_data)
results = ""
display_add = ""
for dict_data in dicts:
if not isinstance(dict_data, dict):
raise Failed(f"Collection Error: {smart} must be either a dictionary or list of dictionaries")
inside_filter, inside_display = _filter(dict_data, fail, is_all=smart == "all", level=level)
display_add += inside_display
results += f"{conjunction if len(results) > 0 else ''}push=1&{inside_filter}pop=1&"
elif smart in ["year", "episode_year"] and smart_mod in [".gt", ".gte", ".lt", ".lte"]:
results, display_add = build_url_arg(util.check_year(smart_data, current_year, smart_final))
elif smart in ["added", "episode_added", "originally_available", "episode_originally_available"] and smart_mod in [".before", ".after"]:
results, display_add = build_url_arg(util.check_date(smart_data, smart_final, return_string=True, plex_date=True))
elif smart in ["added", "episode_added", "originally_available", "episode_originally_available"] and smart_mod in ["", ".not"]:
in_the_last = util.check_number(smart_data, smart_final, minimum=1)
last_text = "is not in the last" if smart_mod == ".not" else "is in the last"
last_mod = "%3E%3E" if smart_mod == "" else "%3C%3C"
results, display_add = build_url_arg(f"-{in_the_last}d", mod=last_mod, arg_s=f"{in_the_last} Days", mod_s=last_text)
elif smart in ["duration"] and smart_mod in [".gt", ".gte", ".lt", ".lte"]:
results, display_add = build_url_arg(util.check_number(smart_data, smart_final, minimum=1) * 60000)
elif smart in ["plays", "episode_plays"] and smart_mod in [".gt", ".gte", ".lt", ".lte"]:
results, display_add = build_url_arg(util.check_number(smart_data, smart_final, minimum=1))
elif smart in ["user_rating", "episode_user_rating", "critic_rating", "audience_rating"] and smart_mod in [".gt", ".gte", ".lt", ".lte"]:
results, display_add = build_url_arg(util.check_number(smart_data, smart_final, number_type="float", minimum=0, maximum=10))
elif smart == "hdr":
if isinstance(smart_data, bool):
hdr_mod = "" if smart_data else "!"
hdr_arg = "true" if smart_data else "false"
results, display_add = build_url_arg(1, mod=hdr_mod, arg_s=hdr_arg, mod_s="is", param_s="HDR")
else:
raise Failed("Collection Error: HDR must be true or false")
else:
if smart in ["title", "episode_title", "studio"] and smart_mod in ["", ".not", ".begins", ".ends"]:
results_list = [(t, t) for t in util.get_list(smart_data, split=False)]
elif smart in plex.tags and smart_mod in ["", ".not", ".begins", ".ends"]:
if smart_final in plex.tmdb_searches:
smart_values = []
for tmdb_value in util.get_list(smart_data):
if tmdb_value.lower() == "tmdb" and "tmdb_person" in self.details:
for tmdb_name in self.details["tmdb_person"]:
smart_values.append(tmdb_name)
else:
smart_values.append(tmdb_value)
else:
smart_values = util.get_list(smart_data)
results_list = []
try:
results_list = self.library.validate_search_list(smart_values, smart, title=False, pairs=True)
except Failed as e:
if fail:
raise
else:
logger.error(e)
elif smart in ["decade", "year", "episode_year"] and smart_mod in ["", ".not"]:
results_list = [(y, y) for y in util.get_year_list(smart_data, current_year, smart_final)]
else:
raise Failed(f"Collection Error: modifier: {smart_mod} not supported with the {smart} plex search attribute")
results = ""
display_add = ""
for og_value, result in results_list:
built_arg = build_url_arg(quote(result) if smart in string_filters else result, arg_s=og_value)
display_add += built_arg[1]
results += f"{conjunction if len(results) > 0 else ''}{built_arg[0]}"
display += display_add
output += f"{conjunction if len(output) > 0 else ''}{results}"
return output, display
base = "all" if "all" in smart_methods else "any"
base_all = base == "all"
if smart_filter[smart_methods[base]] is None:
raise Failed(f"Collection Error: {base} attribute is blank")
if not isinstance(smart_filter[smart_methods[base]], dict):
raise Failed(f"Collection Error: {base} must be a dictionary: {smart_filter[smart_methods[base]]}")
built_filter, filter_text = _filter(smart_filter[smart_methods[base]], validate, is_all=base_all)
self.smart_filter_details = f"{filter_details}Filter:{filter_text}"
final_filter = built_filter[:-1] if base_all else f"push=1&{built_filter}pop=1"
self.smart_url = f"?type={self.smart_type_key}&{f'limit={limit}&' if limit else ''}sort={smart_sorts[smart_sort]}&{final_filter}"
def cant_interact(attr1, attr2, fail=False):
if getattr(self, attr1) and getattr(self, attr2):
message = f"Collection Error: {attr1} & {attr2} attributes cannot go together"
if fail:
raise Failed(message)
else:
setattr(self, attr2, False)
logger.info("")
logger.warning(f"{message} removing {attr2}")
cant_interact("smart_label_collection", "collectionless")
cant_interact("smart_url", "collectionless")
cant_interact("smart_url", "run_again")
cant_interact("smart_label_collection", "smart_url", fail=True)
self.smart = self.smart_url or self.smart_label_collection
for method_key, method_data in self.data.items():
if method_key.lower() in ignored_details:
continue
logger.info("")
logger.info(f"Validating Method: {method_key}")
if "trakt" in method_key.lower() and not config.Trakt: raise Failed(f"Collection Error: {method_key} requires Trakt todo be configured")
elif "imdb" in method_key.lower() and not config.IMDb: raise Failed(f"Collection Error: {method_key} requires TMDb or Trakt to be configured")
elif "radarr" in method_key.lower() and not self.library.Radarr: raise Failed(f"Collection Error: {method_key} requires Radarr to be configured")
elif "sonarr" in method_key.lower() and not self.library.Sonarr: raise Failed(f"Collection Error: {method_key} requires Sonarr to be configured")
elif "tautulli" in method_key.lower() and not self.library.Tautulli: raise Failed(f"Collection Error: {method_key} requires Tautulli to be configured")
elif "mal" in method_key.lower() and not config.MyAnimeList: raise Failed(f"Collection Error: {method_key} requires MyAnimeList to be configured")
elif method_data is not None:
logger.debug(f"Value: {method_data}")
if method_key.lower() in method_alias:
method_name = method_alias[method_key.lower()]
logger.warning(f"Collection Warning: {method_key} attribute will run as {method_name}")
elif method_key.lower() == "add_to_arr":
method_name = "radarr_add" if self.library.is_movie else "sonarr_add"
logger.warning(f"Collection Warning: {method_key} attribute will run as {method_name}")
elif method_key.lower() in ["arr_tag", "arr_folder"]:
method_name = f"{'rad' if self.library.is_movie else 'son'}{method_key.lower()}"
logger.warning(f"Collection Warning: {method_key} attribute will run as {method_name}")
else:
method_name = method_key.lower()
if method_name in show_only_builders and self.library.is_movie:
raise Failed(f"Collection Error: {method_name} attribute only works for show libraries")
elif method_name in movie_only_builders and self.library.is_show:
raise Failed(f"Collection Error: {method_name} attribute only works for movie libraries")
elif method_name in plex.movie_only_searches and self.library.is_show:
raise Failed(f"Collection Error: {method_name} plex search only works for movie libraries")
elif method_name in plex.show_only_searches and self.library.is_movie:
raise Failed(f"Collection Error: {method_name} plex search only works for show libraries")
elif method_name in smart_collection_invalid and self.smart:
raise Failed(f"Collection Error: {method_name} attribute only works with normal collections")
elif method_name not in collectionless_details and self.collectionless:
raise Failed(f"Collection Error: {method_name} attribute does not work for Collectionless collection")
elif self.smart_url and (method_name in all_builders or method_name in smart_url_collection_invalid):
raise Failed(f"Collection Error: {method_name} builder not allowed when using smart_filter")
elif method_name == "summary":
self.summaries[method_name] = method_data
elif method_name == "tmdb_summary":
self.summaries[method_name] = config.TMDb.get_movie_show_or_collection(util.regex_first_int(method_data, "TMDb ID"), self.library.is_movie).overview
elif method_name == "tmdb_description":
self.summaries[method_name] = config.TMDb.get_list(util.regex_first_int(method_data, "TMDb List ID")).description
elif method_name == "tmdb_biography":
self.summaries[method_name] = config.TMDb.get_person(util.regex_first_int(method_data, "TMDb Person ID")).biography
elif method_name == "tvdb_summary":
self.summaries[method_name] = config.TVDb.get_movie_or_show(method_data, self.library.Plex.language, self.library.is_movie).summary
elif method_name == "tvdb_description":
self.summaries[method_name] = config.TVDb.get_list_description(method_data, self.library.Plex.language)
elif method_name == "trakt_description":
self.summaries[method_name] = config.Trakt.standard_list(config.Trakt.validate_trakt(util.get_list(method_data))[0]).description
elif method_name == "letterboxd_description":
self.summaries[method_name] = config.Letterboxd.get_list_description(method_data, self.library.Plex.language)
elif method_name == "collection_mode":
if str(method_data).lower() == "default":
self.details[method_name] = "default"
elif str(method_data).lower() == "hide":
self.details[method_name] = "hide"
elif str(method_data).lower() in ["hide_items", "hideitems"]:
self.details[method_name] = "hideItems"
elif str(method_data).lower() in ["show_items", "showitems"]:
self.details[method_name] = "showItems"
else:
raise Failed(f"Collection Error: {method_data} collection_mode invalid\n\tdefault (Library default)\n\thide (Hide Collection)\n\thide_items (Hide Items in this Collection)\n\tshow_items (Show this Collection and its Items)")
elif method_name == "collection_order":
if str(method_data).lower() == "release":
self.details[method_name] = "release"
elif str(method_data).lower() == "alpha":
self.details[method_name] = "alpha"
else:
raise Failed(f"Collection Error: {method_data} collection_order invalid\n\trelease (Order Collection by release dates)\n\talpha (Order Collection Alphabetically)")
elif method_name == "url_poster":
self.posters[method_name] = method_data
elif method_name == "tmdb_poster":
self.posters[method_name] = f"{config.TMDb.image_url}{config.TMDb.get_movie_show_or_collection(util.regex_first_int(method_data, 'TMDb ID'), self.library.is_movie).poster_path}"
elif method_name == "tmdb_profile":
self.posters[method_name] = f"{config.TMDb.image_url}{config.TMDb.get_person(util.regex_first_int(method_data, 'TMDb Person ID')).profile_path}"
elif method_name == "tvdb_poster":
self.posters[method_name] = f"{config.TVDb.get_movie_or_series(method_data, self.library.Plex.language, self.library.is_movie).poster_path}"
elif method_name == "file_poster":
if os.path.exists(method_data):
self.posters[method_name] = os.path.abspath(method_data)
else:
raise Failed(f"Collection Error: Poster Path Does Not Exist: {os.path.abspath(method_data)}")
elif method_name == "url_background":
self.backgrounds[method_name] = method_data
elif method_name == "tmdb_background":
self.backgrounds[method_name] = f"{config.TMDb.image_url}{config.TMDb.get_movie_show_or_collection(util.regex_first_int(method_data, 'TMDb ID'), self.library.is_movie).poster_path}"
elif method_name == "tvdb_background":
self.posters[method_name] = f"{config.TVDb.get_movie_or_series(method_data, self.library.Plex.language, self.library.is_movie).background_path}"
elif method_name == "file_background":
if os.path.exists(method_data): self.backgrounds[method_name] = os.path.abspath(method_data)
else: raise Failed(f"Collection Error: Background Path Does Not Exist: {os.path.abspath(method_data)}")
elif method_name == "sync_mode":
if str(method_data).lower() in ["append", "sync"]: self.details[method_name] = method_data.lower()
else: raise Failed("Collection Error: sync_mode attribute must be either 'append' or 'sync'")
elif method_name in ["label", "label.remove", "label.sync"]:
if "label" in self.data and "label.sync" in self.data:
raise Failed(f"Collection Error: Cannot use label and label.sync together")
if "label.remove" in self.data and "label.sync" in self.data:
raise Failed(f"Collection Error: Cannot use label.remove and label.sync together")
if method_name == "label" and "label_sync_mode" in self.data and self.data["label_sync_mode"] == "sync":
self.details["label.sync"] = util.get_list(method_data)
else:
self.details[method_name] = util.get_list(method_data)
elif method_name in ["item_label", "item_label.remove", "item_label.sync"]:
if "item_label" in self.data and "item_label.sync" in self.data:
raise Failed(f"Collection Error: Cannot use item_label and item_label.sync together")
if "item_label.remove" in self.data and "item_label.sync" in self.data:
raise Failed(f"Collection Error: Cannot use item_label.remove and item_label.sync together")
self.item_details[method_name] = util.get_list(method_data)
elif method_name in plex.item_advance_keys:
key, options = plex.item_advance_keys[method_name]
if method_name in advance_new_agent and self.library.agent not in plex.new_plex_agents:
logger.error(f"Metadata Error: {method_name} attribute only works for with the New Plex Movie Agent and New Plex TV Agent")
elif method_name in advance_show and not self.library.is_show:
logger.error(f"Metadata Error: {method_name} attribute only works for show libraries")
elif str(method_data).lower() not in options:
logger.error(f"Metadata Error: {method_data} {method_name} attribute invalid")
else:
self.item_details[method_name] = str(method_data).lower()
elif method_name in boolean_details:
self.details[method_name] = util.get_bool(method_name, method_data)
elif method_name in all_details:
self.details[method_name] = method_data
elif method_name == "radarr_add":
self.add_to_radarr = util.get_bool(method_name, method_data)
elif method_name == "radarr_folder":
self.radarr_options["folder"] = method_data
elif method_name in ["radarr_monitor", "radarr_search"]:
self.radarr_options[method_name[7:]] = util.get_bool(method_name, method_data)
elif method_name == "radarr_availability":
if str(method_data).lower() in radarr.availability_translation:
self.radarr_options["availability"] = str(method_data).lower()
else:
raise Failed(f"Collection Error: {method_name} attribute must be either announced, cinemas, released or db")
elif method_name == "radarr_quality":
self.library.Radarr.get_profile_id(method_data)
self.radarr_options["quality"] = method_data
elif method_name == "radarr_tag":
self.radarr_options["tag"] = util.get_list(method_data)
elif method_name == "sonarr_add":
self.add_to_sonarr = util.get_bool(method_name, method_data)
elif method_name == "sonarr_folder":
self.sonarr_options["folder"] = method_data
elif method_name == "sonarr_monitor":
if str(method_data).lower() in sonarr.monitor_translation:
self.sonarr_options["monitor"] = str(method_data).lower()
else:
raise Failed(f"Collection Error: {method_name} attribute must be either all, future, missing, existing, pilot, first, latest or none")
elif method_name == "sonarr_quality":
self.library.Sonarr.get_profile_id(method_data, "quality_profile")
self.sonarr_options["quality"] = method_data
elif method_name == "sonarr_language":
self.library.Sonarr.get_profile_id(method_data, "language_profile")
self.sonarr_options["language"] = method_data
elif method_name == "sonarr_series":
if str(method_data).lower() in sonarr.series_type:
self.sonarr_options["series"] = str(method_data).lower()
else:
raise Failed(f"Collection Error: {method_name} attribute must be either standard, daily, or anime")
elif method_name in ["sonarr_season", "sonarr_search", "sonarr_cutoff_search"]:
self.sonarr_options[method_name[7:]] = util.get_bool(method_name, method_data)
elif method_name == "sonarr_tag":
self.sonarr_options["tag"] = util.get_list(method_data)
elif method_name in ["title", "title.and", "title.not", "title.begins", "studio.ends", "studio", "studio.and", "studio.not", "studio.begins", "studio.ends"]:
self.methods.append(("plex_search", [{method_name: util.get_list(method_data, split=False)}]))
elif method_name in ["year.gt", "year.gte", "year.lt", "year.lte"]:
self.methods.append(("plex_search", [{method_name: util.check_year(method_data, current_year, method_name)}]))
elif method_name in ["added.before", "added.after", "originally_available.before", "originally_available.after"]:
self.methods.append(("plex_search", [{method_name: util.check_date(method_data, method_name, return_string=True, plex_date=True)}]))
elif method_name in ["added", "added.not", "originally_available", "originally_available.not", "duration.gt", "duration.gte", "duration.lt", "duration.lte"]:
self.methods.append(("plex_search", [{method_name: util.check_number(method_data, method_name, minimum=1)}]))
elif method_name in ["user_rating.gt", "user_rating.gte", "user_rating.lt", "user_rating.lte", "critic_rating.gt", "critic_rating.gte", "critic_rating.lt", "critic_rating.lte", "audience_rating.gt", "audience_rating.gte", "audience_rating.lt", "audience_rating.lte"]:
self.methods.append(("plex_search", [{method_name: util.check_number(method_data, method_name, number_type="float", minimum=0, maximum=10)}]))
elif method_name in ["decade", "year", "year.not"]:
self.methods.append(("plex_search", [{method_name: util.get_year_list(method_data, current_year, method_name)}]))
elif method_name in plex.searches:
if method_name in plex.tmdb_searches:
final_values = []
for value in util.get_list(method_data):
if value.lower() == "tmdb" and "tmdb_person" in self.details:
for name in self.details["tmdb_person"]:
final_values.append(name)
else:
final_values.append(value)
else:
final_values = method_data
search = self.library.validate_search_list(final_values, os.path.splitext(method_name)[0])
self.methods.append(("plex_search", [{method_name: search}]))
elif method_name == "plex_all":
self.methods.append((method_name, [""]))
elif method_name == "anidb_popular":
list_count = util.regex_first_int(method_data, "List Size", default=40)
if 1 <= list_count <= 30:
self.methods.append((method_name, [list_count]))
else:
logger.warning("Collection Error: anidb_popular must be an integer between 1 and 30 defaulting to 30")
self.methods.append((method_name, [30]))
elif method_name == "mal_id":
self.methods.append((method_name, util.get_int_list(method_data, "MyAnimeList ID")))
elif method_name in ["anidb_id", "anidb_relation"]:
self.methods.append((method_name, config.AniDB.validate_anidb_list(util.get_int_list(method_data, "AniDB ID"), self.library.Plex.language)))
elif method_name in ["anilist_id", "anilist_relations", "anilist_studio"]:
self.methods.append((method_name, config.AniList.validate_anilist_ids(util.get_int_list(method_data, "AniList ID"), studio=method_name == "anilist_studio")))
elif method_name == "trakt_list":
self.methods.append((method_name, config.Trakt.validate_trakt(util.get_list(method_data))))
elif method_name == "trakt_list_details":
valid_list = config.Trakt.validate_trakt(util.get_list(method_data))
item = config.Trakt.standard_list(valid_list[0])
if hasattr(item, "description") and item.description:
self.summaries[method_name] = item.description
self.methods.append((method_name[:-8], valid_list))
elif method_name in ["trakt_watchlist", "trakt_collection"]:
self.methods.append((method_name, config.Trakt.validate_trakt(util.get_list(method_data), trakt_type=method_name[6:], is_movie=self.library.is_movie)))
elif method_name == "imdb_list":
new_list = []
for imdb_list in util.get_list(method_data, split=False):
if isinstance(imdb_list, dict):
dict_methods = {dm.lower(): dm for dm in imdb_list}
if "url" in dict_methods and imdb_list[dict_methods["url"]]:
imdb_url = config.IMDb.validate_imdb_url(imdb_list[dict_methods["url"]], self.library.Plex.language)
else:
raise Failed("Collection Error: imdb_list attribute url is required")
if "limit" in dict_methods and imdb_list[dict_methods["limit"]]:
list_count = util.regex_first_int(imdb_list[dict_methods["limit"]], "List Limit", default=0)
else:
list_count = 0
else:
imdb_url = config.IMDb.validate_imdb_url(str(imdb_list), self.library.Plex.language)
list_count = 0
new_list.append({"url": imdb_url, "limit": list_count})
self.methods.append((method_name, new_list))
elif method_name == "letterboxd_list":
self.methods.append((method_name, util.get_list(method_data, split=False)))
elif method_name == "letterboxd_list_details":
values = util.get_list(method_data, split=False)
self.summaries[method_name] = config.Letterboxd.get_list_description(values[0], self.library.Plex.language)
self.methods.append((method_name[:-8], values))
elif method_name in dictionary_builders:
if isinstance(method_data, dict):
def get_int(parent, int_method, data_in, methods_in, default_in, minimum=1, maximum=None):
if int_method not in methods_in:
logger.warning(f"Collection Warning: {parent} {int_method} attribute not found using {default_in} as default")
elif not data_in[methods_in[int_method]]:
logger.warning(f"Collection Warning: {parent} {methods_in[int_method]} attribute is blank using {default_in} as default")
elif isinstance(data_in[methods_in[int_method]], int) and data_in[methods_in[int_method]] >= minimum:
if maximum is None or data_in[methods_in[int_method]] <= maximum:
return data_in[methods_in[int_method]]
else:
logger.warning(f"Collection Warning: {parent} {methods_in[int_method]} attribute {data_in[methods_in[int_method]]} invalid must an integer <= {maximum} using {default_in} as default")
else:
logger.warning(f"Collection Warning: {parent} {methods_in[int_method]} attribute {data_in[methods_in[int_method]]} invalid must an integer >= {minimum} using {default_in} as default")
return default_in
if method_name == "filters":
for filter_name, filter_data in method_data.items():
modifier = filter_name[-4:].lower()
modifier = modifier if modifier in [".not", ".lte", ".gte"] else ""
method = filter_name[:-4].lower() if modifier in [".not", ".lte", ".gte"] else filter_name.lower()
if method in method_alias:
filter_method = f"{method_alias[method]}{modifier}"
logger.warning(f"Collection Warning: {filter_name} filter will run as {filter_method}")
else:
filter_method = f"{method}{modifier}"
if filter_method in movie_only_filters and self.library.is_show:
raise Failed(f"Collection Error: {filter_method} filter only works for movie libraries")
elif filter_data is None:
raise Failed(f"Collection Error: {filter_method} filter is blank")
elif filter_method == "year":
valid_data = util.get_year_list(filter_data, current_year, f"{filter_method} filter")
elif filter_method in ["max_age", "duration.gte", "duration.lte", "tmdb_vote_count.gte", "tmdb_vote_count.lte"]:
valid_data = util.check_number(filter_data, f"{filter_method} filter", minimum=1)
elif filter_method in ["year.gte", "year.lte"]:
valid_data = util.check_year(filter_data, current_year, f"{filter_method} filter")
elif filter_method in ["user_rating.gte", "user_rating.lte", "audience_rating.gte", "audience_rating.lte", "critic_rating.gte", "critic_rating.lte"]:
valid_data = util.check_number(filter_data, f"{filter_method} filter", number_type="float", minimum=0.1, maximum=10)
elif filter_method in ["originally_available.gte", "originally_available.lte"]:
valid_data = util.check_date(filter_data, f"{filter_method} filter")
elif filter_method in ["original_language", "original_language.not"]:
valid_data = util.get_list(filter_data, lower=True)
elif filter_method in ["collection", "collection.not"]:
valid_data = filter_data if isinstance(filter_data, list) else [filter_data]
elif filter_method in all_filters:
valid_data = util.get_list(filter_data)
else:
raise Failed(f"Collection Error: {filter_method} filter not supported")
self.filters.append((filter_method, valid_data))
elif method_name == "plex_collectionless":
new_dictionary = {}
dict_methods = {dm.lower(): dm for dm in method_data}
prefix_list = []
if "exclude_prefix" in dict_methods and method_data[dict_methods["exclude_prefix"]]:
if isinstance(method_data[dict_methods["exclude_prefix"]], list):
prefix_list.extend([exclude for exclude in method_data[dict_methods["exclude_prefix"]] if exclude])
else:
prefix_list.append(str(method_data[dict_methods["exclude_prefix"]]))
exact_list = []
if "exclude" in dict_methods and method_data[dict_methods["exclude"]]:
if isinstance(method_data[dict_methods["exclude"]], list):
exact_list.extend([exclude for exclude in method_data[dict_methods["exclude"]] if exclude])
else:
exact_list.append(str(method_data[dict_methods["exclude"]]))
if len(prefix_list) == 0 and len(exact_list) == 0:
raise Failed("Collection Error: you must have at least one exclusion")
exact_list.append(self.name)
new_dictionary["exclude_prefix"] = prefix_list
new_dictionary["exclude"] = exact_list
self.methods.append((method_name, [new_dictionary]))
elif method_name == "plex_search":
searches = {}
validate = True
if "validate" in method_data:
if method_data["validate"] is None:
raise Failed("Collection Error: validate plex search attribute is blank")
if not isinstance(method_data["validate"], bool):
raise Failed("Collection Error: validate plex search attribute must be either true or false")
validate = method_data["validate"]
for search_name, search_data in method_data.items():
search, modifier, search_final = _split(search_name)
if search_name != search_final:
logger.warning(f"Collection Warning: {search_name} plex search attribute will run as {search_final}")
if search_final in plex.movie_only_searches and self.library.is_show:
raise Failed(f"Collection Error: {search_final} plex search attribute only works for movie libraries")
elif search_final in plex.show_only_searches and self.library.is_movie:
raise Failed(f"Collection Error: {search_final} plex search attribute only works for show libraries")
elif search_data is None:
raise Failed(f"Collection Error: {search_final} plex search attribute is blank")
elif search == "sort_by":
if str(search_data).lower() in plex.sorts:
searches[search] = str(search_data).lower()
else:
logger.warning(f"Collection Error: {search_data} is not a valid plex search sort defaulting to title.asc")
elif search == "limit":
if not search_data:
raise Failed(f"Collection Warning: plex search limit attribute is blank")
elif not isinstance(search_data, int) and search_data > 0:
raise Failed(f"Collection Warning: plex search limit attribute: {search_data} must be an integer greater then 0")
else:
searches[search] = search_data
elif search_final not in plex.searches:
raise Failed(f"Collection Error: {search_final} is not a valid plex search attribute")
elif search in ["title", "studio"] and modifier in ["", ".and", ".not", ".begins", ".ends"]:
searches[search_final] = util.get_list(search_data, split=False)
elif search in plex.tags and modifier in ["", ".and", ".not", ".begins", ".ends"]:
if search_final in plex.tmdb_searches:
final_values = []
for value in util.get_list(search_data):
if value.lower() == "tmdb" and "tmdb_person" in self.details:
for name in self.details["tmdb_person"]:
final_values.append(name)
else:
final_values.append(value)
else:
final_values = util.get_list(search_data)
try:
searches[search_final] = self.library.validate_search_list(final_values, search)
except Failed as e:
if validate:
raise
else:
logger.error(e)
elif search == "year" and modifier in [".gt", ".gte", ".lt", ".lte"]:
searches[search_final] = util.check_year(search_data, current_year, search_final)
elif search in ["added", "originally_available"] and modifier in [".before", ".after"]:
searches[search_final] = util.check_date(search_data, search_final, return_string=True, plex_date=True)
elif search in ["added", "originally_available", "duration"] and modifier in ["", ".not", ".gt", ".gte", ".lt", ".lte"]:
searches[search_final] = util.check_number(search_data, search_final, minimum=1)
elif search in ["user_rating", "critic_rating", "audience_rating"] and modifier in [".gt", ".gte", ".lt", ".lte"]:
searches[search_final] = util.check_number(search_data, search_final, number_type="float", minimum=0, maximum=10)
elif search in ["decade", "year"] and modifier in ["", ".not"]:
searches[search_final] = util.get_year_list(search_data, current_year, search_final)
else:
raise Failed(f"Collection Error: modifier: {modifier} not supported with the {search} plex search attribute")
if len(searches) > 0:
self.methods.append((method_name, [searches]))
else:
raise Failed("Collection Error: no valid plex search attributes")
elif method_name == "tmdb_discover":
new_dictionary = {"limit": 100}
for discover_name, discover_data in method_data.items():
discover_final = discover_name.lower()
if discover_data:
if (self.library.is_movie and discover_final in tmdb.discover_movie) or (self.library.is_show and discover_final in tmdb.discover_tv):
if discover_final == "language":
if re.compile("([a-z]{2})-([A-Z]{2})").match(str(discover_data)):
new_dictionary[discover_final] = str(discover_data)
else:
raise Failed(f"Collection Error: {method_name} attribute {discover_final}: {discover_data} must match pattern ([a-z]{{2}})-([A-Z]{{2}}) e.g. en-US")
elif discover_final == "region":
if re.compile("^[A-Z]{2}$").match(str(discover_data)):
new_dictionary[discover_final] = str(discover_data)
else:
raise Failed(f"Collection Error: {method_name} attribute {discover_final}: {discover_data} must match pattern ^[A-Z]{{2}}$ e.g. US")
elif discover_final == "sort_by":
if (self.library.is_movie and discover_data in tmdb.discover_movie_sort) or (self.library.is_show and discover_data in tmdb.discover_tv_sort):
new_dictionary[discover_final] = discover_data
else:
raise Failed(f"Collection Error: {method_name} attribute {discover_final}: {discover_data} is invalid")
elif discover_final == "certification_country":
if "certification" in method_data or "certification.lte" in method_data or "certification.gte" in method_data:
new_dictionary[discover_final] = discover_data
else:
raise Failed(f"Collection Error: {method_name} attribute {discover_final}: must be used with either certification, certification.lte, or certification.gte")
elif discover_final in ["certification", "certification.lte", "certification.gte"]:
if "certification_country" in method_data:
new_dictionary[discover_final] = discover_data
else:
raise Failed(f"Collection Error: {method_name} attribute {discover_final}: must be used with certification_country")
elif discover_final in ["include_adult", "include_null_first_air_dates", "screened_theatrically"]:
if discover_data is True:
new_dictionary[discover_final] = discover_data
elif discover_final in tmdb.discover_dates:
new_dictionary[discover_final] = util.check_date(discover_data, f"{method_name} attribute {discover_final}", return_string=True)
elif discover_final in ["primary_release_year", "year", "first_air_date_year"]:
new_dictionary[discover_final] = util.check_number(discover_data, f"{method_name} attribute {discover_final}", minimum=1800, maximum=current_year + 1)
elif discover_final in ["vote_count.gte", "vote_count.lte", "vote_average.gte", "vote_average.lte", "with_runtime.gte", "with_runtime.lte"]:
new_dictionary[discover_final] = util.check_number(discover_data, f"{method_name} attribute {discover_final}", minimum=1)
elif discover_final in ["with_cast", "with_crew", "with_people", "with_companies", "with_networks", "with_genres", "without_genres", "with_keywords", "without_keywords", "with_original_language", "timezone"]:
new_dictionary[discover_final] = discover_data
else:
raise Failed(f"Collection Error: {method_name} attribute {discover_final} not supported")
elif discover_final == "limit":
if isinstance(discover_data, int) and discover_data > 0:
new_dictionary[discover_final] = discover_data
else:
raise Failed(f"Collection Error: {method_name} attribute {discover_final}: must be a valid number greater then 0")
else:
raise Failed(f"Collection Error: {method_name} attribute {discover_final} not supported")
else:
raise Failed(f"Collection Error: {method_name} parameter {discover_final} is blank")
if len(new_dictionary) > 1:
self.methods.append((method_name, [new_dictionary]))
else:
raise Failed(f"Collection Error: {method_name} had no valid fields")
elif "tautulli" in method_name:
new_dictionary = {}
if method_name == "tautulli_popular":
new_dictionary["list_type"] = "popular"
elif method_name == "tautulli_watched":
new_dictionary["list_type"] = "watched"
else:
raise Failed(f"Collection Error: {method_name} attribute not supported")
dict_methods = {dm.lower(): dm for dm in method_data}
new_dictionary["list_days"] = get_int(method_name, "list_days", method_data, dict_methods, 30)
new_dictionary["list_size"] = get_int(method_name, "list_size", method_data, dict_methods, 10)
new_dictionary["list_buffer"] = get_int(method_name, "list_buffer", method_data, dict_methods, 20)
self.methods.append((method_name, [new_dictionary]))
elif method_name == "mal_season":
new_dictionary = {"sort_by": "anime_num_list_users"}
dict_methods = {dm.lower(): dm for dm in method_data}
if "sort_by" not in dict_methods:
logger.warning("Collection Warning: mal_season sort_by attribute not found using members as default")
elif not method_data[dict_methods["sort_by"]]:
logger.warning("Collection Warning: mal_season sort_by attribute is blank using members as default")
elif method_data[dict_methods["sort_by"]] not in mal.season_sort:
logger.warning(f"Collection Warning: mal_season sort_by attribute {method_data[dict_methods['sort_by']]} invalid must be either 'members' or 'score' using members as default")
else:
new_dictionary["sort_by"] = mal.season_sort[method_data[dict_methods["sort_by"]]]
if current_time.month in [1, 2, 3]: new_dictionary["season"] = "winter"
elif current_time.month in [4, 5, 6]: new_dictionary["season"] = "spring"
elif current_time.month in [7, 8, 9]: new_dictionary["season"] = "summer"
elif current_time.month in [10, 11, 12]: new_dictionary["season"] = "fall"
if "season" not in dict_methods:
logger.warning(f"Collection Warning: mal_season season attribute not found using the current season: {new_dictionary['season']} as default")
elif not method_data[dict_methods["season"]]:
logger.warning(f"Collection Warning: mal_season season attribute is blank using the current season: {new_dictionary['season']} as default")
elif method_data[dict_methods["season"]] not in util.pretty_seasons:
logger.warning(f"Collection Warning: mal_season season attribute {method_data[dict_methods['season']]} invalid must be either 'winter', 'spring', 'summer' or 'fall' using the current season: {new_dictionary['season']} as default")
else:
new_dictionary["season"] = method_data[dict_methods["season"]]
new_dictionary["year"] = get_int(method_name, "year", method_data, dict_methods, current_time.year, minimum=1917, maximum=current_time.year + 1)
new_dictionary["limit"] = get_int(method_name, "limit", method_data, dict_methods, 100, maximum=500)
self.methods.append((method_name, [new_dictionary]))
elif method_name == "mal_userlist":
new_dictionary = {"status": "all", "sort_by": "list_score"}
dict_methods = {dm.lower(): dm for dm in method_data}
if "username" not in dict_methods:
raise Failed("Collection Error: mal_userlist username attribute is required")
elif not method_data[dict_methods["username"]]:
raise Failed("Collection Error: mal_userlist username attribute is blank")
else:
new_dictionary["username"] = method_data[dict_methods["username"]]
if "status" not in dict_methods:
logger.warning("Collection Warning: mal_season status attribute not found using all as default")
elif not method_data[dict_methods["status"]]:
logger.warning("Collection Warning: mal_season status attribute is blank using all as default")
elif method_data[dict_methods["status"]] not in mal.userlist_status:
logger.warning(f"Collection Warning: mal_season status attribute {method_data[dict_methods['status']]} invalid must be either 'all', 'watching', 'completed', 'on_hold', 'dropped' or 'plan_to_watch' using all as default")
else:
new_dictionary["status"] = mal.userlist_status[method_data[dict_methods["status"]]]
if "sort_by" not in dict_methods:
logger.warning("Collection Warning: mal_season sort_by attribute not found using score as default")
elif not method_data[dict_methods["sort_by"]]:
logger.warning("Collection Warning: mal_season sort_by attribute is blank using score as default")
elif method_data[dict_methods["sort_by"]] not in mal.userlist_sort:
logger.warning(f"Collection Warning: mal_season sort_by attribute {method_data[dict_methods['sort_by']]} invalid must be either 'score', 'last_updated', 'title' or 'start_date' using score as default")
else:
new_dictionary["sort_by"] = mal.userlist_sort[method_data[dict_methods["sort_by"]]]
new_dictionary["limit"] = get_int(method_name, "limit", method_data, dict_methods, 100, maximum=1000)
self.methods.append((method_name, [new_dictionary]))
elif "anilist" in method_name:
new_dictionary = {"sort_by": "score"}
dict_methods = {dm.lower(): dm for dm in method_data}
if method_name == "anilist_season":
if current_time.month in [12, 1, 2]: new_dictionary["season"] = "winter"
elif current_time.month in [3, 4, 5]: new_dictionary["season"] = "spring"
elif current_time.month in [6, 7, 8]: new_dictionary["season"] = "summer"
elif current_time.month in [9, 10, 11]: new_dictionary["season"] = "fall"
if "season" not in dict_methods:
logger.warning(f"Collection Warning: anilist_season season attribute not found using the current season: {new_dictionary['season']} as default")
elif not method_data[dict_methods["season"]]:
logger.warning(f"Collection Warning: anilist_season season attribute is blank using the current season: {new_dictionary['season']} as default")
elif method_data[dict_methods["season"]] not in util.pretty_seasons:
logger.warning(f"Collection Warning: anilist_season season attribute {method_data[dict_methods['season']]} invalid must be either 'winter', 'spring', 'summer' or 'fall' using the current season: {new_dictionary['season']} as default")
else:
new_dictionary["season"] = method_data[dict_methods["season"]]
new_dictionary["year"] = get_int(method_name, "year", method_data, dict_methods, current_time.year, minimum=1917, maximum=current_time.year + 1)
elif method_name == "anilist_genre":
if "genre" not in dict_methods:
raise Failed(f"Collection Warning: anilist_genre genre attribute not found")
elif not method_data[dict_methods["genre"]]:
raise Failed(f"Collection Warning: anilist_genre genre attribute is blank")
else:
new_dictionary["genre"] = self.config.AniList.validate_genre(method_data[dict_methods["genre"]])
elif method_name == "anilist_tag":
if "tag" not in dict_methods:
raise Failed(f"Collection Warning: anilist_tag tag attribute not found")
elif not method_data[dict_methods["tag"]]:
raise Failed(f"Collection Warning: anilist_tag tag attribute is blank")
else:
new_dictionary["tag"] = self.config.AniList.validate_tag(method_data[dict_methods["tag"]])
if "sort_by" not in dict_methods:
logger.warning(f"Collection Warning: {method_name} sort_by attribute not found using score as default")
elif not method_data[dict_methods["sort_by"]]:
logger.warning(f"Collection Warning: {method_name} sort_by attribute is blank using score as default")
elif str(method_data[dict_methods["sort_by"]]).lower() not in ["score", "popular"]:
logger.warning(f"Collection Warning: {method_name} sort_by attribute {method_data[dict_methods['sort_by']]} invalid must be either 'score' or 'popular' using score as default")
else:
new_dictionary["sort_by"] = method_data[dict_methods["sort_by"]]
new_dictionary["limit"] = get_int(method_name, "limit", method_data, dict_methods, 0, maximum=500)
self.methods.append((method_name, [new_dictionary]))
else:
raise Failed(f"Collection Error: {method_name} attribute is not a dictionary: {method_data}")
elif method_name in numbered_builders:
list_count = util.regex_first_int(method_data, "List Size", default=10)
if list_count < 1:
logger.warning(f"Collection Warning: {method_name} must be an integer greater then 0 defaulting to 10")
list_count = 10
self.methods.append((method_name, [list_count]))
elif "tvdb" in method_name:
values = util.get_list(method_data)
if method_name[-8:] == "_details":
if method_name == "tvdb_movie_details":
item = config.TVDb.get_movie(self.library.Plex.language, values[0])
if hasattr(item, "description") and item.description:
self.summaries[method_name] = item.description
if hasattr(item, "background_path") and item.background_path:
self.backgrounds[method_name] = f"{config.TMDb.image_url}{item.background_path}"
if hasattr(item, "poster_path") and item.poster_path:
self.posters[method_name] = f"{config.TMDb.image_url}{item.poster_path}"
elif method_name == "tvdb_show_details":
item = config.TVDb.get_series(self.library.Plex.language, values[0])
if hasattr(item, "description") and item.description:
self.summaries[method_name] = item.description
if hasattr(item, "background_path") and item.background_path:
self.backgrounds[method_name] = f"{config.TMDb.image_url}{item.background_path}"
if hasattr(item, "poster_path") and item.poster_path:
self.posters[method_name] = f"{config.TMDb.image_url}{item.poster_path}"
elif method_name == "tvdb_list_details":
self.summaries[method_name] = config.TVDb.get_list_description(values[0], self.library.Plex.language)
self.methods.append((method_name[:-8], values))
else:
self.methods.append((method_name, values))
elif method_name in tmdb.builders:
values = config.TMDb.validate_tmdb_list(util.get_int_list(method_data, f"TMDb {tmdb.type_map[method_name]} ID"), tmdb.type_map[method_name])
if method_name[-8:] == "_details":
if method_name in ["tmdb_collection_details", "tmdb_movie_details", "tmdb_show_details"]:
item = config.TMDb.get_movie_show_or_collection(values[0], self.library.is_movie)
if hasattr(item, "overview") and item.overview:
self.summaries[method_name] = item.overview
if hasattr(item, "backdrop_path") and item.backdrop_path:
self.backgrounds[method_name] = f"{config.TMDb.image_url}{item.backdrop_path}"
if hasattr(item, "poster_path") and item.poster_path:
self.posters[method_name] = f"{config.TMDb.image_url}{item.poster_path}"
elif method_name in ["tmdb_actor_details", "tmdb_crew_details", "tmdb_director_details", "tmdb_producer_details", "tmdb_writer_details"]:
item = config.TMDb.get_person(values[0])
if hasattr(item, "biography") and item.biography:
self.summaries[method_name] = item.biography
if hasattr(item, "profile_path") and item.profile_path:
self.posters[method_name] = f"{config.TMDb.image_url}{item.profile_path}"
else:
item = config.TMDb.get_list(values[0])
if hasattr(item, "description") and item.description:
self.summaries[method_name] = item.description
self.methods.append((method_name[:-8], values))
else:
self.methods.append((method_name, values))
elif method_name in all_builders:
self.methods.append((method_name, util.get_list(method_data)))
elif method_name not in ignored_details:
raise Failed(f"Collection Error: {method_name} attribute not supported")
elif method_key.lower() in all_builders or method_key.lower() in method_alias or method_key.lower() in plex.searches:
raise Failed(f"Collection Error: {method_key} attribute is blank")
else:
logger.warning(f"Collection Warning: {method_key} attribute is blank")
if self.add_to_radarr is None:
self.add_to_radarr = self.library.Radarr.add if self.library.Radarr else False
if self.add_to_sonarr is None:
self.add_to_sonarr = self.library.Sonarr.add if self.library.Sonarr else False
if self.smart_url:
self.add_to_radarr = False
self.add_to_sonarr = False
if self.collectionless:
self.add_to_radarr = False
self.add_to_sonarr = False
self.details["collection_mode"] = "hide"
self.sync = True
if self.build_collection:
try:
self.obj = library.get_collection(self.name)
collection_smart = library.smart(self.obj)
if (self.smart and not collection_smart) or (not self.smart and collection_smart):
logger.info("")
logger.error(f"Collection Error: Converting {self.obj.title} to a {'smart' if self.smart else 'normal'} collection")
library.query(self.obj.delete)
self.obj = None
except Failed:
self.obj = None
self.plex_map = {}
if self.sync and self.obj:
for item in library.get_collection_items(self.obj, self.smart_label_collection):
self.plex_map[item.ratingKey] = item
else:
self.sync = False
self.run_again = False
logger.info("")
logger.info("Validation Successful")
def collect_rating_keys(self, movie_map, show_map):
def add_rating_keys(keys):
if not isinstance(keys, list):
keys = [keys]
self.rating_keys.extend([key for key in keys if key not in self.rating_keys])
for method, values in self.methods:
logger.debug("")
logger.debug(f"Method: {method}")
logger.debug(f"Values: {values}")
for value in values:
def check_map(input_ids):
movie_ids, show_ids = input_ids
items_found_inside = 0
if len(movie_ids) > 0:
items_found_inside += len(movie_ids)
for movie_id in movie_ids:
if movie_id in movie_map:
add_rating_keys(movie_map[movie_id])
elif movie_id not in self.missing_movies:
self.missing_movies.append(movie_id)
if len(show_ids) > 0:
items_found_inside += len(show_ids)
for show_id in show_ids:
if show_id in show_map:
add_rating_keys(show_map[show_id])
elif show_id not in self.missing_shows:
self.missing_shows.append(show_id)
return items_found_inside
logger.debug("")
logger.debug(f"Value: {value}")
logger.info("")
if "plex" in method: add_rating_keys(self.library.get_items(method, value))
elif "tautulli" in method: add_rating_keys(self.library.Tautulli.get_items(self.library, value))
elif "anidb" in method: check_map(self.config.AniDB.get_items(method, value, self.library.Plex.language))
elif "anilist" in method: check_map(self.config.AniList.get_items(method, value))
elif "mal" in method: check_map(self.config.MyAnimeList.get_items(method, value))
elif "tvdb" in method: check_map(self.config.TVDb.get_items(method, value, self.library.Plex.language))
elif "imdb" in method: check_map(self.config.IMDb.get_items(method, value, self.library.Plex.language, self.library.is_movie))
elif "letterboxd" in method: check_map(self.config.Letterboxd.get_items(method, value, self.library.Plex.language))
elif "tmdb" in method: check_map(self.config.TMDb.get_items(method, value, self.library.is_movie))
elif "trakt" in method: check_map(self.config.Trakt.get_items(method, value, self.library.is_movie))
else: logger.error(f"Collection Error: {method} method not supported")
def add_to_collection(self, movie_map):
name, collection_items = self.library.get_collection_name_and_items(self.obj if self.obj else self.name, self.smart_label_collection)
total = len(self.rating_keys)
max_length = len(str(total))
length = 0
for i, item in enumerate(self.rating_keys, 1):
try:
current = self.library.fetchItem(item.ratingKey if isinstance(item, (Movie, Show)) else int(item))
if not isinstance(current, (Movie, Show)):
raise NotFound
except (BadRequest, NotFound):
logger.error(f"Plex Error: Item {item} not found")
continue
match = True
if self.filters:
length = util.print_return(length, f"Filtering {(' ' * (max_length - len(str(i)))) + str(i)}/{total} {current.title}")
for filter_method, filter_data in self.filters:
modifier = filter_method[-4:]
method = filter_method[:-4] if modifier in [".not", ".lte", ".gte"] else filter_method
method_name = filter_translation[method] if method in filter_translation else method
if method_name == "max_age":
threshold_date = datetime.now() - timedelta(days=filter_data)
if current.originallyAvailableAt is None or current.originallyAvailableAt < threshold_date:
match = False
break
elif method_name == "original_language":
movie = None
for key, value in movie_map.items():
if current.ratingKey in value:
try:
movie = self.config.TMDb.get_movie(key)
break
except Failed:
pass
if movie is None:
logger.warning(f"Filter Error: No TMDb ID found for {current.title}")
continue
if (modifier == ".not" and movie.original_language in filter_data) \
or (modifier != ".not" and movie.original_language not in filter_data):
match = False
break
elif method_name == "audio_track_title":
jailbreak = False
for media in current.media:
for part in media.parts:
for audio in part.audioStreams():
for check_title in filter_data:
title = audio.title if audio.title else ""
if check_title.lower() in title.lower():
jailbreak = True
break
if jailbreak: break
if jailbreak: break
if jailbreak: break
if (jailbreak and modifier == ".not") or (not jailbreak and modifier != ".not"):
match = False
break
elif method_name == "filepath":
jailbreak = False
for location in current.locations:
for check_text in filter_data:
if check_text.lower() in location.lower():
jailbreak = True
break
if jailbreak: break
if (jailbreak and modifier == ".not") or (not jailbreak and modifier != ".not"):
match = False
break
elif modifier in [".gte", ".lte"]:
if method_name == "vote_count":
tmdb_item = None
for key, value in movie_map.items():
if current.ratingKey in value:
try:
tmdb_item = self.config.TMDb.get_movie(key) if self.library.is_movie else self.config.TMDb.get_show(key)
break
except Failed:
pass
if tmdb_item is None:
logger.warning(f"Filter Error: No TMDb ID found for {current.title}")
continue
attr = tmdb_item.vote_count
else:
attr = getattr(current, method_name) / 60000 if method_name == "duration" else getattr(current, method_name)
if attr is None or (modifier == ".lte" and attr > filter_data) or (modifier == ".gte" and attr < filter_data):
match = False
break
else:
attrs = []
if method_name in ["video_resolution", "audio_language", "subtitle_language"]:
for media in current.media:
if method_name == "video_resolution":
attrs.extend([media.videoResolution])
for part in media.parts:
if method_name == "audio_language":
attrs.extend([a.language for a in part.audioStreams()])
if method_name == "subtitle_language":
attrs.extend([s.language for s in part.subtitleStreams()])
elif method_name in ["contentRating", "studio", "year", "rating", "originallyAvailableAt"]:
attrs = [str(getattr(current, method_name))]
elif method_name in ["actors", "countries", "directors", "genres", "writers", "collections"]:
attrs = [getattr(x, "tag") for x in getattr(current, method_name)]
else:
raise Failed(f"Filter Error: filter: {method_name} not supported")
if (not list(set(filter_data) & set(attrs)) and modifier != ".not")\
or (list(set(filter_data) & set(attrs)) and modifier == ".not"):
match = False
break
length = util.print_return(length, f"Filtering {(' ' * (max_length - len(str(i)))) + str(i)}/{total} {current.title}")
if match:
logger.info(util.adjust_space(length, f"{name} Collection | {'=' if current in collection_items else '+'} | {current.title}"))
if current in collection_items:
self.plex_map[current.ratingKey] = None
elif self.smart_label_collection:
self.library.query_data(current.addLabel, name)
else:
self.library.query_data(current.addCollection, name)
elif self.details["show_filtered"] is True:
logger.info(f"{name} Collection | X | {current.title}")
media_type = f"{'Movie' if self.library.is_movie else 'Show'}{'s' if total > 1 else ''}"
util.print_end(length)
logger.info("")
logger.info(f"{total} {media_type} Processed")
def run_missing(self):
arr_filters = []
for filter_method, filter_data in self.filters:
if (filter_method.startswith("original_language") and self.library.is_movie) or filter_method.startswith("tmdb_vote_count"):
arr_filters.append((filter_method, filter_data))
if len(self.missing_movies) > 0:
missing_movies_with_names = []
for missing_id in self.missing_movies:
try:
movie = self.config.TMDb.get_movie(missing_id)
except Failed as e:
logger.error(e)
continue
match = True
for filter_method, filter_data in arr_filters:
if (filter_method == "original_language" and movie.original_language not in filter_data) \
or (filter_method == "original_language.not" and movie.original_language in filter_data) \
or (filter_method == "tmdb_vote_count.gte" and movie.vote_count < filter_data) \
or (filter_method == "tmdb_vote_count.lte" and movie.vote_count > filter_data):
match = False
break
if match:
missing_movies_with_names.append((movie.title, missing_id))
if self.details["show_missing"] is True:
logger.info(f"{self.name} Collection | ? | {movie.title} (TMDb: {missing_id})")
elif self.details["show_filtered"] is True:
logger.info(f"{self.name} Collection | X | {movie.title} (TMDb: {missing_id})")
logger.info("")
logger.info(f"{len(missing_movies_with_names)} Movie{'s' if len(missing_movies_with_names) > 1 else ''} Missing")
if self.details["save_missing"] is True:
self.library.add_missing(self.name, missing_movies_with_names, True)
if (self.add_to_radarr and self.library.Radarr) or self.run_again:
missing_tmdb_ids = [missing_id for title, missing_id in missing_movies_with_names]
if self.add_to_radarr and self.library.Radarr:
try:
self.library.Radarr.add_tmdb(missing_tmdb_ids, **self.radarr_options)
except Failed as e:
logger.error(e)
if self.run_again:
self.run_again_movies.extend(missing_tmdb_ids)
if len(self.missing_shows) > 0 and self.library.is_show:
missing_shows_with_names = []
for missing_id in self.missing_shows:
try:
title = str(self.config.TVDb.get_series(self.library.Plex.language, missing_id).title.encode("ascii", "replace").decode())
except Failed as e:
logger.error(e)
continue
match = True
if arr_filters:
show = self.config.TMDb.get_show(self.config.Convert.tvdb_to_tmdb(missing_id))
for filter_method, filter_data in arr_filters:
if (filter_method == "tmdb_vote_count.gte" and show.vote_count < filter_data) \
or (filter_method == "tmdb_vote_count.lte" and show.vote_count > filter_data):
match = False
break
if match:
missing_shows_with_names.append((title, missing_id))
if self.details["show_missing"] is True:
logger.info(f"{self.name} Collection | ? | {title} (TVDB: {missing_id})")
elif self.details["show_filtered"] is True:
logger.info(f"{self.name} Collection | X | {title} (TVDb: {missing_id})")
logger.info("")
logger.info(f"{len(missing_shows_with_names)} Show{'s' if len(missing_shows_with_names) > 1 else ''} Missing")
if self.details["save_missing"] is True:
self.library.add_missing(self.name, missing_shows_with_names, False)
if (self.add_to_sonarr and self.library.Sonarr) or self.run_again:
missing_tvdb_ids = [missing_id for title, missing_id in missing_shows_with_names]
if self.add_to_sonarr and self.library.Sonarr:
try:
self.library.Sonarr.add_tvdb(missing_tvdb_ids, **self.sonarr_options)
except Failed as e:
logger.error(e)
if self.run_again:
self.run_again_shows.extend(missing_tvdb_ids)
def sync_collection(self):
count_removed = 0
for ratingKey, item in self.plex_map.items():
if item is not None:
if count_removed == 0:
logger.info("")
util.separator(f"Removed from {self.name} Collection", space=False, border=False)
logger.info("")
logger.info(f"{self.name} Collection | - | {item.title}")
if self.smart_label_collection:
self.library.query_data(item.removeLabel, self.name)
else:
self.library.query_data(item.removeCollection, self.name)
count_removed += 1
if count_removed > 0:
logger.info("")
logger.info(f"{count_removed} {'Movie' if self.library.is_movie else 'Show'}{'s' if count_removed == 1 else ''} Removed")
def update_details(self):
if not self.obj and self.smart_url:
self.library.create_smart_collection(self.name, self.smart_type_key, self.smart_url)
elif self.smart_label_collection:
try:
smart_type, self.smart_url = self.library.smart_label_url(self.name, self.smart_sort)
if not self.obj:
self.library.create_smart_collection(self.name, smart_type, self.smart_url)
except Failed:
raise Failed(f"Collection Error: Label: {self.name} was not added to any items in the Library")
self.obj = self.library.get_collection(self.name)
if self.smart_url and self.smart_url != self.library.smart_filter(self.obj):
self.library.update_smart_collection(self.obj, self.smart_url)
logger.info(f"Detail: Smart Filter updated to {self.smart_url}")
edits = {}
def get_summary(summary_method, summaries):
logger.info(f"Detail: {summary_method} updated Collection Summary")
return summaries[summary_method]
if "summary" in self.summaries: summary = get_summary("summary", self.summaries)
elif "tmdb_description" in self.summaries: summary = get_summary("tmdb_description", self.summaries)
elif "letterboxd_description" in self.summaries: summary = get_summary("letterboxd_description", self.summaries)
elif "tmdb_summary" in self.summaries: summary = get_summary("tmdb_summary", self.summaries)
elif "tvdb_summary" in self.summaries: summary = get_summary("tvdb_summary", self.summaries)
elif "tmdb_biography" in self.summaries: summary = get_summary("tmdb_biography", self.summaries)
elif "tmdb_person" in self.summaries: summary = get_summary("tmdb_person", self.summaries)
elif "tmdb_collection_details" in self.summaries: summary = get_summary("tmdb_collection_details", self.summaries)
elif "trakt_list_details" in self.summaries: summary = get_summary("trakt_list_details", self.summaries)
elif "tmdb_list_details" in self.summaries: summary = get_summary("tmdb_list_details", self.summaries)
elif "tmdb_actor_details" in self.summaries: summary = get_summary("tmdb_actor_details", self.summaries)
elif "tmdb_crew_details" in self.summaries: summary = get_summary("tmdb_crew_details", self.summaries)
elif "tmdb_director_details" in self.summaries: summary = get_summary("tmdb_director_details", self.summaries)
elif "tmdb_producer_details" in self.summaries: summary = get_summary("tmdb_producer_details", self.summaries)
elif "tmdb_writer_details" in self.summaries: summary = get_summary("tmdb_writer_details", self.summaries)
elif "tmdb_movie_details" in self.summaries: summary = get_summary("tmdb_movie_details", self.summaries)
elif "tvdb_movie_details" in self.summaries: summary = get_summary("tvdb_movie_details", self.summaries)
elif "tvdb_show_details" in self.summaries: summary = get_summary("tvdb_show_details", self.summaries)
elif "tmdb_show_details" in self.summaries: summary = get_summary("tmdb_show_details", self.summaries)
else: summary = None
if summary:
if str(summary) != str(self.obj.summary):
edits["summary.value"] = summary
edits["summary.locked"] = 1
if "sort_title" in self.details:
if str(self.details["sort_title"]) != str(self.obj.titleSort):
edits["titleSort.value"] = self.details["sort_title"]
edits["titleSort.locked"] = 1
logger.info(f"Detail: sort_title updated Collection Sort Title to {self.details['sort_title']}")
if "content_rating" in self.details:
if str(self.details["content_rating"]) != str(self.obj.contentRating):
edits["contentRating.value"] = self.details["content_rating"]
edits["contentRating.locked"] = 1
logger.info(f"Detail: content_rating updated Collection Content Rating to {self.details['content_rating']}")
if "collection_mode" in self.details:
if int(self.obj.collectionMode) not in plex.collection_mode_keys\
or plex.collection_mode_keys[int(self.obj.collectionMode)] != self.details["collection_mode"]:
self.library.collection_mode_query(self.obj, self.details["collection_mode"])
logger.info(f"Detail: collection_mode updated Collection Mode to {self.details['collection_mode']}")
if "collection_order" in self.details:
if int(self.obj.collectionSort) not in plex.collection_order_keys\
or plex.collection_order_keys[int(self.obj.collectionSort)] != self.details["collection_order"]:
self.library.collection_order_query(self.obj, self.details["collection_order"])
logger.info(f"Detail: collection_order updated Collection Order to {self.details['collection_order']}")
add_tags = self.details["label"] if "label" in self.details else None
remove_tags = self.details["label.remove"] if "label.remove" in self.details else None
sync_tags = self.details["label.sync"] if "label.sync" in self.details else None
self.library.edit_tags("label", self.obj, add_tags=add_tags, remove_tags=remove_tags, sync_tags=sync_tags)
if len(self.item_details) > 0:
add_tags = self.item_details["item_label"] if "item_label" in self.item_details else None
remove_tags = self.item_details["item_label.remove"] if "item_label.remove" in self.item_details else None
sync_tags = self.item_details["item_label.sync"] if "item_label.sync" in self.item_details else None
for item in self.library.get_collection_items(self.obj, self.smart_label_collection):
self.library.edit_tags("label", item, add_tags=add_tags, remove_tags=remove_tags, sync_tags=sync_tags)
advance_edits = {}
for method_name, method_data in self.item_details.items():
if method_name in plex.item_advance_keys:
key, options = plex.item_advance_keys[method_name]
if getattr(item, key) != options[method_data]:
advance_edits[key] = options[method_data]
self.library.edit_item(item, item.title, "Movie" if self.library.is_movie else "Show", advance_edits, advanced=True)
if len(edits) > 0:
logger.debug(edits)
self.library.edit_query(self.obj, edits)
logger.info("Details: have been updated")
if self.library.asset_directory:
name_mapping = self.name
if "name_mapping" in self.details:
if self.details["name_mapping"]: name_mapping = self.details["name_mapping"]
else: logger.error("Collection Error: name_mapping attribute is blank")
poster_image, background_image = self.library.update_item_from_assets(self.obj, collection_mode=True, upload=False, name=name_mapping)
if poster_image:
self.posters["asset_directory"] = poster_image
if background_image:
self.backgrounds["asset_directory"] = background_image
def set_image(image_method, images, is_background=False):
message = f"{'background' if is_background else 'poster'} to [{'File' if image_method in image_file_details else 'URL'}] {images[image_method]}"
try:
self.library.upload_image(self.obj, images[image_method], poster=not is_background, url=image_method not in image_file_details)
logger.info(f"Detail: {image_method} updated collection {message}")
except BadRequest:
logger.error(f"Detail: {image_method} failed to update {message}")
if len(self.posters) > 1:
logger.info(f"{len(self.posters)} posters found:")
for p in self.posters:
logger.info(f"Method: {p} Poster: {self.posters[p]}")
if "url_poster" in self.posters: set_image("url_poster", self.posters)
elif "file_poster" in self.posters: set_image("file_poster", self.posters)
elif "tmdb_poster" in self.posters: set_image("tmdb_poster", self.posters)
elif "tmdb_profile" in self.posters: set_image("tmdb_profile", self.posters)
elif "tvdb_poster" in self.posters: set_image("tvdb_poster", self.posters)
elif "asset_directory" in self.posters: set_image("asset_directory", self.posters)
elif "tmdb_person" in self.posters: set_image("tmdb_person", self.posters)
elif "tmdb_collection_details" in self.posters: set_image("tmdb_collection_details", self.posters)
elif "tmdb_actor_details" in self.posters: set_image("tmdb_actor_details", self.posters)
elif "tmdb_crew_details" in self.posters: set_image("tmdb_crew_details", self.posters)
elif "tmdb_director_details" in self.posters: set_image("tmdb_director_details", self.posters)
elif "tmdb_producer_details" in self.posters: set_image("tmdb_producer_details", self.posters)
elif "tmdb_writer_details" in self.posters: set_image("tmdb_writer_details", self.posters)
elif "tmdb_movie_details" in self.posters: set_image("tmdb_movie_details", self.posters)
elif "tvdb_movie_details" in self.posters: set_image("tvdb_movie_details", self.posters)
elif "tvdb_show_details" in self.posters: set_image("tvdb_show_details", self.posters)
elif "tmdb_show_details" in self.posters: set_image("tmdb_show_details", self.posters)
else: logger.info("No poster to update")
if len(self.backgrounds) > 1:
logger.info(f"{len(self.backgrounds)} backgrounds found:")
for b in self.backgrounds:
logger.info(f"Method: {b} Background: {self.backgrounds[b]}")
if "url_background" in self.backgrounds: set_image("url_background", self.backgrounds, is_background=True)
elif "file_background" in self.backgrounds: set_image("file_background", self.backgrounds, is_background=True)
elif "tmdb_background" in self.backgrounds: set_image("tmdb_background", self.backgrounds, is_background=True)
elif "tvdb_background" in self.backgrounds: set_image("tvdb_background", self.backgrounds, is_background=True)
elif "asset_directory" in self.backgrounds: set_image("asset_directory", self.backgrounds, is_background=True)
elif "tmdb_collection_details" in self.backgrounds: set_image("tmdb_collection_details", self.backgrounds, is_background=True)
elif "tmdb_movie_details" in self.backgrounds: set_image("tmdb_movie_details", self.backgrounds, is_background=True)
elif "tvdb_movie_details" in self.backgrounds: set_image("tvdb_movie_details", self.backgrounds, is_background=True)
elif "tvdb_show_details" in self.backgrounds: set_image("tvdb_show_details", self.backgrounds, is_background=True)
elif "tmdb_show_details" in self.backgrounds: set_image("tmdb_show_details", self.backgrounds, is_background=True)
else: logger.info("No background to update")
def run_collections_again(self, movie_map, show_map):
self.obj = self.library.get_collection(self.name)
name, collection_items = self.library.get_collection_name_and_items(self.obj, self.smart_label_collection)
rating_keys = []
for mm in self.run_again_movies:
if mm in movie_map:
rating_keys.extend(movie_map[mm])
if self.library.is_show:
for sm in self.run_again_shows:
if sm in show_map:
rating_keys.extend(show_map[sm])
if len(rating_keys) > 0:
for rating_key in rating_keys:
try:
current = self.library.fetchItem(int(rating_key))
except (BadRequest, NotFound):
logger.error(f"Plex Error: Item {rating_key} not found")
continue
if current in collection_items:
logger.info(f"{name} Collection | = | {current.title}")
elif self.smart_label_collection:
self.library.query_data(current.addLabel, name)
else:
self.library.query_data(current.addCollection, name)
logger.info(f"{name} Collection | + | {current.title}")
logger.info(f"{len(rating_keys)} {'Movie' if self.library.is_movie else 'Show'}{'s' if len(rating_keys) > 1 else ''} Processed")
if len(self.run_again_movies) > 0:
logger.info("")
for missing_id in self.run_again_movies:
if missing_id not in movie_map:
try:
movie = self.config.TMDb.get_movie(missing_id)
except Failed as e:
logger.error(e)
continue
if self.details["show_missing"] is True:
logger.info(f"{name} Collection | ? | {movie.title} (TMDb: {missing_id})")
logger.info("")
logger.info(f"{len(self.run_again_movies)} Movie{'s' if len(self.run_again_movies) > 1 else ''} Missing")
if len(self.run_again_shows) > 0 and self.library.is_show:
logger.info("")
for missing_id in self.run_again_shows:
if missing_id not in show_map:
try:
title = str(self.config.TVDb.get_series(self.library.Plex.language, missing_id).title.encode("ascii", "replace").decode())
except Failed as e:
logger.error(e)
continue
if self.details["show_missing"] is True:
logger.info(f"{name} Collection | ? | {title} (TVDb: {missing_id})")
logger.info(f"{len(self.run_again_shows)} Show{'s' if len(self.run_again_shows) > 1 else ''} Missing")
Reference in a new issue View git blame Copy permalink