Mirrored_Repos/plex-prerolls
1
0
Fork 0
mirror of https://github.com/nwithan8/plex-prerolls synced 2024-08-30 16:52:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

time precedence, test run ability, added typing

Browse Source
This commit is contained in:
Brian Lindner 2021-01-06 22:33:14 -05:00
parent d6bb7c61c5
commit 309f4abc6a
No known key found for this signature in database
GPG Key ID: 8A53187BAA2C7197
4 changed files with 231 additions and 149 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -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/

12
README.md
View File

@ -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
```
---

29
plexutil.py
View File

@ -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

335
schedule_preroll.py
View File

@ -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,7 +380,7 @@ 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():
@ -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)