mirror of
https://github.com/pkkid/python-plexapi
synced 2024-11-10 14:14:19 +00:00
604 lines
21 KiB
Python
604 lines
21 KiB
Python
# -*- coding: utf-8 -*-
|
|
import logging, re
|
|
import os
|
|
from datetime import datetime
|
|
from plexapi.compat import quote, urlencode, string_type
|
|
import requests
|
|
|
|
from plexapi.exceptions import NotFound, UnknownType, Unsupported
|
|
from plexapi.exceptions import NotFound, NotImplementedError, UnknownType, Unsupported
|
|
from threading import Thread
|
|
|
|
|
|
# Search Types - Plex uses these to filter specific media types when searching.
|
|
SEARCHTYPES = {'movie': 1, 'show': 2, 'season': 3, 'episode': 4,
|
|
'artist': 8, 'album': 9, 'track': 10}
|
|
LIBRARY_TYPES = {}
|
|
|
|
|
|
def register_libtype(cls):
|
|
""" Registry of library types we may come across when parsing XML. This allows us to
|
|
define a few helper functions to dynamically convery the XML into objects. See
|
|
buildItem() below for an example.
|
|
"""
|
|
LIBRARY_TYPES[cls.TYPE] = cls
|
|
return cls
|
|
|
|
|
|
class NA(object):
|
|
""" This used to be a simple variable equal to '__NA__'. There has been need to
|
|
compare NA against None in some use cases. This object allows the internals
|
|
of PlexAPI to distinguish between unfetched values and fetched, but non-existent
|
|
values. (NA == None results to True; NA is None results to False)
|
|
"""
|
|
|
|
def __bool__(self):
|
|
return False
|
|
|
|
def __eq__(self, other):
|
|
return isinstance(other, NA) or other in [None, '__NA__']
|
|
|
|
def __nonzero__(self):
|
|
return False
|
|
|
|
def __repr__(self):
|
|
return '__NA__'
|
|
|
|
|
|
class SecretsFilter(logging.Filter):
|
|
""" Logging filter to hide secrets. """
|
|
def __init__(self, secrets=None):
|
|
self.secrets = secrets or set()
|
|
|
|
def add_secret(self, secret):
|
|
self.secrets.add(secret)
|
|
|
|
def filter(self, record):
|
|
cleanargs = list(record.args)
|
|
for i in range(len(cleanargs)):
|
|
if isinstance(cleanargs[i], string_type):
|
|
for secret in self.secrets:
|
|
cleanargs[i] = cleanargs[i].replace(secret, '<hidden>')
|
|
record.args = tuple(cleanargs)
|
|
return True
|
|
|
|
|
|
class PlexPartialObject(object):
|
|
""" Not all objects in the Plex listings return the complete list of elements
|
|
for the object. This object will allow you to assume each object is complete,
|
|
and if the specified value you request is None it will fetch the full object
|
|
automatically and update itself.
|
|
|
|
Attributes:
|
|
data (ElementTree): Response from PlexServer used to build this object (optional).
|
|
initpath (str): Relative path requested when retrieving specified `data` (optional).
|
|
server (:class:`~plexapi.server.PlexServer`): PlexServer object this is from.
|
|
"""
|
|
def __init__(self, data, initpath, server=None):
|
|
self.server = server
|
|
self.initpath = initpath
|
|
self._loadData(data)
|
|
self._reloaded = False
|
|
|
|
def __eq__(self, other):
|
|
return other is not None and self.key == other.key
|
|
|
|
def __repr__(self):
|
|
clsname = self.__class__.__name__
|
|
key = self.key.replace('/library/metadata/', '') if self.key else 'NA'
|
|
title = self.title.replace(' ', '.')[0:20].encode('utf8')
|
|
return '<%s:%s:%s>' % (clsname, key, title)
|
|
|
|
def __getattr__(self, attr):
|
|
# Auto reload self, from the full key (path) when needed.
|
|
if attr == 'key' or self.__dict__.get(attr) or self.isFullObject():
|
|
return self.__dict__.get(attr, NA)
|
|
print('reload because of %s' % attr)
|
|
self.reload()
|
|
return self.__dict__.get(attr, NA)
|
|
|
|
def __setattr__(self, attr, value):
|
|
if value != NA or self.isFullObject():
|
|
self.__dict__[attr] = value
|
|
|
|
def _loadData(self, data):
|
|
raise NotImplementedError('Abstract method not implemented.')
|
|
|
|
def isFullObject(self):
|
|
""" Retruns True if this is already a full object. A full object means all attributes
|
|
were populated from the api path representing only this item. For example, the
|
|
search result for a movie often only contain a portion of the attributes a full
|
|
object (main url) for that movie contain.
|
|
"""
|
|
return not self.key or self.key == self.initpath
|
|
|
|
def isPartialObject(self):
|
|
""" Returns True if this is NOT a full object. """
|
|
return not self.isFullObject()
|
|
|
|
def reload(self):
|
|
""" Reload the data for this object from PlexServer XML. """
|
|
data = self.server.query(self.key)
|
|
self.initpath = self.key
|
|
self._loadData(data[0])
|
|
self._reloaded = True
|
|
return self
|
|
|
|
|
|
class Playable(object):
|
|
""" This is a general place to store functions specific to media that is Playable.
|
|
Things were getting mixed up a bit when dealing with Shows, Season, Artists,
|
|
Albums which are all not playable.
|
|
|
|
Attributes:
|
|
player (:class:`~plexapi.client.PlexClient`): Client object playing this item (for active sessions).
|
|
playlistItemID (int): Playlist item ID (only populated for :class:`~plexapi.playlist.Playlist` items).
|
|
sessionKey (int): Active session key.
|
|
transcodeSession (:class:`~plexapi.media.TranscodeSession`): Transcode Session object
|
|
if item is being transcoded (None otherwise).
|
|
username (str): Username of the person playing this item (for active sessions).
|
|
viewedAt (datetime): Datetime item was last viewed (history).
|
|
"""
|
|
|
|
def _loadData(self, data):
|
|
# Load data for active sessions (/status/sessions)
|
|
self.sessionKey = cast(int, data.attrib.get('sessionKey', NA))
|
|
self.username = findUsername(data)
|
|
self.player = findPlayer(self.server, data)
|
|
self.transcodeSession = findTranscodeSession(self.server, data)
|
|
# Load data for history details (/status/sessions/history/all)
|
|
self.viewedAt = toDatetime(data.attrib.get('viewedAt', NA))
|
|
# Load data for playlist items
|
|
self.playlistItemID = cast(int, data.attrib.get('playlistItemID', NA))
|
|
|
|
def getStreamURL(self, **params):
|
|
""" Returns a stream url that may be used by external applications such as VLC.
|
|
|
|
Parameters:
|
|
**params (dict): optional parameters to manipulate the playback when accessing
|
|
the stream. A few known parameters include: maxVideoBitrate, videoResolution
|
|
offset, copyts, protocol, mediaIndex, platform.
|
|
|
|
Raises:
|
|
Unsupported: When the item doesn't support fetching a stream URL.
|
|
"""
|
|
if self.TYPE not in ('movie', 'episode', 'track'):
|
|
raise Unsupported('Fetching stream URL for %s is unsupported.' % self.TYPE)
|
|
mvb = params.get('maxVideoBitrate')
|
|
vr = params.get('videoResolution', '')
|
|
params = {
|
|
'path': self.key,
|
|
'offset': params.get('offset', 0),
|
|
'copyts': params.get('copyts', 1),
|
|
'protocol': params.get('protocol'),
|
|
'mediaIndex': params.get('mediaIndex', 0),
|
|
'X-Plex-Platform': params.get('platform', 'Chrome'),
|
|
'maxVideoBitrate': max(mvb, 64) if mvb else None,
|
|
'videoResolution': vr if re.match('^\d+x\d+$', vr) else None
|
|
}
|
|
# remove None values
|
|
params = {k: v for k, v in params.items() if v is not None}
|
|
streamtype = 'audio' if self.TYPE in ('track', 'album') else 'video'
|
|
# sort the keys since the randomness fucks with my tests..
|
|
sorted_params = sorted(params.items(), key=lambda val: val[0])
|
|
return self.server.url('/%s/:/transcode/universal/start.m3u8?%s' % (streamtype, urlencode(sorted_params)))
|
|
|
|
def iterParts(self):
|
|
""" Iterates over the parts of this media item. """
|
|
for item in self.media:
|
|
for part in item.parts:
|
|
yield part
|
|
|
|
def play(self, client):
|
|
""" Start playback on the specified client.
|
|
|
|
Parameters:
|
|
client (:class:`~plexapi.client.PlexClient`): Client to start playing on.
|
|
"""
|
|
client.playMedia(self)
|
|
|
|
def download(self, savepath=None, keep_orginal_name=False, **kwargs):
|
|
"""Download a episode. If kwargs are passed your can download a trancoded file.
|
|
|
|
Args:
|
|
savepath (str): Abs path to savefolder
|
|
keep_orginal_name (bool): Use the mediafiles orginal name
|
|
|
|
kwargs:
|
|
See getStreamURL docs.
|
|
|
|
"""
|
|
downloaded = []
|
|
locs = [i for i in self.iterParts() if i]
|
|
for loc in locs:
|
|
if keep_orginal_name is False:
|
|
name = '%s.%s' % (self._prettyfilename(), loc.container)
|
|
else:
|
|
name = loc.file
|
|
|
|
# So this seems to be a alot slower but allows transcode.
|
|
if kwargs:
|
|
download_url = self.getStreamURL(**kwargs)
|
|
else:
|
|
download_url = self.server.url('%s?download=1' % loc.key)
|
|
|
|
dl = download(download_url, filename=name, savepath=savepath, session=self.server.session)
|
|
if dl:
|
|
downloaded.append(dl)
|
|
|
|
return downloaded
|
|
|
|
|
|
def buildItem(server, elem, initpath, bytag=False):
|
|
""" Factory function to build the objects used within the PlexAPI.
|
|
|
|
Parameters:
|
|
server (:class:`~plexapi.server.PlexServer`): PlexServer object this is from.
|
|
elem (ElementTree): XML data needed to build the object.
|
|
initpath (str): Relative path requested when retrieving specified `data` (optional).
|
|
bytag (bool): Creates the object from the name specified by the tag instead of the
|
|
default which builds the object specified by the type attribute. <tag type='foo' />
|
|
|
|
Raises:
|
|
UnknownType: Unknown library type.
|
|
"""
|
|
libtype = elem.tag if bytag else elem.attrib.get('type')
|
|
if libtype == 'photo' and elem.tag == 'Directory':
|
|
libtype = 'photoalbum'
|
|
if libtype in LIBRARY_TYPES:
|
|
cls = LIBRARY_TYPES[libtype]
|
|
return cls(server, elem, initpath)
|
|
raise UnknownType('Unknown library type: %s' % libtype)
|
|
|
|
|
|
def cast(func, value):
|
|
""" Cast the specified value to the specified type (returned by func).
|
|
|
|
Parameters:
|
|
func (func): Calback function to used cast to type (int, bool, float, etc).
|
|
value (any): value to be cast and returned.
|
|
"""
|
|
if value not in [None, NA]:
|
|
if func == bool:
|
|
return bool(int(value))
|
|
elif func in [int, float]:
|
|
try:
|
|
return func(value)
|
|
except ValueError:
|
|
return float('nan')
|
|
return func(value)
|
|
return value
|
|
|
|
|
|
def findKey(server, key):
|
|
""" Finds and builds a object based on ratingKey.
|
|
|
|
Parameters:
|
|
server (:class:`~plexapi.server.PlexServer`): PlexServer object this is from.
|
|
key (int): ratingKey to find and return.
|
|
|
|
Raises:
|
|
NotFound: Unable to find key
|
|
"""
|
|
path = '/library/metadata/{0}'.format(key)
|
|
try:
|
|
# Item seems to be the first sub element
|
|
elem = server.query(path)[0]
|
|
return buildItem(server, elem, path)
|
|
except:
|
|
raise NotFound('Unable to find key: %s' % key)
|
|
|
|
|
|
def findItem(server, path, title):
|
|
""" Finds and builds a object based on title.
|
|
|
|
Parameters:
|
|
server (:class:`~plexapi.server.PlexServer`): PlexServer object this is from.
|
|
path (str): API path that returns item to search title for.
|
|
title (str): Title of the item to find and return.
|
|
|
|
Raises:
|
|
NotFound: Unable to find item.
|
|
"""
|
|
for elem in server.query(path):
|
|
if elem.attrib.get('title').lower() == title.lower():
|
|
return buildItem(server, elem, path)
|
|
raise NotFound('Unable to find item: %s' % title)
|
|
|
|
|
|
def findLocations(data, single=False):
|
|
""" Returns a list of filepaths from a location tag.
|
|
|
|
Parameters:
|
|
data (ElementTree): XML object to search for locations in.
|
|
single (bool): Set True to only return the first location found.
|
|
Return type will be a string if this is set to True.
|
|
"""
|
|
locations = []
|
|
for elem in data:
|
|
if elem.tag == 'Location':
|
|
locations.append(elem.attrib.get('path'))
|
|
if single:
|
|
return locations[0] if locations else None
|
|
return locations
|
|
|
|
|
|
def findPlayer(server, data):
|
|
""" Returns the :class:`~plexapi.client.PlexClient` object found in the specified data.
|
|
|
|
Parameters:
|
|
server (:class:`~plexapi.server.PlexServer`): PlexServer object this is from.
|
|
data (ElementTree): XML data to find Player in.
|
|
"""
|
|
elem = data.find('Player')
|
|
if elem is not None:
|
|
from plexapi.client import PlexClient
|
|
baseurl = 'http://%s:%s' % (elem.attrib.get('address'), elem.attrib.get('port'))
|
|
return PlexClient(baseurl, server=server, data=elem)
|
|
return None
|
|
|
|
|
|
def findStreams(media, streamtype):
|
|
""" Returns a list of streams (str) found in media that match the specified streamtype.
|
|
|
|
Parameters:
|
|
media (:class:`~plexapi.utils.Playable`): Item to search for streams (show, movie, episode).
|
|
streamtype (str): Streamtype to return (videostream, audiostream, subtitlestream).
|
|
"""
|
|
streams = []
|
|
for mediaitem in media:
|
|
for part in mediaitem.parts:
|
|
for stream in part.streams:
|
|
if stream.TYPE == streamtype:
|
|
streams.append(stream)
|
|
return streams
|
|
|
|
|
|
def findTranscodeSession(server, data):
|
|
""" Returns a :class:`~plexapi.media.TranscodeSession` object if found within the specified
|
|
XML data.
|
|
|
|
Parameters:
|
|
server (:class:`~plexapi.server.PlexServer`): PlexServer object this is from.
|
|
data (ElementTree): XML data to find TranscodeSession in.
|
|
"""
|
|
|
|
elem = data.find('TranscodeSession')
|
|
if elem is not None:
|
|
from plexapi import media
|
|
return media.TranscodeSession(server, elem)
|
|
return None
|
|
|
|
|
|
def findUsername(data):
|
|
""" Returns the username if found in the specified XML data. Returns None if not found.
|
|
|
|
Parameters:
|
|
data (ElementTree): XML data to find username in.
|
|
"""
|
|
elem = data.find('User')
|
|
if elem is not None:
|
|
return elem.attrib.get('title')
|
|
return None
|
|
|
|
|
|
def isInt(str):
|
|
""" Returns True if the specified string passes as an int. """
|
|
try:
|
|
int(str)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def joinArgs(args):
|
|
""" Returns a query string (uses for HTTP URLs) where only the value is URL encoded.
|
|
Example return value: '?genre=action&type=1337'.
|
|
|
|
Parameters:
|
|
args (dict): Arguments to include in query string.
|
|
"""
|
|
if not args:
|
|
return ''
|
|
arglist = []
|
|
for key in sorted(args, key=lambda x: x.lower()):
|
|
value = str(args[key])
|
|
arglist.append('%s=%s' % (key, quote(value)))
|
|
return '?%s' % '&'.join(arglist)
|
|
|
|
|
|
<<<<<<< HEAD
|
|
def listChoices(server, path):
|
|
""" Returns a dict of {title:key} for all simple choices in a search filter.
|
|
=======
|
|
def listChoices(server, path): # pragma: no cover # Dont think its is used
|
|
"""ListChoices is by _cleanSort etc.
|
|
>>>>>>> more cov
|
|
|
|
Parameters:
|
|
server (:class:`~plexapi.server.PlexServer`): PlexServer object this is from.
|
|
path (str): Relative path to request XML data from.
|
|
"""
|
|
return {c.attrib['title']: c.attrib['key'] for c in server.query(path)}
|
|
|
|
|
|
def listItems(server, path, libtype=None, watched=None, bytag=False):
|
|
""" Returns a list of object built from :func:`~plexapi.utils.buildItem()` found
|
|
within the specified path.
|
|
|
|
Parameters:
|
|
server (:class:`~plexapi.server.PlexServer`): PlexServer object this is from.
|
|
path (str): Relative path to request XML data from.
|
|
libtype (str): Optionally return only the specified library type.
|
|
watched (bool): Optionally return only watched or unwatched items.
|
|
bytag (bool): Set true if libtype is found in the XML tag (and not the 'type' attribute).
|
|
"""
|
|
items = []
|
|
for elem in server.query(path):
|
|
if libtype and elem.attrib.get('type') != libtype:
|
|
continue
|
|
if watched is True and int(elem.attrib.get('viewCount', 0)) == 0:
|
|
continue
|
|
if watched is False and int(elem.attrib.get('viewCount', 0)) >= 1:
|
|
continue
|
|
try:
|
|
items.append(buildItem(server, elem, path, bytag))
|
|
except UnknownType:
|
|
pass
|
|
return items
|
|
|
|
|
|
<<<<<<< HEAD
|
|
def rget(obj, attrstr, default=None, delim='.'):
|
|
""" Returns the value at the specified attrstr location within a nexted tree of
|
|
dicts, lists, tuples, functions, classes, etc. The lookup is done recursivley
|
|
for each key in attrstr (split by by the delimiter) This function is heavily
|
|
influenced by the lookups used in Django templates.
|
|
|
|
Parameters:
|
|
obj (any): Object to start the lookup in (dict, obj, list, tuple, etc).
|
|
attrstr (str): String to lookup (ex: 'foo.bar.baz.value')
|
|
default (any): Default value to return if not found.
|
|
delim (str): Delimiter separating keys in attrstr.
|
|
"""
|
|
=======
|
|
def rget(obj, attrstr, default=None, delim='.'): # pragma: no cover # Dont think its is used
|
|
>>>>>>> more cov
|
|
try:
|
|
parts = attrstr.split(delim, 1)
|
|
attr = parts[0]
|
|
attrstr = parts[1] if len(parts) == 2 else None
|
|
if isinstance(obj, dict):
|
|
value = obj[attr]
|
|
elif isinstance(obj, list):
|
|
value = obj[int(attr)]
|
|
elif isinstance(obj, tuple):
|
|
value = obj[int(attr)]
|
|
elif isinstance(obj, object):
|
|
value = getattr(obj, attr)
|
|
if attrstr:
|
|
return rget(value, attrstr, default, delim)
|
|
return value
|
|
except:
|
|
return default
|
|
|
|
|
|
def searchType(libtype):
|
|
""" Returns the integer value of the library string type.
|
|
|
|
Parameters:
|
|
libtype (str): Library type to lookup (movie, show, season, episode,
|
|
artist, album, track)
|
|
|
|
Raises:
|
|
NotFound: Unknown libtype
|
|
"""
|
|
libtype = str(libtype)
|
|
if libtype in [str(v) for v in SEARCHTYPES.values()]:
|
|
return libtype
|
|
if SEARCHTYPES.get(libtype) is not None:
|
|
return SEARCHTYPES[libtype]
|
|
raise NotFound('Unknown libtype: %s' % libtype)
|
|
|
|
|
|
def threaded(callback, listargs):
|
|
""" Returns the result of <callback> for each set of \*args in listargs. Each call
|
|
to <callback. is called concurrently in their own separate threads.
|
|
|
|
Parameters:
|
|
callback (func): Callback function to apply to each set of \*args.
|
|
listargs (list): List of lists; \*args to pass each thread.
|
|
"""
|
|
threads, results = [], []
|
|
for args in listargs:
|
|
args += [results, len(results)]
|
|
results.append(None)
|
|
threads.append(Thread(target=callback, args=args))
|
|
threads[-1].start()
|
|
for thread in threads:
|
|
thread.join()
|
|
|
|
return results
|
|
|
|
|
|
def toDatetime(value, format=None):
|
|
""" Returns a datetime object from the specified value.
|
|
|
|
Parameters:
|
|
value (str): value to return as a datetime
|
|
format (str): Format to pass strftime (optional; if value is a str).
|
|
"""
|
|
if value and value != NA:
|
|
if format:
|
|
value = datetime.strptime(value, format)
|
|
else:
|
|
value = datetime.fromtimestamp(int(value))
|
|
return value
|
|
|
|
|
|
def download(url, filename=None, savepath=None, session=None, chunksize=4024, mocked=False):
|
|
"""Helper to download a thumb, videofile or something.
|
|
|
|
Args:
|
|
url (str): url where the content be reached
|
|
filename (str): Filename of the downloaded file, default None
|
|
savepath (str): Defaults to current working dir
|
|
chunksize (int): What chunksize read/write at the time
|
|
mocked (bool): Helper to do evertything except write the file.
|
|
|
|
Example:
|
|
|
|
>>> download(a_episode.getStreamURL(), a_episode.location)
|
|
/path/to/file
|
|
|
|
Returns:
|
|
/path/to/file or None
|
|
|
|
"""
|
|
session = session or requests.Session()
|
|
print('Mocked download %s' % mocked)
|
|
|
|
if savepath is None:
|
|
savepath = os.getcwd()
|
|
else:
|
|
# Make sure the user supplied path exists
|
|
try:
|
|
os.makedirs(savepath)
|
|
except OSError:
|
|
if not os.path.isdir(savepath): # pragma: no cover
|
|
raise
|
|
|
|
filename = os.path.basename(filename)
|
|
fullpath = os.path.join(savepath, filename)
|
|
|
|
try:
|
|
response = session.get(url, stream=True)
|
|
|
|
# images dont have a extention so we try
|
|
# to guess it from content-type
|
|
ext = os.path.splitext(fullpath)[-1]
|
|
if ext:
|
|
ext = ''
|
|
else:
|
|
cp = response.headers.get('content-type')
|
|
|
|
if cp:
|
|
if 'image' in cp:
|
|
ext = '.%s' % cp.split('/')[1]
|
|
|
|
fullpath = '%s%s' % (fullpath, ext)
|
|
|
|
if mocked:
|
|
return fullpath
|
|
|
|
with open(fullpath, 'wb') as f:
|
|
for chunk in response.iter_content(chunk_size=chunksize):
|
|
if chunk:
|
|
f.write(chunk)
|
|
|
|
log.debug('Downloaded %s to %s from %s' % (filename, fullpath, url))
|
|
|
|
return fullpath
|
|
|
|
except Exception as e: # pragma: no cover
|
|
log.exception('Failed to download %s to %s %s' % (url, fullpath, e))
|