Mirrors/Plex-Meta-Manager
2
0
Fork 0
mirror of https://github.com/meisnate12/Plex-Meta-Manager synced 2024-11-10 06:54:21 +00:00
Code Issues Projects Releases Packages Wiki Activity
Plex-Meta-Manager/modules/config.py
meisnate12 188063cb19 [3] Added the mass_added_at_update operation
2024-05-09 09:04:04 -04:00

1358 lines
96 KiB
Python
Raw Blame History

import base64, os, re, requests
from datetime import datetime
from lxml import html
from modules import util, radarr, sonarr, operations
from modules.anidb import AniDB
from modules.anilist import AniList
from modules.cache import Cache
from modules.convert import Convert
from modules.ergast import Ergast
from modules.icheckmovies import ICheckMovies
from modules.imdb import IMDb
from modules.github import GitHub
from modules.letterboxd import Letterboxd
from modules.mal import MyAnimeList
from modules.meta import PlaylistFile
from modules.mojo import BoxOfficeMojo
from modules.notifiarr import Notifiarr
from modules.gotify import Gotify
from modules.omdb import OMDb
from modules.overlays import Overlays
from modules.plex import Plex
from modules.radarr import Radarr
from modules.sonarr import Sonarr
from modules.reciperr import Reciperr
from modules.mdblist import MDBList
from modules.tautulli import Tautulli
from modules.tmdb import TMDb
from modules.trakt import Trakt
from modules.tvdb import TVDb
from modules.util import Failed, NotScheduled, NotScheduledRange, YAML
from modules.webhooks import Webhooks
from retrying import retry
logger = util.logger
mediastingers_url = "https://raw.githubusercontent.com/Kometa-Team/Mediastingers/master/stingers.yml"
run_order_options = {
"collections": "Represents Collection Updates",
"metadata": "Represents Metadata Updates",
"overlays": "Represents Overlay Updates",
"operations": "Represents Operations Updates"
}
sync_modes = {"append": "Only Add Items to the Collection or Playlist", "sync": "Add & Remove Items from the Collection or Playlist"}
filetype_list = {
"jpg": "Use JPG files for saving Overlays",
"png": "Use PNG files for saving Overlays",
"webp_lossy": "Use Lossy WEBP files for saving Overlays",
"webp_lossless": "Use Lossless WEBP files for saving Overlays"
}
imdb_label_options = {
"remove": "Remove All IMDb Parental Labels",
"none": "Add IMDb Parental Labels for None, Mild, Moderate, or Severe",
"mild": "Add IMDb Parental Labels for Mild, Moderate, or Severe",
"moderate": "Add IMDb Parental Labels for Moderate or Severe",
"severe": "Add IMDb Parental Labels for Severe"
}
mass_genre_options = {
"lock": "Lock Genre", "unlock": "Unlock Genre", "remove": "Remove and Lock Genre", "reset": "Remove and Unlock Genre",
"tmdb": "Use TMDb Genres", "imdb": "Use IMDb Genres", "omdb": "Use IMDb Genres through OMDb", "tvdb": "Use TVDb Genres",
"mal": "Use MyAnimeList Genres", "anidb": "Use AniDB Main Tags",
"anidb_3_0": "Use AniDB Main Tags and All 3 Star Tags and above", "anidb_2_5": "Use AniDB Main Tags and All 2.5 Star Tags and above",
"anidb_2_0": "Use AniDB Main Tags and All 2 Star Tags and above", "anidb_1_5": "Use AniDB Main Tags and All 1.5 Star Tags and above",
"anidb_1_0": "Use AniDB Main Tags and All 1 Star Tags and above", "anidb_0_5": "Use AniDB Main Tags and All 0.5 Star Tags and above"
}
mass_content_options = {
"lock": "Lock Rating", "unlock": "Unlock Rating", "remove": "Remove and Lock Rating", "reset": "Remove and Unlock Rating",
"omdb": "Use IMDb Rating through OMDb", "mdb": "Use MDBList Rating",
"mdb_commonsense": "Use Commonsense Rating through MDBList", "mdb_commonsense0": "Use Commonsense Rating with Zero Padding through MDBList",
"mdb_age_rating": "Use MDBList Age Rating", "mdb_age_rating0": "Use MDBList Age Rating with Zero Padding",
"mal": "Use MyAnimeList Rating"
}
mass_collection_content_options = {
"lock": "Lock Rating", "unlock": "Unlock Rating", "remove": "Remove and Lock Rating", "reset": "Remove and Unlock Rating",
"highest": "Highest Rating in the collection", "lowest": "Lowest Rating in the collection",
"average": "Highest Rating in the collection"
}
content_rating_default = {
1: [
]
}
mass_studio_options = {
"lock": "Lock Rating", "unlock": "Unlock Rating", "remove": "Remove and Lock Rating", "reset": "Remove and Unlock Rating",
"tmdb": "Use TMDb Studio", "anidb": "Use AniDB Animation Work", "mal": "Use MyAnimeList Studio"
}
mass_original_title_options = {
"lock": "Lock Original Title", "unlock": "Unlock Original Title", "remove": "Remove and Lock Original Title", "reset": "Remove and Unlock Original Title",
"anidb": "Use AniDB Main Title", "anidb_official": "Use AniDB Official Title based on the language attribute in the config file",
"mal": "Use MyAnimeList Main Title", "mal_english": "Use MyAnimeList English Title", "mal_japanese": "Use MyAnimeList Japanese Title",
}
mass_available_options = {
"lock": "Lock Originally Available", "unlock": "Unlock Originally Available", "remove": "Remove and Lock Originally Available", "reset": "Remove and Unlock Originally Available",
"tmdb": "Use TMDb Release", "omdb": "Use IMDb Release through OMDb", "mdb": "Use MDBList Release", "mdb_digital": "Use MDBList Digital Release", "tvdb": "Use TVDb Release",
"anidb": "Use AniDB Release", "mal": "Use MyAnimeList Release"
}
mass_image_options = {
"lock": "Lock Image", "unlock": "Unlock Image", "plex": "Use Plex Images", "tmdb": "Use TMDb Images"
}
mass_episode_rating_options = {
"lock": "Lock Rating", "unlock": "Unlock Rating", "remove": "Remove and Lock Rating", "reset": "Remove and Unlock Rating",
"tmdb": "Use TMDb Rating", "imdb": "Use IMDb Rating"
}
mass_rating_options = {
"lock": "Lock Rating",
"unlock": "Unlock Rating",
"remove": "Remove and Lock Rating",
"reset": "Remove and Unlock Rating",
"tmdb": "Use TMDb Rating",
"imdb": "Use IMDb Rating",
"trakt_user": "Use Trakt User Rating",
"omdb": "Use IMDb Rating through OMDb",
"mdb": "Use MDBList Score",
"mdb_average": "Use MDBList Average Score",
"mdb_imdb": "Use IMDb Rating through MDBList",
"mdb_metacritic": "Use Metacritic Rating through MDBList",
"mdb_metacriticuser": "Use Metacritic User Rating through MDBList",
"mdb_trakt": "Use Trakt Rating through MDBList",
"mdb_tomatoes": "Use Rotten Tomatoes Rating through MDBList",
"mdb_tomatoesaudience": "Use Rotten Tomatoes Audience Rating through MDBList",
"mdb_tmdb": "Use TMDb Rating through MDBList",
"mdb_letterboxd": "Use Letterboxd Rating through MDBList",
"mdb_myanimelist": "Use MyAnimeList Rating through MDBList",
"anidb_rating": "Use AniDB Rating",
"anidb_average": "Use AniDB Average",
"anidb_score": "Use AniDB Review Dcore",
"mal": "Use MyAnimeList Rating"
}
reset_overlay_options = {"tmdb": "Reset to TMDb poster", "plex": "Reset to Plex Poster"}
library_operations = {
"assets_for_all": "bool", "split_duplicates": "bool", "update_blank_track_titles": "bool", "remove_title_parentheses": "bool",
"radarr_add_all_existing": "bool", "radarr_remove_by_tag": "str", "sonarr_add_all_existing": "bool", "sonarr_remove_by_tag": "str",
"mass_content_rating_update": mass_content_options, "mass_collection_content_rating_update": "dict",
"mass_genre_update": mass_genre_options, "mass_studio_update": mass_studio_options,
"mass_audience_rating_update": mass_rating_options, "mass_episode_audience_rating_update": mass_episode_rating_options,
"mass_critic_rating_update": mass_rating_options, "mass_episode_critic_rating_update": mass_episode_rating_options,
"mass_user_rating_update": mass_rating_options, "mass_episode_user_rating_update": mass_episode_rating_options,
"mass_original_title_update": mass_original_title_options, "mass_imdb_parental_labels": imdb_label_options,
"mass_originally_available_update": mass_available_options, "mass_added_at_update": mass_available_options,
"mass_collection_mode": "mass_collection_mode", "mass_poster_update": "dict", "mass_background_update": "dict",
"metadata_backup": "dict", "delete_collections": "dict", "genre_mapper": "dict", "content_rating_mapper": "dict",
}
class ConfigFile:
def __init__(self, default_dir, attrs, secrets):
logger.info("Locating config...")
config_file = attrs["config_file"]
if config_file and os.path.exists(config_file): self.config_path = os.path.abspath(config_file)
elif config_file and not os.path.exists(config_file): raise Failed(f"Config Error: config not found at {os.path.abspath(config_file)}")
elif os.path.exists(os.path.join(default_dir, "config.yml")): self.config_path = os.path.abspath(os.path.join(default_dir, "config.yml"))
else: raise Failed(f"Config Error: config not found at {os.path.abspath(default_dir)}")
logger.info(f"Using {self.config_path} as config")
logger.clear_errors()
self._mediastingers = None
self.default_dir = default_dir
self.secrets = secrets
self.version = attrs["version"] if "version" in attrs else None
self.branch = attrs["branch"] if "branch" in attrs else None
self.read_only = attrs["read_only"] if "read_only" in attrs else False
self.no_missing = attrs["no_missing"] if "no_missing" in attrs else None
self.no_report = attrs["no_report"] if "no_report" in attrs else None
self.ignore_schedules = attrs["ignore_schedules"] if "ignore_schedules" in attrs else False
self.start_time = attrs["time_obj"]
self.run_hour = datetime.strptime(attrs["time"], "%H:%M").hour
self.requested_collections = None
if "collections" in attrs and attrs["collections"]:
self.requested_collections = [s.strip() for s in attrs["collections"].split("|")]
self.requested_libraries = None
if "libraries" in attrs and attrs["libraries"]:
self.requested_libraries = [s.strip() for s in attrs["libraries"].split("|")]
self.requested_files = None
if "files" in attrs and attrs["files"]:
self.requested_files = []
for s in attrs["files"].split("|"):
s = s.strip()
if s:
if s.endswith(".yml"):
self.requested_files.append(s[:-4])
elif s.endswith(".yaml"):
self.requested_files.append(s[:-5])
else:
self.requested_files.append(s)
self.collection_only = attrs["collection_only"] if "collection_only" in attrs else False
self.metadata_only = attrs["metadata_only"] if "metadata_only" in attrs else False
self.operations_only = attrs["operations_only"] if "operations_only" in attrs else False
self.overlays_only = attrs["overlays_only"] if "overlays_only" in attrs else False
self.env_plex_url = attrs["plex_url"] if "plex_url" in attrs else ""
self.env_plex_token = attrs["plex_token"] if "plex_token" in attrs else ""
self.tpdb_timer = None
current_time = datetime.now()
with open(self.config_path, encoding="utf-8") as fp:
logger.separator("Redacted Config", space=False, border=False, debug=True)
for line in fp.readlines():
logger.debug(re.sub(r"(token|client.*|url|api_*key|secret|error|delete|run_start|run_end|version|changes|username|password): .+", r"\1: (redacted)", line.strip("\r\n")))
logger.debug("")
self.data = YAML(self.config_path).data
def replace_attr(all_data, in_attr, par):
if "settings" not in all_data:
all_data["settings"] = {}
if par in all_data and all_data[par] and in_attr in all_data[par] and in_attr not in all_data["settings"]:
all_data["settings"][in_attr] = all_data[par][in_attr]
del all_data[par][in_attr]
if "libraries" not in self.data:
self.data["libraries"] = {}
if "settings" not in self.data:
self.data["settings"] = {}
if "tmdb" not in self.data:
self.data["tmdb"] = {}
replace_attr(self.data, "cache", "cache")
replace_attr(self.data, "cache_expiration", "cache")
if "config" in self.data:
del self.data["cache"]
replace_attr(self.data, "asset_directory", "plex")
replace_attr(self.data, "sync_mode", "plex")
replace_attr(self.data, "show_unmanaged", "plex")
replace_attr(self.data, "show_filtered", "plex")
replace_attr(self.data, "show_missing", "plex")
replace_attr(self.data, "save_missing", "plex")
if self.data["libraries"]:
for library in self.data["libraries"]:
if not self.data["libraries"][library]:
continue
if "metadata_path" in self.data["libraries"][library]:
logger.warning("Config Warning: metadata_path has been deprecated and split into collection_files and metadata_files, Please visit the wiki to learn more about this transition.")
path_dict = self.data["libraries"][library].pop("metadata_path")
if "collection_files" not in self.data["libraries"][library]:
self.data["libraries"][library]["collection_files"] = path_dict
if "metadata_files" not in self.data["libraries"][library]:
self.data["libraries"][library]["metadata_files"] = path_dict
if "overlay_path" in self.data["libraries"][library]:
logger.warning("Config Warning: overlay_path has been deprecated in favor of overlay_files, Please visit the wiki to learn more about this transition.")
self.data["libraries"][library]["overlay_files"] = self.data["libraries"][library].pop("overlay_path")
if "radarr_add_all" in self.data["libraries"][library]:
self.data["libraries"][library]["radarr_add_all_existing"] = self.data["libraries"][library].pop("radarr_add_all")
if "sonarr_add_all" in self.data["libraries"][library]:
self.data["libraries"][library]["sonarr_add_all_existing"] = self.data["libraries"][library].pop("sonarr_add_all")
if "plex" in self.data["libraries"][library] and self.data["libraries"][library]["plex"]:
replace_attr(self.data["libraries"][library], "asset_directory", "plex")
replace_attr(self.data["libraries"][library], "sync_mode", "plex")
replace_attr(self.data["libraries"][library], "show_unmanaged", "plex")
replace_attr(self.data["libraries"][library], "show_filtered", "plex")
replace_attr(self.data["libraries"][library], "show_missing", "plex")
replace_attr(self.data["libraries"][library], "save_missing", "plex")
if "settings" in self.data["libraries"][library] and self.data["libraries"][library]["settings"]:
if "collection_minimum" in self.data["libraries"][library]["settings"]:
self.data["libraries"][library]["settings"]["minimum_items"] = self.data["libraries"][library]["settings"].pop("collection_minimum")
if "save_missing" in self.data["libraries"][library]["settings"]:
self.data["libraries"][library]["settings"]["save_report"] = self.data["libraries"][library]["settings"].pop("save_missing")
if "radarr" in self.data["libraries"][library] and self.data["libraries"][library]["radarr"]:
if "monitor" in self.data["libraries"][library]["radarr"] and isinstance(self.data["libraries"][library]["radarr"]["monitor"], bool):
self.data["libraries"][library]["radarr"]["monitor"] = True if self.data["libraries"][library]["radarr"]["monitor"] else False
if "add" in self.data["libraries"][library]["radarr"]:
self.data["libraries"][library]["radarr"]["add_missing"] = self.data["libraries"][library]["radarr"].pop("add")
if "sonarr" in self.data["libraries"][library] and self.data["libraries"][library]["sonarr"]:
if "add" in self.data["libraries"][library]["sonarr"]:
self.data["libraries"][library]["sonarr"]["add_missing"] = self.data["libraries"][library]["sonarr"].pop("add")
if "operations" in self.data["libraries"][library] and self.data["libraries"][library]["operations"]:
if "radarr_add_all" in self.data["libraries"][library]["operations"]:
self.data["libraries"][library]["operations"]["radarr_add_all_existing"] = self.data["libraries"][library]["operations"].pop("radarr_add_all")
if "sonarr_add_all" in self.data["libraries"][library]["operations"]:
self.data["libraries"][library]["operations"]["sonarr_add_all_existing"] = self.data["libraries"][library]["operations"].pop("sonarr_add_all")
if "mass_imdb_parental_labels" in self.data["libraries"][library]["operations"] and self.data["libraries"][library]["operations"]["mass_imdb_parental_labels"]:
if self.data["libraries"][library]["operations"]["mass_imdb_parental_labels"] == "with_none":
self.data["libraries"][library]["operations"]["mass_imdb_parental_labels"] = "none"
elif self.data["libraries"][library]["operations"]["mass_imdb_parental_labels"] == "without_none":
self.data["libraries"][library]["operations"]["mass_imdb_parental_labels"] = "mild"
if "webhooks" in self.data["libraries"][library] and self.data["libraries"][library]["webhooks"] and "collection_changes" not in self.data["libraries"][library]["webhooks"]:
changes = []
def hooks(hook_attr):
if hook_attr in self.data["libraries"][library]["webhooks"]:
changes.extend([w for w in util.get_list(self.data["libraries"][library]["webhooks"].pop(hook_attr), split=False) if w not in changes])
hooks("collection_creation")
hooks("collection_addition")
hooks("collection_removal")
hooks("collection_changes")
self.data["libraries"][library]["webhooks"]["changes"] = None if not changes else changes if len(changes) > 1 else changes[0]
if "libraries" in self.data: self.data["libraries"] = self.data.pop("libraries")
if "playlist_files" in self.data: self.data["playlist_files"] = self.data.pop("playlist_files")
if "settings" in self.data:
temp = self.data.pop("settings")
if "collection_minimum" in temp:
temp["minimum_items"] = temp.pop("collection_minimum")
if "playlist_sync_to_user" in temp:
temp["playlist_sync_to_users"] = temp.pop("playlist_sync_to_user")
if "save_missing" in temp:
temp["save_report"] = temp.pop("save_missing")
self.data["settings"] = temp
if "webhooks" in self.data:
temp = self.data.pop("webhooks")
if "changes" not in temp:
changes = []
def hooks(hook_attr):
if hook_attr in temp:
items = util.get_list(temp.pop(hook_attr), split=False)
if items:
changes.extend([w for w in items if w not in changes])
hooks("collection_creation")
hooks("collection_addition")
hooks("collection_removal")
hooks("collection_changes")
temp["changes"] = None if not changes else changes if len(changes) > 1 else changes[0]
self.data["webhooks"] = temp
if "github" in self.data: self.data["github"] = self.data.pop("github")
if "plex" in self.data: self.data["plex"] = self.data.pop("plex")
if "tmdb" in self.data: self.data["tmdb"] = self.data.pop("tmdb")
if "tautulli" in self.data: self.data["tautulli"] = self.data.pop("tautulli")
if "omdb" in self.data: self.data["omdb"] = self.data.pop("omdb")
if "mdblist" in self.data: self.data["mdblist"] = self.data.pop("mdblist")
if "notifiarr" in self.data: self.data["notifiarr"] = self.data.pop("notifiarr")
if "gotify" in self.data: self.data["gotify"] = self.data.pop("gotify")
if "anidb" in self.data: self.data["anidb"] = self.data.pop("anidb")
if "radarr" in self.data:
if "monitor" in self.data["radarr"] and isinstance(self.data["radarr"]["monitor"], bool):
self.data["radarr"]["monitor"] = True if self.data["radarr"]["monitor"] else False
temp = self.data.pop("radarr")
if temp and "add" in temp:
temp["add_missing"] = temp.pop("add")
self.data["radarr"] = temp
if "sonarr" in self.data:
temp = self.data.pop("sonarr")
if temp and "add" in temp:
temp["add_missing"] = temp.pop("add")
self.data["sonarr"] = temp
if "trakt" in self.data: self.data["trakt"] = self.data.pop("trakt")
if "mal" in self.data: self.data["mal"] = self.data.pop("mal")
def check_next(next_data):
if isinstance(next_data, dict):
for d in next_data:
out = check_next(next_data[d])
if out:
next_data[d] = out
elif isinstance(next_data, list):
for d in next_data:
check_next(d)
else:
for secret, secret_value in self.secrets.items():
for test in [secret, secret.upper().replace("-", "_")]:
if f"<<{test}>>" in str(next_data):
return str(next_data).replace(f"<<{test}>>", secret_value)
return next_data
if self.secrets:
check_next(self.data)
def check_for_attribute(data, attribute, parent=None, test_list=None, translations=None, default=None, do_print=True, default_is_none=False, req_default=False, var_type="str", throw=False, save=True, int_min=0, int_max=None):
endline = ""
if parent is not None:
if data and parent in data:
data = data[parent]
else:
data = None
do_print = False
save = False
final_value = data[attribute] if data and attribute in data else None
if translations and final_value in translations:
final_value = translations[final_value]
if self.read_only:
save = False
text = f"{attribute} attribute" if parent is None else f"{parent} sub-attribute {attribute}"
if data is None or attribute not in data:
message = f"{text} not found"
if parent and save is True:
yaml = YAML(self.config_path)
endline = f"\n{parent} sub-attribute {attribute} added to config"
if parent not in yaml.data or not yaml.data[parent]: yaml.data[parent] = {attribute: default}
elif attribute not in yaml.data[parent]: yaml.data[parent][attribute] = default
else: endline = ""
yaml.save()
if default_is_none and var_type in ["list", "int_list", "lower_list", "list_path"]: return default if default else []
elif final_value is None:
if default_is_none and var_type in ["list", "int_list", "lower_list", "list_path"]: return default if default else []
elif default_is_none: return None
else: message = f"{text} is blank"
elif var_type == "url":
if final_value.endswith(("\\", "/")): return final_value[:-1]
else: return final_value
elif var_type == "bool":
if isinstance(final_value, bool): return final_value
else: message = f"{text} must be either true or false"
elif var_type == "int":
if isinstance(final_value, int) and final_value >= int_min and (not int_max or final_value <= int_max):
return final_value
else:
message = f"{text} must an integer greater than or equal to {int_min}{f' and less than or equal to {int_max}'}"
elif var_type == "path":
if os.path.exists(os.path.abspath(final_value)): return final_value
else: message = f"Path {os.path.abspath(final_value)} does not exist"
elif var_type in ["list", "lower_list", "int_list"]:
output_list = []
for output_item in util.get_list(final_value, lower=var_type == "lower_list", split=var_type != "list", int_list=var_type == "int_list"):
if output_item not in output_list:
output_list.append(output_item)
failed_items = [o for o in output_list if o not in test_list] if test_list else []
if failed_items:
message = f"{text}: {', '.join(failed_items)} is an invalid input"
else:
return output_list
elif var_type == "list_path":
temp_list = []
warning_message = ""
for p in util.get_list(final_value, split=False):
if os.path.exists(os.path.abspath(p)):
temp_list.append(p)
else:
if len(warning_message) > 0:
warning_message += "\n"
warning_message += f"Config Warning: Path does not exist: {os.path.abspath(p)}"
if do_print and warning_message:
logger.warning(warning_message)
if len(temp_list) > 0: return temp_list
else: message = "No Paths exist"
elif test_list is None or final_value in test_list: return final_value
else: message = f"{text}: {final_value} is an invalid input"
if var_type == "path" and default and os.path.exists(os.path.abspath(default)):
return default
elif var_type == "path" and default:
if final_value:
message = f"neither {final_value} or the default path {default} could be found"
else:
message = f"no {text} found and the default path {default} could not be found"
default = None
if default is not None or default_is_none:
message = message + f" using {default} as default"
message = message + endline
if req_default and default is None:
raise Failed(f"Config Error: {attribute} attribute must be set under {parent} globally or under this specific Library")
options = ""
if test_list:
for test_option, test_description in test_list.items():
if len(options) > 0:
options = f"{options}\n"
options = f"{options} {test_option} ({test_description})"
if (default is None and not default_is_none) or throw:
if len(options) > 0:
message = message + "\n" + options
raise Failed(f"Config Error: {message}")
if do_print:
logger.warning(f"Config Warning: {message}")
if final_value and test_list is not None and final_value not in test_list:
logger.warning(options)
return default
self.general = {
"run_order": check_for_attribute(self.data, "run_order", parent="settings", var_type="lower_list", test_list=run_order_options, default=["operations", "metadata", "collections", "overlays"]),
"cache": check_for_attribute(self.data, "cache", parent="settings", var_type="bool", default=True),
"cache_expiration": check_for_attribute(self.data, "cache_expiration", parent="settings", var_type="int", default=60, int_min=1),
"asset_directory": check_for_attribute(self.data, "asset_directory", parent="settings", var_type="list_path", default_is_none=True),
"asset_folders": check_for_attribute(self.data, "asset_folders", parent="settings", var_type="bool", default=True),
"asset_depth": check_for_attribute(self.data, "asset_depth", parent="settings", var_type="int", default=0),
"create_asset_folders": check_for_attribute(self.data, "create_asset_folders", parent="settings", var_type="bool", default=False),
"prioritize_assets": check_for_attribute(self.data, "prioritize_assets", parent="settings", var_type="bool", default=False),
"dimensional_asset_rename": check_for_attribute(self.data, "dimensional_asset_rename", parent="settings", var_type="bool", default=False),
"download_url_assets": check_for_attribute(self.data, "download_url_assets", parent="settings", var_type="bool", default=False),
"show_missing_assets": check_for_attribute(self.data, "show_missing_assets", parent="settings", var_type="bool", default=True),
"show_missing_season_assets": check_for_attribute(self.data, "show_missing_season_assets", parent="settings", var_type="bool", default=False),
"show_missing_episode_assets": check_for_attribute(self.data, "show_missing_episode_assets", parent="settings", var_type="bool", default=False),
"show_asset_not_needed": check_for_attribute(self.data, "show_asset_not_needed", parent="settings", var_type="bool", default=True),
"sync_mode": check_for_attribute(self.data, "sync_mode", parent="settings", default="append", test_list=sync_modes),
"default_collection_order": check_for_attribute(self.data, "default_collection_order", parent="settings", default_is_none=True),
"minimum_items": check_for_attribute(self.data, "minimum_items", parent="settings", var_type="int", default=1),
"item_refresh_delay": check_for_attribute(self.data, "item_refresh_delay", parent="settings", var_type="int", default=0),
"delete_below_minimum": check_for_attribute(self.data, "delete_below_minimum", parent="settings", var_type="bool", default=False),
"delete_not_scheduled": check_for_attribute(self.data, "delete_not_scheduled", parent="settings", var_type="bool", default=False),
"run_again_delay": check_for_attribute(self.data, "run_again_delay", parent="settings", var_type="int", default=0),
"missing_only_released": check_for_attribute(self.data, "missing_only_released", parent="settings", var_type="bool", default=False),
"only_filter_missing": check_for_attribute(self.data, "only_filter_missing", parent="settings", var_type="bool", default=False),
"show_unmanaged": check_for_attribute(self.data, "show_unmanaged", parent="settings", var_type="bool", default=True),
"show_unconfigured": check_for_attribute(self.data, "show_unconfigured", parent="settings", var_type="bool", default=True),
"show_filtered": check_for_attribute(self.data, "show_filtered", parent="settings", var_type="bool", default=False),
"show_options": check_for_attribute(self.data, "show_options", parent="settings", var_type="bool", default=False),
"show_missing": check_for_attribute(self.data, "show_missing", parent="settings", var_type="bool", default=True),
"save_report": check_for_attribute(self.data, "save_report", parent="settings", var_type="bool", default=False),
"tvdb_language": check_for_attribute(self.data, "tvdb_language", parent="settings", default="default"),
"ignore_ids": check_for_attribute(self.data, "ignore_ids", parent="settings", var_type="int_list", default_is_none=True),
"ignore_imdb_ids": check_for_attribute(self.data, "ignore_imdb_ids", parent="settings", var_type="lower_list", default_is_none=True),
"playlist_sync_to_users": check_for_attribute(self.data, "playlist_sync_to_users", parent="settings", default="all", default_is_none=True),
"playlist_exclude_users": check_for_attribute(self.data, "playlist_exclude_users", parent="settings", default_is_none=True),
"playlist_report": check_for_attribute(self.data, "playlist_report", parent="settings", var_type="bool", default=True),
"verify_ssl": check_for_attribute(self.data, "verify_ssl", parent="settings", var_type="bool", default=True),
"custom_repo": check_for_attribute(self.data, "custom_repo", parent="settings", default_is_none=True),
"overlay_artwork_filetype": check_for_attribute(self.data, "overlay_artwork_filetype", parent="settings", test_list=filetype_list, translations={"webp": "webp_lossy"}, default="jpg"),
"overlay_artwork_quality": check_for_attribute(self.data, "overlay_artwork_quality", parent="settings", var_type="int", default_is_none=True, int_min=1, int_max=100),
"assets_for_all": check_for_attribute(self.data, "assets_for_all", parent="settings", var_type="bool", default=False, save=False, do_print=False)
}
self.custom_repo = None
if self.general["custom_repo"]:
repo = self.general["custom_repo"]
if "https://github.com/" in repo:
repo = repo.replace("https://github.com/", "https://raw.githubusercontent.com/").replace("/tree/", "/")
self.custom_repo = repo
self.latest_version = util.current_version(self.version, branch=self.branch)
add_operations = True if "operations" not in self.general["run_order"] else False
add_metadata = True if "metadata" not in self.general["run_order"] else False
add_collection = True if "collections" not in self.general["run_order"] else False
add_overlays = True if "overlays" not in self.general["run_order"] else False
if add_operations or add_metadata or add_collection or add_overlays:
new_run_order = []
for run_order in self.general["run_order"]:
if add_operations and not new_run_order:
new_run_order.append("operations")
if add_metadata:
new_run_order.append("metadata")
if add_collection:
new_run_order.append("collections")
new_run_order.append(run_order)
if add_metadata and run_order == "operations":
new_run_order.append("metadata")
if add_collection and (run_order == "metadata" or (run_order == "operations" and add_metadata)):
new_run_order.append("collections")
if add_overlays:
new_run_order.append("overlays")
self.general["run_order"] = new_run_order
yaml = YAML(self.config_path)
if "settings" not in yaml.data or not yaml.data["settings"]:
yaml.data["settings"] = {}
yaml.data["settings"]["run_order"] = new_run_order
yaml.save()
self.session = requests.Session()
if not self.general["verify_ssl"]:
self.session.verify = False
if self.session.verify is False:
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
if self.general["cache"]:
logger.separator()
self.Cache = Cache(self.config_path, self.general["cache_expiration"])
else:
self.Cache = None
self.GitHub = GitHub(self, {"token": check_for_attribute(self.data, "token", parent="github", default_is_none=True)})
logger.separator()
self.NotifiarrFactory = None
if "notifiarr" in self.data:
logger.info("Connecting to Notifiarr...")
try:
self.NotifiarrFactory = Notifiarr(self, {"apikey": check_for_attribute(self.data, "apikey", parent="notifiarr", throw=True)})
except Failed as e:
if str(e).endswith("is blank"):
logger.warning(e)
else:
logger.stacktrace()
logger.error(e)
logger.info(f"Notifiarr Connection {'Failed' if self.NotifiarrFactory is None else 'Successful'}")
else:
logger.info("notifiarr attribute not found")
self.GotifyFactory = None
if "gotify" in self.data:
logger.info("Connecting to Gotify...")
try:
self.GotifyFactory = Gotify(self, {
"url": check_for_attribute(self.data, "url", parent="gotify", throw=True),
"token": check_for_attribute(self.data, "token", parent="gotify", throw=True)
})
except Failed as e:
if str(e).endswith("is blank"):
logger.warning(e)
else:
logger.stacktrace()
logger.error(e)
logger.info(f"Gotify Connection {'Failed' if self.GotifyFactory is None else 'Successful'}")
else:
logger.info("gotify attribute not found")
self.webhooks = {
"error": check_for_attribute(self.data, "error", parent="webhooks", var_type="list", default_is_none=True),
"version": check_for_attribute(self.data, "version", parent="webhooks", var_type="list", default_is_none=True),
"run_start": check_for_attribute(self.data, "run_start", parent="webhooks", var_type="list", default_is_none=True),
"run_end": check_for_attribute(self.data, "run_end", parent="webhooks", var_type="list", default_is_none=True),
"changes": check_for_attribute(self.data, "changes", parent="webhooks", var_type="list", default_is_none=True),
"delete": check_for_attribute(self.data, "delete", parent="webhooks", var_type="list", default_is_none=True)
}
self.Webhooks = Webhooks(self, self.webhooks, notifiarr=self.NotifiarrFactory, gotify=self.GotifyFactory)
try:
self.Webhooks.start_time_hooks(self.start_time)
if self.version[0] != "Unknown" and self.latest_version[0] != "Unknown" and self.version[1] != self.latest_version[1] or (self.version[2] and self.version[2] < self.latest_version[2]):
self.Webhooks.version_hooks(self.version, self.latest_version)
except Failed as e:
logger.stacktrace()
logger.error(f"Webhooks Error: {e}")
logger.save_errors = True
logger.separator()
try:
self.TMDb = None
if "tmdb" in self.data:
logger.info("Connecting to TMDb...")
self.TMDb = TMDb(self, {
"apikey": check_for_attribute(self.data, "apikey", parent="tmdb", throw=True),
"language": check_for_attribute(self.data, "language", parent="tmdb", default="en"),
"expiration": check_for_attribute(self.data, "cache_expiration", parent="tmdb", var_type="int", default=60, int_min=1)
})
regions = {k.upper(): v for k, v in self.TMDb.iso_3166_1.items()}
region = check_for_attribute(self.data, "region", parent="tmdb", test_list=regions, default_is_none=True)
self.TMDb.region = str(region).upper() if region else region
logger.info(f"TMDb Connection {'Failed' if self.TMDb is None else 'Successful'}")
else:
raise Failed("Config Error: tmdb attribute not found")
logger.separator()
self.OMDb = None
if "omdb" in self.data:
logger.info("Connecting to OMDb...")
try:
self.OMDb = OMDb(self, {
"apikey": check_for_attribute(self.data, "apikey", parent="omdb", throw=True),
"expiration": check_for_attribute(self.data, "cache_expiration", parent="omdb", var_type="int", default=60, int_min=1)
})
except Failed as e:
if str(e).endswith("is blank"):
logger.warning(e)
else:
logger.error(e)
logger.info(f"OMDb Connection {'Failed' if self.OMDb is None else 'Successful'}")
else:
logger.info("omdb attribute not found")
logger.separator()
self.MDBList = MDBList(self)
if "mdblist" in self.data:
logger.info("Connecting to MDBList...")
try:
self.MDBList.add_key(
check_for_attribute(self.data, "apikey", parent="mdblist", throw=True),
check_for_attribute(self.data, "cache_expiration", parent="mdblist", var_type="int", default=60, int_min=1)
)
logger.info("MDBList Connection Successful")
except Failed as e:
if str(e).endswith("is blank"):
logger.warning(e)
else:
logger.error(e)
logger.info("MDBList Connection Failed")
else:
logger.info("mdblist attribute not found")
logger.separator()
self.Trakt = None
if "trakt" in self.data:
logger.info("Connecting to Trakt...")
try:
self.Trakt = Trakt(self, {
"client_id": check_for_attribute(self.data, "client_id", parent="trakt", throw=True),
"client_secret": check_for_attribute(self.data, "client_secret", parent="trakt", throw=True),
"pin": check_for_attribute(self.data, "pin", parent="trakt", default_is_none=True),
"config_path": self.config_path,
"authorization": self.data["trakt"]["authorization"] if "authorization" in self.data["trakt"] else None
})
except Failed as e:
if str(e).endswith("is blank"):
logger.warning(e)
else:
logger.error(e)
logger.info(f"Trakt Connection {'Failed' if self.Trakt is None else 'Successful'}")
else:
logger.info("trakt attribute not found")
logger.separator()
self.MyAnimeList = None
if "mal" in self.data:
logger.info("Connecting to My Anime List...")
try:
self.MyAnimeList = MyAnimeList(self, {
"client_id": check_for_attribute(self.data, "client_id", parent="mal", throw=True),
"client_secret": check_for_attribute(self.data, "client_secret", parent="mal", throw=True),
"localhost_url": check_for_attribute(self.data, "localhost_url", parent="mal", default_is_none=True),
"cache_expiration": check_for_attribute(self.data, "cache_expiration", parent="mal", var_type="int", default=60, int_min=1),
"config_path": self.config_path,
"authorization": self.data["mal"]["authorization"] if "authorization" in self.data["mal"] else None
})
except Failed as e:
if str(e).endswith("is blank"):
logger.warning(e)
else:
logger.error(e)
logger.info(f"My Anime List Connection {'Failed' if self.MyAnimeList is None else 'Successful'}")
else:
logger.info("mal attribute not found")
self.AniDB = AniDB(self, {"language": check_for_attribute(self.data, "language", parent="anidb", default="en")})
if "anidb" in self.data:
logger.separator()
logger.info("Connecting to AniDB...")
try:
self.AniDB.authorize(
check_for_attribute(self.data, "client", parent="anidb", throw=True),
check_for_attribute(self.data, "version", parent="anidb", var_type="int", throw=True),
check_for_attribute(self.data, "cache_expiration", parent="anidb", var_type="int", default=60, int_min=1)
)
except Failed as e:
if str(e).endswith("is blank"):
logger.warning(e)
else:
logger.error(e)
logger.info(f"AniDB API Connection {'Successful' if self.AniDB.is_authorized else 'Failed'}")
try:
self.AniDB.login(
check_for_attribute(self.data, "username", parent="anidb", throw=True),
check_for_attribute(self.data, "password", parent="anidb", throw=True)
)
except Failed as e:
if str(e).endswith("is blank"):
logger.warning(e)
else:
logger.error(e)
logger.info(f"AniDB Login {'Successful' if self.AniDB.username else 'Failed Continuing as Guest'}")
logger.separator()
self.playlist_names = []
self.playlist_files = []
if "playlist_files" not in self.data:
logger.info("playlist_files attribute not found")
elif not self.data["playlist_files"]:
logger.info("playlist_files attribute is blank")
else:
logger.info("")
logger.info("Reading in Playlist Files")
files, had_scheduled = util.load_files(self.data["playlist_files"], "playlist_files", schedule=(current_time, self.run_hour, self.ignore_schedules))
if not files and not had_scheduled:
raise Failed("Config Error: No Paths Found for playlist_files")
for file_type, playlist_file, temp_vars, asset_directory in files:
try:
playlist_obj = PlaylistFile(self, file_type, playlist_file, temp_vars, asset_directory)
self.playlist_names.extend([p for p in playlist_obj.playlists])
self.playlist_files.append(playlist_obj)
except Failed as e:
logger.info("Playlist File Failed To Load")
logger.error(e)
except NotScheduled as e:
logger.info("")
logger.separator(f"Skipping {e} Playlist File")
self.TVDb = TVDb(self, self.general["tvdb_language"], self.general["cache_expiration"])
self.IMDb = IMDb(self)
self.Convert = Convert(self)
self.AniList = AniList(self)
self.ICheckMovies = ICheckMovies(self)
self.Letterboxd = Letterboxd(self)
self.BoxOfficeMojo = BoxOfficeMojo(self)
self.Reciperr = Reciperr(self)
self.Ergast = Ergast(self)
logger.separator()
logger.info("Connecting to Plex Libraries...")
self.general["plex"] = {
"url": check_for_attribute(self.data, "url", parent="plex", var_type="url", default_is_none=True),
"token": check_for_attribute(self.data, "token", parent="plex", default_is_none=True),
"timeout": check_for_attribute(self.data, "timeout", parent="plex", var_type="int", default=60),
"verify_ssl": check_for_attribute(self.data, "verify_ssl", parent="plex", var_type="bool", default_is_none=True),
"db_cache": check_for_attribute(self.data, "db_cache", parent="plex", var_type="int", default_is_none=True)
}
for attr in ["clean_bundles", "empty_trash", "optimize"]:
try:
self.general["plex"][attr] = check_for_attribute(self.data, attr, parent="plex", var_type="bool", default=False, throw=True)
except Failed as e:
if "plex" in self.data and attr in self.data["plex"] and self.data["plex"][attr]:
self.general["plex"][attr] = self.data["plex"][attr]
else:
self.general["plex"][attr] = False
logger.warning(str(e).replace("Error", "Warning"))
self.general["radarr"] = {
"url": check_for_attribute(self.data, "url", parent="radarr", var_type="url", default_is_none=True),
"token": check_for_attribute(self.data, "token", parent="radarr", default_is_none=True),
"add_missing": check_for_attribute(self.data, "add_missing", parent="radarr", var_type="bool", default=False),
"add_existing": check_for_attribute(self.data, "add_existing", parent="radarr", var_type="bool", default=False),
"upgrade_existing": check_for_attribute(self.data, "upgrade_existing", parent="radarr", var_type="bool", default=False),
"monitor_existing": check_for_attribute(self.data, "monitor_existing", parent="radarr", var_type="bool", default=False),
"ignore_cache": check_for_attribute(self.data, "ignore_cache", parent="radarr", var_type="bool", default=False),
"root_folder_path": check_for_attribute(self.data, "root_folder_path", parent="radarr", default_is_none=True),
"monitor": check_for_attribute(self.data, "monitor", parent="radarr", var_type="bool", default=True),
"availability": check_for_attribute(self.data, "availability", parent="radarr", test_list=radarr.availability_descriptions, default="announced"),
"quality_profile": check_for_attribute(self.data, "quality_profile", parent="radarr", default_is_none=True),
"tag": check_for_attribute(self.data, "tag", parent="radarr", var_type="lower_list", default_is_none=True),
"search": check_for_attribute(self.data, "search", parent="radarr", var_type="bool", default=False),
"radarr_path": check_for_attribute(self.data, "radarr_path", parent="radarr", default_is_none=True),
"plex_path": check_for_attribute(self.data, "plex_path", parent="radarr", default_is_none=True)
}
self.general["sonarr"] = {
"url": check_for_attribute(self.data, "url", parent="sonarr", var_type="url", default_is_none=True),
"token": check_for_attribute(self.data, "token", parent="sonarr", default_is_none=True),
"add_missing": check_for_attribute(self.data, "add_missing", parent="sonarr", var_type="bool", default=False),
"add_existing": check_for_attribute(self.data, "add_existing", parent="sonarr", var_type="bool", default=False),
"upgrade_existing": check_for_attribute(self.data, "upgrade_existing", parent="sonarr", var_type="bool", default=False),
"monitor_existing": check_for_attribute(self.data, "monitor_existing", parent="sonarr", var_type="bool", default=False),
"ignore_cache": check_for_attribute(self.data, "ignore_cache", parent="sonarr", var_type="bool", default=False),
"root_folder_path": check_for_attribute(self.data, "root_folder_path", parent="sonarr", default_is_none=True),
"monitor": check_for_attribute(self.data, "monitor", parent="sonarr", test_list=sonarr.monitor_descriptions, default="all"),
"quality_profile": check_for_attribute(self.data, "quality_profile", parent="sonarr", default_is_none=True),
"language_profile": check_for_attribute(self.data, "language_profile", parent="sonarr", default_is_none=True),
"series_type": check_for_attribute(self.data, "series_type", parent="sonarr", test_list=sonarr.series_type_descriptions, default="standard"),
"season_folder": check_for_attribute(self.data, "season_folder", parent="sonarr", var_type="bool", default=True),
"tag": check_for_attribute(self.data, "tag", parent="sonarr", var_type="lower_list", default_is_none=True),
"search": check_for_attribute(self.data, "search", parent="sonarr", var_type="bool", default=False),
"cutoff_search": check_for_attribute(self.data, "cutoff_search", parent="sonarr", var_type="bool", default=False),
"sonarr_path": check_for_attribute(self.data, "sonarr_path", parent="sonarr", default_is_none=True),
"plex_path": check_for_attribute(self.data, "plex_path", parent="sonarr", default_is_none=True)
}
self.general["tautulli"] = {
"url": check_for_attribute(self.data, "url", parent="tautulli", var_type="url", default_is_none=True),
"apikey": check_for_attribute(self.data, "apikey", parent="tautulli", default_is_none=True)
}
self.libraries = []
libs = check_for_attribute(self.data, "libraries", throw=True)
for library_name, lib in libs.items():
if self.requested_libraries and library_name not in self.requested_libraries:
continue
params = {o: None for o in library_operations}
params["mapping_name"] = str(library_name)
params["name"] = str(lib["library_name"]) if lib and "library_name" in lib and lib["library_name"] else str(library_name)
display_name = f"{params['name']} ({params['mapping_name']})" if lib and "library_name" in lib and lib["library_name"] else params["mapping_name"]
logger.separator(f"{display_name} Configuration")
logger.info("")
logger.info(f"Connecting to {display_name} Library...")
params["run_order"] = check_for_attribute(lib, "run_order", parent="settings", var_type="lower_list", default=self.general["run_order"], do_print=False, save=False)
params["asset_directory"] = check_for_attribute(lib, "asset_directory", parent="settings", var_type="list_path", default=self.general["asset_directory"], default_is_none=True, do_print=False, save=False)
params["asset_folders"] = check_for_attribute(lib, "asset_folders", parent="settings", var_type="bool", default=self.general["asset_folders"], do_print=False, save=False)
params["asset_depth"] = check_for_attribute(lib, "asset_depth", parent="settings", var_type="int", default=self.general["asset_depth"], do_print=False, save=False)
params["sync_mode"] = check_for_attribute(lib, "sync_mode", parent="settings", test_list=sync_modes, default=self.general["sync_mode"], do_print=False, save=False)
params["default_collection_order"] = check_for_attribute(lib, "default_collection_order", parent="settings", default=self.general["default_collection_order"], default_is_none=True, do_print=False, save=False)
params["show_unmanaged"] = check_for_attribute(lib, "show_unmanaged", parent="settings", var_type="bool", default=self.general["show_unmanaged"], do_print=False, save=False)
params["show_unconfigured"] = check_for_attribute(lib, "show_unconfigured", parent="settings", var_type="bool", default=self.general["show_unconfigured"], do_print=False, save=False)
params["show_filtered"] = check_for_attribute(lib, "show_filtered", parent="settings", var_type="bool", default=self.general["show_filtered"], do_print=False, save=False)
params["show_options"] = check_for_attribute(lib, "show_options", parent="settings", var_type="bool", default=self.general["show_options"], do_print=False, save=False)
params["show_missing"] = check_for_attribute(lib, "show_missing", parent="settings", var_type="bool", default=self.general["show_missing"], do_print=False, save=False)
params["show_missing_assets"] = check_for_attribute(lib, "show_missing_assets", parent="settings", var_type="bool", default=self.general["show_missing_assets"], do_print=False, save=False)
params["save_report"] = check_for_attribute(lib, "save_report", parent="settings", var_type="bool", default=self.general["save_report"], do_print=False, save=False)
params["missing_only_released"] = check_for_attribute(lib, "missing_only_released", parent="settings", var_type="bool", default=self.general["missing_only_released"], do_print=False, save=False)
params["only_filter_missing"] = check_for_attribute(lib, "only_filter_missing", parent="settings", var_type="bool", default=self.general["only_filter_missing"], do_print=False, save=False)
params["create_asset_folders"] = check_for_attribute(lib, "create_asset_folders", parent="settings", var_type="bool", default=self.general["create_asset_folders"], do_print=False, save=False)
params["dimensional_asset_rename"] = check_for_attribute(lib, "dimensional_asset_rename", parent="settings", var_type="bool", default=self.general["dimensional_asset_rename"], do_print=False, save=False)
params["prioritize_assets"] = check_for_attribute(lib, "prioritize_assets", parent="settings", var_type="bool", default=self.general["prioritize_assets"], do_print=False, save=False)
params["download_url_assets"] = check_for_attribute(lib, "download_url_assets", parent="settings", var_type="bool", default=self.general["download_url_assets"], do_print=False, save=False)
params["show_missing_season_assets"] = check_for_attribute(lib, "show_missing_season_assets", parent="settings", var_type="bool", default=self.general["show_missing_season_assets"], do_print=False, save=False)
params["show_missing_episode_assets"] = check_for_attribute(lib, "show_missing_episode_assets", parent="settings", var_type="bool", default=self.general["show_missing_episode_assets"], do_print=False, save=False)
params["show_asset_not_needed"] = check_for_attribute(lib, "show_asset_not_needed", parent="settings", var_type="bool", default=self.general["show_asset_not_needed"], do_print=False, save=False)
params["minimum_items"] = check_for_attribute(lib, "minimum_items", parent="settings", var_type="int", default=self.general["minimum_items"], do_print=False, save=False)
params["item_refresh_delay"] = check_for_attribute(lib, "item_refresh_delay", parent="settings", var_type="int", default=self.general["item_refresh_delay"], do_print=False, save=False)
params["delete_below_minimum"] = check_for_attribute(lib, "delete_below_minimum", parent="settings", var_type="bool", default=self.general["delete_below_minimum"], do_print=False, save=False)
params["delete_not_scheduled"] = check_for_attribute(lib, "delete_not_scheduled", parent="settings", var_type="bool", default=self.general["delete_not_scheduled"], do_print=False, save=False)
params["ignore_ids"] = check_for_attribute(lib, "ignore_ids", parent="settings", var_type="int_list", default_is_none=True, do_print=False, save=False)
params["ignore_ids"].extend([i for i in self.general["ignore_ids"] if i not in params["ignore_ids"]])
params["ignore_imdb_ids"] = check_for_attribute(lib, "ignore_imdb_ids", parent="settings", var_type="lower_list", default_is_none=True, do_print=False, save=False)
params["ignore_imdb_ids"].extend([i for i in self.general["ignore_imdb_ids"] if i not in params["ignore_imdb_ids"]])
params["overlay_artwork_filetype"] = check_for_attribute(lib, "overlay_artwork_filetype", parent="settings", test_list=filetype_list, translations={"webp": "webp_lossy"}, default=self.general["overlay_artwork_filetype"], do_print=False, save=False)
params["overlay_artwork_quality"] = check_for_attribute(lib, "overlay_artwork_quality", parent="settings", var_type="int", default=self.general["overlay_artwork_quality"], default_is_none=True, int_min=1, int_max=100, do_print=False, save=False)
params["changes_webhooks"] = check_for_attribute(lib, "changes", parent="webhooks", var_type="list", default=self.webhooks["changes"], do_print=False, save=False, default_is_none=True)
params["report_path"] = None
if lib and "report_path" in lib and lib["report_path"]:
if os.path.exists(os.path.dirname(os.path.abspath(lib["report_path"]))):
params["report_path"] = lib["report_path"]
else:
logger.error(f"Config Error: Folder {os.path.dirname(os.path.abspath(lib['report_path']))} does not exist")
if lib and "operations" in lib and lib["operations"]:
final_operations = {}
logger.separator("Operation Configuration", space=False, border=False)
config_ops = util.parse("Config", "operations", lib["operations"], datatype="listdict")
op_size = len(config_ops)
for i, config_op in enumerate(config_ops, 1):
logger.info("")
logger.info(f"Operation {i}/{op_size}")
for k, v in config_op.items():
logger.info(f" {k}: {v}")
if "schedule" in config_op and not self.ignore_schedules:
if not config_op["schedule"]:
logger.error("Config Error: schedule attribute is blank")
else:
try:
util.schedule_check("schedule", config_op["schedule"], current_time, self.run_hour)
except NotScheduled:
logger.info(f"Skipping Operation Not Scheduled for {config_op['schedule']}")
continue
if "delete_collections" not in config_op and ("delete_unmanaged_collections" in config_op or "delete_collections_with_less" in config_op):
config_op["delete_collections"] = {}
if "delete_unmanaged_collections" in config_op:
config_op["delete_collections"]["unmanaged"] = check_for_attribute(config_op, "delete_unmanaged_collections", var_type="bool", default=False, save=False)
if "delete_collections_with_less" in config_op:
config_op["delete_collections"]["less"] = check_for_attribute(config_op, "delete_collections_with_less", var_type="int", default_is_none=True, save=False)
section_final = {}
for op, data_type in library_operations.items():
if op not in config_op:
continue
if op == "mass_imdb_parental_labels":
section_final[op] = check_for_attribute(config_op, op, test_list=data_type, default_is_none=True, save=False)
elif isinstance(data_type, dict):
try:
if not config_op[op]:
raise Failed("is blank")
input_list = config_op[op] if isinstance(config_op[op], list) else [config_op[op]]
final_list = []
for list_attr in input_list:
if not list_attr:
raise Failed(f"has a blank value")
if str(list_attr).lower() in data_type:
final_list.append(str(list_attr).lower())
elif op in ["mass_content_rating_update", "mass_studio_update", "mass_original_title_update"]:
final_list.append(str(list_attr))
elif op == "mass_genre_update":
final_list.append(list_attr if isinstance(list_attr, list) else [list_attr])
elif op in ["mass_originally_available_update", "mass_added_at_update"]:
final_list.append(util.validate_date(list_attr))
elif op.endswith("rating_update"):
final_list.append(util.check_int(list_attr, datatype="float", minimum=0, maximum=10, throw=True))
else:
raise Failed(f"has an invalid value: {list_attr}")
section_final[op] = final_list
except Failed as e:
logger.error(f"Config Error: {op} {e}")
elif op == "mass_collection_mode":
section_final[op] = util.check_collection_mode(config_op[op])
elif data_type == "dict":
input_dict = config_op[op]
if op in ["mass_poster_update", "mass_background_update", "mass_collection_content_rating_update"] and input_dict and not isinstance(input_dict, dict):
input_dict = {"source": input_dict}
if not input_dict or not isinstance(input_dict, dict):
raise Failed(f"Config Error: {op} must be a dictionary")
elif op in ["mass_poster_update", "mass_background_update"]:
section_final[op] = {
"source": check_for_attribute(input_dict, "source", test_list=mass_image_options, default_is_none=True, save=False),
"seasons": check_for_attribute(input_dict, "seasons", var_type="bool", default=True, save=False),
"episodes": check_for_attribute(input_dict, "episodes", var_type="bool", default=True, save=False),
}
elif op == "metadata_backup":
default_path = os.path.join(default_dir, f"{str(library_name)}_Metadata_Backup.yml")
if "path" not in input_dict:
logger.warning(f"Config Warning: path attribute not found using default: {default_path}")
elif "path" in input_dict and not input_dict["path"]:
logger.warning(f"Config Warning: path attribute blank using default: {default_path}")
else:
default_path = input_dict["path"]
section_final[op] = {
"path": default_path,
"exclude": check_for_attribute(input_dict, "exclude", var_type="lower_list", default_is_none=True, save=False),
"sync_tags": check_for_attribute(input_dict, "sync_tags", var_type="bool", default=False, save=False),
"add_blank_entries": check_for_attribute(input_dict, "add_blank_entries", var_type="bool", default=True, save=False)
}
elif "mapper" in op:
section_final[op] = {}
for old_value, new_value in input_dict.items():
if not old_value:
logger.warning("Config Warning: The key cannot be empty")
elif new_value and str(old_value) == str(new_value):
logger.warning(f"Config Warning: {op} value '{new_value}' ignored as it cannot be mapped to itself")
else:
section_final[op][str(old_value)] = str(new_value) if new_value else None # noqa
elif op == "delete_collections":
section_final[op] = {
"managed": check_for_attribute(input_dict, "managed", var_type="bool", default_is_none=True, save=False),
"configured": check_for_attribute(input_dict, "configured", var_type="bool", default_is_none=True, save=False),
"less": check_for_attribute(input_dict, "less", var_type="int", default_is_none=True, save=False, int_min=1),
}
elif op == "mass_collection_content_rating_update":
section_final[op] = {
"source": check_for_attribute(input_dict, "source", test_list=mass_collection_content_options, default_is_none=True, save=False),
"ranking": check_for_attribute(input_dict, "ranking", var_type="list", default=content_rating_default, save=False),
}
else:
section_final[op] = check_for_attribute(config_op, op, var_type=data_type, default=False, save=False)
for k, v in section_final.items():
if k not in final_operations:
final_operations[k] = v
else:
logger.warning(f"Config Warning: Operation {k} already scheduled")
for k, v in final_operations.items():
params[k] = v
for mass_key in operations.meta_operations:
if not params[mass_key]:
continue
sources = params[mass_key]["source"] if isinstance(params[mass_key], dict) else params[mass_key]
if not isinstance(sources, list):
sources = [sources]
try:
for source in sources:
if source and source == "omdb" and self.OMDb is None:
raise Failed(f"{source} without a successful OMDb Connection")
if source and str(source).startswith("mdb") and not self.MDBList.has_key:
raise Failed(f"{source} without a successful MDBList Connection")
if source and str(source).startswith("anidb") and not self.AniDB.is_authorized:
raise Failed(f"{source} without a successful AniDB Connection")
if source and str(source).startswith("mal") and self.MyAnimeList is None:
raise Failed(f"{source} without a successful MyAnimeList Connection")
if source and str(source).startswith("trakt") and self.Trakt is None:
raise Failed(f"{source} without a successful Trakt Connection")
except Failed as e:
logger.error(f"Config Error: {mass_key} cannot use {e}")
params[mass_key] = None
lib_vars = {}
if lib and "template_variables" in lib and lib["template_variables"] and isinstance(lib["template_variables"], dict):
lib_vars = lib["template_variables"]
params["collection_files"] = []
try:
if lib and "collection_files" in lib:
logger.info("")
logger.info("Reading in Collection Files")
if not lib["collection_files"]:
raise Failed("Config Error: collection_files attribute is blank")
files, had_scheduled = util.load_files(lib["collection_files"], "collection_files", schedule=(current_time, self.run_hour, self.ignore_schedules), lib_vars=lib_vars)
if files:
params["collection_files"] = files
elif not had_scheduled:
raise Failed("Config Error: No Paths Found for collection_files")
except Failed as e:
logger.error(e)
params["metadata_files"] = []
try:
if lib and "metadata_files" in lib:
logger.info("")
logger.info("Reading in Metadata Files")
if not lib["metadata_files"]:
raise Failed("Config Error: metadata_files attribute is blank")
files, had_scheduled = util.load_files(lib["metadata_files"], "metadata_files", schedule=(current_time, self.run_hour, self.ignore_schedules), lib_vars=lib_vars)
if files:
params["metadata_files"] = files
elif not had_scheduled:
raise Failed("Config Error: No Paths Found for metadata_files")
except Failed as e:
logger.error(e)
params["default_dir"] = default_dir
params["skip_library"] = False
if lib and "schedule" in lib and not self.requested_libraries and not self.ignore_schedules:
if not lib["schedule"]:
logger.error(f"Config Error: schedule attribute is blank")
else:
logger.debug(f"Value: {lib['schedule']}")
try:
util.schedule_check("schedule", lib["schedule"], current_time, self.run_hour)
except NotScheduled:
params["skip_library"] = True
old_reset = None
old_schedule = None
params["overlay_files"] = []
params["remove_overlays"] = False
params["reapply_overlays"] = False
params["reset_overlays"] = None
if lib and "overlay_files" in lib:
try:
logger.info("")
logger.info("Reading in Overlay Files")
if not lib["overlay_files"]:
raise Failed("Config Error: overlay_files attribute is blank")
files, _ = util.load_files(lib["overlay_files"], "overlay_files", lib_vars=lib_vars)
for file in util.get_list(lib["overlay_files"], split=False):
if isinstance(file, dict):
if ("remove_overlays" in file and file["remove_overlays"] is True) \
or ("remove_overlay" in file and file["remove_overlay"] is True) \
or ("revert_overlays" in file and file["revert_overlays"] is True):
logger.warning("Config Warning: remove_overlays under overlay_files is deprecated it now goes directly under the library attribute.")
params["remove_overlays"] = True
if ("reapply_overlays" in file and file["reapply_overlays"] is True) \
or ("reapply_overlay" in file and file["reapply_overlay"] is True):
logger.warning("Config Warning: reapply_overlays under overlay_files is deprecated it now goes directly under the library attribute.")
params["reapply_overlays"] = True
if "reset_overlays" in file or "reset_overlay" in file:
attr = f"reset_overlay{'s' if 'reset_overlays' in file else ''}"
logger.warning("Config Warning: reset_overlays under overlay_files is deprecated it now goes directly under the library attribute.")
old_reset = file[attr]
if "schedule" in file and file["schedule"]:
logger.warning("Config Warning: schedule under overlay_files is deprecated it now goes directly under the library attribute as schedule_overlays.")
old_schedule = file["schedule"]
params["overlay_files"] = files
except Failed as e:
logger.error(e)
if lib:
if ("remove_overlays" in lib and lib["remove_overlays"] is True) \
or ("remove_overlay" in lib and lib["remove_overlay"] is True) \
or ("revert_overlays" in lib and lib["revert_overlays"] is True):
params["remove_overlays"] = True
if ("reapply_overlays" in lib and lib["reapply_overlays"] is True) \
or ("reapply_overlay" in lib and lib["reapply_overlay"] is True):
params["reapply_overlays"] = True
if "reset_overlays" in lib or "reset_overlay" in lib:
attr = f"reset_overlay{'s' if 'reset_overlays' in lib else ''}"
old_reset = lib[attr]
if old_reset is not None:
reset_options = old_reset if isinstance(old_reset, list) else [old_reset]
final_list = []
for reset_option in reset_options:
if reset_option and reset_option in reset_overlay_options:
final_list.append(reset_option)
else:
final_text = f"Config Error: reset_overlays attribute {reset_option} invalid. Options: "
for option, description in reset_overlay_options.items():
final_text = f"{final_text}\n {option} ({description})"
logger.error(final_text)
if final_list:
params["reset_overlays"] = final_list
else:
final_text = f"Config Error: No proper reset_overlays option found. {old_reset}. Options: "
for option, description in reset_overlay_options.items():
final_text = f"{final_text}\n {option} ({description})"
logger.error(final_text)
if "schedule_overlays" in lib or "schedule_overlay" in lib:
attr = f"schedule_overlay{'s' if 'schedule_overlays' in lib else ''}"
old_schedule = lib[attr]
if old_schedule is not None:
logger.debug(f"Value: {old_schedule}")
err = None
try:
util.schedule_check("schedule_overlays", old_schedule, current_time, self.run_hour)
except NotScheduledRange as e:
err = e
except NotScheduled as e:
if not self.ignore_schedules:
err = e
if err:
logger.info("")
logger.info(f"Overlay Schedule:{err}\n\nOverlays not scheduled to run")
params["overlay_files"] = []
params["remove_overlays"] = False
if lib and "overlay_files" in lib and not params["overlay_files"] and params["remove_overlays"] is False and params["reset_overlays"] is False:
logger.error("Config Error: No Paths Found for overlay_files")
params["image_files"] = []
try:
if lib and "image_files" in lib:
if not lib["image_files"]:
raise Failed("Config Error: image_files attribute is blank")
files, _ = util.load_files(lib["image_files"], "image_files")
if not files:
raise Failed("Config Error: No Paths Found for image_files")
params["image_files"] = files
except Failed as e:
logger.error(e)
try:
logger.info("")
logger.separator("Plex Configuration", space=False, border=False)
params["plex"] = {
"url": check_for_attribute(lib, "url", parent="plex", var_type="url", default=self.general["plex"]["url"], req_default=True, save=False),
"token": check_for_attribute(lib, "token", parent="plex", default=self.general["plex"]["token"], req_default=True, save=False),
"timeout": check_for_attribute(lib, "timeout", parent="plex", var_type="int", default=self.general["plex"]["timeout"], save=False),
"verify_ssl": check_for_attribute(lib, "verify_ssl", parent="plex", var_type="bool", default=self.general["plex"]["verify_ssl"], default_is_none=True, save=False),
"db_cache": check_for_attribute(lib, "db_cache", parent="plex", var_type="int", default=self.general["plex"]["db_cache"], default_is_none=True, save=False)
}
for attr in ["clean_bundles", "empty_trash", "optimize"]:
try:
params["plex"][attr] = check_for_attribute(lib, attr, parent="plex", var_type="bool", save=False, throw=True)
except Failed as er:
test = lib["plex"][attr] if "plex" in lib and attr in lib["plex"] and lib["plex"][attr] else self.general["plex"][attr]
params["plex"][attr] = False
if test is not True and test is not False:
try:
util.schedule_check(attr, test, current_time, self.run_hour)
params["plex"][attr] = True
except NotScheduled:
logger.info(f"Skipping Operation Not Scheduled for {test}")
if params["plex"]["url"].lower() == "env":
params["plex"]["url"] = self.env_plex_url
if params["plex"]["token"].lower() == "env":
params["plex"]["token"] = self.env_plex_token
library = Plex(self, params)
logger.info("")
logger.info(f"{display_name} Library Connection Successful")
logger.info("")
logger.separator("Scanning Files", space=False, border=False)
library.scan_files(self.operations_only, self.overlays_only, self.collection_only, self.metadata_only)
if not library.collection_files and not library.metadata_files and not library.overlay_files and not library.library_operation and not library.images_files and not self.playlist_files:
raise Failed("Config Error: No valid collection file, metadata file, overlay file, image file, playlist file, or library operations found")
except Failed as e:
logger.stacktrace()
logger.error(e)
logger.info("")
logger.info(f"{display_name} Library Connection Failed")
continue
if self.general["radarr"]["url"] or (lib and "radarr" in lib):
logger.info("")
logger.separator("Radarr Configuration", space=False, border=False)
logger.info("")
logger.info(f"Connecting to {display_name} library's Radarr...")
logger.info("")
try:
library.Radarr = Radarr(self, library, {
"url": check_for_attribute(lib, "url", parent="radarr", var_type="url", default=self.general["radarr"]["url"], req_default=True, save=False),
"token": check_for_attribute(lib, "token", parent="radarr", default=self.general["radarr"]["token"], req_default=True, save=False),
"add_missing": check_for_attribute(lib, "add_missing", parent="radarr", var_type="bool", default=self.general["radarr"]["add_missing"], save=False),
"add_existing": check_for_attribute(lib, "add_existing", parent="radarr", var_type="bool", default=self.general["radarr"]["add_existing"], save=False),
"upgrade_existing": check_for_attribute(lib, "upgrade_existing", parent="radarr", var_type="bool", default=self.general["radarr"]["upgrade_existing"], save=False),
"monitor_existing": check_for_attribute(lib, "monitor_existing", parent="radarr", var_type="bool", default=self.general["radarr"]["monitor_existing"], save=False),
"ignore_cache": check_for_attribute(lib, "ignore_cache", parent="radarr", var_type="bool", default=self.general["radarr"]["ignore_cache"], save=False),
"root_folder_path": check_for_attribute(lib, "root_folder_path", parent="radarr", default=self.general["radarr"]["root_folder_path"], req_default=True, save=False),
"monitor": check_for_attribute(lib, "monitor", parent="radarr", var_type="bool", default=self.general["radarr"]["monitor"], save=False),
"availability": check_for_attribute(lib, "availability", parent="radarr", test_list=radarr.availability_descriptions, default=self.general["radarr"]["availability"], save=False),
"quality_profile": check_for_attribute(lib, "quality_profile", parent="radarr", default=self.general["radarr"]["quality_profile"], req_default=True, save=False),
"tag": check_for_attribute(lib, "tag", parent="radarr", var_type="lower_list", default=self.general["radarr"]["tag"], default_is_none=True, save=False),
"search": check_for_attribute(lib, "search", parent="radarr", var_type="bool", default=self.general["radarr"]["search"], save=False),
"radarr_path": check_for_attribute(lib, "radarr_path", parent="radarr", default=self.general["radarr"]["radarr_path"], default_is_none=True, save=False),
"plex_path": check_for_attribute(lib, "plex_path", parent="radarr", default=self.general["radarr"]["plex_path"], default_is_none=True, save=False)
})
except Failed as e:
logger.stacktrace()
logger.error(e)
logger.info("")
logger.info(f"{display_name} library's Radarr Connection {'Failed' if library.Radarr is None else 'Successful'}")
if self.general["sonarr"]["url"] or (lib and "sonarr" in lib):
logger.info("")
logger.separator("Sonarr Configuration", space=False, border=False)
logger.info("")
logger.info(f"Connecting to {display_name} library's Sonarr...")
logger.info("")
try:
library.Sonarr = Sonarr(self, library, {
"url": check_for_attribute(lib, "url", parent="sonarr", var_type="url", default=self.general["sonarr"]["url"], req_default=True, save=False),
"token": check_for_attribute(lib, "token", parent="sonarr", default=self.general["sonarr"]["token"], req_default=True, save=False),
"add_missing": check_for_attribute(lib, "add_missing", parent="sonarr", var_type="bool", default=self.general["sonarr"]["add_missing"], save=False),
"add_existing": check_for_attribute(lib, "add_existing", parent="sonarr", var_type="bool", default=self.general["sonarr"]["add_existing"], save=False),
"upgrade_existing": check_for_attribute(lib, "upgrade_existing", parent="sonarr", var_type="bool", default=self.general["sonarr"]["upgrade_existing"], save=False),
"monitor_existing": check_for_attribute(lib, "monitor_existing", parent="sonarr", var_type="bool", default=self.general["sonarr"]["monitor_existing"], save=False),
"ignore_cache": check_for_attribute(lib, "ignore_cache", parent="sonarr", var_type="bool", default=self.general["sonarr"]["ignore_cache"], save=False),
"root_folder_path": check_for_attribute(lib, "root_folder_path", parent="sonarr", default=self.general["sonarr"]["root_folder_path"], req_default=True, save=False),
"monitor": check_for_attribute(lib, "monitor", parent="sonarr", test_list=sonarr.monitor_descriptions, default=self.general["sonarr"]["monitor"], save=False),
"quality_profile": check_for_attribute(lib, "quality_profile", parent="sonarr", default=self.general["sonarr"]["quality_profile"], req_default=True, save=False),
"language_profile": check_for_attribute(lib, "language_profile", parent="sonarr", default=self.general["sonarr"]["language_profile"], save=False) if self.general["sonarr"]["language_profile"] else check_for_attribute(lib, "language_profile", parent="sonarr", default_is_none=True, save=False),
"series_type": check_for_attribute(lib, "series_type", parent="sonarr", test_list=sonarr.series_type_descriptions, default=self.general["sonarr"]["series_type"], save=False),
"season_folder": check_for_attribute(lib, "season_folder", parent="sonarr", var_type="bool", default=self.general["sonarr"]["season_folder"], save=False),
"tag": check_for_attribute(lib, "tag", parent="sonarr", var_type="lower_list", default=self.general["sonarr"]["tag"], default_is_none=True, save=False),
"search": check_for_attribute(lib, "search", parent="sonarr", var_type="bool", default=self.general["sonarr"]["search"], save=False),
"cutoff_search": check_for_attribute(lib, "cutoff_search", parent="sonarr", var_type="bool", default=self.general["sonarr"]["cutoff_search"], save=False),
"sonarr_path": check_for_attribute(lib, "sonarr_path", parent="sonarr", default=self.general["sonarr"]["sonarr_path"], default_is_none=True, save=False),
"plex_path": check_for_attribute(lib, "plex_path", parent="sonarr", default=self.general["sonarr"]["plex_path"], default_is_none=True, save=False)
})
except Failed as e:
logger.stacktrace()
logger.error(e)
logger.info("")
logger.info(f"{display_name} library's Sonarr Connection {'Failed' if library.Sonarr is None else 'Successful'}")
if self.general["tautulli"]["url"] or (lib and "tautulli" in lib):
logger.info("")
logger.separator("Tautulli Configuration", space=False, border=False)
logger.info("")
logger.info(f"Connecting to {display_name} library's Tautulli...")
logger.info("")
try:
library.Tautulli = Tautulli(self, library, {
"url": check_for_attribute(lib, "url", parent="tautulli", var_type="url", default=self.general["tautulli"]["url"], req_default=True, save=False),
"apikey": check_for_attribute(lib, "apikey", parent="tautulli", default=self.general["tautulli"]["apikey"], req_default=True, save=False)
})
except Failed as e:
logger.stacktrace()
logger.error(e)
logger.info("")
logger.info(f"{display_name} library's Tautulli Connection {'Failed' if library.Tautulli is None else 'Successful'}")
library.Webhooks = Webhooks(self, {}, library=library, notifiarr=self.NotifiarrFactory, gotify=self.GotifyFactory)
library.Overlays = Overlays(self, library)
logger.info("")
self.libraries.append(library)
logger.separator()
self.library_map = {_l.original_mapping_name: _l for _l in self.libraries}
if len(self.libraries) > 0:
logger.info(f"{len(self.libraries)} Plex Library Connection{'s' if len(self.libraries) > 1 else ''} Successful")
else:
raise Failed("Config Error: No libraries were found in config")
logger.separator()
if logger.saved_errors:
self.notify(logger.saved_errors)
except Exception as e:
logger.stacktrace()
self.notify(logger.saved_errors + [e])
logger.save_errors = False
logger.clear_errors()
raise
def notify(self, text, server=None, library=None, collection=None, playlist=None, critical=True):
for error in util.get_list(text, split=False):
try:
self.Webhooks.error_hooks(error, server=server, library=library, collection=collection, playlist=playlist, critical=critical)
except Failed as e:
logger.stacktrace()
logger.error(f"Webhooks Error: {e}")
def notify_delete(self, message, server=None, library=None):
try:
self.Webhooks.delete_hooks(message, server=server, library=library)
except Failed as e:
logger.stacktrace()
logger.error(f"Webhooks Error: {e}")
def get_html(self, url, headers=None, params=None):
return html.fromstring(self.get(url, headers=headers, params=params).content)
def get_json(self, url, json=None, headers=None, params=None):
response = self.get(url, json=json, headers=headers, params=params)
try:
return response.json()
except ValueError:
logger.error(str(response.content))
raise
@retry(stop_max_attempt_number=6, wait_fixed=10000)
def get(self, url, json=None, headers=None, params=None):
return self.session.get(url, json=json, headers=headers, params=params)
def get_image_encoded(self, url):
return base64.b64encode(self.get(url).content).decode('utf-8')
def post_html(self, url, data=None, json=None, headers=None):
return html.fromstring(self.post(url, data=data, json=json, headers=headers).content)
def post_json(self, url, data=None, json=None, headers=None):
response = self.post(url, data=data, json=json, headers=headers)
try:
return response.json()
except ValueError:
logger.error(str(response.content))
raise
@retry(stop_max_attempt_number=6, wait_fixed=10000)
def post(self, url, data=None, json=None, headers=None):
return self.session.post(url, data=data, json=json, headers=headers)
def load_yaml(self, url):
return YAML(input_data=self.get(url).content).data
@property
def mediastingers(self):
if self._mediastingers is None:
self._mediastingers = self.load_yaml(mediastingers_url)
return self._mediastingers
Reference in a new issue View git blame Copy permalink