python-plexapi/plexapi/utils.py

# -*- 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


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