From 309f4abc6a2f443421867e9246ff371ec814ea21 Mon Sep 17 00:00:00 2001 From: Brian Lindner <8153326+BrianLindner@users.noreply.github.com> Date: Wed, 6 Jan 2021 22:33:14 -0500 Subject: [PATCH] time precedence, test run ability, added typing --- .gitignore | 2 + README.md | 12 +- plexutil.py | 29 ++-- schedule_preroll.py | 337 +++++++++++++++++++++++++++----------------- 4 files changed, 231 insertions(+), 149 deletions(-) diff --git a/.gitignore b/.gitignore index d05a0fe..f11e527 100644 --- a/.gitignore +++ b/.gitignore @@ -59,6 +59,7 @@ coverage.xml *.py,cover .hypothesis/ .pytest_cache/ +test/ # Translations *.mo @@ -137,6 +138,7 @@ venv.bak/ .mypy_cache/ .dmypy.json dmypy.json +mypy.ini # Pyre type checker .pyre/ diff --git a/README.md b/README.md index 4e8a01a..3fb117d 100644 --- a/README.md +++ b/README.md @@ -31,7 +31,6 @@ Schedule priority for a given Date: 2. **date_range** \ Include listing for the specified Start/End date range that include the given Date \ -Multipe ranges may apply at same time (append) \ Range can be specified as a Date or DateTime \ **overrides usage of *week/month/default* listings @@ -41,7 +40,7 @@ Include listing for the specified WEEK of the year for the given Date \ 4. **monthly** \ Include listing for the specified MONTH of the year for the given Date \ -**overrides usage of *week/month/default* listings +**overrides usage of *default* listings 5. **default** \ Default listing used of none of above apply to the given Date @@ -129,7 +128,8 @@ default: path: /path/to/file.mp4;/path/to/file.mp4 ``` -#### Date Range Section: +#### Date Range Section + Use it for *Day* or *Ranges of Time* needs \ Now with Time support! (optional) @@ -158,7 +158,7 @@ python schedule_preroll.py - -h : help information - -c : config.ini (local or PlexAPI system central) for Connection Info (see [config.ini.sample](config.ini.sample)) - -s : preroll_schedules.yaml for various scheduling information (see [spreroll_schedules.yaml.sample](preroll_schedules.yaml.sample)) -- -l : location of custom logger.conf config file \ +- -lc : location of custom logger.conf config file \ See: - Sample [logger config](logging.conf) - Logger usage [Examples](https://github.com/amilstead/python-logging-examples/blob/master/configuration/fileConfig/config.ini) @@ -174,7 +174,7 @@ Automate scheduling of pre-roll intros for Plex optional arguments: -h, --help show this help message and exit -v, --version show the version number and exit - -l LOG_CONFIG_FILE, --logconfig-path LOG_CONFIG_FILE + -lc LOG_CONFIG_FILE, --logconfig-path LOG_CONFIG_FILE Path to logging config file. [Default: ./logging.conf] -c CONFIG_FILE, --config-path CONFIG_FILE Path to Config.ini to use for Plex Server info. [Default: ./config.ini] @@ -188,7 +188,7 @@ optional arguments: python schedule_preroll.py \ -c path/to/custom/config.ini \ -s path/to/custom/preroll_schedules.yaml \ - -l path/to/custom/logger.conf + -lc path/to/custom/logger.conf ``` --- diff --git a/plexutil.py b/plexutil.py index e6a548e..054b3e9 100644 --- a/plexutil.py +++ b/plexutil.py @@ -4,13 +4,12 @@ Raises: FileNotFoundError: [description] KeyError: [description] - KeyError: [description] - ConfigError: [description] """ import os import sys import logging import logging.config +from typing import Dict, List from configparser import ConfigParser from plexapi.server import PlexServer, CONFIG @@ -19,15 +18,15 @@ logger = logging.getLogger(__name__) filename = os.path.basename(sys.argv[0]) SCRIPT_NAME = os.path.splitext(filename)[0] -def getPlexConfig(config_file=None): +def getPlexConfig(config_file: str='') -> Dict[str,str]: """Return Plex Config paramaters for connection info {PLEX_URL, PLEX_TOKEN}\n - Attempts to use either:\n + Attempts to use one of either:\n * supplier path/to/config file (INI Format) * local config.ini (primary) * PlexAPI system config.ini (secondary) Args: - config_file (string): path/to/config.ini style config file (INI Format) + config_file (str): path/to/config.ini style config file (INI Format) Raises: KeyError: Config Params not found in config file(s) @@ -37,22 +36,20 @@ def getPlexConfig(config_file=None): dict: Dict of config params {PLEX_URL, PLEX_TOKEN} """ - cfg = {} + cfg = {} # type: Dict[str, str] plex_url = '' plex_token = '' + filename = '' use_local_config = False use_plexapi_config = False # Look for a local Config.ini file, use settings if present local_config = ConfigParser() - if config_file != None: - if os.path.exists(config_file): - filename = config_file - else: - raise FileNotFoundError('Config file "{}" not found'.format(config_file)) - else: + if config_file == None or config_file == '': filename = 'config.ini' + else: + filename = str(config_file) #try reading a local file local_config.read(filename) @@ -98,21 +95,21 @@ def getPlexConfig(config_file=None): raise KeyError(msg) if not use_local_config and not use_plexapi_config: - msg = 'No Plex config information found {server_baseurl, server_token}' + msg = 'ConfigFile Error: No Plex config information found {server_baseurl, server_token}' logger.error(msg) - raise ConfigError(msg) + raise FileNotFoundError(msg) cfg['PLEX_URL'] = plex_url cfg['PLEX_TOKEN']= plex_token return cfg -def setupLogger(log_config): +def setupLogger(log_config: str) -> None: """load and configure a program logger using a supplier logging configuration file \n if possible the program will attempt to create log folders if not already existing Args: - log_config (string): path/to/logging.(conf|ini) style config file (INI Format) + log_config (str): path/to/logging.(conf|ini) style config file (INI Format) Raises: KeyError: Problems processing logging config files diff --git a/schedule_preroll.py b/schedule_preroll.py index acd90a1..8a8fc44 100644 --- a/schedule_preroll.py +++ b/schedule_preroll.py @@ -8,7 +8,7 @@ Set it and forget it! Optional Arguments: -h, --help show this help message and exit -v, --version show the version number and exit - -l LOG_CONFIG_FILE, --logconfig-path LOG_CONFIG_FILE + -lc LOG_CONFIG_FILE, --logconfig-path LOG_CONFIG_FILE Path to logging config file. [Default: ./logging.conf] -c CONFIG_FILE, --config-path CONFIG_FILE Path to Config.ini to use for Plex Server info. [Default: ./config.ini] @@ -34,10 +34,10 @@ import os import sys import logging import requests -import datetime -from datetime import datetime, date, time, timedelta +from datetime import datetime, date, timedelta import yaml -from argparse import ArgumentParser +from typing import NamedTuple, Union, Optional, Tuple, List, Dict +from argparse import Namespace, ArgumentParser from configparser import ConfigParser from configparser import Error as ConfigError from plexapi.server import PlexServer, CONFIG @@ -50,7 +50,17 @@ logger = logging.getLogger(__name__) filename = os.path.basename(sys.argv[0]) SCRIPT_NAME = os.path.splitext(filename)[0] -def getArguments(): +#ScheduleEntry = Dict[str, Union[str, bool, date, datetime]] +class ScheduleEntry(NamedTuple): + type: str + startdate: Union[date,datetime] + enddate: Union[date,datetime] + force: bool + path: str + +ScheduleType = Dict[str, ScheduleEntry] + +def getArguments() -> Namespace: """Return command line arguments See https://docs.python.org/3/howto/argparse.html @@ -58,18 +68,22 @@ def getArguments(): argparse.Namespace: Namespace object """ description = 'Automate scheduling of pre-roll intros for Plex' - version = '0.8.0' + version = '0.9.0' - config_default = './config.ini' + config_default = None #'./config.ini' log_config_default = './logging.conf' schedule_default = './preroll_schedules.yaml' parser = ArgumentParser(description='{}'.format(description)) parser.add_argument('-v', '--version', action='version', version='%(prog)s {}'.format(version), help='show the version number and exit') - parser.add_argument('-l', '--logconfig-path', + parser.add_argument('-lc', '--logconfig-path', dest='log_config_file', action='store', default=log_config_default, help='Path to logging config file. [Default: {}]'.format(log_config_default)) + parser.add_argument('-t', '--test-run', + dest='do_test_run', action='store_true', + default=False, + help='Perform a test run, display output but dont save') parser.add_argument('-c', '--config-path', dest='config_file', action='store', help='Path to Config.ini to use for Plex Server info. [Default: {}]'.format(config_default)) @@ -80,17 +94,21 @@ def getArguments(): return args -def getYAMLSchema(): +def getYAMLSchema() -> Dict[str, List[ScheduleEntry]]: """Return the main schema layout of the preroll_schedules.yaml file Returns: - dict: Dict of main schema items + Dict (List[ScheduleType]): Dict of main schema items """ - schema = {'default': None, 'monthly': None, - 'weekly': None, 'date_range': None, 'misc': None} + schema = {'default': [], + 'monthly': [], + 'weekly': [], + 'date_range': [], + 'misc': [] + } # type: Dict[str, List[ScheduleEntry]] return schema -def getWeekRange(year, weeknum): +def getWeekRange(year:int, weeknum:int) -> Tuple[date, date]: """Return the starting/ending date range of a given year/week Args: @@ -107,7 +125,7 @@ def getWeekRange(year, weeknum): return start, end -def getMonthRange(year, monthnum): +def getMonthRange(year:int, monthnum:int) -> Tuple[date, date]: """Return the starting/ending date range of a given year/month Args: @@ -124,11 +142,31 @@ def getMonthRange(year, monthnum): return start, end -def getPrerollSchedule(schedule_file=None): +def duration_seconds(start:Union[date,datetime], end:Union[date,datetime]) -> float: + """Return length of time between two date/datetime in seconds + + Args: + start (date/datetime): [description] + end (date/datetime): [description] + + Returns: + float: Length in time seconds + """ + if not isinstance(start, datetime): + start = datetime.combine(start, datetime.min.time()) + if not isinstance(end, datetime): + end = datetime.combine(end, datetime.max.time()) + + delta = end - start + + logger.debug('duration_second[] Start: {} End: {} Duration: {}'.format(start, end, delta.total_seconds())) + return delta.total_seconds() + +def getPrerollSchedule(schedule_file:Optional[str]=None) -> List[ScheduleEntry]: """Return a listing of defined preroll schedules for searching/use Args: - schedule_file (string): path/to/schedule_preroll.yaml style config file (YAML Format) + schedule_file (str): path/to/schedule_preroll.yaml style config file (YAML Format) Raises: FileNotFoundError: If no schedule config file exists @@ -139,8 +177,8 @@ def getPrerollSchedule(schedule_file=None): default_files = ['preroll_schedules.yaml', 'preroll_schedules.yml'] filename = None - if schedule_file: - if os.path.exists(schedule_file): + if schedule_file != '' and schedule_file != None: + if os.path.exists(str(schedule_file)): filename = schedule_file else: msg = 'Pre-roll Schedule file "{}" not found'.format(schedule_file) @@ -158,145 +196,160 @@ def getPrerollSchedule(schedule_file=None): raise FileNotFoundError(msg) with open(filename, 'r') as file: - #contents = yaml.load(file, Loader=yaml.SafeLoader) - contents = yaml.load(file, Loader=yaml.FullLoader) + contents = yaml.load(file, Loader=yaml.SafeLoader) today = date.today() - schedule = [] - for schedule_type in getYAMLSchema(): - if schedule_type == 'weekly': + schedule = [] # type: List[ScheduleEntry] + for schedule_section in getYAMLSchema(): + if schedule_section == 'weekly': try: - use = contents[schedule_type]['enabled'] + use = contents[schedule_section]['enabled'] if use: for i in range(1,53): try: - path = contents[schedule_type][i] + path = contents[schedule_section][i] if path: - entry = {} start, end = getWeekRange(today.year, i) - entry['Type'] = schedule_type - entry['StartDate'] = start - entry['EndDate'] = end - entry['Path'] = path + + entry = ScheduleEntry(type=schedule_section, + force=False, + startdate=start, + enddate=end, + path=path) schedule.append(entry) - except KeyError as e: + except KeyError as ke: # skip KeyError for missing Weeks - msg = 'Key Value not found: "{}"->"{}", skipping week'.format(schedule_type, i) + msg = 'Key Value not found: "{}"->"{}", skipping week'.format(schedule_section, i) logger.debug(msg) - continue - except KeyError as e: - msg = 'Key Value not found in "{}" section'.format(schedule_type) - logger.error(msg, exc_info=e) - raise e - elif schedule_type == 'monthly': + pass + except KeyError as ke: + msg = 'Key Value not found in "{}" section'.format(schedule_section) + logger.error(msg, exc_info=ke) + raise + elif schedule_section == 'monthly': try: - use = contents[schedule_type]['enabled'] + use = contents[schedule_section]['enabled'] if use: for i in range(1,13): month_abrev = date(today.year, i, 1).strftime('%b').lower() try: - path = contents[schedule_type][month_abrev] + path = contents[schedule_section][month_abrev] if path: - entry = {} start, end = getMonthRange(today.year, i) - entry['Type'] = schedule_type - entry['StartDate'] = start - entry['EndDate'] = end - entry['Path'] = path + + entry = ScheduleEntry(type=schedule_section, + force=False, + startdate=start, + enddate=end, + path=path) schedule.append(entry) - except KeyError as e: + except KeyError as ke: # skip KeyError for missing Months - msg = 'Key Value not found: "{}"->"{}", skipping month'.format(schedule_type, month_abrev) + msg = 'Key Value not found: "{}"->"{}", skipping month'.format(schedule_section, month_abrev) logger.warning(msg) - continue - except KeyError as e: - msg = 'Key Value not found in "{}" section'.format(schedule_type) - logger.error(msg, exc_info=e) - raise e - elif schedule_type == 'date_range': + pass + except KeyError as ke: + msg = 'Key Value not found in "{}" section'.format(schedule_section) + logger.error(msg, exc_info=ke) + raise + elif schedule_section == 'date_range': try: - use = contents[schedule_type]['enabled'] + use = contents[schedule_section]['enabled'] if use: - for r in contents[schedule_type]['ranges']: + for r in contents[schedule_section]['ranges']: try: path = r['path'] if path: - entry = {} - entry['Type'] = schedule_type - entry['StartDate'] = r['start_date'] - entry['EndDate'] = r['end_date'] - entry['Path'] = path + try: + force = r['force'] + except KeyError as ke: + # special case Optional, ignore + force = False + pass + + start = r['start_date'] + end = r['end_date'] + + entry = ScheduleEntry(type=schedule_section, + force=force, + startdate=start, + enddate=end, + path=path) schedule.append(entry) - except KeyError as e: - #logger.error('Key Value not found: "{}"'.format(schedule_type), exc_info=e) - raise e - except KeyError as e: - msg = 'Key Value not found in "{}" section'.format(schedule_type) - logger.error(msg, exc_info=e) - raise e - elif schedule_type == 'misc': + except KeyError as ke: + msg = 'Key Value not found for entry: "{}"'.format(entry) + logger.error(msg, exc_info=ke) + raise + except KeyError as ke: + msg = 'Key Value not found in "{}" section'.format(schedule_section) + logger.error(msg, exc_info=ke) + raise + elif schedule_section == 'misc': try: - use = contents[schedule_type]['enabled'] + use = contents[schedule_section]['enabled'] if use: try: - path = contents[schedule_type]['always_use'] + path = contents[schedule_section]['always_use'] if path: - entry = {} - entry['Type'] = schedule_type - entry['StartDate'] = date(today.year, today.month, today.day) - entry['EndDate'] = date(today.year, today.month, today.day) - entry['Path'] = path + entry = ScheduleEntry(type=schedule_section, + force=False, + startdate=date(today.year, today.month, today.day), + enddate=date(today.year, today.month, today.day), + path=path) schedule.append(entry) - except KeyError as e: - #logger.error('Key Value not found: "{}"'.format(schedule_type), exc_info=e) - raise e - except KeyError as e: - msg = 'Key Value not found in "{}" section'.format(schedule_type) - logger.error(msg, exc_info=e) - raise e - elif schedule_type == 'default': + except KeyError as ke: + msg = 'Key Value not found for entry: "{}"'.format(entry) + logger.error(msg, exc_info=ke) + raise + except KeyError as ke: + msg = 'Key Value not found in "{}" section'.format(schedule_section) + logger.error(msg, exc_info=ke) + raise + elif schedule_section == 'default': try: - use = contents[schedule_type]['enabled'] + use = contents[schedule_section]['enabled'] if use: try: - path = contents[schedule_type]['path'] + path = contents[schedule_section]['path'] if path: - entry = {} - entry['Type'] = schedule_type - entry['StartDate'] = date(today.year, today.month, today.day) - entry['EndDate'] = date(today.year, today.month, today.day) - - entry['Path'] = path + entry = ScheduleEntry(type=schedule_section, + force=False, + startdate=date(today.year, today.month, today.day), + enddate=date(today.year, today.month, today.day), + path=path) schedule.append(entry) - except KeyError as e: - #logger.error('Key Value not found: "{}"'.format(schedule_type), exc_info=e) - raise e - except KeyError as e: - msg = 'Key Value not found in "{}" section'.format(schedule_type) - logger.error(msg, exc_info=e) - raise e - + except KeyError as ke: + msg = 'Key Value not found for entry: "{}"'.format(entry) + logger.error(msg, exc_info=ke) + raise + except KeyError as ke: + msg = 'Key Value not found in "{}" section'.format(schedule_section) + logger.error(msg, exc_info=ke) + raise else: - continue + msg = 'Unknown schedule_section "{}" detected'.format(schedule_section) + logger.error(msg) + raise ValueError(msg) # Sort list so most recent Ranges appear first - schedule.sort(reverse=True, key=lambda x:x['StartDate']) + schedule.sort(reverse=True, key=lambda x:x.startdate) + #schedule.sort(reverse=False, key=lambda x:duration_seconds(x['startdate'], x['enddate'])) return schedule -def buildListingString(items, play_all=False): +def buildListingString(items:List[str], play_all:bool=False) -> str: """Build the Plex formatted string of preroll paths Args: @@ -310,17 +363,16 @@ def buildListingString(items, play_all=False): # use , to play all entries listing = ','.join(items) else: - pass #use ; to play random selection listing = ';'.join(items) return listing -def getPrerollListingString(schedule, for_datetime=None): +def getPrerollListing(schedule:List[ScheduleEntry], for_datetime:Optional[datetime]=None) -> str: """Return listing of preroll videos to be used by Plex Args: - schedule (list): List of schedule entries (See: getPrerollSchedule) + schedule (List[ScheduleEntry]): List of schedule entries (See: getPrerollSchedule) for_datetime (datetime, optional): Date to process pre-roll string for [Default: Today] Useful if wanting to test what different schedules produce @@ -328,11 +380,11 @@ def getPrerollListingString(schedule, for_datetime=None): string: listing of preroll video paths to be used for Extras. CSV style: (;|,) """ listing = '' - entries = {} + entries = getYAMLSchema() # prep the storage lists for y in getYAMLSchema(): - entries[y] = [] + entries[y] = [] # determine which date to build the listing for if for_datetime: @@ -346,8 +398,8 @@ def getPrerollListingString(schedule, for_datetime=None): # process the schedule for the given date for entry in schedule: try: - entry_start = entry['StartDate'] - entry_end = entry['EndDate'] + entry_start = entry.startdate #['startdate'] + entry_end = entry.enddate #['enddate'] if not isinstance(entry_start, datetime): entry_start = datetime.combine(entry_start, datetime.min.time()) if not isinstance(entry_end, datetime): @@ -357,52 +409,80 @@ def getPrerollListingString(schedule, for_datetime=None): logger.debug(msg) if entry_start <= check_datetime <= entry_end: - entry_type = entry['Type'] - path = entry['Path'] + entry_type = entry.type #['type'] + entry_path = entry.path #['path'] + entry_force = False + try: + entry_force = entry.force #['force'] + except KeyError as ke: + # special case Optional, ignore + pass - msg = 'pass: Using "{}" - "{}"'.format(entry_start, entry_end) + msg = 'Check PASS: Using "{}" - "{}"'.format(entry_start, entry_end) logger.debug(msg) - if path: - entries[entry_type].append(path) + if entry_path: + found = False + # check new schedule item against exist list + for e in entries[entry_type]: + duration_new = duration_seconds(entry_start, entry_end) + duration_curr = duration_seconds(e.startdate, e.enddate) #['startdate'], e['enddate']) + + # only the narrowest timeframe should stay + # disregard if a force entry is there + if duration_new < duration_curr and e.force != True: #['force'] != True: + entries[entry_type].remove(e) + found = True + else: + found = True + + # prep for use if New, or is a force Usage + if not found or entry_force == True: + entries[entry_type].append(entry) except KeyError as ke: msg = 'KeyError with entry "{}"'.format(entry) logger.warning(msg, exc_info=ke) - continue + raise # Build the merged output based or order of Priority merged_list = [] if entries['misc']: - merged_list.extend(entries['misc']) + merged_list.extend([p.path for p in entries['misc']]) if entries['date_range']: - merged_list.extend(entries['date_range']) + merged_list.extend([p.path for p in entries['date_range']]) if entries['weekly'] and not entries['date_range']: - merged_list.extend(entries['weekly']) + merged_list.extend([p.path for p in entries['weekly']]) if entries['monthly'] \ and not entries['weekly'] and not entries['date_range']: - merged_list.extend(entries['monthly']) + merged_list.extend([p.path for p in entries['monthly']]) if entries['default'] \ and not entries['monthly'] and not entries['weekly'] and not entries['date_range']: - merged_list.extend(entries['default']) + merged_list.extend([p.path for p in entries['default']]) listing = buildListingString(merged_list) return listing -def savePrerollList(plex, preroll_listing): +def savePrerollList(plex:PlexServer, preroll_listing:Union[str, List[str]]) -> None: """Save Plex Preroll info to PlexServer settings Args: plex (PlexServer): Plex server to update - preroll_listing (string, list): csv listing or List of preroll paths to save + preroll_listing (str, list[str]): csv listing or List of preroll paths to save """ # if happend to send in an Iterable List, merge to a string if type(preroll_listing) is list: - preroll_listing = buildListingString(preroll_listing) + preroll_listing = buildListingString(list(preroll_listing)) + + msg = 'Attempting save of pre-rolls: "{}"'.format(preroll_listing) + logger.debug(msg) plex.settings.get('cinemaTrailersPrerollID').set(preroll_listing) plex.settings.save() + msg = 'Saved Pre-Rolls: Server: "{}" Pre-Rolls: "{}"'.format(plex.friendlyName, preroll_listing) + logger.info(msg) + if __name__ == '__main__': args = getArguments() @@ -430,9 +510,12 @@ if __name__ == '__main__': raise e schedule = getPrerollSchedule(args.schedule_file) - prerolls = getPrerollListingString(schedule) + prerolls = getPrerollListing(schedule) - savePrerollList(plex, prerolls) + if args.do_test_run: + msg = 'Test Run of Plex Pre-Rolls: **Nothing being saved**\n{}\n'.format(prerolls) + logger.debug(msg) + print(msg) + else: + savePrerollList(plex, prerolls) - msg = 'Saved pre-roll list to server: "{}"'.format(prerolls) - logger.info(msg)