mirror of
https://github.com/meisnate12/Plex-Meta-Manager
synced 2024-11-22 12:33:10 +00:00
multiple metadata paths
This commit is contained in:
meisnate12
parent
72ac502116
commit
e9a18c23cc
6 changed files with 459 additions and 77 deletions
|
|
@ -17,9 +17,6 @@ class AniDBAPI:
|
|||
"relation": "/relation/graph"
|
||||
}
|
||||
|
||||
def get_AniDB_IDs(self):
|
||||
return html.fromstring(requests.get("https://raw.githubusercontent.com/Anime-Lists/anime-lists/master/anime-list-master.xml").content)
|
||||
|
||||
@retry(stop_max_attempt_number=6, wait_fixed=10000)
|
||||
def send_request(self, url, language):
|
||||
return html.fromstring(requests.get(url, headers={"Accept-Language": language, "User-Agent": "Mozilla/5.0 x64"}).content)
|
||||
|
|
|
|||
|
|
@ -813,18 +813,18 @@ class CollectionBuilder:
|
|||
self.methods.append((method_name[:-8], values))
|
||||
elif method_name in dictionary_builders:
|
||||
if isinstance(method_data, dict):
|
||||
def get_int(parent, method, data_in, methods_in, default_in, minimum=1, maximum=None):
|
||||
if method not in methods_in:
|
||||
logger.warning(f"Collection Warning: {parent} {method} attribute not found using {default_in} as default")
|
||||
elif not data_in[methods_in[method]]:
|
||||
logger.warning(f"Collection Warning: {parent} {methods_in[method]} attribute is blank using {default_in} as default")
|
||||
elif isinstance(data_in[methods_in[method]], int) and data_in[methods_in[method]] >= minimum:
|
||||
if maximum is None or data_in[methods_in[method]] <= maximum:
|
||||
return data_in[methods_in[method]]
|
||||
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[method]} attribute {data_in[methods_in[method]]} invalid must an integer <= {maximum} using {default_in} as default")
|
||||
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[method]} attribute {data_in[methods_in[method]]} invalid must an integer >= {minimum} using {default_in} as default")
|
||||
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():
|
||||
|
|
|
|||
|
|
@ -144,7 +144,7 @@ class Config:
|
|||
else: message = f"Path {os.path.abspath(data[attribute])} does not exist"
|
||||
elif var_type == "list": return util.get_list(data[attribute])
|
||||
elif var_type == "list_path":
|
||||
temp_list = [path for path in util.get_list(data[attribute], split=True) if os.path.exists(os.path.abspath(path))]
|
||||
temp_list = [p for p in util.get_list(data[attribute], split=True) if os.path.exists(os.path.abspath(p))]
|
||||
if len(temp_list) > 0: return temp_list
|
||||
else: message = "No Paths exist"
|
||||
elif var_type == "lower_list": return util.get_list(data[attribute], lower=True)
|
||||
|
|
@ -266,7 +266,6 @@ class Config:
|
|||
self.IMDb = IMDbAPI(self)
|
||||
self.AniDB = AniDBAPI(self)
|
||||
self.Arms = ArmsAPI(self)
|
||||
self.AniDBIDs = self.AniDB.get_AniDB_IDs()
|
||||
self.AniList = AniListAPI(self)
|
||||
self.Letterboxd = LetterboxdAPI(self)
|
||||
|
||||
|
|
@ -383,7 +382,28 @@ class Config:
|
|||
params["mass_audience_rating_update"] = None
|
||||
|
||||
try:
|
||||
params["metadata_path"] = check_for_attribute(lib, "metadata_path", var_type="path", default=os.path.join(default_dir, f"{library_name}.yml"), throw=True)
|
||||
if lib and "metadata_path" in lib:
|
||||
params["metadata_path"] = []
|
||||
if lib["metadata_path"] is None:
|
||||
raise Failed("Config Error: metadata_path attribute is blank")
|
||||
paths_to_check = lib["metadata_path"] if isinstance(lib["metadata_path"], list) else [lib["metadata_path"]]
|
||||
for path in paths_to_check:
|
||||
if isinstance(path, dict):
|
||||
if "url" in path:
|
||||
if path["url"] is None:
|
||||
logger.error("Config Error: metadata_path url is blank")
|
||||
else:
|
||||
params["metadata_path"].append(("URL", path["url"]))
|
||||
if "git" in path:
|
||||
if path["git"] is None:
|
||||
logger.error("Config Error: metadata_path git is blank")
|
||||
else:
|
||||
params["metadata_path"].append(("Git", path['git']))
|
||||
else:
|
||||
params["metadata_path"].append(("File", path))
|
||||
else:
|
||||
params["metadata_path"] = [("File", os.path.join(default_dir, f"{library_name}.yml"))]
|
||||
params["default_dir"] = default_dir
|
||||
params["plex"] = {}
|
||||
params["plex"]["url"] = check_for_attribute(lib, "url", parent="plex", default=self.general["plex"]["url"], req_default=True, save=False)
|
||||
params["plex"]["token"] = check_for_attribute(lib, "token", parent="plex", default=self.general["plex"]["token"], req_default=True, save=False)
|
||||
|
|
@ -474,26 +494,29 @@ class Config:
|
|||
util.separator(f"Mapping {library.name} Library")
|
||||
logger.info("")
|
||||
movie_map, show_map = self.map_guids(library)
|
||||
if not test and not resume_from:
|
||||
if library.mass_update:
|
||||
self.mass_metadata(library, movie_map, show_map)
|
||||
try: library.update_metadata(self.TMDb, test)
|
||||
except Failed as e: logger.error(e)
|
||||
logger.info("")
|
||||
util.separator(f"{library.name} Library {'Test ' if test else ''}Collections")
|
||||
collections = {c: library.collections[c] for c in util.get_list(requested_collections) if c in library.collections} if requested_collections else library.collections
|
||||
if resume_from and resume_from not in collections:
|
||||
logger.warning(f"Collection: {resume_from} not in {library.name}")
|
||||
continue
|
||||
if collections:
|
||||
resume_from = self.run_collection(library, collections, test, resume_from, movie_map, show_map)
|
||||
if not test and not resume_from and library.mass_update:
|
||||
self.mass_metadata(library, movie_map, show_map)
|
||||
for metadata in library.metadata_files:
|
||||
logger.info("")
|
||||
util.separator(f"Running Metadata File\n{metadata.path}")
|
||||
if not test and not resume_from:
|
||||
try: metadata.update_metadata(self.TMDb, test)
|
||||
except Failed as e: logger.error(e)
|
||||
logger.info("")
|
||||
util.separator(f"{'Test ' if test else ''}Collections")
|
||||
collections = metadata.get_collections(requested_collections)
|
||||
if resume_from and resume_from not in collections:
|
||||
logger.warning(f"Collection: {resume_from} not in Metadata File: {metadata.path}")
|
||||
continue
|
||||
if collections:
|
||||
resume_from = self.run_collection(library, metadata, collections, test, resume_from, movie_map, show_map)
|
||||
|
||||
if library.show_unmanaged is True and not test and not requested_collections:
|
||||
logger.info("")
|
||||
util.separator(f"Unmanaged Collections in {library.name} Library")
|
||||
logger.info("")
|
||||
unmanaged_count = 0
|
||||
collections_in_plex = [str(plex_col) for plex_col in collections]
|
||||
collections_in_plex = [str(plex_col) for plex_col in library.collections]
|
||||
for col in library.get_all_collections():
|
||||
if col.title not in collections_in_plex:
|
||||
logger.info(col.title)
|
||||
|
|
@ -529,23 +552,19 @@ class Config:
|
|||
logger.info("")
|
||||
util.separator(f"{library.name} Library Run Again")
|
||||
logger.info("")
|
||||
collections = {c: library.collections[c] for c in util.get_list(requested_collections) if c in library.collections} if requested_collections else library.collections
|
||||
if collections:
|
||||
util.separator(f"Mapping {library.name} Library")
|
||||
movie_map, show_map = self.map_guids(library)
|
||||
for builder in library.run_again:
|
||||
logger.info("")
|
||||
movie_map, show_map = self.map_guids(library)
|
||||
for builder in library.run_again:
|
||||
logger.info("")
|
||||
util.separator(f"{builder.name} Collection")
|
||||
logger.info("")
|
||||
try:
|
||||
collection_obj = library.get_collection(builder.name)
|
||||
except Failed as e:
|
||||
util.print_multiline(e, error=True)
|
||||
continue
|
||||
builder.run_collections_again(collection_obj, movie_map, show_map)
|
||||
util.separator(f"{builder.name} Collection")
|
||||
logger.info("")
|
||||
try:
|
||||
collection_obj = library.get_collection(builder.name)
|
||||
except Failed as e:
|
||||
util.print_multiline(e, error=True)
|
||||
continue
|
||||
builder.run_collections_again(collection_obj, movie_map, show_map)
|
||||
|
||||
def run_collection(self, library, collections, test, resume_from, movie_map, show_map):
|
||||
def run_collection(self, library, metadata, collections, test, resume_from, movie_map, show_map):
|
||||
for mapping_name, collection_attrs in collections.items():
|
||||
if test and ("test" not in collection_attrs or collection_attrs["test"] is not True):
|
||||
no_template_test = True
|
||||
|
|
@ -553,11 +572,11 @@ class Config:
|
|||
for data_template in util.get_list(collection_attrs["template"], split=False):
|
||||
if "name" in data_template \
|
||||
and data_template["name"] \
|
||||
and library.templates \
|
||||
and data_template["name"] in library.templates \
|
||||
and library.templates[data_template["name"]] \
|
||||
and "test" in library.templates[data_template["name"]] \
|
||||
and library.templates[data_template["name"]]["test"] is True:
|
||||
and metadata.templates \
|
||||
and data_template["name"] in metadata.templates \
|
||||
and metadata.templates[data_template["name"]] \
|
||||
and "test" in metadata.templates[data_template["name"]] \
|
||||
and metadata.templates[data_template["name"]]["test"] is True:
|
||||
no_template_test = False
|
||||
if no_template_test:
|
||||
continue
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
import json, logging, re, requests, secrets, webbrowser
|
||||
import logging, re, requests, secrets, webbrowser
|
||||
from modules import util
|
||||
from modules.util import Failed, TimeoutExpired
|
||||
from retrying import retry
|
||||
|
|
|
|||
377
modules/meta.py
Normal file
377
modules/meta.py
Normal file
|
|
@ -0,0 +1,377 @@
|
|||
import logging, os, re, requests
|
||||
from datetime import datetime
|
||||
from modules import plex, util
|
||||
from modules.util import Failed
|
||||
from plexapi.exceptions import NotFound
|
||||
from ruamel import yaml
|
||||
|
||||
logger = logging.getLogger("Plex Meta Manager")
|
||||
|
||||
class Metadata:
|
||||
def __init__(self, library, file_type, path):
|
||||
self.library = library
|
||||
self.type = file_type
|
||||
self.path = path
|
||||
self.github_base = "https://raw.githubusercontent.com/meisnate12/Plex-Meta-Manager-Configs/master/"
|
||||
logger.info("")
|
||||
logger.info(f"Loading Metadata {file_type}: {path}")
|
||||
def get_dict(attribute, attr_data, check_list=None):
|
||||
if attribute in attr_data:
|
||||
if attr_data[attribute]:
|
||||
if isinstance(attr_data[attribute], dict):
|
||||
if check_list:
|
||||
new_dict = {}
|
||||
for a_name, a_data in attr_data[attribute].items():
|
||||
if a_name in check_list:
|
||||
logger.error(f"Config Warning: Skipping duplicate {attribute[:-1] if attribute[-1] == 's' else attribute}: {a_name}")
|
||||
else:
|
||||
new_dict[a_name] = a_data
|
||||
return new_dict
|
||||
else:
|
||||
return attr_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
|
||||
try:
|
||||
if file_type == "URL":
|
||||
content = requests.get(path).content
|
||||
elif file_type == "Git":
|
||||
content = requests.get(f"{self.github_base}{path}.yml").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)
|
||||
|
||||
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}")
|
||||
|
||||
def get_collections(self, requested_collections):
|
||||
if requested_collections:
|
||||
return {c: self.collections[c] for c in util.get_list(requested_collections) if c in self.collections}
|
||||
else:
|
||||
return self.collections
|
||||
|
||||
def update_metadata(self, TMDb, test):
|
||||
logger.info("")
|
||||
util.separator(f"Running Metadata")
|
||||
logger.info("")
|
||||
if not self.metadata:
|
||||
raise Failed("No metadata to edit")
|
||||
for mapping_name, meta in self.metadata.items():
|
||||
methods = {mm.lower(): mm for mm in meta}
|
||||
if test and ("test" not in methods or meta[methods["test"]] is not True):
|
||||
continue
|
||||
|
||||
updated = False
|
||||
edits = {}
|
||||
advance_edits = {}
|
||||
|
||||
def add_edit(name, current, group, alias, key=None, value=None, var_type="str"):
|
||||
if value or name in alias:
|
||||
if value or group[alias[name]]:
|
||||
if key is None: key = name
|
||||
if value is None: value = group[alias[name]]
|
||||
try:
|
||||
if var_type == "date":
|
||||
final_value = util.check_date(value, name, return_string=True, plex_date=True)
|
||||
elif var_type == "float":
|
||||
final_value = util.check_number(value, name, number_type="float", minimum=0, maximum=10)
|
||||
else:
|
||||
final_value = value
|
||||
if str(current) != str(final_value):
|
||||
edits[f"{key}.value"] = final_value
|
||||
edits[f"{key}.locked"] = 1
|
||||
logger.info(f"Detail: {name} updated to {final_value}")
|
||||
except Failed as ee:
|
||||
logger.error(ee)
|
||||
else:
|
||||
logger.error(f"Metadata Error: {name} attribute is blank")
|
||||
|
||||
def add_advanced_edit(attr, obj, group, alias, show_library=False, new_agent=False):
|
||||
key, options = plex.advance_keys[attr]
|
||||
if attr in alias:
|
||||
if new_agent and self.library.agent not in plex.new_plex_agents:
|
||||
logger.error(f"Metadata Error: {attr} attribute only works for with the New Plex Movie Agent and New Plex TV Agent")
|
||||
elif show_library and not self.library.is_show:
|
||||
logger.error(f"Metadata Error: {attr} attribute only works for show libraries")
|
||||
elif group[alias[attr]]:
|
||||
method_data = str(group[alias[attr]]).lower()
|
||||
if method_data not in options:
|
||||
logger.error(f"Metadata Error: {group[alias[attr]]} {attr} attribute invalid")
|
||||
elif getattr(obj, key) != options[method_data]:
|
||||
advance_edits[key] = options[method_data]
|
||||
logger.info(f"Detail: {attr} updated to {method_data}")
|
||||
else:
|
||||
logger.error(f"Metadata Error: {attr} attribute is blank")
|
||||
|
||||
def edit_tags(attr, obj, group, alias, key=None, extra=None, movie_library=False):
|
||||
if key is None:
|
||||
key = f"{attr}s"
|
||||
if attr in alias and f"{attr}.sync" in alias:
|
||||
logger.error(f"Metadata Error: Cannot use {attr} and {attr}.sync together")
|
||||
elif attr in alias or f"{attr}.sync" in alias:
|
||||
attr_key = attr if attr in alias else f"{attr}.sync"
|
||||
if movie_library and not self.library.is_movie:
|
||||
logger.error(f"Metadata Error: {attr_key} attribute only works for movie libraries")
|
||||
elif group[alias[attr_key]] or extra:
|
||||
item_tags = [item_tag.tag for item_tag in getattr(obj, key)]
|
||||
input_tags = []
|
||||
if group[alias[attr_key]]:
|
||||
input_tags.extend(util.get_list(group[alias[attr_key]]))
|
||||
if extra:
|
||||
input_tags.extend(extra)
|
||||
if f"{attr}.sync" in alias:
|
||||
remove_method = getattr(obj, f"remove{attr.capitalize()}")
|
||||
for tag in (t for t in item_tags if t not in input_tags):
|
||||
updated = True
|
||||
remove_method(tag)
|
||||
logger.info(f"Detail: {attr.capitalize()} {tag} removed")
|
||||
add_method = getattr(obj, f"add{attr.capitalize()}")
|
||||
for tag in (t for t in input_tags if t not in item_tags):
|
||||
updated = True
|
||||
add_method(tag)
|
||||
logger.info(f"Detail: {attr.capitalize()} {tag} added")
|
||||
else:
|
||||
logger.error(f"Metadata Error: {attr} attribute is blank")
|
||||
|
||||
def set_image(attr, obj, group, alias, poster=True, url=True):
|
||||
if group[alias[attr]]:
|
||||
message = f"{'poster' if poster else 'background'} to [{'URL' if url else 'File'}] {group[alias[attr]]}"
|
||||
self.library.upload_image(obj, group[alias[attr]], poster=poster, url=url)
|
||||
logger.info(f"Detail: {attr} updated {message}")
|
||||
else:
|
||||
logger.error(f"Metadata Error: {attr} attribute is blank")
|
||||
|
||||
def set_images(obj, group, alias):
|
||||
if "url_poster" in alias:
|
||||
set_image("url_poster", obj, group, alias)
|
||||
elif "file_poster" in alias:
|
||||
set_image("file_poster", obj, group, alias, url=False)
|
||||
if "url_background" in alias:
|
||||
set_image("url_background", obj, group, alias, poster=False)
|
||||
elif "file_background" in alias:
|
||||
set_image("file_background", obj, group, alias, poster=False, url=False)
|
||||
|
||||
logger.info("")
|
||||
util.separator()
|
||||
logger.info("")
|
||||
year = None
|
||||
if "year" in methods:
|
||||
year = util.check_number(meta[methods["year"]], "year", minimum=1800, maximum=datetime.now().year + 1)
|
||||
|
||||
title = mapping_name
|
||||
if "title" in methods:
|
||||
if meta[methods["title"]] is None:
|
||||
logger.error("Metadata Error: title attribute is blank")
|
||||
else:
|
||||
title = meta[methods["title"]]
|
||||
|
||||
item = self.library.search_item(title, year=year)
|
||||
|
||||
if item is None:
|
||||
item = self.library.search_item(f"{title} (SUB)", year=year)
|
||||
|
||||
if item is None and "alt_title" in methods:
|
||||
if meta[methods["alt_title"]] is None:
|
||||
logger.error("Metadata Error: alt_title attribute is blank")
|
||||
else:
|
||||
alt_title = meta["alt_title"]
|
||||
item = self.library.search_item(alt_title, year=year)
|
||||
|
||||
if item is None:
|
||||
logger.error(f"Plex Error: Item {mapping_name} not found")
|
||||
logger.error(f"Skipping {mapping_name}")
|
||||
continue
|
||||
|
||||
item_type = "Movie" if self.library.is_movie else "Show"
|
||||
logger.info(f"Updating {item_type}: {title}...")
|
||||
|
||||
tmdb_item = None
|
||||
tmdb_is_movie = None
|
||||
if ("tmdb_show" in methods or "tmdb_id" in methods) and "tmdb_movie" in methods:
|
||||
logger.error("Metadata Error: Cannot use tmdb_movie and tmdb_show when editing the same metadata item")
|
||||
|
||||
if "tmdb_show" in methods or "tmdb_id" in methods or "tmdb_movie" in methods:
|
||||
try:
|
||||
if "tmdb_show" in methods or "tmdb_id" in methods:
|
||||
data = meta[methods["tmdb_show" if "tmdb_show" in methods else "tmdb_id"]]
|
||||
if data is None:
|
||||
logger.error("Metadata Error: tmdb_show attribute is blank")
|
||||
else:
|
||||
tmdb_is_movie = False
|
||||
tmdb_item = TMDb.get_show(util.regex_first_int(data, "Show"))
|
||||
elif "tmdb_movie" in methods:
|
||||
if meta[methods["tmdb_movie"]] is None:
|
||||
logger.error("Metadata Error: tmdb_movie attribute is blank")
|
||||
else:
|
||||
tmdb_is_movie = True
|
||||
tmdb_item = TMDb.get_movie(util.regex_first_int(meta[methods["tmdb_movie"]], "Movie"))
|
||||
except Failed as e:
|
||||
logger.error(e)
|
||||
|
||||
originally_available = None
|
||||
original_title = None
|
||||
rating = None
|
||||
studio = None
|
||||
tagline = None
|
||||
summary = None
|
||||
genres = []
|
||||
if tmdb_item:
|
||||
originally_available = tmdb_item.release_date if tmdb_is_movie else tmdb_item.first_air_date
|
||||
if tmdb_item and tmdb_is_movie is True and tmdb_item.original_title != tmdb_item.title:
|
||||
original_title = tmdb_item.original_title
|
||||
elif tmdb_item and tmdb_is_movie is False and tmdb_item.original_name != tmdb_item.name:
|
||||
original_title = tmdb_item.original_name
|
||||
rating = tmdb_item.vote_average
|
||||
if tmdb_is_movie is True and tmdb_item.production_companies:
|
||||
studio = tmdb_item.production_companies[0].name
|
||||
elif tmdb_is_movie is False and tmdb_item.networks:
|
||||
studio = tmdb_item.networks[0].name
|
||||
tagline = tmdb_item.tagline if len(tmdb_item.tagline) > 0 else None
|
||||
summary = tmdb_item.overview
|
||||
genres = [genre.name for genre in tmdb_item.genres]
|
||||
|
||||
edits = {}
|
||||
add_edit("title", item.title, meta, methods, value=title)
|
||||
add_edit("sort_title", item.titleSort, meta, methods, key="titleSort")
|
||||
add_edit("originally_available", str(item.originallyAvailableAt)[:-9], meta, methods,
|
||||
key="originallyAvailableAt", value=originally_available, var_type="date")
|
||||
add_edit("critic_rating", item.rating, meta, methods, value=rating, key="rating", var_type="float")
|
||||
add_edit("audience_rating", item.audienceRating, meta, methods, key="audienceRating", var_type="float")
|
||||
add_edit("content_rating", item.contentRating, meta, methods, key="contentRating")
|
||||
add_edit("original_title", item.originalTitle, meta, methods, key="originalTitle", value=original_title)
|
||||
add_edit("studio", item.studio, meta, methods, value=studio)
|
||||
add_edit("tagline", item.tagline, meta, methods, value=tagline)
|
||||
add_edit("summary", item.summary, meta, methods, value=summary)
|
||||
self.library.edit_item(item, mapping_name, item_type, edits)
|
||||
|
||||
advance_edits = {}
|
||||
add_advanced_edit("episode_sorting", item, meta, methods, show_library=True)
|
||||
add_advanced_edit("keep_episodes", item, meta, methods, show_library=True)
|
||||
add_advanced_edit("delete_episodes", item, meta, methods, show_library=True)
|
||||
add_advanced_edit("season_display", item, meta, methods, show_library=True)
|
||||
add_advanced_edit("episode_ordering", item, meta, methods, show_library=True)
|
||||
add_advanced_edit("metadata_language", item, meta, methods, new_agent=True)
|
||||
add_advanced_edit("use_original_title", item, meta, methods, new_agent=True)
|
||||
self.library.edit_item(item, mapping_name, item_type, advance_edits, advanced=True)
|
||||
|
||||
edit_tags("genre", item, meta, methods, extra=genres)
|
||||
edit_tags("label", item, meta, methods)
|
||||
edit_tags("collection", item, meta, methods)
|
||||
edit_tags("country", item, meta, methods, key="countries", movie_library=True)
|
||||
edit_tags("director", item, meta, methods, movie_library=True)
|
||||
edit_tags("producer", item, meta, methods, movie_library=True)
|
||||
edit_tags("writer", item, meta, methods, movie_library=True)
|
||||
|
||||
logger.info(f"{item_type}: {mapping_name} Details Update {'Complete' if updated else 'Not Needed'}")
|
||||
|
||||
set_images(item, meta, methods)
|
||||
|
||||
if "seasons" in methods and self.library.is_show:
|
||||
if meta[methods["seasons"]]:
|
||||
for season_id in meta[methods["seasons"]]:
|
||||
updated = False
|
||||
logger.info("")
|
||||
logger.info(f"Updating season {season_id} of {mapping_name}...")
|
||||
if isinstance(season_id, int):
|
||||
season = None
|
||||
for s in item.seasons():
|
||||
if s.index == season_id:
|
||||
season = s
|
||||
break
|
||||
if season is None:
|
||||
logger.error(f"Metadata Error: Season: {season_id} not found")
|
||||
else:
|
||||
season_dict = meta[methods["seasons"]][season_id]
|
||||
season_methods = {sm.lower(): sm for sm in season_dict}
|
||||
|
||||
if "title" in season_methods and season_dict[season_methods["title"]]:
|
||||
title = season_dict[season_methods["title"]]
|
||||
else:
|
||||
title = season.title
|
||||
if "sub" in season_methods:
|
||||
if season_dict[season_methods["sub"]] is None:
|
||||
logger.error("Metadata Error: sub attribute is blank")
|
||||
elif season_dict[season_methods["sub"]] is True and "(SUB)" not in title:
|
||||
title = f"{title} (SUB)"
|
||||
elif season_dict[season_methods["sub"]] is False and title.endswith(" (SUB)"):
|
||||
title = title[:-6]
|
||||
else:
|
||||
logger.error("Metadata Error: sub attribute must be True or False")
|
||||
|
||||
edits = {}
|
||||
add_edit("title", season.title, season_dict, season_methods, value=title)
|
||||
add_edit("summary", season.summary, season_dict, season_methods)
|
||||
self.library.edit_item(season, season_id, "Season", edits)
|
||||
set_images(season, season_dict, season_methods)
|
||||
else:
|
||||
logger.error(f"Metadata Error: Season: {season_id} invalid, it must be an integer")
|
||||
logger.info(f"Season {season_id} of {mapping_name} Details Update {'Complete' if updated else 'Not Needed'}")
|
||||
else:
|
||||
logger.error("Metadata Error: seasons attribute is blank")
|
||||
elif "seasons" in methods:
|
||||
logger.error("Metadata Error: seasons attribute only works for show libraries")
|
||||
|
||||
if "episodes" in methods and self.library.is_show:
|
||||
if meta[methods["episodes"]]:
|
||||
for episode_str in meta[methods["episodes"]]:
|
||||
updated = False
|
||||
logger.info("")
|
||||
match = re.search("[Ss]\\d+[Ee]\\d+", episode_str)
|
||||
if match:
|
||||
output = match.group(0)[1:].split("E" if "E" in match.group(0) else "e")
|
||||
season_id = int(output[0])
|
||||
episode_id = int(output[1])
|
||||
logger.info(f"Updating episode S{season_id}E{episode_id} of {mapping_name}...")
|
||||
try:
|
||||
episode = item.episode(season=season_id, episode=episode_id)
|
||||
except NotFound:
|
||||
logger.error(f"Metadata Error: episode {episode_id} of season {season_id} not found")
|
||||
else:
|
||||
episode_dict = meta[methods["episodes"]][episode_str]
|
||||
episode_methods = {em.lower(): em for em in episode_dict}
|
||||
|
||||
if "title" in episode_methods and episode_dict[episode_methods["title"]]:
|
||||
title = episode_dict[episode_methods["title"]]
|
||||
else:
|
||||
title = episode.title
|
||||
if "sub" in episode_dict:
|
||||
if episode_dict[episode_methods["sub"]] is None:
|
||||
logger.error("Metadata Error: sub attribute is blank")
|
||||
elif episode_dict[episode_methods["sub"]] is True and "(SUB)" not in title:
|
||||
title = f"{title} (SUB)"
|
||||
elif episode_dict[episode_methods["sub"]] is False and title.endswith(" (SUB)"):
|
||||
title = title[:-6]
|
||||
else:
|
||||
logger.error("Metadata Error: sub attribute must be True or False")
|
||||
edits = {}
|
||||
add_edit("title", episode.title, episode_dict, episode_methods, value=title)
|
||||
add_edit("sort_title", episode.titleSort, episode_dict, episode_methods,
|
||||
key="titleSort")
|
||||
add_edit("rating", episode.rating, episode_dict, episode_methods)
|
||||
add_edit("originally_available", str(episode.originallyAvailableAt)[:-9],
|
||||
episode_dict, episode_methods, key="originallyAvailableAt")
|
||||
add_edit("summary", episode.summary, episode_dict, episode_methods)
|
||||
self.library.edit_item(episode, f"{season_id} Episode: {episode_id}", "Season", edits)
|
||||
edit_tags("director", episode, episode_dict, episode_methods)
|
||||
edit_tags("writer", episode, episode_dict, episode_methods)
|
||||
set_images(episode, episode_dict, episode_methods)
|
||||
logger.info(f"Episode S{episode_id}E{season_id} of {mapping_name} Details Update {'Complete' if updated else 'Not Needed'}")
|
||||
else:
|
||||
logger.error(f"Metadata Error: episode {episode_str} invalid must have S##E## format")
|
||||
else:
|
||||
logger.error("Metadata Error: episodes attribute is blank")
|
||||
elif "episodes" in methods:
|
||||
logger.error("Metadata Error: episodes attribute only works for show libraries")
|
||||
|
|
@ -1,6 +1,7 @@
|
|||
import glob, logging, os, re, requests
|
||||
from datetime import datetime, timedelta
|
||||
from modules import util
|
||||
from modules.meta import Metadata
|
||||
from modules.util import Failed
|
||||
from plexapi import utils
|
||||
from plexapi.exceptions import BadRequest, NotFound, Unauthorized
|
||||
|
|
@ -308,33 +309,21 @@ class PlexAPI:
|
|||
self.agent = self.Plex.agent
|
||||
self.is_movie = self.Plex.type == "movie"
|
||||
self.is_show = self.Plex.type == "show"
|
||||
self.collections = []
|
||||
self.metadatas = []
|
||||
|
||||
logger.info(f"Using Metadata File: {params['metadata_path']}")
|
||||
try:
|
||||
self.data, ind, bsi = yaml.util.load_yaml_guess_indent(open(params["metadata_path"], encoding="utf-8"))
|
||||
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}")
|
||||
self.metadata_files = []
|
||||
for file_type, metadata_file in params["metadata_path"]:
|
||||
try:
|
||||
meta_obj = Metadata(self, file_type, metadata_file)
|
||||
self.collections.extend([c for c in meta_obj.collections])
|
||||
self.metadatas.extend([c for c in meta_obj.metadata])
|
||||
self.metadata_files.append(meta_obj)
|
||||
except Failed as e:
|
||||
logger.error(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")
|
||||
return None
|
||||
|
||||
self.metadata = get_dict("metadata")
|
||||
self.templates = get_dict("templates")
|
||||
self.collections = get_dict("collections")
|
||||
|
||||
if self.metadata is None and self.collections is None:
|
||||
raise Failed("YAML Error: metadata or collections attribute is required")
|
||||
if len(self.metadata_files) == 0:
|
||||
raise Failed("Metadata File Error: No valid metadata files found")
|
||||
|
||||
if params["asset_directory"]:
|
||||
for ad in params["asset_directory"]:
|
||||
|
|
@ -346,7 +335,7 @@ class PlexAPI:
|
|||
self.Sonarr = None
|
||||
self.Tautulli = None
|
||||
self.name = params["name"]
|
||||
self.missing_path = os.path.join(os.path.dirname(os.path.abspath(params["metadata_path"])), f"{os.path.splitext(os.path.basename(params['metadata_path']))[0]}_missing.yml")
|
||||
self.missing_path = os.path.join(params["default_dir"], f"{self.name}_missing.yml")
|
||||
self.metadata_path = params["metadata_path"]
|
||||
self.asset_directory = params["asset_directory"]
|
||||
self.asset_folders = params["asset_folders"]
|
||||
|
|
|
|||
Loading…
Reference in a new issue