mirror of
https://github.com/meisnate12/Plex-Meta-Manager
synced 2024-11-10 06:54:21 +00:00
#527 collection_name will have its commas removed if used for arr tags
This commit is contained in:
meisnate12
parent
4aa536b27a
commit
10cd2b1b01
3 changed files with 208 additions and 241 deletions
|
|
@ -245,119 +245,10 @@ class CollectionBuilder:
|
|||
if "template" in methods:
|
||||
logger.debug("")
|
||||
logger.debug("Validating Method: template")
|
||||
if not self.metadata.templates:
|
||||
raise Failed(f"{self.Type} Error: No templates found")
|
||||
elif not self.data[methods["template"]]:
|
||||
raise Failed(f"{self.Type} 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(f"{self.Type} Error: template attribute is not a dictionary")
|
||||
elif "name" not in variables:
|
||||
raise Failed(f"{self.Type} Error: template sub-attribute name is required")
|
||||
elif not variables["name"]:
|
||||
raise Failed(f"{self.Type} Error: template sub-attribute name is blank")
|
||||
elif variables["name"] not in self.metadata.templates:
|
||||
raise Failed(f"{self.Type} Error: template {variables['name']} not found")
|
||||
elif not isinstance(self.metadata.templates[variables["name"]], dict):
|
||||
raise Failed(f"{self.Type} Error: template {variables['name']} is not a dictionary")
|
||||
else:
|
||||
for tm in variables:
|
||||
if not variables[tm]:
|
||||
raise Failed(f"{self.Type} 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"{self.Type} Error: template default sub-attribute {dv} is blank")
|
||||
else:
|
||||
raise Failed(f"{self.Type} Error: template sub-attribute default is not a dictionary")
|
||||
else:
|
||||
raise Failed(f"{self.Type} Error: template sub-attribute default is blank")
|
||||
|
||||
optional = []
|
||||
if "optional" in template:
|
||||
if template["optional"]:
|
||||
for op in util.get_list(template["optional"]):
|
||||
if op not in default:
|
||||
optional.append(str(op))
|
||||
else:
|
||||
logger.warning(f"Template Warning: variable {op} cannot be optional if it has a default")
|
||||
else:
|
||||
raise Failed(f"{self.Type} Error: template sub-attribute optional is blank")
|
||||
|
||||
if "move_collection_prefix" in template:
|
||||
if template["move_collection_prefix"]:
|
||||
for op in util.get_list(template["move_collection_prefix"]):
|
||||
variables["collection_name"] = variables["collection_name"].replace(f"{str(op).strip()} ", "") + f", {str(op).strip()}"
|
||||
else:
|
||||
raise Failed(f"{self.Type} Error: template sub-attribute move_collection_prefix 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", "move_collection_prefix"]:
|
||||
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
|
||||
new_attributes = self.metadata.apply_template(self.name, self.data, self.data[methods["template"]])
|
||||
for attr in new_attributes:
|
||||
self.data[attr] = new_attributes[attr]
|
||||
methods[attr.lower()] = attr
|
||||
|
||||
if "delete_not_scheduled" in methods:
|
||||
logger.debug("")
|
||||
|
|
@ -439,19 +330,6 @@ class CollectionBuilder:
|
|||
else:
|
||||
raise Failed(f"{self.Type} Error: {self.data[methods['collection_order']]} collection_order invalid\n\trelease (Order Collection by release dates)\n\talpha (Order Collection Alphabetically)\n\tcustom (Custom Order Collection)\n\tOther sorting options can be found at https://github.com/meisnate12/Plex-Meta-Manager/wiki/Smart-Builders#sort-options")
|
||||
|
||||
self.sort_by = None
|
||||
if "sort_by" in methods and not self.playlist:
|
||||
logger.debug("")
|
||||
logger.debug("Validating Method: sort_by")
|
||||
if self.data[methods["sort_by"]] is None:
|
||||
raise Failed(f"{self.Type} Error: sort_by attribute is blank")
|
||||
else:
|
||||
logger.debug(f"Value: {self.data[methods['sort_by']]}")
|
||||
if (self.library.is_movie and self.data[methods["sort_by"]] not in plex.movie_sorts) or (self.library.is_show and self.data[methods["sort_by"]] not in plex.show_sorts):
|
||||
raise Failed(f"{self.Type} Error: sort_by attribute {self.data[methods['sort_by']]} invalid")
|
||||
else:
|
||||
self.sort_by = self.data[methods["sort_by"]]
|
||||
|
||||
self.collection_level = "movie" if self.library.is_movie else "show"
|
||||
if self.playlist:
|
||||
self.collection_level = "item"
|
||||
|
|
|
|||
253
modules/meta.py
253
modules/meta.py
|
|
@ -9,65 +9,205 @@ logger = logging.getLogger("Plex Meta Manager")
|
|||
|
||||
github_base = "https://raw.githubusercontent.com/meisnate12/Plex-Meta-Manager-Configs/master/"
|
||||
|
||||
class MetadataFile:
|
||||
def __init__(self, config, library, file_type, path):
|
||||
|
||||
def get_dict(attribute, attr_data, check_list=None):
|
||||
if check_list is None:
|
||||
check_list = []
|
||||
if attr_data and attribute in attr_data:
|
||||
if attr_data[attribute]:
|
||||
if isinstance(attr_data[attribute], dict):
|
||||
new_dict = {}
|
||||
for _name, _data in attr_data[attribute].items():
|
||||
if _name in check_list:
|
||||
logger.error(
|
||||
f"Config Warning: Skipping duplicate {attribute[:-1] if attribute[-1] == 's' else attribute}: {_name}")
|
||||
elif _data is None:
|
||||
logger.error(
|
||||
f"Config Warning: {attribute[:-1] if attribute[-1] == 's' else attribute}: {_name} has no data")
|
||||
elif not isinstance(_data, dict):
|
||||
logger.error(
|
||||
f"Config Warning: {attribute[:-1] if attribute[-1] == 's' else attribute}: {_name} must be a dictionary")
|
||||
else:
|
||||
new_dict[str(_name)] = _data
|
||||
return new_dict
|
||||
else:
|
||||
logger.warning(f"Config Warning: {attribute} must be a dictionary")
|
||||
else:
|
||||
logger.warning(f"Config Warning: {attribute} attribute is blank")
|
||||
return None
|
||||
|
||||
|
||||
class DataFile:
|
||||
def __init__(self, config, file_type, path):
|
||||
self.config = config
|
||||
self.library = library
|
||||
self.type = file_type
|
||||
self.path = path
|
||||
def get_dict(attribute, attr_data, check_list=None):
|
||||
if check_list is None:
|
||||
check_list = []
|
||||
if attr_data and attribute in attr_data:
|
||||
if attr_data[attribute]:
|
||||
if isinstance(attr_data[attribute], dict):
|
||||
new_dict = {}
|
||||
for _name, _data in attr_data[attribute].items():
|
||||
if _name in check_list:
|
||||
logger.error(f"Config Warning: Skipping duplicate {attribute[:-1] if attribute[-1] == 's' else attribute}: {_name}")
|
||||
elif _data is None:
|
||||
logger.error(f"Config Warning: {attribute[:-1] if attribute[-1] == 's' else attribute}: {_name} has no data")
|
||||
elif not isinstance(_data, dict):
|
||||
logger.error(f"Config Warning: {attribute[:-1] if attribute[-1] == 's' else attribute}: {_name} must be a dictionary")
|
||||
else:
|
||||
new_dict[str(_name)] = _data
|
||||
return new_dict
|
||||
else:
|
||||
logger.warning(f"Config Warning: {attribute} must be a dictionary")
|
||||
self.data_type = ""
|
||||
self.templates = {}
|
||||
|
||||
def load_file(self):
|
||||
try:
|
||||
if self.type in ["URL", "Git"]:
|
||||
content_path = self.path if self.type == "URL" else f"{github_base}{self.path}.yml"
|
||||
response = self.config.get(content_path)
|
||||
if response.status_code >= 400:
|
||||
raise Failed(f"URL Error: No file found at {content_path}")
|
||||
content = response.content
|
||||
elif os.path.exists(os.path.abspath(self.path)):
|
||||
content = open(self.path, encoding="utf-8")
|
||||
else:
|
||||
raise Failed(f"File Error: File does not exist {self.path}")
|
||||
data, _, _ = yaml.util.load_yaml_guess_indent(content)
|
||||
return data
|
||||
except yaml.scanner.ScannerError as ye:
|
||||
raise Failed(f"YAML Error: {util.tab_new_lines(ye)}")
|
||||
except Exception as e:
|
||||
util.print_stacktrace()
|
||||
raise Failed(f"YAML Error: {e}")
|
||||
|
||||
def apply_template(self, name, data, template):
|
||||
if not self.templates:
|
||||
raise Failed(f"{self.data_type} Error: No templates found")
|
||||
elif not template:
|
||||
raise Failed(f"{self.data_type} Error: template attribute is blank")
|
||||
else:
|
||||
logger.debug(f"Value: {template}")
|
||||
for variables in util.get_list(template, split=False):
|
||||
if not isinstance(variables, dict):
|
||||
raise Failed(f"{self.data_type} Error: template attribute is not a dictionary")
|
||||
elif "name" not in variables:
|
||||
raise Failed(f"{self.data_type} Error: template sub-attribute name is required")
|
||||
elif not variables["name"]:
|
||||
raise Failed(f"{self.data_type} Error: template sub-attribute name is blank")
|
||||
elif variables["name"] not in self.templates:
|
||||
raise Failed(f"{self.data_type} Error: template {variables['name']} not found")
|
||||
elif not isinstance(self.templates[variables["name"]], dict):
|
||||
raise Failed(f"{self.data_type} Error: template {variables['name']} is not a dictionary")
|
||||
else:
|
||||
logger.warning(f"Config Warning: {attribute} attribute is blank")
|
||||
return None
|
||||
for tm in variables:
|
||||
if not variables[tm]:
|
||||
raise Failed(f"{self.data_type} Error: template sub-attribute {tm} is blank")
|
||||
if self.data_type == "Collection" and "collection_name" not in variables:
|
||||
variables["collection_name"] = str(name)
|
||||
if self.data_type == "Playlist" and "playlist_name" not in variables:
|
||||
variables["playlist_name"] = str(name)
|
||||
|
||||
template_name = variables["name"]
|
||||
template = self.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"{self.data_type} Error: template default sub-attribute {dv} is blank")
|
||||
else:
|
||||
raise Failed(f"{self.data_type} Error: template sub-attribute default is not a dictionary")
|
||||
else:
|
||||
raise Failed(f"{self.data_type} Error: template sub-attribute default is blank")
|
||||
|
||||
optional = []
|
||||
if "optional" in template:
|
||||
if template["optional"]:
|
||||
for op in util.get_list(template["optional"]):
|
||||
if op not in default:
|
||||
optional.append(str(op))
|
||||
else:
|
||||
logger.warning(f"Template Warning: variable {op} cannot be optional if it has a default")
|
||||
else:
|
||||
raise Failed(f"{self.data_type} Error: template sub-attribute optional is blank")
|
||||
|
||||
if "move_collection_prefix" in template:
|
||||
if template["move_collection_prefix"]:
|
||||
for op in util.get_list(template["move_collection_prefix"]):
|
||||
variables["collection_name"] = variables["collection_name"].replace(f"{str(op).strip()} ", "") + f", {str(op).strip()}"
|
||||
else:
|
||||
raise Failed(f"{self.data_type} Error: template sub-attribute move_collection_prefix is blank")
|
||||
|
||||
def check_data(_method, _data):
|
||||
if isinstance(_data, dict):
|
||||
final_data = {}
|
||||
for sm, sd in _data.items():
|
||||
try:
|
||||
final_data[sm] = check_data(_method, sd)
|
||||
except Failed:
|
||||
continue
|
||||
elif isinstance(_data, list):
|
||||
final_data = []
|
||||
for li in _data:
|
||||
try:
|
||||
final_data.append(check_data(_method, 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 == "collection_name" or variable == "playlist_name") and _method in ["radarr_tag", "item_radarr_tag", "sonarr_tag", "item_sonarr_tag"]:
|
||||
txt = scan_text(txt, variable, variable_data.replace(",", ""))
|
||||
elif 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
|
||||
|
||||
new_attributes = {}
|
||||
for method_name, attr_data in template.items():
|
||||
if method_name not in data and method_name not in ["default", "optional", "move_collection_prefix"]:
|
||||
if attr_data is None:
|
||||
logger.error(f"Template Error: template attribute {method_name} is blank")
|
||||
continue
|
||||
try:
|
||||
new_attributes[method_name] = check_data(method_name, attr_data)
|
||||
except Failed:
|
||||
continue
|
||||
return new_attributes
|
||||
|
||||
|
||||
class MetadataFile(DataFile):
|
||||
def __init__(self, config, library, file_type, path):
|
||||
super().__init__(config, file_type, path)
|
||||
self.data_type = "Collection"
|
||||
self.library = library
|
||||
if file_type == "Data":
|
||||
self.metadata = None
|
||||
self.collections = get_dict("collections", path, library.collections)
|
||||
self.templates = get_dict("templates", path)
|
||||
else:
|
||||
try:
|
||||
logger.info("")
|
||||
logger.info(f"Loading Metadata {file_type}: {path}")
|
||||
if file_type in ["URL", "Git"]:
|
||||
content_path = path if file_type == "URL" else f"{github_base}{path}.yml"
|
||||
response = self.config.get(content_path)
|
||||
if response.status_code >= 400:
|
||||
raise Failed(f"URL Error: No file found at {content_path}")
|
||||
content = response.content
|
||||
elif os.path.exists(os.path.abspath(path)):
|
||||
content = open(path, encoding="utf-8")
|
||||
else:
|
||||
raise Failed(f"File Error: File does not exist {path}")
|
||||
data, ind, bsi = yaml.util.load_yaml_guess_indent(content)
|
||||
self.metadata = get_dict("metadata", data, library.metadatas)
|
||||
self.templates = get_dict("templates", data)
|
||||
self.collections = get_dict("collections", data, library.collections)
|
||||
logger.info("")
|
||||
logger.info(f"Loading Metadata {file_type}: {path}")
|
||||
data = self.load_file()
|
||||
self.metadata = get_dict("metadata", data, library.metadatas)
|
||||
self.templates = get_dict("templates", data)
|
||||
self.collections = get_dict("collections", data, library.collections)
|
||||
|
||||
if self.metadata is None and self.collections is None:
|
||||
raise Failed("YAML Error: metadata or collections attribute is required")
|
||||
logger.info(f"Metadata File Loaded Successfully")
|
||||
except yaml.scanner.ScannerError as ye:
|
||||
raise Failed(f"YAML Error: {util.tab_new_lines(ye)}")
|
||||
except Exception as e:
|
||||
util.print_stacktrace()
|
||||
raise Failed(f"YAML Error: {e}")
|
||||
if self.metadata is None and self.collections is None:
|
||||
raise Failed("YAML Error: metadata or collections attribute is required")
|
||||
logger.info(f"Metadata File Loaded Successfully")
|
||||
|
||||
def get_collections(self, requested_collections):
|
||||
if requested_collections:
|
||||
|
|
@ -403,3 +543,18 @@ class MetadataFile:
|
|||
logger.error("Metadata Error: episodes attribute is blank")
|
||||
elif "episodes" in methods:
|
||||
logger.error("Metadata Error: episodes attribute only works for show libraries")
|
||||
|
||||
|
||||
class PlaylistFile(DataFile):
|
||||
def __init__(self, config, file_type, path):
|
||||
super().__init__(config, file_type, path)
|
||||
self.data_type = "Playlist"
|
||||
self.playlists = {}
|
||||
logger.info("")
|
||||
logger.info(f"Loading Playlist File {file_type}: {path}")
|
||||
data = self.load_file()
|
||||
self.playlists = get_dict("playlists", data, self.config.playlist_names)
|
||||
self.templates = get_dict("templates", data)
|
||||
if not self.playlists:
|
||||
raise Failed("YAML Error: playlists attribute is required")
|
||||
logger.info(f"Playlist File Loaded Successfully")
|
||||
|
|
|
|||
|
|
@ -1,66 +0,0 @@
|
|||
import logging, os, re
|
||||
from datetime import datetime
|
||||
from modules import plex, util
|
||||
from modules.util import Failed, ImageData
|
||||
from plexapi.exceptions import NotFound
|
||||
from ruamel import yaml
|
||||
|
||||
logger = logging.getLogger("Plex Meta Manager")
|
||||
|
||||
github_base = "https://raw.githubusercontent.com/meisnate12/Plex-Meta-Manager-Configs/master/"
|
||||
|
||||
class PlaylistFile:
|
||||
def __init__(self, config, file_type, path):
|
||||
self.config = config
|
||||
self.type = file_type
|
||||
self.path = path
|
||||
self.playlists = {}
|
||||
self.templates = {}
|
||||
try:
|
||||
logger.info("")
|
||||
logger.info(f"Loading Playlist File {file_type}: {path}")
|
||||
if file_type in ["URL", "Git"]:
|
||||
content_path = path if file_type == "URL" else f"{github_base}{path}.yml"
|
||||
response = self.config.get(content_path)
|
||||
if response.status_code >= 400:
|
||||
raise Failed(f"URL Error: No file found at {content_path}")
|
||||
content = response.content
|
||||
elif os.path.exists(os.path.abspath(path)):
|
||||
content = open(path, encoding="utf-8")
|
||||
else:
|
||||
raise Failed(f"File Error: File does not exist {path}")
|
||||
data, ind, bsi = yaml.util.load_yaml_guess_indent(content)
|
||||
if data and "playlists" in data:
|
||||
if data["playlists"]:
|
||||
if isinstance(data["playlists"], dict):
|
||||
for _name, _data in data["playlists"].items():
|
||||
if _name in self.config.playlist_names:
|
||||
logger.error(f"Config Warning: Skipping duplicate playlist: {_name}")
|
||||
elif _data is None:
|
||||
logger.error(f"Config Warning: playlist: {_name} has no data")
|
||||
elif not isinstance(_data, dict):
|
||||
logger.error(f"Config Warning: playlist: {_name} must be a dictionary")
|
||||
else:
|
||||
self.playlists[str(_name)] = _data
|
||||
else:
|
||||
logger.warning(f"Config Warning: playlists must be a dictionary")
|
||||
else:
|
||||
logger.warning(f"Config Warning: playlists attribute is blank")
|
||||
if not self.playlists:
|
||||
raise Failed("YAML Error: playlists attribute is required")
|
||||
if data and "templates" in data:
|
||||
if data["templates"]:
|
||||
if isinstance(data["templates"], dict):
|
||||
for _name, _data in data["templates"].items():
|
||||
self.templates[str(_name)] = _data
|
||||
else:
|
||||
logger.warning(f"Config Warning: templates must be a dictionary")
|
||||
else:
|
||||
logger.warning(f"Config Warning: templates attribute is blank")
|
||||
|
||||
logger.info(f"Playlist File Loaded Successfully")
|
||||
except yaml.scanner.ScannerError as ye:
|
||||
raise Failed(f"YAML Error: {util.tab_new_lines(ye)}")
|
||||
except Exception as e:
|
||||
util.print_stacktrace()
|
||||
raise Failed(f"YAML Error: {e}")
|
||||
Loading…
Reference in a new issue