mirror of
https://github.com/nwithan8/plex-prerolls
synced 2024-08-30 16:52:17 +00:00
522 lines
21 KiB
Python
522 lines
21 KiB
Python
#!/usr/bin/python
|
|
"""Schedule Plex server related Pre-roll Intro videos
|
|
A helper script to automate management of Plex pre-rolls.
|
|
Define when you want different pre-rolls to play throughout the year.
|
|
|
|
Set it and forget it!
|
|
|
|
Optional Arguments:
|
|
-h, --help show this help message and exit
|
|
-v, --version show the version number and exit
|
|
-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]
|
|
-s SCHEDULE_FILE, --schedule-path SCHEDULE_FILE
|
|
Path to pre-roll schedule file (YAML) to be use. [Default: ./preroll_schedules.yaml]
|
|
|
|
Requirements:
|
|
- See Requirements.txt for Python modules
|
|
|
|
Scheduling:
|
|
Add to system scheduler such as:
|
|
> crontab -e
|
|
> 0 0 * * * python path/to/schedule_preroll.py >/dev/null 2>&1
|
|
|
|
Raises:
|
|
FileNotFoundError: [description]
|
|
KeyError: [description]
|
|
ConfigError: [description]
|
|
FileNotFoundError: [description]
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import logging
|
|
import requests
|
|
from datetime import datetime, date, timedelta
|
|
import yaml
|
|
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
|
|
|
|
# import local util modules
|
|
import plexutil
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
filename = os.path.basename(sys.argv[0])
|
|
SCRIPT_NAME = os.path.splitext(filename)[0]
|
|
|
|
#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
|
|
|
|
Returns:
|
|
argparse.Namespace: Namespace object
|
|
"""
|
|
description = 'Automate scheduling of pre-roll intros for Plex'
|
|
version = '0.9.2'
|
|
|
|
config_default = '' # './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('-lc', '--logconfig-file',
|
|
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-file',
|
|
dest='config_file', action='store',
|
|
help='Path to Config.ini to use for Plex Server info. [Default: {}]'.format(config_default))
|
|
parser.add_argument('-s', '--schedule-file',
|
|
dest='schedule_file', action='store',
|
|
help='Path to pre-roll schedule file (YAML) to be use. [Default: {}]'.format(schedule_default))
|
|
args = parser.parse_args()
|
|
|
|
return args
|
|
|
|
def getYAMLSchema() -> Dict[str, List[ScheduleEntry]]:
|
|
"""Return the main schema layout of the preroll_schedules.yaml file
|
|
|
|
Returns:
|
|
Dict (List[ScheduleType]): Dict of main schema items
|
|
"""
|
|
schema = {'default': [],
|
|
'monthly': [],
|
|
'weekly': [],
|
|
'date_range': [],
|
|
'misc': []
|
|
} # type: Dict[str, List[ScheduleEntry]]
|
|
return schema
|
|
|
|
def getWeekRange(year:int, weeknum:int) -> Tuple[date, date]:
|
|
"""Return the starting/ending date range of a given year/week
|
|
|
|
Args:
|
|
year (int): Year to calc range for
|
|
weeknum (int): Month of the year (1-12)
|
|
|
|
Returns:
|
|
Date: Start date of the Year/Month
|
|
Date: End date of the Year/Month
|
|
"""
|
|
start = datetime.strptime('{}-W{}-0'.format(year, int(weeknum)-1),
|
|
"%Y-W%W-%w").date()
|
|
end = start + timedelta(days=6)
|
|
|
|
return start, end
|
|
|
|
def getMonthRange(year:int, monthnum:int) -> Tuple[date, date]:
|
|
"""Return the starting/ending date range of a given year/month
|
|
|
|
Args:
|
|
year (int): Year to calc range for
|
|
monthnum (int): Month of the year (1-12)
|
|
|
|
Returns:
|
|
Date: Start date of the Year/Month
|
|
Date: End date of the Year/Month
|
|
"""
|
|
start = date(year, monthnum, 1)
|
|
next_month = start.replace(day=28) + timedelta(days=4)
|
|
end = next_month - timedelta(days=next_month.day)
|
|
|
|
return start, end
|
|
|
|
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 (str): path/to/schedule_preroll.yaml style config file (YAML Format)
|
|
|
|
Raises:
|
|
FileNotFoundError: If no schedule config file exists
|
|
|
|
Returns:
|
|
list: list of schedules (Dict: {Type, StartDate, EndDate, Path})
|
|
"""
|
|
default_files = ['preroll_schedules.yaml', 'preroll_schedules.yml']
|
|
|
|
filename = None
|
|
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)
|
|
raise FileNotFoundError(msg)
|
|
else:
|
|
for f in default_files:
|
|
if os.path.exists(f):
|
|
filename = f
|
|
break
|
|
|
|
# if we still cant find a schedule file, we abort
|
|
if not filename:
|
|
msg = 'Missing schedule file: "{}"'.format('" / "'.join(default_files))
|
|
logger.critical(msg)
|
|
raise FileNotFoundError(msg)
|
|
|
|
with open(filename, 'r') as file:
|
|
contents = yaml.load(file, Loader=yaml.SafeLoader)
|
|
|
|
today = date.today()
|
|
schedule = [] # type: List[ScheduleEntry]
|
|
for schedule_section in getYAMLSchema():
|
|
if schedule_section == 'weekly':
|
|
try:
|
|
use = contents[schedule_section]['enabled']
|
|
|
|
if use:
|
|
for i in range(1,53):
|
|
try:
|
|
path = contents[schedule_section][i]
|
|
|
|
if path:
|
|
start, end = getWeekRange(today.year, i)
|
|
|
|
entry = ScheduleEntry(type=schedule_section,
|
|
force=False,
|
|
startdate=start,
|
|
enddate=end,
|
|
path=path)
|
|
|
|
schedule.append(entry)
|
|
except KeyError as ke:
|
|
# skip KeyError for missing Weeks
|
|
msg = 'Key Value not found: "{}"->"{}", skipping week'.format(schedule_section, i)
|
|
logger.debug(msg)
|
|
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_section]['enabled']
|
|
|
|
if use:
|
|
for i in range(1,13):
|
|
month_abrev = date(today.year, i, 1).strftime('%b').lower()
|
|
try:
|
|
path = contents[schedule_section][month_abrev]
|
|
|
|
if path:
|
|
start, end = getMonthRange(today.year, i)
|
|
|
|
entry = ScheduleEntry(type=schedule_section,
|
|
force=False,
|
|
startdate=start,
|
|
enddate=end,
|
|
path=path)
|
|
|
|
schedule.append(entry)
|
|
except KeyError as ke:
|
|
# skip KeyError for missing Months
|
|
msg = 'Key Value not found: "{}"->"{}", skipping month'.format(schedule_section, month_abrev)
|
|
logger.warning(msg)
|
|
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_section]['enabled']
|
|
if use:
|
|
for r in contents[schedule_section]['ranges']:
|
|
try:
|
|
path = r['path']
|
|
|
|
if 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 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_section]['enabled']
|
|
if use:
|
|
try:
|
|
path = contents[schedule_section]['always_use']
|
|
|
|
if 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 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_section]['enabled']
|
|
if use:
|
|
try:
|
|
path = contents[schedule_section]['path']
|
|
|
|
if 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 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:
|
|
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=False, key=lambda x:duration_seconds(x['startdate'], x['enddate']))
|
|
|
|
return schedule
|
|
|
|
def buildListingString(items:List[str], play_all:bool=False) -> str:
|
|
"""Build the Plex formatted string of preroll paths
|
|
|
|
Args:
|
|
items (list): List of preroll video paths to place into a string listing
|
|
play_all (bool, optional): Play all videos. [Default: False (Random choice)]
|
|
|
|
Returns:
|
|
string: CSV Listing (, or ;) based on play_all param of preroll video paths
|
|
"""
|
|
if play_all:
|
|
# use , to play all entries
|
|
listing = ','.join(items)
|
|
else:
|
|
#use ; to play random selection
|
|
listing = ';'.join(items)
|
|
|
|
return listing
|
|
|
|
def getPrerollListing(schedule:List[ScheduleEntry], for_datetime:Optional[datetime]=None) -> str:
|
|
"""Return listing of preroll videos to be used by Plex
|
|
|
|
Args:
|
|
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
|
|
|
|
Returns:
|
|
string: listing of preroll video paths to be used for Extras. CSV style: (;|,)
|
|
"""
|
|
listing = ''
|
|
entries = getYAMLSchema()
|
|
|
|
# prep the storage lists
|
|
for y in getYAMLSchema():
|
|
entries[y] = []
|
|
|
|
# determine which date to build the listing for
|
|
if for_datetime:
|
|
if isinstance(for_datetime, datetime):
|
|
check_datetime = for_datetime
|
|
else:
|
|
check_datetime = datetime.combine(for_datetime, datetime.now().time())
|
|
else:
|
|
check_datetime = datetime.now()
|
|
|
|
# process the schedule for the given date
|
|
for entry in schedule:
|
|
try:
|
|
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):
|
|
entry_end = datetime.combine(entry_end, datetime.max.time())
|
|
|
|
msg = 'checking "{}" against: "{}" - "{}"'.format(check_datetime, entry_start, entry_end)
|
|
logger.debug(msg)
|
|
|
|
if entry_start <= check_datetime <= entry_end:
|
|
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 = 'Check PASS: Using "{}" - "{}"'.format(entry_start, entry_end)
|
|
logger.debug(msg)
|
|
|
|
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.error(msg, exc_info=ke)
|
|
raise
|
|
|
|
# Build the merged output based or order of Priority
|
|
merged_list = []
|
|
if entries['misc']:
|
|
merged_list.extend([p.path for p in entries['misc']])
|
|
if 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([p.path for p in entries['weekly']])
|
|
if entries['monthly'] \
|
|
and not entries['weekly'] and not entries['date_range']:
|
|
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([p.path for p in entries['default']])
|
|
|
|
listing = buildListingString(merged_list)
|
|
|
|
return 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 (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(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()
|
|
|
|
plexutil.setupLogger(args.log_config_file)
|
|
|
|
cfg = plexutil.getPlexConfig(args.config_file)
|
|
|
|
# Initialize Session information
|
|
sess = requests.Session()
|
|
# Ignore verifying the SSL certificate
|
|
sess.verify = False # '/path/to/certfile'
|
|
# If verify is set to a path of a directory (not a cert file),
|
|
# the directory needs to be processed with the c_rehash utility
|
|
# from OpenSSL.
|
|
if sess.verify is False:
|
|
# Disable the warning that the request is insecure, we know that...
|
|
import urllib3
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
try:
|
|
plex = PlexServer(cfg['PLEX_URL'], cfg['PLEX_TOKEN'], session=sess)
|
|
except Exception as e:
|
|
msg = 'Error connecting to Plex'
|
|
logger.error(msg, exc_info=e)
|
|
raise e
|
|
|
|
schedule = getPrerollSchedule(args.schedule_file)
|
|
prerolls = getPrerollListing(schedule)
|
|
|
|
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)
|
|
|