"""ShareLink Plugin.""" import re from ..plugins import SoCoPlugin from ..exceptions import SoCoException class ShareClass: """Base class for supported services.""" def canonical_uri(self, uri): """Recognize a share link and return its canonical representation. Args: uri (str): A URI like "https://tidal.com/browse/album/157273956". Returns: str: The canonical URI or None if not recognized. """ raise NotImplementedError def service_number(self): """Return the service number. Returns: int: A number identifying the supported music service. """ raise NotImplementedError @staticmethod def magic(): """Return magic. Returns: dict: Magic prefix/key/class values for each share type. """ return { "album": { "prefix": "x-rincon-cpcontainer:1004206c", "key": "00040000", "class": "object.container.album.musicAlbum", }, "episode": { "prefix": "", "key": "00032020", "class": "object.item.audioItem.musicTrack", }, "track": { "prefix": "", "key": "00032020", "class": "object.item.audioItem.musicTrack", }, "show": { "prefix": "x-rincon-cpcontainer:1006206c", "key": "1006206c", "class": "object.container.playlistContainer", }, "song": { "prefix": "", "key": "10032020", "class": "object.item.audioItem.musicTrack", }, "playlist": { "prefix": "x-rincon-cpcontainer:1006206c", "key": "1006206c", "class": "object.container.playlistContainer", }, } def extract(self, uri): """Extract the share type and encoded URI from a share link. Returns: share_type: The shared type, like "album" or "track". encoded_uri: An escaped URI with a service-specific format. """ raise NotImplementedError class SpotifyShare(ShareClass): """Spotify share class.""" def canonical_uri(self, uri): match = re.search( r"spotify.*[:/](album|episode|playlist|show|track)[:/](\w+)", uri ) if match: return "spotify:" + match.group(1) + ":" + match.group(2) return None def service_number(self): return 2311 def extract(self, uri): spotify_uri = self.canonical_uri(uri) share_type = spotify_uri.split(":")[1] encoded_uri = spotify_uri.replace(":", "%3a") return (share_type, encoded_uri) class SpotifyUSShare(SpotifyShare): """Spotify US share class.""" def service_number(self): return 3079 class TIDALShare(ShareClass): """TIDAL share class.""" def canonical_uri(self, uri): match = re.search(r"https://tidal.*[:/](album|track|playlist)[:/]([\w-]+)", uri) if match: return "tidal:" + match.group(1) + ":" + match.group(2) return None def service_number(self): return 44551 def extract(self, uri): tidal_uri = self.canonical_uri(uri) share_type = tidal_uri.split(":")[1] encoded_uri = tidal_uri.replace("tidal:", "").replace(":", "%2f") return (share_type, encoded_uri) class DeezerShare(ShareClass): """Deezer share class.""" def canonical_uri(self, uri): match = re.search( r"https://www.deezer.*[:/](album|track|playlist)[:/]([\w-]+)", uri ) if match: return "deezer:" + match.group(1) + ":" + match.group(2) return None def service_number(self): return 519 def extract(self, uri): deezer_uri = self.canonical_uri(uri) share_type = deezer_uri.split(":")[1] encoded_uri = deezer_uri.replace("deezer:", "").replace(":", "-") return (share_type, encoded_uri) class AppleMusicShare(ShareClass): """Apple Music share class.""" def canonical_uri(self, uri): # https://music.apple.com/dk/album/black-velvet/217502930?i=217503142 match = re.search( r"https://music\.apple\.com/\w+/album/[^/]+/\d+\?i=(\d+)", uri ) if match: return "song:" + match.group(1) # https://music.apple.com/dk/album/amused-to-death/975952384 match = re.search(r"https://music\.apple\.com/\w+/album/[^/]+/(\d+)", uri) if match: return "album:" + match.group(1) # Apple-created playlist # https://music.apple.com/dk/playlist/power-ballads-essentials/pl.92e04ee75ed64804b9df468b5f45a161 # User-created playlist # https://music.apple.com/de/playlist/unnamed-playlist/pl.u-rR2PCrLdLJk match = re.search( r"https://music\.apple\.com/\w+/playlist/[^/]+/(pl\.[-a-zA-Z0-9]+)", uri ) if match: return "playlist:" + match.group(1) return None def service_number(self): return 52231 def extract(self, uri): uri = self.canonical_uri(uri) share_type = uri.split(":")[0] encoded_uri = uri.replace(":", "%3a") return (share_type, encoded_uri) class ShareLinkPlugin(SoCoPlugin): """A SoCo plugin for playing music service share links.""" def __init__(self, soco): """Initialize the plugin.""" super().__init__(soco) self.services = [ SpotifyShare(), SpotifyUSShare(), TIDALShare(), DeezerShare(), AppleMusicShare(), ] @property def name(self): return "ShareLink Plugin" def is_share_link(self, uri): """bool: Is the URI for a supported music service.""" for service in self.services: if service.canonical_uri(uri): return True return False def add_share_link_to_queue(self, uri, position=0, as_next=False, **kwargs): """Add a Spotify/Tidal/... item to the queue. This is similar to soco.add_uri_to_queue() but will work with music service share links that do not directly point to sound files. Args: uri (str): A URI like "spotify:album:6wiUBliPe76YAVpNEdidpY". position (int): The index (1-based) at which the URI should be added. Default is 0 (add URI at the end of the queue). as_next (bool): Whether this URI should be played as the next track in shuffle mode. This only works if "play_mode=SHUFFLE". **kwargs: any keyword arguments. If ``dc_title`` is specified, this will be used as the dc:title of the metadata_template. If not specified, the dc:title will be empty. Returns: int: The index of the new item in the queue. """ fault = SoCoException("Unsupported URI: " + uri) for service in self.services: if service.canonical_uri(uri): share_type, encoded_uri = service.extract(uri) magic = service.magic() enqueue_uri = magic[share_type]["prefix"] + encoded_uri dc_title = kwargs.pop("dc_title", "") metadata_template = ( '{title}{item_class}SA_RINCON{sn}_X_#Svc{sn}' "-0-Token" ) metadata = metadata_template.format( item_id=magic[share_type]["key"] + encoded_uri, title=dc_title, item_class=magic[share_type]["class"], sn=service.service_number(), ) try: response = self.soco.avTransport.AddURIToQueue( [ ("InstanceID", 0), ("EnqueuedURI", enqueue_uri), ("EnqueuedURIMetaData", metadata), ("DesiredFirstTrackNumberEnqueued", position), ("EnqueueAsNext", int(as_next)), ], **kwargs, ) qnumber = response["FirstTrackNumberEnqueued"] return int(qnumber) except SoCoException as err: # Try remaining services on failure but keep the exception # around in case nothing succeeds. fault = err raise fault