plex_search update!

This commit is contained in:
meisnate12 2021-03-26 01:43:11 -04:00
parent eeb2320f62
commit ad31c59cff
5 changed files with 199 additions and 114 deletions

View file

@ -305,23 +305,37 @@ class CollectionBuilder:
else: raise Failed(f"Collection Error: {method_name} attribute must be either true or false")
elif method_name in util.all_details:
self.details[method_name] = 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 ["decade", "year.greater", "year.less"]:
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 ["duration.greater", "duration.less", "rating.greater", "rating.less"]:
self.methods.append(("plex_search", [{method_name: util.check_number(method_data, method_name, minimum=0)}]))
elif method_name in ["year", "year.not"]:
self.methods.append(("plex_search", [[(method_name, util.get_year_list(self.data[m], method_name))]]))
elif method_name in ["decade", "decade.not"]:
self.methods.append(("plex_search", [[(method_name, util.get_int_list(self.data[m], util.remove_not(method_name)))]]))
self.methods.append(("plex_search", [{method_name: util.get_year_list(method_data, current_year, method_name)}]))
elif method_name in util.tmdb_searches:
final_values = []
for value in util.get_list(self.data[m]):
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)
self.methods.append(("plex_search", [[(method_name, final_values)]]))
elif method_name == "title":
self.methods.append(("plex_search", [[(method_name, util.get_list(self.data[m], split=False))]]))
self.methods.append(("plex_search", [{method_name: self.library.validate_search_list(final_values, os.path.splitext(method_name)[0])}]))
elif method_name in util.plex_searches:
self.methods.append(("plex_search", [[(method_name, util.get_list(self.data[m]))]]))
if method_name in util.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
self.methods.append(("plex_search", [{method_name: self.library.validate_search_list(final_values, os.path.splitext(method_name)[0])}]))
elif method_name == "plex_all":
self.methods.append((method_name, [""]))
elif method_name == "plex_collection":
@ -436,31 +450,63 @@ class CollectionBuilder:
new_dictionary["exclude"] = exact_list
self.methods.append((method_name, [new_dictionary]))
elif method_name == "plex_search":
searches = []
used = []
for s in self.data[m]:
if s in util.method_alias or (s.endswith(".not") and s[:-4] in util.method_alias):
search = (util.method_alias[s[:-4]] + s[-4:]) if s.endswith(".not") else util.method_alias[s]
logger.warning(f"Collection Warning: {s} plex search attribute will run as {search}")
searches = {}
for search_name, search_data in method_data:
search, modifier = os.path.splitext(str(search_name).lower())
if search in util.method_alias:
search = util.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}"
if search_final in util.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_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 util.plex_sort:
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 == "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", "producer", "subtitle_language", "writer"] and modifier in ["", ".and", ".not"]) \
or (search == "resolution" and modifier in [""]):
if search_final in util.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 = search_data
searches[search_final] = self.library.validate_search_list(final_values, search)
elif (search == "decade" and modifier in [""]) \
or (search == "year" and modifier in [".greater", ".less"]):
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 ["duration", "rating"] and modifier in [".greater", ".less"]:
searches[search_final] = util.check_number(search_data, search_final, minimum=0)
elif search == "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", "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 [".before", ".after"]) \
or (search in ["duration", "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:
search = s
if search in util.movie_only_searches and self.library.is_show:
raise Failed(f"Collection Error: {search} plex search attribute only works for movie libraries")
elif util.remove_not(search) in used:
raise Failed(f"Collection Error: Only one instance of {search} can be used try using it as a filter instead")
elif search in ["year", "year.not"]:
years = util.get_year_list(self.data[m][s], search)
if len(years) > 0:
used.append(util.remove_not(search))
searches.append((search, util.get_int_list(self.data[m][s], util.remove_not(search))))
elif search == "title":
used.append(util.remove_not(search))
searches.append((search, util.get_list(self.data[m][s], split=False)))
elif search in util.plex_searches:
used.append(util.remove_not(search))
searches.append((search, util.get_list(self.data[m][s])))
else:
logger.error(f"Collection Error: {search} plex search attribute not supported")
raise Failed(f"Collection Error: {search_final} plex search attribute not supported")
self.methods.append((method_name, [searches]))
elif method_name == "tmdb_discover":
new_dictionary = {"limit": 100}
@ -758,38 +804,31 @@ class CollectionBuilder:
items_found += len(items)
elif method == "plex_search":
search_terms = {}
title_searches = None
has_processed = False
for search_method, search_list in value:
if search_method == "title":
search_limit = None
search_sort = None
for search_method, search_data in value:
if search_method == "limit":
search_limit = search_data
elif search_method == "sort_by":
search_sort = util.plex_sort[search_data]
else:
search, modifier = os.path.splitext(str(search_method).lower())
final_search = util.search_alias[search] if search in util.search_alias else search
final_mod = util.plex_modifiers[modifier] if modifier in util.plex_modifiers else ""
final_method = f"{final_search}{final_mod}"
search_terms[final_method] = search_data * 60000 if final_search == "duration" else search_data
ors = ""
for o, param in enumerate(search_list):
ors += f"{' OR ' if o > 0 else ''}{param}"
title_searches = search_list
logger.info(f"Processing {pretty}: title({ors})")
has_processed = True
break
for search_method, search_list in value:
if search_method != "title":
final_method = search_method[:-4] + "!" if search_method[-4:] == ".not" else search_method
if self.library.is_show:
final_method = "show." + final_method
search_terms[final_method] = search_list
ors = ""
for o, param in enumerate(search_list):
or_des = " OR " if o > 0 else f"{search_method}("
conjunction = " AND " if final_mod == "&" else " OR "
for o, param in enumerate(search_data):
or_des = conjunction if o > 0 else f"{search_method}("
ors += f"{or_des}{param}"
if has_processed:
logger.info(f"\t\t AND {ors})")
else:
logger.info(f"Processing {pretty}: {ors})")
has_processed = True
if title_searches:
items = []
for title_search in title_searches:
items.extend(self.library.Plex.search(title_search, **search_terms))
else:
items = self.library.Plex.search(**search_terms)
items = self.library.Plex.search(sort=search_sort, maxresults=search_limit, **search_terms)
items_found += len(items)
elif method == "plex_collectionless":
good_collections = []

View file

@ -106,8 +106,8 @@ class Config:
if isinstance(data[attribute], bool): return data[attribute]
else: message = f"{text} must be either true or false"
elif var_type == "int":
if isinstance(data[attribute], int) and data[attribute] > 0: return data[attribute]
else: message = f"{text} must an integer > 0"
if isinstance(data[attribute], int) and data[attribute] >= 0: return data[attribute]
else: message = f"{text} must an integer >= 0"
elif var_type == "path":
if os.path.exists(os.path.abspath(data[attribute])): return data[attribute]
else: message = f"Path {os.path.abspath(data[attribute])} does not exist"

View file

@ -14,25 +14,34 @@ logger = logging.getLogger("Plex Meta Manager")
class PlexAPI:
def __init__(self, params, TMDb, TVDb):
try: self.PlexServer = PlexServer(params["plex"]["url"], params["plex"]["token"], timeout=params["plex"]["timeout"])
except Unauthorized: raise Failed("Plex Error: Plex token is invalid")
except ValueError as e: raise Failed(f"Plex Error: {e}")
try:
self.PlexServer = PlexServer(params["plex"]["url"], params["plex"]["token"], timeout=params["plex"]["timeout"])
except Unauthorized:
raise Failed("Plex Error: Plex token is invalid")
except ValueError as e:
raise Failed(f"Plex Error: {e}")
except requests.exceptions.ConnectionError:
util.print_stacktrace()
raise Failed("Plex Error: Plex url is invalid")
self.is_movie = params["library_type"] == "movie"
self.is_show = params["library_type"] == "show"
self.Plex = next((s for s in self.PlexServer.library.sections() if s.title == params["name"] and ((self.is_movie and isinstance(s, MovieSection)) or (self.is_show and isinstance(s, ShowSection)))), None)
if not self.Plex: raise Failed(f"Plex Error: Plex Library {params['name']} not found")
try: self.data, ind, bsi = yaml.util.load_yaml_guess_indent(open(params["metadata_path"], encoding="utf-8"))
except yaml.scanner.ScannerError as e: raise Failed(f"YAML Error: {util.tab_new_lines(e)}")
if not self.Plex:
raise Failed(f"Plex Error: Plex Library {params['name']} not found")
try:
self.data, ind, bsi = yaml.util.load_yaml_guess_indent(open(params["metadata_path"], encoding="utf-8"))
except yaml.scanner.ScannerError as e:
raise Failed(f"YAML Error: {util.tab_new_lines(e)}")
def get_dict(attribute):
if attribute in self.data:
if self.data[attribute]:
if isinstance(self.data[attribute], dict): return self.data[attribute]
else: logger.warning(f"Config Warning: {attribute} must be a dictionary")
else: logger.warning(f"Config Warning: {attribute} attribute is blank")
if isinstance(self.data[attribute], dict):
return self.data[attribute]
else:
logger.warning(f"Config Warning: {attribute} must be a dictionary")
else:
logger.warning(f"Config Warning: {attribute} attribute is blank")
return None
self.metadata = get_dict("metadata")
@ -81,6 +90,21 @@ class PlexAPI:
def server_search(self, data):
return self.PlexServer.search(data)
def get_search_choices(self, search_name, key=False):
if key: return {c.key.lower(): c.key for c in self.Plex.listFilterChoices(search_name)}
else: return {c.title.lower(): c.title for c in self.Plex.listFilterChoices(search_name)}
def validate_search_list(self, data, search_name):
final_search = util.search_alias[search_name] if search_name in util.search_alias else search_name
search_choices = self.get_search_choices(final_search, key=final_search.endswith("Language"))
valid_list = []
for value in util.get_list(data):
if str(value).lower in search_choices:
valid_list.append(search_choices[str(value).lower])
else:
raise Failed(f"Plex Error: No {search_name}: {value} found")
return valid_list
def get_all_collections(self):
return self.Plex.search(libtype="collection")

View file

@ -29,11 +29,20 @@ method_alias = {
"decades": "decade",
"directors": "director",
"genres": "genre",
"labels": "label",
"studios": "studio", "network": "studio", "networks": "studio",
"producers": "producer",
"writers": "writer",
"years": "year"
}
search_alias = {
"audio_language": "audioLanguage",
"content_rating": "contentRating",
"subtitle_language": "subtitleLanguage",
"added": "addedAt",
"originally_available": "originallyAvailableAt",
"rating": "userRating"
}
filter_alias = {
"actor": "actors",
"collection": "collections",
@ -330,18 +339,6 @@ dictionary_lists = [
"tautulli_watched",
"tmdb_discover"
]
plex_searches = [
"actor", #"actor.not", # Waiting on PlexAPI to fix issue
"country", #"country.not",
"decade", #"decade.not",
"director", #"director.not",
"genre", #"genre.not",
"producer", #"producer.not",
"studio", #"studio.not",
"title",
"writer", #"writer.not"
"year" #"year.not",
]
show_only_lists = [
"tmdb_network",
"tmdb_show",
@ -360,20 +357,6 @@ movie_only_lists = [
"tvdb_movie",
"tvdb_movie_details"
]
movie_only_searches = [
"actor", "actor.not",
"country", "country.not",
"decade", "decade.not",
"director", "director.not",
"producer", "producer.not",
"writer", "writer.not"
]
tmdb_searches = [
"actor", "actor.not",
"director", "director.not",
"producer", "producer.not",
"writer", "writer.not"
]
count_lists = [
"anidb_popular",
"anilist_popular",
@ -452,6 +435,59 @@ tmdb_type = {
"tmdb_writer": "Person",
"tmdb_writer_details": "Person"
}
plex_searches = [
"title", "title.and", "title.not", "title.begins", "title.ends",
"studio", "studio.and", "studio.not", "studio.begins", "studio.ends",
"actor", "actor.and", "actor.not",
"audio_language", "audio_language.and", "audio_language.not",
"collection", "collection.and", "collection.not",
"content_rating", "content_rating.and", "content_rating.not",
"country", "country.and", "country.not",
"director", "director.and", "director.not",
"genre", "genre.and", "genre.not",
"label", "label.and", "label.not",
"producer", "producer.and", "producer.not",
"subtitle_language", "subtitle_language.and", "subtitle_language.not",
"writer", "writer.and", "writer.not",
"decade", "resolution",
"added.before", "added.after",
"originally_available.before", "originally_available.after",
"duration.greater", "duration.less",
"rating.greater", "rating.less",
"year", "year.not", "year.greater", "year.less"
]
plex_sort = {
"title.asc": "titleSort:asc", "title.desc": "titleSort:desc",
"originally_available.asc": "originallyAvailableAt:asc", "originally_available.desc": "originallyAvailableAt:desc",
"critic_rating.asc": "rating:asc", "critic_rating.desc": "rating:desc",
"audience_rating.asc": "audienceRating:asc", "audience_rating.desc": "audienceRating:desc",
"duration.asc": "duration:asc", "duration.desc": "duration:desc",
"added.asc": "addedAt:asc", "added.desc": "addedAt:desc"
}
plex_modifiers = {
".and": "&",
".not": "!",
".begins": "<",
".ends": ">",
".before": "<<",
".after": ">>",
".greater": ">>",
".less": "<<"
}
movie_only_searches = [
"audio_language", "audio_language.and", "audio_language.not",
"country", "country.and", "country.not",
"subtitle_language", "subtitle_language.and", "subtitle_language.not",
"decade", "resolution",
"originally_available.before", "originally_available.after",
"duration.greater", "duration.less"
]
tmdb_searches = [
"actor", "actor.and", "actor.not",
"director", "director.and", "director.not",
"producer", "producer.and", "producer.not",
"writer", "writer.and", "writer.not"
]
all_filters = [
"actor", "actor.not",
"audio_language", "audio_language.not",
@ -612,25 +648,11 @@ def get_int_list(data, id_type):
except Failed as e: logger.error(e)
return int_values
def get_year_list(data, method):
values = get_list(data)
def get_year_list(data, current_year, method):
final_years = []
current_year = datetime.now().year
values = get_list(data)
for value in values:
try:
if "-" in value:
year_range = re.search("(\\d{4})-(\\d{4}|NOW)", str(value))
start = check_year(year_range.group(1), current_year, method)
end = current_year if year_range.group(2) == "NOW" else check_year(year_range.group(2), current_year, method)
if int(start) > int(end):
raise Failed(f"Collection Error: {method} starting year: {start} cannot be greater then ending year {end}")
else:
for i in range(int(start), int(end) + 1):
final_years.append(int(i))
else:
final_years.append(check_year(value, current_year, method))
except AttributeError:
raise Failed(f"Collection Error: {method} failed to parse year from {value}")
final_years.append(check_year(value, current_year, method))
return final_years
def check_year(year, current_year, method):
@ -653,9 +675,9 @@ def check_number(value, method, number_type="int", minimum=None, maximum=None):
else:
return num_value
def check_date(date_text, method, return_string=False):
try: date_obg = datetime.strptime(str(date_text), "%m/%d/%Y")
except ValueError: raise Failed(f"Collection Error: {method}: {date_text} must match pattern MM/DD/YYYY e.g. 12/25/2020")
def check_date(date_text, method, return_string=False, plex_date=False):
try: date_obg = datetime.strptime(str(date_text), "%Y/%m/%d" if plex_date else "%m/%d/%Y")
except ValueError: raise Failed(f"Collection Error: {method}: {date_text} must match pattern {'YYYY/MM/DD e.g. 2020/12/25' if plex_date else 'MM/DD/YYYY e.g. 12/25/2020'}")
return str(date_text) if return_string else date_obg
def logger_input(prompt, timeout=60):

View file

@ -1,6 +1,6 @@
# Remove
# Less common, pinned
PlexAPI==4.5.0
PlexAPI==4.5.1
tmdbv3api==1.7.5
trakt.py==4.3.0
# More common, flexible