# -*- coding: utf-8 -*- import re from itertools import groupby from pathlib import Path from urllib.parse import quote_plus, unquote from plexapi import media, utils from plexapi.base import Playable, PlexPartialObject, cached_data_property from plexapi.exceptions import BadRequest, NotFound, Unsupported from plexapi.library import LibrarySection, MusicSection from plexapi.mixins import SmartFilterMixin, ArtMixin, PosterMixin, PlaylistEditMixins from plexapi.utils import deprecated @utils.registerPlexObject class Playlist( PlexPartialObject, Playable, SmartFilterMixin, ArtMixin, PosterMixin, PlaylistEditMixins ): """ Represents a single Playlist. Attributes: TAG (str): 'Playlist' TYPE (str): 'playlist' addedAt (datetime): Datetime the playlist was added to the server. allowSync (bool): True if you allow syncing playlists. composite (str): URL to composite image (/playlist//composite/) content (str): The filter URI string for smart playlists. duration (int): Duration of the playlist in milliseconds. durationInSeconds (int): Duration of the playlist in seconds. fields (List<:class:`~plexapi.media.Field`>): List of field objects. guid (str): Plex GUID for the playlist (com.plexapp.agents.none://XXXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXX). icon (str): Icon URI string for smart playlists. key (str): API URL (/playlist/). leafCount (int): Number of items in the playlist view. librarySectionID (int): Library section identifier (radio only) librarySectionKey (str): Library section key (radio only) librarySectionTitle (str): Library section title (radio only) playlistType (str): 'audio', 'video', or 'photo' radio (bool): If this playlist represents a radio station ratingKey (int): Unique key identifying the playlist. smart (bool): True if the playlist is a smart playlist. summary (str): Summary of the playlist. title (str): Name of the playlist. titleSort (str): Title to use when sorting (defaults to title). type (str): 'playlist' updatedAt (datetime): Datetime the playlist was updated. """ TAG = 'Playlist' TYPE = 'playlist' def _loadData(self, data): """ Load attribute values from Plex XML response. """ Playable._loadData(self, data) self.addedAt = utils.toDatetime(data.attrib.get('addedAt')) self.allowSync = utils.cast(bool, data.attrib.get('allowSync')) self.composite = data.attrib.get('composite') # url to thumbnail self.content = data.attrib.get('content') self.duration = utils.cast(int, data.attrib.get('duration')) self.durationInSeconds = utils.cast(int, data.attrib.get('durationInSeconds')) self.guid = data.attrib.get('guid') self.icon = data.attrib.get('icon') self.key = data.attrib.get('key', '').replace('/items', '') # FIX_BUG_50 self.leafCount = utils.cast(int, data.attrib.get('leafCount')) self.librarySectionID = utils.cast(int, data.attrib.get('librarySectionID')) self.librarySectionKey = data.attrib.get('librarySectionKey') self.librarySectionTitle = data.attrib.get('librarySectionTitle') self.playlistType = data.attrib.get('playlistType') self.radio = utils.cast(bool, data.attrib.get('radio', 0)) self.ratingKey = utils.cast(int, data.attrib.get('ratingKey')) self.smart = utils.cast(bool, data.attrib.get('smart')) self.summary = data.attrib.get('summary') self.title = data.attrib.get('title') self.titleSort = data.attrib.get('titleSort', self.title) self.type = data.attrib.get('type') self.updatedAt = utils.toDatetime(data.attrib.get('updatedAt')) @cached_data_property def fields(self): return self.findItems(self._data, media.Field) def __len__(self): # pragma: no cover return len(self.items()) def __iter__(self): # pragma: no cover for item in self.items(): yield item def __contains__(self, other): # pragma: no cover return any(i.key == other.key for i in self.items()) def __getitem__(self, key): # pragma: no cover return self.items()[key] @property def thumb(self): """ Alias to self.composite. """ return self.composite @property def metadataType(self): """ Returns the type of metadata in the playlist (movie, track, or photo). """ if self.isVideo: return 'movie' elif self.isAudio: return 'track' elif self.isPhoto: return 'photo' else: raise Unsupported('Unexpected playlist type') @property def isVideo(self): """ Returns True if this is a video playlist. """ return self.playlistType == 'video' @property def isAudio(self): """ Returns True if this is an audio playlist. """ return self.playlistType == 'audio' @property def isPhoto(self): """ Returns True if this is a photo playlist. """ return self.playlistType == 'photo' def _getPlaylistItemID(self, item): """ Match an item to a playlist item and return the item playlistItemID. """ for _item in self.items(): if _item.ratingKey == item.ratingKey: return _item.playlistItemID raise NotFound(f'Item with title "{item.title}" not found in the playlist') @cached_data_property def _filters(self): """ Cache for filters. """ return self._parseFilters(self.content) def filters(self): """ Returns the search filter dict for smart playlist. The filter dict be passed back into :func:`~plexapi.library.LibrarySection.search` to get the list of items. """ return self._filters @cached_data_property def _section(self): """ Cache for section. """ if not self.smart: raise BadRequest('Regular playlists are not associated with a library.') # Try to parse the library section from the content URI string match = re.search(r'/library/sections/(\d+)/all', unquote(self.content or '')) if match: sectionKey = int(match.group(1)) return self._server.library.sectionByID(sectionKey) # Try to get the library section from the first item in the playlist if self.items(): return self.items()[0].section() raise Unsupported('Unable to determine the library section') def section(self): """ Returns the :class:`~plexapi.library.LibrarySection` this smart playlist belongs to. Raises: :class:`plexapi.exceptions.BadRequest`: When trying to get the section for a regular playlist. :class:`plexapi.exceptions.Unsupported`: When unable to determine the library section. """ return self._section def item(self, title): """ Returns the item in the playlist that matches the specified title. Parameters: title (str): Title of the item to return. Raises: :class:`plexapi.exceptions.NotFound`: When the item is not found in the playlist. """ for item in self.items(): if item.title.lower() == title.lower(): return item raise NotFound(f'Item with title "{title}" not found in the playlist') @cached_data_property def _items(self): """ Cache for items. """ if self.radio: return [] key = f'{self.key}/items' items = self.fetchItems(key) # Cache server connections to avoid reconnecting for each item _servers = {} for item in items: if item.sourceURI: serverID = item.sourceURI.split('/')[2] if serverID not in _servers: try: _servers[serverID] = self._server.myPlexAccount().resource(serverID).connect() except NotFound: # Override the server connection with None if the server is not found _servers[serverID] = None item._server = _servers[serverID] return items def items(self): """ Returns a list of all items in the playlist. """ return self._items def get(self, title): """ Alias to :func:`~plexapi.playlist.Playlist.item`. """ return self.item(title) def addItems(self, items): """ Add items to the playlist. Parameters: items (List): List of :class:`~plexapi.audio.Audio`, :class:`~plexapi.video.Video`, or :class:`~plexapi.photo.Photo` objects to be added to the playlist. Raises: :class:`plexapi.exceptions.BadRequest`: When trying to add items to a smart playlist. """ if self.smart: raise BadRequest('Cannot add items to a smart playlist.') if items and not isinstance(items, (list, tuple)): items = [items] # Group items by server to maintain order when adding items from multiple servers for server, _items in groupby(items, key=lambda item: item._server): ratingKeys = [] for item in _items: if item.listType != self.playlistType: # pragma: no cover raise BadRequest(f'Can not mix media types when building a playlist: ' f'{self.playlistType} and {item.listType}') ratingKeys.append(str(item.ratingKey)) ratingKeys = ','.join(ratingKeys) uri = f'{server._uriRoot()}/library/metadata/{ratingKeys}' args = {'uri': uri} key = f"{self.key}/items{utils.joinArgs(args)}" self._server.query(key, method=self._server._session.put) return self @deprecated('use "removeItems" instead') def removeItem(self, item): self.removeItems(item) def removeItems(self, items): """ Remove items from the playlist. Parameters: items (List): List of :class:`~plexapi.audio.Audio`, :class:`~plexapi.video.Video`, or :class:`~plexapi.photo.Photo` objects to be removed from the playlist. Raises: :class:`plexapi.exceptions.BadRequest`: When trying to remove items from a smart playlist. :class:`plexapi.exceptions.NotFound`: When the item does not exist in the playlist. """ if self.smart: raise BadRequest('Cannot remove items from a smart playlist.') if items and not isinstance(items, (list, tuple)): items = [items] for item in items: playlistItemID = self._getPlaylistItemID(item) key = f'{self.key}/items/{playlistItemID}' self._server.query(key, method=self._server._session.delete) return self def moveItem(self, item, after=None): """ Move an item to a new position in the playlist. Parameters: items (obj): :class:`~plexapi.audio.Audio`, :class:`~plexapi.video.Video`, or :class:`~plexapi.photo.Photo` objects to be moved in the playlist. after (obj): :class:`~plexapi.audio.Audio`, :class:`~plexapi.video.Video`, or :class:`~plexapi.photo.Photo` objects to move the item after in the playlist. Raises: :class:`plexapi.exceptions.BadRequest`: When trying to move items in a smart playlist. :class:`plexapi.exceptions.NotFound`: When the item or item after does not exist in the playlist. """ if self.smart: raise BadRequest('Cannot move items in a smart playlist.') playlistItemID = self._getPlaylistItemID(item) key = f'{self.key}/items/{playlistItemID}/move' if after: afterPlaylistItemID = self._getPlaylistItemID(after) key += f'?after={afterPlaylistItemID}' self._server.query(key, method=self._server._session.put) return self def updateFilters(self, limit=None, sort=None, filters=None, **kwargs): """ Update the filters for a smart playlist. Parameters: limit (int): Limit the number of items in the playlist. sort (str or list, optional): A string of comma separated sort fields or a list of sort fields in the format ``column:dir``. See :func:`~plexapi.library.LibrarySection.search` for more info. filters (dict): A dictionary of advanced filters. See :func:`~plexapi.library.LibrarySection.search` for more info. **kwargs (dict): Additional custom filters to apply to the search results. See :func:`~plexapi.library.LibrarySection.search` for more info. Raises: :class:`plexapi.exceptions.BadRequest`: When trying update filters for a regular playlist. """ if not self.smart: raise BadRequest('Cannot update filters for a regular playlist.') section = self.section() searchKey = section._buildSearchKey( sort=sort, libtype=section.METADATA_TYPE, limit=limit, filters=filters, **kwargs) uri = f'{self._server._uriRoot()}{searchKey}' args = {'uri': uri} key = f"{self.key}/items{utils.joinArgs(args)}" self._server.query(key, method=self._server._session.put) return self def _edit(self, **kwargs): """ Actually edit the playlist. """ if isinstance(self._edits, dict): self._edits.update(kwargs) return self key = f'{self.key}{utils.joinArgs(kwargs)}' self._server.query(key, method=self._server._session.put) return self @deprecated('use "editTitle" and "editSummary" instead') def edit(self, title=None, summary=None): """ Edit the playlist. Parameters: title (str, optional): The title of the playlist. summary (str, optional): The summary of the playlist. """ args = {} if title: args['title'] = title if summary: args['summary'] = summary return self._edit(**args) def delete(self): """ Delete the playlist. """ self._server.query(self.key, method=self._server._session.delete) @classmethod def _create(cls, server, title, items): """ Create a regular playlist. """ if not items: raise BadRequest('Must include items to add when creating new playlist.') if items and not isinstance(items, (list, tuple)): items = [items] listType = items[0].listType ratingKeys = [] for item in items: if item.listType != listType: # pragma: no cover raise BadRequest('Can not mix media types when building a playlist.') ratingKeys.append(str(item.ratingKey)) ratingKeys = ','.join(ratingKeys) uri = f'{server._uriRoot()}/library/metadata/{ratingKeys}' args = {'uri': uri, 'type': listType, 'title': title, 'smart': 0} key = f"/playlists{utils.joinArgs(args)}" data = server.query(key, method=server._session.post)[0] return cls(server, data, initpath=key) @classmethod def _createSmart(cls, server, title, section, limit=None, libtype=None, sort=None, filters=None, **kwargs): """ Create a smart playlist. """ if not isinstance(section, LibrarySection): section = server.library.section(section) libtype = libtype or section.METADATA_TYPE searchKey = section._buildSearchKey( sort=sort, libtype=libtype, limit=limit, filters=filters, **kwargs) uri = f'{server._uriRoot()}{searchKey}' args = {'uri': uri, 'type': section.CONTENT_TYPE, 'title': title, 'smart': 1} key = f"/playlists{utils.joinArgs(args)}" data = server.query(key, method=server._session.post)[0] return cls(server, data, initpath=key) @classmethod def _createFromM3U(cls, server, title, section, m3ufilepath): """ Create a playlist from uploading an m3u file. """ if not isinstance(section, LibrarySection): section = server.library.section(section) if not isinstance(section, MusicSection): raise BadRequest('Can only create playlists from m3u files in a music library.') args = {'sectionID': section.key, 'path': m3ufilepath} key = f"/playlists/upload{utils.joinArgs(args)}" server.query(key, method=server._session.post) try: return server.playlists(sectionId=section.key, guid__endswith=m3ufilepath)[0].editTitle(title).reload() except IndexError: raise BadRequest('Failed to create playlist from m3u file.') from None @classmethod def create(cls, server, title, section=None, items=None, smart=False, limit=None, libtype=None, sort=None, filters=None, m3ufilepath=None, **kwargs): """ Create a playlist. Parameters: server (:class:`~plexapi.server.PlexServer`): Server to create the playlist on. title (str): Title of the playlist. section (:class:`~plexapi.library.LibrarySection`, str): Smart playlists and m3u import only, the library section to create the playlist in. items (List): Regular playlists only, list of :class:`~plexapi.audio.Audio`, :class:`~plexapi.video.Video`, or :class:`~plexapi.photo.Photo` objects to be added to the playlist. smart (bool): True to create a smart playlist. Default False. limit (int): Smart playlists only, limit the number of items in the playlist. libtype (str): Smart playlists only, the specific type of content to filter (movie, show, season, episode, artist, album, track, photoalbum, photo). sort (str or list, optional): Smart playlists only, a string of comma separated sort fields or a list of sort fields in the format ``column:dir``. See :func:`~plexapi.library.LibrarySection.search` for more info. filters (dict): Smart playlists only, a dictionary of advanced filters. See :func:`~plexapi.library.LibrarySection.search` for more info. m3ufilepath (str): Music playlists only, the full file path to an m3u file to import. Note: This will overwrite any playlist previously created from the same m3u file. **kwargs (dict): Smart playlists only, additional custom filters to apply to the search results. See :func:`~plexapi.library.LibrarySection.search` for more info. Raises: :class:`plexapi.exceptions.BadRequest`: When no items are included to create the playlist. :class:`plexapi.exceptions.BadRequest`: When mixing media types in the playlist. :class:`plexapi.exceptions.BadRequest`: When attempting to import m3u file into non-music library. :class:`plexapi.exceptions.BadRequest`: When failed to import m3u file. Returns: :class:`~plexapi.playlist.Playlist`: A new instance of the created Playlist. """ if m3ufilepath: return cls._createFromM3U(server, title, section, m3ufilepath) elif smart: if items: raise BadRequest('Cannot create a smart playlist with items.') return cls._createSmart(server, title, section, limit, libtype, sort, filters, **kwargs) else: return cls._create(server, title, items) def copyToUser(self, user): """ Copy playlist to another user account. Parameters: user (:class:`~plexapi.myplex.MyPlexUser` or str): `MyPlexUser` object, username, email, or user id of the user to copy the playlist to. """ userServer = self._server.switchUser(user) return self.create(server=userServer, title=self.title, items=self.items()) def sync(self, videoQuality=None, photoResolution=None, audioBitrate=None, client=None, clientId=None, limit=None, unwatched=False, title=None): """ Add the playlist as a sync item for the specified device. See :func:`~plexapi.myplex.MyPlexAccount.sync` for possible exceptions. Parameters: videoQuality (int): idx of quality of the video, one of VIDEO_QUALITY_* values defined in :mod:`~plexapi.sync` module. Used only when playlist contains video. photoResolution (str): maximum allowed resolution for synchronized photos, see PHOTO_QUALITY_* values in the module :mod:`~plexapi.sync`. Used only when playlist contains photos. audioBitrate (int): maximum bitrate for synchronized music, better use one of MUSIC_BITRATE_* values from the module :mod:`~plexapi.sync`. Used only when playlist contains audio. client (:class:`~plexapi.myplex.MyPlexDevice`): sync destination, see :func:`~plexapi.myplex.MyPlexAccount.sync`. clientId (str): sync destination, see :func:`~plexapi.myplex.MyPlexAccount.sync`. limit (int): maximum count of items to sync, unlimited if `None`. unwatched (bool): if `True` watched videos wouldn't be synced. title (str): descriptive title for the new :class:`~plexapi.sync.SyncItem`, if empty the value would be generated from metadata of current photo. Raises: :exc:`~plexapi.exceptions.BadRequest`: When playlist is not allowed to sync. :exc:`~plexapi.exceptions.Unsupported`: When playlist content is unsupported. Returns: :class:`~plexapi.sync.SyncItem`: A new instance of the created sync item. """ if not self.allowSync: raise BadRequest('The playlist is not allowed to sync') from plexapi.sync import SyncItem, Policy, MediaSettings myplex = self._server.myPlexAccount() sync_item = SyncItem(self._server, None) sync_item.title = title if title else self.title sync_item.rootTitle = self.title sync_item.contentType = self.playlistType sync_item.metadataType = self.metadataType sync_item.machineIdentifier = self._server.machineIdentifier sync_item.location = f'playlist:///{quote_plus(self.guid)}' sync_item.policy = Policy.create(limit, unwatched) if self.isVideo: sync_item.mediaSettings = MediaSettings.createVideo(videoQuality) elif self.isAudio: sync_item.mediaSettings = MediaSettings.createMusic(audioBitrate) elif self.isPhoto: sync_item.mediaSettings = MediaSettings.createPhoto(photoResolution) else: raise Unsupported('Unsupported playlist content') return myplex.sync(sync_item, client=client, clientId=clientId) def _getWebURL(self, base=None): """ Get the Plex Web URL with the correct parameters. """ return self._server._buildWebURL(base=base, endpoint='playlist', key=self.key) @property def metadataDirectory(self): """ Returns the Plex Media Server data directory where the metadata is stored. """ guid_hash = utils.sha1hash(self.guid) return str(Path('Metadata') / 'Playlists' / guid_hash[0] / f'{guid_hash[1:]}.bundle')