Plex
/
Plex-Meta-Manager
mirror of https://github.com/meisnate12/Plex-Meta-Manager.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

#226 added `smart_filter`

Browse Source
pull/240/head
meisnate12 3 years ago
parent 8f36901153
commit 72ac502116
3 changed files with 419 additions and 102 deletions
Show Stats Download Patch File Download Diff File

269
modules/builder.py
Unescape View File

@ -3,9 +3,11 @@ 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 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"]
@ -24,6 +26,7 @@ method_alias = {
"writers": "writer",
"years": "year"
}
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",
@ -89,7 +92,8 @@ smart_url_collection_invalid = [
"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"
"sonarr_series", "sonarr_season", "sonarr_tag", "sonarr_search", "sonarr_cutoff_search",
"filters"
]
all_details = [
"sort_title", "content_rating", "collection_mode", "collection_order",
@ -108,6 +112,7 @@ collectionless_details = [
"name_mapping", "label", "label_sync_mode", "test"
]
ignored_details = [
"smart_filter",
"smart_label",
"smart_url",
"run_again",
@ -158,6 +163,15 @@ movie_only_filters = [
"writer", "writer.not"
]
def split_attribute(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, name, data):
self.config = config
@ -356,56 +370,204 @@ class CollectionBuilder:
self.run_again = "run_again" in methods
self.collectionless = "plex_collectionless" in methods
self.smart_sort = "title.asc"
if "tmdb_person" in methods:
if 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']]}")
else:
raise Failed("Collection Error: tmdb_person attribute is blank")
self.smart_sort = "random"
self.smart_label_collection = False
if "smart_label" in methods:
self.smart_label_collection = True
if self.data[methods["smart_label"]]:
if str(self.data[methods["smart_label"]]).lower() in plex.smart_sorts:
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.info("")
logger.warning(f"Collection Error: smart_label attribute: {self.data[methods['smart_label']]} is invalid defaulting to title.asc")
logger.warning(f"Collection Error: smart_label attribute: {self.data[methods['smart_label']]} is invalid defaulting to random")
else:
logger.info("")
logger.warning("Collection Error: smart_label attribute is blank defaulting to title.asc")
logger.warning("Collection Error: smart_label attribute is blank defaulting to random")
self.smart_url = None
self.smart_url_collection = False
self.smart_type_key = None
if "smart_url" in methods:
if self.data[methods["smart_url"]]:
self.smart_url_collection = True
try:
self.smart_url = library.get_smart_filter_from_uri(self.data[methods["smart_url"]])
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")
else:
raise Failed("Collection Error: smart_url attribute is blank")
if self.smart_label_collection and self.collectionless:
raise Failed(f"Collection Error: plex_collectionless & smart_label_collection attributes cannot go together")
if "smart_filter" in methods:
logger.info("")
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}")
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 self.smart_url_collection and self.run_again:
raise Failed(f"Collection Error: run_again & smart_url_collection attributes cannot go together")
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"
logger.info(f"Smart {smart_type.capitalize()[:-1]} Filter")
self.smart_type_key, smart_sorts = plex.smart_types[smart_type]
self.smart = self.smart_url_collection or self.smart_label_collection
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"]]
logger.info(f"Sort By: {smart_sort}")
if "tmdb_person" in methods:
if 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
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"]]
logger.info(f"Limit: {limit}")
def _filter(filter_dict, 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_attribute(smart_key)
def build_url_arg(arg, mod=None, arg_s=None, mod_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.search_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]
display_line = f"{indent}{smart.title().replace('_', ' ')} {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, 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))
else:
if smart in ["title", "episode_title"] 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:
final_tmdb_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"]:
final_tmdb_values.append(tmdb_name)
else:
final_tmdb_values.append(tmdb_value)
elif smart == "studio":
final_tmdb_values = util.get_list(smart_data, split=False)
else:
final_tmdb_values = util.get_list(smart_data)
results_list = self.library.validate_search_list(final_tmdb_values, smart, fail=True, title=False, pairs=True)
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]], is_all=base_all)
util.print_multiline(f"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:
raise Failed(f"Collection Error: No valid TMDb Person IDs in {self.data[methods['tmdb_person']]}")
else:
raise Failed("Collection Error: tmdb_person attribute is blank")
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 "trakt" in method_key.lower() and not config.Trakt: raise Failed(f"Collection Error: {method_key} requires Trakt todo be configured")
@ -441,9 +603,9 @@ class CollectionBuilder:
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_collection and method_name in all_builders:
elif self.smart_url and method_name in all_builders:
raise Failed(f"Collection Error: {method_name} builder not allowed when using smart_url")
elif self.smart_url_collection and method_name in smart_url_collection_invalid:
elif self.smart_url and method_name in smart_url_collection_invalid:
raise Failed(f"Collection Error: {method_name} detail not allowed when using smart_url")
elif method_name == "summary":
self.summaries[method_name] = method_data
@ -571,13 +733,13 @@ class CollectionBuilder:
self.sonarr_options["tag"] = util.get_list(method_data)
elif method_name in ["title", "title.and", "title.not", "title.begins", "title.ends"]:
self.methods.append(("plex_search", [{method_name: util.get_list(method_data, split=False)}]))
elif method_name in ["year.greater", "year.less"]:
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.greater", "duration.less"]:
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.greater", "user_rating.less", "critic_rating.greater", "critic_rating.less", "audience_rating.greater", "audience_rating.less"]:
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)}]))
@ -721,15 +883,15 @@ class CollectionBuilder:
elif method_name == "plex_search":
searches = {}
for search_name, search_data in method_data.items():
search, modifier = os.path.splitext(str(search_name).lower())
if search in method_alias:
search = method_alias[search]
logger.warning(f"Collection Warning: {str(search_name).lower()} plex search attribute will run as {search}{modifier if modifier else ''}")
search_final = f"{search}{modifier}"
search, modifier, search_final = split_attribute(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")
if search_final in plex.show_only_searches and self.library.is_movie:
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_final not in plex.searches:
raise Failed(f"Collection Error: {search_final} is not a valid plex search attribute")
elif search_data is None:
raise Failed(f"Collection Error: {search_final} plex search attribute is blank")
elif search == "sort_by":
@ -746,9 +908,7 @@ class CollectionBuilder:
searches[search] = search_data
elif search == "title" and modifier in ["", ".and", ".not", ".begins", ".ends"]:
searches[search_final] = util.get_list(search_data, split=False)
elif (search == "studio" and modifier in ["", ".and", ".not", ".begins", ".ends"]) \
or (search in ["actor", "audio_language", "collection", "content_rating", "country", "director", "genre", "label", "network", "producer", "subtitle_language", "writer"] and modifier in ["", ".and", ".not"]) \
or (search == "resolution" and modifier in [""]):
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):
@ -758,31 +918,24 @@ class CollectionBuilder:
else:
final_values.append(value)
else:
final_values = search_data
final_values = util.get_list(search_data)
valid_values = self.library.validate_search_list(final_values, search)
if valid_values:
searches[search_final] = valid_values
else:
logger.warning(f"Collection Warning: No valid {search} values found in {final_values}")
elif search == "year" and modifier in [".greater", ".less"]:
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"] and modifier in ["", ".not"]) or (search in ["duration"] and modifier in [".greater", ".less"]):
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 [".greater", ".less"]:
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 == "decade" and modifier in [""]) or (search == "year" and modifier in ["", ".not"]):
elif search in ["decade", "year"] and modifier in ["", ".not"]:
searches[search_final] = util.get_year_list(search_data, current_year, search_final)
elif (search in ["title", "studio"] and modifier not in ["", ".and", ".not", ".begins", ".ends"]) \
or (search in ["actor", "audio_language", "collection", "content_rating", "country", "director", "genre", "label", "network", "producer", "subtitle_language", "writer"] and modifier not in ["", ".and", ".not"]) \
or (search in ["resolution", "decade"] and modifier not in [""]) \
or (search in ["added", "originally_available"] and modifier not in ["", ".not", ".before", ".after"]) \
or (search in ["duration", "user_rating", "critic_rating", "audience_rating"] and modifier not in [".greater", ".less"]) \
or (search in ["year"] and modifier not in ["", ".not", ".greater", ".less"]):
raise Failed(f"Collection Error: modifier: {modifier} not supported with the {search} plex search attribute")
else:
raise Failed(f"Collection Error: {search_final} plex search attribute not supported")
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:
@ -1041,7 +1194,7 @@ class CollectionBuilder:
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_collection:
if self.smart_url:
self.add_to_radarr = False
self.add_to_sonarr = False
@ -1079,8 +1232,9 @@ class CollectionBuilder:
else:
missing_shows.append(show_id)
return items_found_inside
logger.info("")
logger.debug("")
logger.debug(f"Value: {value}")
logger.info("")
if "plex" in method:
items = self.library.get_items(method, value)
items_found += len(items)
@ -1097,6 +1251,7 @@ class CollectionBuilder:
elif "trakt" in method: items_found += check_map(self.config.Trakt.get_items(method, value, self.library.is_movie))
else: logger.error(f"Collection Error: {method} method not supported")
logger.info("")
if len(items) > 0:
rating_key_map = self.library.add_to_collection(collection_obj if collection_obj else collection_name, items, self.filters, self.details["show_filtered"], self.smart_label_collection, rating_key_map, movie_map, show_map)
else:

7
modules/config.py
Unescape View File

@ -576,6 +576,7 @@ class Config:
try:
builder = CollectionBuilder(self, library, mapping_name, collection_attrs)
except Failed as f:
util.print_stacktrace()
util.print_multiline(f, error=True)
continue
except Exception as e:
@ -614,12 +615,12 @@ class Config:
logger.info("")
logger.info(f"Collection Filter {f[0]}: {f[1]}")
if not builder.smart_url_collection:
if not builder.smart_url:
builder.run_methods(collection_obj, collection_name, rating_key_map, movie_map, show_map)
try:
if not collection_obj and builder.smart_url_collection:
library.create_smart_collection(collection_name, builder.smart_url)
if not collection_obj and builder.smart_url:
library.create_smart_collection(collection_name, builder.smart_type_key, builder.smart_url)
elif not collection_obj and builder.smart_label_collection:
library.create_smart_labels(collection_name, sort=builder.smart_sort)
plex_collection = library.get_collection(collection_name)

245
modules/plex.py
Unescape View File

@ -22,7 +22,26 @@ search_translation = {
"originally_available": "originallyAvailableAt",
"audience_rating": "audienceRating",
"critic_rating": "rating",
"user_rating": "userRating"
"user_rating": "userRating",
"plays": "viewCount",
"episode_title": "episode.title",
"episode_added": "episode.addedAt",
"episode_originally_available": "episode.originallyAvailableAt",
"episode_year": "episode.year",
"episode_user_rating": "episode.userRating",
"episode_plays": "episode.viewCount"
}
modifier_translation = {
"": "",
".not": "!",
".gt": "%3E%3E",
".gte": "%3E",
".lt": "%3C%3C",
".lte": "%3C",
".before": "%3C%3C",
".after": "%3E%3E",
".begins": "%3C",
".ends": "%3E"
}
episode_sorting_options = {"default": "-1", "oldest": "0", "newest": "1"}
keep_episodes_options = {"all": 0, "5_latest": 5, "3_latest": 3, "latest": 1, "past_3": -3, "past_7": -7, "past_30": -30}
@ -87,13 +106,14 @@ searches = [
"subtitle_language", "subtitle_language.and", "subtitle_language.not",
"writer", "writer.and", "writer.not",
"decade", "resolution",
"added.before", "added.after",
"added", "added.not", "added.before", "added.after",
"originally_available", "originally_available.not",
"originally_available.before", "originally_available.after",
"duration.greater", "duration.less",
"user_rating.greater", "user_rating.less",
"audience_rating.greater", "audience_rating.less",
"critic_rating.greater", "critic_rating.less",
"year", "year.not", "year.greater", "year.less"
"duration.gt", "duration.gte", "duration.lt", "duration.lte",
"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"\
"year", "year.not", "year.gt", "year.gte", "year.lt", "year.lte"
]
movie_only_searches = [
"audio_language", "audio_language.and", "audio_language.not",
@ -101,7 +121,7 @@ movie_only_searches = [
"subtitle_language", "subtitle_language.and", "subtitle_language.not",
"decade", "resolution",
"originally_available.before", "originally_available.after",
"duration.greater", "duration.less"
"duration.gt", "duration.gte", "duration.lt", "duration.lte"
]
show_only_searches = [
"network", "network.and", "network.not",
@ -112,19 +132,6 @@ tmdb_searches = [
"producer", "producer.and", "producer.not",
"writer", "writer.and", "writer.not"
]
smart_sorts = {
"title.asc": "titleSort", "title.desc": "titleSort%3Adesc",
"year.asc": "year", "year.desc": "year%3Adesc",
"originally_available.asc": "originallyAvailableAt", "originally_available.desc": "originallyAvailableAt%3Adesc",
"critic_rating.asc": "rating", "critic_rating.desc": "rating%3Adesc",
"audience_rating.asc": "audienceRating", "audience_rating.desc": "audienceRating%3Adesc",
"user_rating.asc": "userRating", "user_rating.desc": "userRating%3Adesc",
"content_rating.asc": "contentRating", "content_rating.desc": "contentRating%3Adesc",
"duration.asc": "duration", "duration.desc": "duration%3Adesc",
"plays.asc": "viewCount", "plays.desc": "viewCount%3Adesc",
"added.asc": "addedAt", "added.desc": "addedAt%3Adesc",
"random": "random"
}
sorts = {
None: None,
"title.asc": "titleSort:asc", "title.desc": "titleSort:desc",
@ -141,8 +148,144 @@ modifiers = {
".ends": ">",
".before": "<<",
".after": ">>",
".greater": ">>",
".less": "<<"
".gt": ">>",
".gte": "__gte",
".lt": "<<",
".lte": "__lte"
}
mod_displays = {
"": "is",
".not": "is not",
".begins": "begins with",
".ends": "ends with",
".before": "is before",
".after": "is after",
".gt": "is greater than",
".gte": "is greater than or equal",
".lt": "is less than",
".lte": "is less than or equal"
}
tags = [
"actor",
"audio_language",
"collection",
"content_rating",
"country",
"director",
"genre",
"label",
"network",
"producer",
"resolution",
"studio",
"subtitle_language",
"writer"
]
smart_searches = [
"all", "any",
"title", "title.not", "title.begins", "title.ends",
"studio", "studio.not", "studio.begins", "studio.ends",
"actor", "actor.not",
"audio_language", "audio_language.not",
"collection", "collection.not",
"content_rating", "content_rating.not",
"country", "country.not",
"director", "director.not",
"genre", "genre.not",
"label", "label.not",
"network", "network.not",
"producer", "producer.not",
"subtitle_language", "subtitle_language.not",
"writer", "writer.not",
"decade", "resolution",
"added", "added.not", "added.before", "added.after",
"originally_available", "originally_available.not",
"originally_available.before", "originally_available.after",
"plays.gt", "plays.gte", "plays.lt", "plays.lte",
"duration.gt", "duration.gte", "duration.lt", "duration.lte",
"user_rating.gt", "user_rating.gte", "user_rating.lt", "user_rating.lte",
"audience_rating.gt", "audience_rating.gte", "audience_rating.lt","audience_rating.lte",
"critic_rating.gt", "critic_rating.gte", "critic_rating.lt","critic_rating.lte",
"year", "year.not", "year.gt", "year.gte", "year.lt","year.lte",
"episode_title", "episode_title.not", "episode_title.begins", "episode_title.ends",
"episode_added", "episode_added.not", "episode_added.before", "episode_added.after",
"episode_originally_available", "episode_originally_available.not",
"episode_originally_available.before", "episode_originally_available.after",
"episode_year", "episode_year.not", "episode_year.gt", "episode_year.gte", "episode_year.lt","episode_year.lte",
"episode_user_rating.gt", "episode_user_rating.gte", "episode_user_rating.lt","episode_user_rating.lte",
"episode_plays.gt", "episode_plays.gte", "episode_plays.lt", "episode_plays.lte"
]
movie_only_smart_searches = [
"country", "country.not",
"director", "director.not",
"producer", "producer.not",
"writer", "writer.not",
"decade",
"originally_available", "originally_available.not",
"originally_available.before", "originally_available.after",
"plays.gt", "plays.gte", "plays.lt", "plays.lte",
"duration.gt", "duration.gte", "duration.lt", "duration.lte"
]
show_only_smart_searches = [
"episode_title", "episode_title.not", "episode_title.begins", "episode_title.ends",
"episode_added", "episode_added.not", "episode_added.before", "episode_added.after",
"episode_originally_available", "episode_originally_available.not",
"episode_originally_available.before", "episode_originally_available.after",
"episode_year", "episode_year.not", "episode_year.gt", "episode_year.gte", "episode_year.lt","episode_year.lte",
"episode_user_rating.gt", "episode_user_rating.gte", "episode_user_rating.lt","episode_user_rating.lte",
"episode_plays.gt", "episode_plays.gte", "episode_plays.lt", "episode_plays.lte"
]
movie_smart_sorts = {
"title.asc": "titleSort", "title.desc": "titleSort%3Adesc",
"year.asc": "year", "year.desc": "year%3Adesc",
"originally_available.asc": "originallyAvailableAt", "originally_available.desc": "originallyAvailableAt%3Adesc",
"critic_rating.asc": "rating", "critic_rating.desc": "rating%3Adesc",
"audience_rating.asc": "audienceRating", "audience_rating.desc": "audienceRating%3Adesc",
"user_rating.asc": "userRating", "user_rating.desc": "userRating%3Adesc",
"content_rating.asc": "contentRating", "content_rating.desc": "contentRating%3Adesc",
"duration.asc": "duration", "duration.desc": "duration%3Adesc",
"plays.asc": "viewCount", "plays.desc": "viewCount%3Adesc",
"added.asc": "addedAt", "added.desc": "addedAt%3Adesc",
"random": "random"
}
show_smart_sorts = {
"title.asc": "titleSort", "title.desc": "titleSort%3Adesc",
"year.asc": "year", "year.desc": "year%3Adesc",
"originally_available.asc": "originallyAvailableAt", "originally_available.desc": "originallyAvailableAt%3Adesc",
"critic_rating.asc": "rating", "critic_rating.desc": "rating%3Adesc",
"audience_rating.asc": "audienceRating", "audience_rating.desc": "audienceRating%3Adesc",
"user_rating.asc": "userRating", "user_rating.desc": "userRating%3Adesc",
"content_rating.asc": "contentRating", "content_rating.desc": "contentRating%3Adesc",
"added.asc": "addedAt", "added.desc": "addedAt%3Adesc",
"episode_added.asc": "episode.addedAt", "episode_added.desc": "episode.addedAt%3Adesc",
"random": "random"
}
season_smart_sorts = {
"season.asc": "season.index%2Cseason.titleSort", "season.desc": "season.index%3Adesc%2Cseason.titleSort",
"show.asc": "show.titleSort%2Cindex", "show.desc": "show.titleSort%3Adesc%2Cindex",
"user_rating.asc": "userRating", "user_rating.desc": "userRating%3Adesc",
"added.asc": "addedAt", "added.desc": "addedAt%3Adesc",
"random": "random"
}
episode_smart_sorts = {
"title.asc": "titleSort", "title.desc": "titleSort%3Adesc",
"show.asc": "show.titleSort%2Cseason.index%3AnullsLast%2Cepisode.index%3AnullsLast%2Cepisode.originallyAvailableAt%3AnullsLast%2Cepisode.titleSort%2Cepisode.id",
"show.desc": "show.titleSort%3Adesc%2Cseason.index%3AnullsLast%2Cepisode.index%3AnullsLast%2Cepisode.originallyAvailableAt%3AnullsLast%2Cepisode.titleSort%2Cepisode.id",
"year.asc": "year", "year.desc": "year%3Adesc",
"originally_available.asc": "originallyAvailableAt", "originally_available.desc": "originallyAvailableAt%3Adesc",
"critic_rating.asc": "rating", "critic_rating.desc": "rating%3Adesc",
"audience_rating.asc": "audienceRating", "audience_rating.desc": "audienceRating%3Adesc",
"user_rating.asc": "userRating", "user_rating.desc": "userRating%3Adesc",
"duration.asc": "duration", "duration.desc": "duration%3Adesc",
"plays.asc": "viewCount", "plays.desc": "viewCount%3Adesc",
"added.asc": "addedAt", "added.desc": "addedAt%3Adesc",
"random": "random"
}
smart_types = {
"movies": (1, movie_smart_sorts),
"shows": (2, show_smart_sorts),
"seasons": (3, season_smart_sorts),
"episodes": (4, episode_smart_sorts),
}
class PlexAPI:
@ -282,12 +425,12 @@ class PlexAPI:
item.uploadArt(filepath=location)
@retry(stop_max_attempt_number=6, wait_fixed=10000, retry_on_exception=util.retry_if_not_failed)
def get_search_choices(self, search_name):
def get_search_choices(self, search_name, title=True):
try:
choices = {}
for choice in self.Plex.listFilterChoices(search_name):
choices[choice.title.lower()] = choice.title
choices[choice.key.lower()] = choice.title
choices[choice.title.lower()] = choice.title if title else choice.key
choices[choice.key.lower()] = choice.title if title else choice.key
return choices
except NotFound:
raise Failed(f"Collection Error: plex search attribute: {search_name} only supported with Plex's New TV Agent")
@ -307,12 +450,14 @@ class PlexAPI:
labels = self.get_labels()
if title not in labels:
raise Failed(f"Plex Error: Label: {title} does not exist")
uri_args = f"?type=1&sort={smart_sorts[sort]}&label={labels[title]}"
self.create_smart_collection(title, uri_args)
smart_type = 1 if self.is_movie else 2
sort_type = movie_smart_sorts[sort] if self.is_movie else show_smart_sorts[sort]
uri_args = f"?type={smart_type}&sort={sort_type}&label={labels[title]}"
self.create_smart_collection(title, smart_type, uri_args)
def create_smart_collection(self, title, uri_args):
def create_smart_collection(self, title, smart_type, uri_args):
args = {
"type": 1,
"type": smart_type,
"title": title,
"smart": 1,
"sectionId": self.Plex.key,
@ -322,7 +467,8 @@ class PlexAPI:
def get_smart_filter_from_uri(self, uri):
smart_filter = parse.parse_qs(parse.urlparse(uri.replace("/#!/", "/")).query)["key"][0]
return self.build_smart_filter(smart_filter[smart_filter.index("?"):])
args = smart_filter[smart_filter.index("?"):]
return self.build_smart_filter(args), int(args[args.index("type=") + 5:args.index("type=") + 6])
def build_smart_filter(self, uri_args):
return f"server://{self.PlexServer.machineIdentifier}/com.plexapp.plugins.library/library/sections/{self.Plex.key}/all{uri_args}"
@ -337,13 +483,18 @@ class PlexAPI:
smart_filter = self.get_collection(collection)._data.attrib.get('content')
return smart_filter[smart_filter.index("?"):]
def validate_search_list(self, data, search_name):
def validate_search_list(self, data, search_name, fail=False, title=True, pairs=False):
final_search = search_translation[search_name] if search_name in search_translation else search_name
search_choices = self.get_search_choices(final_search)
search_choices = self.get_search_choices(final_search, title=title)
valid_list = []
for value in util.get_list(data):
if str(value).lower() in search_choices:
valid_list.append(search_choices[str(value).lower()])
if pairs:
valid_list.append((value, search_choices[str(value).lower()]))
else:
valid_list.append(search_choices[str(value).lower()])
elif fail:
raise Failed(f"Plex Error: {search_name}: {value} not found")
else:
logger.error(f"Plex Error: {search_name}: {value} not found")
return valid_list
@ -400,9 +551,9 @@ class PlexAPI:
final_mod = ">>"
elif search in ["added", "originally_available"] and modifier == ".not":
final_mod = "<<"
elif search in ["critic_rating", "audience_rating"] and modifier == ".greater":
final_mod = "__gte"
elif search in ["critic_rating", "audience_rating"] and modifier == ".less":
elif search in ["critic_rating", "audience_rating"] and modifier == ".gt":
final_mod = "__gt"
elif search in ["critic_rating", "audience_rating"] and modifier == ".lt":
final_mod = "__lt"
else:
final_mod = modifiers[modifier] if modifier in modifiers else ""
@ -416,7 +567,7 @@ class PlexAPI:
search_terms[final_method] = search_data
if status_message:
if search in ["added", "originally_available"] or modifier in [".greater", ".less", ".before", ".after"]:
if search in ["added", "originally_available"] or modifier in [".gt", ".gte", ".lt", ".lte", ".before", ".after"]:
ors = f"{search_method}({search_data}"
else:
ors = ""
@ -438,22 +589,32 @@ class PlexAPI:
return self.search(sort=sorts[search_sort], maxresults=search_limit, **search_terms)
elif method == "plex_collectionless":
good_collections = []
if status_message:
logger.info("Collections Excluded")
for col in self.get_all_collections():
keep_collection = True
for pre in data["exclude_prefix"]:
if col.title.startswith(pre) or (col.titleSort and col.titleSort.startswith(pre)):
keep_collection = False
logger.info(f"Collection Excluded: {col.title} by prefix {pre}")
if status_message:
logger.info(f"{col.title} excluded by prefix match {pre}")
break
if keep_collection:
for ext in data["exclude"]:
if col.title == ext or (col.titleSort and col.titleSort == ext):
keep_collection = False
logger.info(f"Collection Excluded: {col.title} by exact match")
if status_message:
logger.info(f"{col.title} excluded by exact match")
break
if keep_collection:
logger.info(f"Collection Passed: {col.title}")
good_collections.append(col.index)
good_collections.append(col)
if status_message:
logger.info("")
logger.info("Collections Not Excluded (Items in these collections are not added to Collectionless)")
for col in good_collections:
logger.info(col.title)
collection_indexes = [c.index for c in good_collections]
all_items = self.get_all()
length = 0
for i, item in enumerate(all_items, 1):
@ -461,7 +622,7 @@ class PlexAPI:
add_item = True
self.query(item.reload)
for collection in item.collections:
if collection.id in good_collections:
if collection.id in collection_indexes:
add_item = False
break
if add_item:

Write Preview
Loading…
Cancel
Save