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

[92] update YAML read/writes

Browse Source
pull/877/head
meisnate12 3 years ago
parent da6d31c48b
commit 2c9ce0dfb4
10 changed files with 183 additions and 182 deletions
Show Stats Download Patch File Download Diff File

2
VERSION
Unescape View File

@ -1 +1 @@
1.16.5-develop91
1.16.5-develop92

1
docs/metadata/details/setting.md
Unescape View File

@ -30,4 +30,5 @@ All the following attributes serve various functions as how the collection/playl
| `changes_webhooks` | **Description:** Used to specify a collection/playlist changes webhook for just this collection/playlist.<br>**Values:** List of webhooks |
| `sync_to_trakt_list` | **Description:** Used to specify a trakt list you want the collection/playlist synced to.<br>**Values:** Trakt List Slug you want to sync to |
| `sync_missing_to_trakt_list` | **Description:** Used to also sync missing items to the Trakt List specified by `sync_to_trakt_list`.<br>**Values:** `ture` or `false` |
| `allowed_library_types` | **Description:** Used to specify the types of libraries that this definition can work with.<br>Multiple can be used for one definition as a list or comma separated string.<br>**Values:** `movie`, `show`, or `artist` |

220
modules/config.py
Unescape View File

@ -26,10 +26,9 @@ 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
from modules.util import Failed, NotScheduled, NotScheduledRange, YAML
from modules.webhooks import Webhooks
from retrying import retry
from ruamel import yaml
logger = util.logger
@ -85,120 +84,113 @@ class ConfigFile:
self.overlays_only = attrs["overlays_only"] if "overlays_only" in attrs else False
current_time = datetime.now()
yaml.YAML().allow_duplicate_keys = True
try:
new_config, _, _ = yaml.util.load_yaml_guess_indent(open(self.config_path, encoding="utf-8"))
def replace_attr(all_data, attr, par):
if "settings" not in all_data:
all_data["settings"] = {}
if par in all_data and all_data[par] and attr in all_data[par] and attr not in all_data["settings"]:
all_data["settings"][attr] = all_data[par][attr]
del all_data[par][attr]
if "libraries" not in new_config:
new_config["libraries"] = {}
if "settings" not in new_config:
new_config["settings"] = {}
if "tmdb" not in new_config:
new_config["tmdb"] = {}
replace_attr(new_config, "cache", "cache")
replace_attr(new_config, "cache_expiration", "cache")
if "config" in new_config:
del new_config["cache"]
replace_attr(new_config, "asset_directory", "plex")
replace_attr(new_config, "sync_mode", "plex")
replace_attr(new_config, "show_unmanaged", "plex")
replace_attr(new_config, "show_filtered", "plex")
replace_attr(new_config, "show_missing", "plex")
replace_attr(new_config, "save_missing", "plex")
if new_config["libraries"]:
for library in new_config["libraries"]:
if not new_config["libraries"][library]:
continue
if "radarr_add_all" in new_config["libraries"][library]:
new_config["libraries"][library]["radarr_add_all_existing"] = new_config["libraries"][library].pop("radarr_add_all")
if "sonarr_add_all" in new_config["libraries"][library]:
new_config["libraries"][library]["sonarr_add_all_existing"] = new_config["libraries"][library].pop("sonarr_add_all")
if "plex" in new_config["libraries"][library] and new_config["libraries"][library]["plex"]:
replace_attr(new_config["libraries"][library], "asset_directory", "plex")
replace_attr(new_config["libraries"][library], "sync_mode", "plex")
replace_attr(new_config["libraries"][library], "show_unmanaged", "plex")
replace_attr(new_config["libraries"][library], "show_filtered", "plex")
replace_attr(new_config["libraries"][library], "show_missing", "plex")
replace_attr(new_config["libraries"][library], "save_missing", "plex")
if "settings" in new_config["libraries"][library] and new_config["libraries"][library]["settings"]:
if "collection_minimum" in new_config["libraries"][library]["settings"]:
new_config["libraries"][library]["settings"]["minimum_items"] = new_config["libraries"][library]["settings"].pop("collection_minimum")
if "radarr" in new_config["libraries"][library] and new_config["libraries"][library]["radarr"]:
if "add" in new_config["libraries"][library]["radarr"]:
new_config["libraries"][library]["radarr"]["add_missing"] = new_config["libraries"][library]["radarr"].pop("add")
if "sonarr" in new_config["libraries"][library] and new_config["libraries"][library]["sonarr"]:
if "add" in new_config["libraries"][library]["sonarr"]:
new_config["libraries"][library]["sonarr"]["add_missing"] = new_config["libraries"][library]["sonarr"].pop("add")
if "operations" in new_config["libraries"][library] and new_config["libraries"][library]["operations"]:
if "radarr_add_all" in new_config["libraries"][library]["operations"]:
new_config["libraries"][library]["operations"]["radarr_add_all_existing"] = new_config["libraries"][library]["operations"].pop("radarr_add_all")
if "sonarr_add_all" in new_config["libraries"][library]["operations"]:
new_config["libraries"][library]["operations"]["sonarr_add_all_existing"] = new_config["libraries"][library]["operations"].pop("sonarr_add_all")
if "webhooks" in new_config["libraries"][library] and new_config["libraries"][library]["webhooks"] and "collection_changes" not in new_config["libraries"][library]["webhooks"]:
changes = []
def hooks(attr):
if attr in new_config["libraries"][library]["webhooks"]:
changes.extend([w for w in util.get_list(new_config["libraries"][library]["webhooks"].pop(attr), split=False) if w not in changes])
hooks("collection_creation")
hooks("collection_addition")
hooks("collection_removal")
hooks("collection_changes")
new_config["libraries"][library]["webhooks"]["changes"] = None if not changes else changes if len(changes) > 1 else changes[0]
if "libraries" in new_config: new_config["libraries"] = new_config.pop("libraries")
if "playlist_files" in new_config: new_config["playlist_files"] = new_config.pop("playlist_files")
if "settings" in new_config:
temp = new_config.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")
new_config["settings"] = temp
if "webhooks" in new_config:
temp = new_config.pop("webhooks")
if "changes" not in temp:
loaded_yaml = YAML(self.config_path)
self.data = loaded_yaml.data
def replace_attr(all_data, attr, par):
if "settings" not in all_data:
all_data["settings"] = {}
if par in all_data and all_data[par] and attr in all_data[par] and attr not in all_data["settings"]:
all_data["settings"][attr] = all_data[par][attr]
del all_data[par][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 "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 "radarr" in self.data["libraries"][library] and self.data["libraries"][library]["radarr"]:
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 "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(attr):
if attr in temp:
items = util.get_list(temp.pop(attr), split=False)
if items:
changes.extend([w for w in items if w not in changes])
if attr in self.data["libraries"][library]["webhooks"]:
changes.extend([w for w in util.get_list(self.data["libraries"][library]["webhooks"].pop(attr), split=False) 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]
new_config["webhooks"] = temp
if "plex" in new_config: new_config["plex"] = new_config.pop("plex")
if "tmdb" in new_config: new_config["tmdb"] = new_config.pop("tmdb")
if "tautulli" in new_config: new_config["tautulli"] = new_config.pop("tautulli")
if "omdb" in new_config: new_config["omdb"] = new_config.pop("omdb")
if "mdblist" in new_config: new_config["mdblist"] = new_config.pop("mdblist")
if "notifiarr" in new_config: new_config["notifiarr"] = new_config.pop("notifiarr")
if "anidb" in new_config: new_config["anidb"] = new_config.pop("anidb")
if "radarr" in new_config:
temp = new_config.pop("radarr")
if temp and "add" in temp:
temp["add_missing"] = temp.pop("add")
new_config["radarr"] = temp
if "sonarr" in new_config:
temp = new_config.pop("sonarr")
if temp and "add" in temp:
temp["add_missing"] = temp.pop("add")
new_config["sonarr"] = temp
if "trakt" in new_config: new_config["trakt"] = new_config.pop("trakt")
if "mal" in new_config: new_config["mal"] = new_config.pop("mal")
self.data = new_config
except yaml.scanner.ScannerError as e:
logger.stacktrace()
raise Failed(f"YAML Error: {util.tab_new_lines(e)}")
except Exception as e:
logger.stacktrace()
raise Failed(f"YAML Error: {e}")
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")
self.data["settings"] = temp
if "webhooks" in self.data:
temp = self.data.pop("webhooks")
if "changes" not in temp:
changes = []
def hooks(attr):
if attr in temp:
items = util.get_list(temp.pop(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 "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 "anidb" in self.data: self.data["anidb"] = self.data.pop("anidb")
if "radarr" in self.data:
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_for_attribute(data, attribute, parent=None, test_list=None, default=None, do_print=True, default_is_none=False, req_default=False, var_type="str", throw=False, save=True):
endline = ""
@ -215,12 +207,12 @@ class ConfigFile:
if data is None or attribute not in data:
message = f"{text} not found"
if parent and save is True:
loaded_config, _, _ = yaml.util.load_yaml_guess_indent(open(self.config_path, encoding="utf-8"))
yaml = YAML(self.config_path)
endline = f"\n{parent} sub-attribute {attribute} added to config"
if parent not in loaded_config or not loaded_config[parent]: loaded_config[parent] = {attribute: default}
elif attribute not in loaded_config[parent]: loaded_config[parent][attribute] = default
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.round_trip_dump(loaded_config, open(self.config_path, "w"), block_seq_indent=2)
yaml.save()
if default_is_none and var_type in ["list", "int_list", "comma_list"]: return default if default else []
elif data[attribute] is None:
if default_is_none and var_type in ["list", "int_list", "comma_list"]: return default if default else []

10
modules/library.py
Unescape View File

@ -3,8 +3,7 @@ from abc import ABC, abstractmethod
from modules import util
from modules.meta import MetadataFile, OverlayFile
from modules.operations import Operations
from modules.util import Failed
from ruamel import yaml
from modules.util import Failed, YAML
logger = util.logger
@ -258,10 +257,9 @@ class Library(ABC):
self.report_data[collection][other].append(title)
with open(self.report_path, "w"): pass
try:
yaml.round_trip_dump(self.report_data, open(self.report_path, "w", encoding="utf-8"))
except yaml.scanner.ScannerError as e:
logger.error(f"YAML Error: {util.tab_new_lines(e)}")
yaml = YAML(self.report_path)
yaml.data = self.report_data
yaml.save()
def cache_items(self):
logger.info("")

12
modules/mal.py
Unescape View File

@ -1,7 +1,6 @@
import math, re, secrets, time, webbrowser
import re, secrets, time, webbrowser
from modules import util
from modules.util import Failed, TimeoutExpired
from ruamel import yaml
from modules.util import Failed, TimeoutExpired, YAML
logger = util.logger
@ -138,16 +137,15 @@ class MyAnimeList:
def _save(self, authorization):
if authorization is not None and "access_token" in authorization and authorization["access_token"] and self._check(authorization):
if self.authorization != authorization and not self.config.read_only:
yaml.YAML().allow_duplicate_keys = True
config, ind, bsi = yaml.util.load_yaml_guess_indent(open(self.config_path, encoding="utf-8"))
config["mal"]["authorization"] = {
yaml = YAML(self.config_path)
yaml.data["mal"]["authorization"] = {
"access_token": authorization["access_token"],
"token_type": authorization["token_type"],
"expires_in": authorization["expires_in"],
"refresh_token": authorization["refresh_token"]
}
logger.info(f"Saving authorization information to {self.config_path}")
yaml.round_trip_dump(config, open(self.config_path, "w"), indent=ind, block_seq_indent=bsi)
yaml.save()
logger.secret(authorization["access_token"])
self.authorization = authorization
return True

38
modules/meta.py
Unescape View File

@ -1,9 +1,8 @@
import math, operator, os, re, requests
from datetime import datetime
from modules import plex, ergast, util
from modules.util import Failed
from modules.util import Failed, YAML
from plexapi.exceptions import NotFound, BadRequest
from ruamel import yaml
logger = util.logger
@ -81,28 +80,19 @@ class DataFile:
return data
def load_file(self, file_type, file_path):
try:
if file_type in ["URL", "Git", "Repo"]:
if file_type == "Repo" and not self.config.custom_repo:
raise Failed("Config Error: No custom_repo defined")
content_path = file_path if file_type == "URL" else f"{self.config.custom_repo if file_type == 'Repo' else util.github_base}{file_path}.yml"
response = self.config.get(content_path)
if response.status_code >= 400:
raise Failed(f"URL Error: No file found at {content_path}")
content = response.content
elif os.path.exists(os.path.abspath(file_path)):
content = open(file_path, encoding="utf-8")
else:
raise Failed(f"File Error: File does not exist {os.path.abspath(file_path)}")
data, _, _ = yaml.util.load_yaml_guess_indent(content)
except yaml.scanner.ScannerError as ye:
raise Failed(f"YAML Error: {util.tab_new_lines(ye)}")
except Exception as e:
logger.stacktrace()
raise Failed(f"YAML Error: {e}")
if not data or not isinstance(data, dict):
raise Failed("YAML Error: File is empty")
return data
if file_type in ["URL", "Git", "Repo"]:
if file_type == "Repo" and not self.config.custom_repo:
raise Failed("Config Error: No custom_repo defined")
content_path = file_path if file_type == "URL" else f"{self.config.custom_repo if file_type == 'Repo' else util.github_base}{file_path}.yml"
response = self.config.get(content_path)
if response.status_code >= 400:
raise Failed(f"URL Error: No file found at {content_path}")
yaml = YAML(input_data=response.content, check_empty=True)
elif os.path.exists(os.path.abspath(file_path)):
yaml = YAML(path=os.path.abspath(file_path), check_empty=True)
else:
raise Failed(f"File Error: File does not exist {os.path.abspath(file_path)}")
return yaml.data
def apply_template(self, name, data, template_call):
if not self.templates:

32
modules/operations.py
Unescape View File

@ -1,10 +1,9 @@
import os, re
from datetime import datetime
from modules import plex, util
from modules.util import Failed
from modules.util import Failed, YAML
from plexapi.audio import Artist
from plexapi.video import Show
from ruamel import yaml
logger = util.logger
@ -455,24 +454,24 @@ class Operations:
logger.info("")
logger.info(f"Metadata Backup Path: {self.library.metadata_backup['path']}")
logger.info("")
meta = None
yaml = None
if os.path.exists(self.library.metadata_backup["path"]):
try:
meta, _, _ = yaml.util.load_yaml_guess_indent(open(self.library.metadata_backup["path"], encoding="utf-8"))
except yaml.scanner.ScannerError as e:
logger.error(f"YAML Error: {util.tab_new_lines(e)}")
yaml = YAML(path=self.library.metadata_backup["path"])
except Failed as e:
logger.error(e)
filename, file_extension = os.path.splitext(self.library.metadata_backup["path"])
i = 1
while os.path.exists(f"{filename}{i}{file_extension}"):
i += 1
os.rename(self.library.metadata_backup["path"], f"{filename}{i}{file_extension}")
logger.error(f"Backup failed to load saving copy to {filename}{i}{file_extension}")
if not meta:
meta = {}
if "metadata" not in meta:
meta["metadata"] = {}
if not yaml:
yaml = YAML(path=self.library.metadata_backup["path"], create=True)
if "metadata" not in yaml.data:
yaml.data["metadata"] = {}
special_names = {}
for mk, mv in meta["metadata"].items():
for mk, mv in yaml.data["metadata"].items():
if "title" in mv:
special_names[mv["title"]] = mk
if "year" in mv:
@ -484,7 +483,7 @@ class Operations:
map_key, attrs = self.library.get_locked_attributes(item, titles)
if map_key in special_names:
map_key = special_names[map_key]
og_dict = meta["metadata"][map_key] if map_key in meta["metadata"] and meta["metadata"][map_key] else {}
og_dict = yaml.data["metadata"][map_key] if map_key in yaml.data["metadata"] and yaml.data["metadata"][map_key] else {}
if attrs or (self.library.metadata_backup["add_blank_entries"] and not og_dict):
def get_dict(attrs_dict):
return {ak: get_dict(av) if isinstance(av, dict) else av for ak, av in attrs_dict.items()}
@ -494,13 +493,10 @@ class Operations:
for lk, lv in looping.items():
dest_dict[lk] = loop_dict(lv, dest_dict[lk] if lk in dest_dict and dest_dict[lk] else {}) if isinstance(lv, dict) else lv
return dest_dict
meta["metadata"][map_key] = loop_dict(get_dict(attrs), og_dict)
yaml.data["metadata"][map_key] = loop_dict(get_dict(attrs), og_dict)
logger.exorcise()
try:
yaml.round_trip_dump(meta, open(self.library.metadata_backup["path"], "w", encoding="utf-8"), block_seq_indent=2)
logger.info(f"{len(meta['metadata'])} {self.library.type.capitalize()}{'s' if len(meta['metadata']) > 1 else ''} Backed Up")
except yaml.scanner.ScannerError as e:
logger.error(f"YAML Error: {util.tab_new_lines(e)}")
yaml.save()
logger.info(f"{len(yaml.data['metadata'])} {self.library.type.capitalize()}{'s' if len(yaml.data['metadata']) > 1 else ''} Backed Up")
operation_run_time = str(datetime.now() - operation_start).split('.')[0]
logger.info("")

12
modules/trakt.py
Unescape View File

@ -1,8 +1,7 @@
import requests, time, webbrowser
from modules import util
from modules.util import Failed, TimeoutExpired
from modules.util import Failed, TimeoutExpired, YAML
from retrying import retry
from ruamel import yaml
logger = util.logger
@ -171,10 +170,9 @@ class Trakt:
def _save(self, authorization):
if authorization and self._check(authorization):
if self.authorization != authorization and not self.config.read_only:
yaml.YAML().allow_duplicate_keys = True
config, ind, bsi = yaml.util.load_yaml_guess_indent(open(self.config_path, encoding="utf-8"))
config["trakt"]["pin"] = None
config["trakt"]["authorization"] = {
yaml = YAML(self.config_path)
yaml.data["trakt"]["pin"] = None
yaml.data["trakt"]["authorization"] = {
"access_token": authorization["access_token"],
"token_type": authorization["token_type"],
"expires_in": authorization["expires_in"],
@ -183,7 +181,7 @@ class Trakt:
"created_at": authorization["created_at"]
}
logger.info(f"Saving authorization information to {self.config_path}")
yaml.round_trip_dump(config, open(self.config_path, "w"), indent=ind, block_seq_indent=bsi)
yaml.save()
self.authorization = authorization
return True
return False

37
modules/util.py
Unescape View File

@ -1,4 +1,4 @@
import glob, logging, os, re, requests, signal, sys, time
import glob, logging, os, re, requests, ruamel.yaml, signal, sys, time
from datetime import datetime, timedelta
from pathvalidate import is_valid_filename, sanitize_filename
from plexapi.audio import Album, Track
@ -89,9 +89,6 @@ github_base = "https://raw.githubusercontent.com/meisnate12/Plex-Meta-Manager-Co
previous_time = None
start_time = None
def tab_new_lines(data):
return str(data).replace("\n", "\n ") if "\n" in str(data) else str(data)
def make_ordinal(n):
return f"{n}{'th' if 11 <= (n % 100) <= 13 else ['th', 'st', 'nd', 'rd', 'th'][min(n % 10, 4)]}"
@ -778,3 +775,35 @@ def check_time(message, end=False):
else:
logger.debug(f"{message}: {current_time - previous_time}")
previous_time = None if end else current_time
class YAML:
def __init__(self, path=None, input_data=None, check_empty=False, create=False):
self.path = path
self.input_data = input_data
self.yaml = ruamel.yaml.YAML()
self.yaml.indent(mapping=2, sequence=2)
try:
if input_data:
self.data = self.yaml.load(input_data)
else:
if create and not os.path.exists(self.path):
with open(self.path, 'w'):
pass
self.data = {}
else:
with open(self.path, encoding="utf-8") as fp:
self.data = self.yaml.load(fp)
except ruamel.yaml.error.YAMLError as e:
e = str(e).replace("\n", "\n ")
raise Failed(f"YAML Error: {e}")
except Exception as e:
raise Failed(f"YAML Error: {e}")
if not self.data or not isinstance(self.data, dict):
if check_empty:
raise Failed("YAML Error: File is empty")
self.data = {}
def save(self):
if self.path:
with open(self.path, 'w') as fp:
self.yaml.dump(self.data, fp)

1
plex_meta_manager.py
Unescape View File

@ -6,7 +6,6 @@ try:
from modules.logs import MyLogger
from plexapi.exceptions import NotFound
from plexapi.video import Show, Season
from ruamel import yaml
except ModuleNotFoundError:
print("Requirements Error: Requirements are not installed")
sys.exit(0)

Write Preview
Loading…
Cancel
Save