Added support for parsing Tidal API response

Not tested yet

qobuz_dl_rewrite/clients.py

@@ -1,13 +1,14 @@
+import base64
 import hashlib
 import json
 import logging
-import base64
 import time
 from abc import ABC, abstractmethod
 from typing import Generator, Tuple, Union
 
 import requests
 import tidalapi
+
 try:
     from urlparse import urljoin
 except ImportError:
@@ -30,9 +31,9 @@ QOBUZ_BASE = "https://www.qobuz.com/api.json/0.2"
 
 # Tidal
 TIDAL_Q_IDS = {
-    4: tidalapi.Quality.low,  # AAC
-    5: tidalapi.Quality.high,  # AAC
-    6: tidalapi.Quality.lossless,  # Lossless, but it also could be MQA
+    4: "LOW",  # AAC
+    5: "MEDIUM",  # AAC
+    6: "HIGH",  # Lossless, but it also could be MQA
 }
 
 
@@ -392,7 +393,7 @@ class TidalClient(SecureClientInterface):
         :raises ValueError: if field value is invalid
         """
 
-        return self.session.search(media_type, query, limit)
+        return self._search(query, media_type, limit=limit)
 
     def get(self, meta_id: Union[str, int], media_type: str = "album"):
         """Get metadata.
@@ -412,31 +413,33 @@ class TidalClient(SecureClientInterface):
         :type quality: int
         """
         # Not tested
-        self.session._config.quality = TIDAL_Q_IDS[quality]
-        return self.session.get_track_url(meta_id)
+        return self._get_file_url(meta_id, quality=quality)
 
-    def _get_album(self, album_id):
-        album_info = self.session.get_album(album_id)
-        album_tracks = self.session.get_album_tracks(album_id)
-        album = album_info.__dict__
-        album['tracklist'] = album_tracks
-        return album
+    def _search(self, query, media_type="album", **kwargs):
+        params = {
+            "query": query,
+            "limit": kwargs.get("limit", 50),
+        }
+        return self.session.request("GET", f"search/{media_type}s", params).json()
 
-    def _get(self, media_id, media_type='album'):
-        if media_type == 'album':
-            info = self.session.request('GET', f"albums/{media_id}")
-            tracklist = self.session.request('GET', f"albums/{media_id}/tracks")
+    def _get(self, media_id, media_type="album"):
+        if media_type == "album":
+            info = self.session.request("GET", f"albums/{media_id}")
+            tracklist = self.session.request("GET", f"albums/{media_id}/tracks")
             album = info.json()
-            album['tracklist'] = tracklist.json()
+            album["tracks"] = tracklist.json()
             return album
 
-        elif media_type == 'track':
-            return self.session.request('GET', f"tracks/{media_id}").json()
-        elif media_type == 'playlist':
+        elif media_type == "track":
+            return self.session.request("GET", f"tracks/{media_id}").json()
+        elif media_type == "playlist":
             return self.session.request("GET", f"playlists/{media_id}/tracks").json()
-        elif media_type == 'artist':
+        elif media_type == "artist":
             return self.session.request("GET", f"artists/{media_id}/albums").json()
         else:
             raise ValueError
 
-
+    def _get_file_url(self, track_id, quality=6):
+        params = {"soundQuality": TIDAL_Q_IDS[quality]}
+        resp = self.session.request("GET", f"tracks/{track_id}/streamUrl", params)
+        return resp.json()

qobuz_dl_rewrite/downloader.py

@@ -13,12 +13,23 @@ from pathvalidate import sanitize_filename
 from . import converter
 from .clients import ClientInterface
 from .constants import EXT, FLAC_MAX_BLOCKSIZE
-from .exceptions import InvalidQuality, NonStreamable, TooLargeCoverArt
+from .exceptions import (
+    InvalidQuality,
+    InvalidSourceError,
+    NonStreamable,
+    TooLargeCoverArt,
+)
 from .metadata import TrackMetadata
 from .util import quality_id, safe_get, tqdm_download
 
 logger = logging.getLogger(__name__)
 
+# TODO: add the other quality options
+TIDAL_Q_MAP = {
+    "LOSSLESS": 6,
+    "HI_RES": 7,
+}
+
 
 class Track:
     """Represents a downloadable track returned by the qobuz api.
@@ -80,7 +91,17 @@ class Track:
         assert hasattr(self, "id"), "id must be set before loading metadata"
 
         track_meta = self.client.get(self.id, media_type="track")
-        self.meta = TrackMetadata(track=track_meta)  # meta dict -> TrackMetadata object
+        self.meta = TrackMetadata(
+            track=track_meta, source=self.client.source
+        )  # meta dict -> TrackMetadata object
+
+    @staticmethod
+    def _get_tracklist(resp, client):
+        if client.source in ("qobuz", "tidal"):
+            return resp["tracks"]["items"]
+        else:
+            # TODO: implement deezer
+            raise NotImplementedError
 
     def download(
         self,
@@ -154,8 +175,9 @@ class Track:
         :type client: ClientInterface
         :raises IndexError
         """
-        track = album.get("tracks", {}).get("items", [])[pos]
-        meta = TrackMetadata(album=album, track=track)
+
+        track = cls._get_tracklist(album)[pos]
+        meta = TrackMetadata(album=album, track=track, source=client.source)
         meta.add_track_meta(album["tracks"]["items"][pos])
         return cls(client=client, meta=meta, id=track["id"])
 
@@ -335,6 +357,21 @@ class Tracklist(list):
         for track in self:
             track.convert(codec, **kwargs)
 
+    @classmethod
+    def from_api(cls, item: dict, client: ClientInterface):
+        """Create an Album object from the api response of Qobuz, Tidal,
+        or Deezer.
+
+        :param resp: response dict
+        :type resp: dict
+        :param source: in ('qobuz', 'deezer', 'tidal')
+        :type source: str
+        """
+        info = cls._load_get_response(item)
+
+        # equivalent to Album(client=client, **info)
+        return cls(client=client, **info)
+
     @staticmethod
     def get_cover_obj(cover_path: str, quality: int) -> Union[Picture, APIC]:
         """Given the path to an image and a quality id, return an initialized
@@ -366,6 +403,10 @@ class Tracklist(list):
 
         return cover_obj
 
+    @staticmethod
+    def _parse_get_resp(item, client):
+        raise NotImplementedError
+
 
 class Album(Tracklist):
     """Represents a downloadable Qobuz album."""
@@ -389,18 +430,60 @@ class Album(Tracklist):
 
     def load_meta(self):
         self.meta = self.client.get(self.id, media_type="album")
-        self.title = self.meta.get("title")
-        self._artist = self.meta.get("artist") or self.meta.get("performer")
-        self.albumartist = self._artist.get("name")
-        self.version = self.meta.get("version")
-        self.cover_urls = self.meta.get("image")
-        self.streamable = self.meta.get("streamable")
+        # update attributes based on response
+        self.__dict__.update(self._parse_get_resp(self.meta))
 
         if not self.get("streamable", False):
             raise NonStreamable(f"This album is not streamable ({self.id} ID)")
 
         self._load_tracks()
 
+    @staticmethod
+    def _parse_get_resp(resp: dict, client: ClientInterface) -> dict:
+        """Parse information from a client.get(query, 'album') call.
+
+        :param resp:
+        :type resp: dict
+        :rtype: dict
+        """
+        if client.source == "qobuz":
+            info = {
+                "title": resp.get("title"),
+                "_artist": resp.get("artist") or resp.get("performer"),
+                "albumartist": resp.get("name"),
+                "version": resp.get("version"),
+                "cover_urls": resp.get("image"),
+                "streamable": resp.get("streamable"),
+                "quality": quality_id(
+                    resp.get("maximum_bit_depth"), resp.get("maximum_sampling_rate")
+                ),
+                "tracktotal": resp.get("tracks_count"),
+            }
+        elif client.source == "tidal":
+            info = {
+                "title": resp.get("title"),
+                "_artist": safe_get(resp, "artist", "name"),
+                "albumartist": safe_get(resp, "artist", "name"),
+                "version": resp.get("version"),
+                "cover_urls": [resp.get("cover")],
+                "streamable": resp.get("allowStreaming"),
+                "quality": TIDAL_Q_MAP[resp.get("audioQuality")],
+                "tracktotal": resp.get("numberOfTracks"),
+            }
+        elif client.source == "deezer":
+            info = {
+                "title": resp.get("title"),
+                "albumartist": safe_get(resp, "artist", "name"),
+                "_artist": safe_get(resp, "artist", "name"),
+                "id": resp.get("id"),
+                "url": resp.get("link"),
+                "quality": 6,  # all tracks are 16/44.1 streamable
+            }
+        else:
+            raise NotImplementedError
+
+        return info
+
     def _load_tracks(self):
         """Given an album metadata dict returned by the API, append all of its
         tracks to `self`.
@@ -408,56 +491,12 @@ class Album(Tracklist):
         This uses a classmethod to convert an item into a Track object, which
         stores the metadata inside a TrackMetadata object.
         """
-        for i in range(self.meta.get("tracks", {}).get("total", [])):
+        for i in range(self.tracktotal):
             # append method inherited from superclass list
             self.append(
                 Track.from_album_meta(album=self.meta, pos=i, client=self.client)
             )
 
-    @classmethod
-    def from_api(cls, item: dict, client: ClientInterface, source: str = "qobuz"):
-        """Create an Album object from the api response of Qobuz, Tidal,
-        or Deezer.
-
-        :param resp: response dict
-        :type resp: dict
-        :param source: in ('qobuz', 'deezer', 'tidal')
-        :type source: str
-        """
-        if source == "qobuz":
-            # only collect minimal information for identification purposes
-            info = {
-                "title": item.get("title"),
-                "albumartist": item.get("artist", {}).get("name")
-                or item.get("performer", {}).get("name"),  # KeyError
-                "id": item.get("id"),  # this is the important part
-                "version": item.get("version"),  # KeyError
-                "url": item.get("url"),
-                "quality": quality_id(
-                    item.get("maximum_bit_depth"), item.get("maximum_sampling_rate")
-                ),
-                "streamable": item.get("streamable", False),
-            }
-        elif source == "tidal":
-            info = {
-                "title": item.name,
-                "id": item.id,
-                "albumartist": item.artist.name,
-            }
-        elif source == "deezer":
-            info = {
-                "title": item.get("title"),
-                "albumartist": item.get("artist", {}).get("name"),
-                "id": item.get("id"),
-                "url": item.get("link"),
-                "quality": 6,
-            }
-        else:
-            raise ValueError(f"invalid source '{source}'")
-
-        # equivalent to Album(client=client, **info)
-        return cls(client=client, **info)
-
     @property
     def title(self) -> str:
         """Return the title of the album.

qobuz_dl_rewrite/exceptions.py

@@ -26,6 +26,10 @@ class InvalidContainerError(Exception):
     pass
 
 
+class InvalidSourceError(Exception):
+    pass
+
+
 class ParsingError(Exception):
     pass
 

qobuz_dl_rewrite/metadata.py

@@ -41,7 +41,9 @@ class TrackMetadata:
 
     """
 
-    def __init__(self, track: Optional[dict] = None, album: Optional[dict] = None):
+    def __init__(
+        self, track: Optional[dict] = None, album: Optional[dict] = None, source="qobuz"
+    ):
         """Creates a TrackMetadata object optionally initialized with
         dicts returned by the Qobuz API.
 
@@ -64,6 +66,8 @@ class TrackMetadata:
         self.tracknumber = None
         self.discnumber = None
 
+        self.__source = source  # not included in tags
+
         if (track and album) is None:
             return
 
@@ -82,17 +86,28 @@ class TrackMetadata:
 
         :param dict album: from the Qobuz API
         """
-        self.album = album.get("title")
-        self.tracktotal = str(album.get("tracks_count", 1))
-        self.genre = album.get("genres_list", [])
-        self.date = album.get("release_date_original") or album.get("release_date")
-        self.copyright = album.get("copyright")
-        self.albumartist = album.get("artist", {}).get("name")
+        if self.__source == "qobuz":
+            self.album = album.get("title")
+            self.tracktotal = str(album.get("tracks_count", 1))
+            self.genre = album.get("genres_list", [])
+            self.date = album.get("release_date_original") or album.get("release_date")
+            self.copyright = album.get("copyright")
+            self.albumartist = album.get("artist", {}).get("name")
 
-        self.label = album.get("label")
+            self.label = album.get("label")
 
-        if isinstance(self.label, dict):
-            self.label = self.label.get("name")
+            if isinstance(self.label, dict):
+                self.label = self.label.get("name")
+        elif self.__source == "tidal":
+            self.album = album.get("title")
+            self.tracktotal = album.get("numberOfTracks")
+            self.date = album.get("releaseDate")
+            self.copyright = album.get("copyright")
+            self.albumartist = album.get("artist", {}).get("name")
+        elif self.__source == "deezer":
+            raise NotImplementedError
+        else:
+            raise ValueError
 
     def add_track_meta(self, track: dict):
         """Parse the metadata from a track dict returned by the
@@ -100,23 +115,36 @@ class TrackMetadata:
 
         :param track:
         """
-        self.title = track.get("title").strip()
-        if track.get("version"):
-            logger.debug("Version found: %s", track["version"])
-            self.title = f"{self.title} ({track['version']})"
-        if track.get("work"):
-            logger.debug("Work found: %s", track["work"])
-            self.title = f"{track['work']}: {self.title}"
+        if self.__source == "qobuz":
+            self.title = track.get("title").strip()
+            self._mod_title(track.get("version"), track.get("work"))
+            self.composer = track.get("composer", {}).get("name")
 
-        self.composer = track.get("composer", {}).get("name")
+            self.tracknumber = str(track.get("track_number", 1))
+            self.discnumber = str(track.get("media_number", 1))
+            try:
+                self.artist = track["performer"]["name"]
+            except KeyError:
+                if hasattr(self, "albumartist"):
+                    self.artist = self.albumartist
+        elif self.__source == "tidal":
+            self.title = track.get("title").strip()
+            self._mod_title(track.get("version"), None)
+            self.tracknumber = track.get("trackNumber")
+            self.discnumber = track.get("volumeNumber")
+            self.artist = track.get("artist", {}).get("name")
+        elif self.__source == "deezer":
+            raise NotImplementedError
+        else:
+            raise ValueError
 
-        self.tracknumber = str(track.get("track_number", 1))
-        self.discnumber = str(track.get("media_number", 1))
-        try:
-            self.artist = track["performer"]["name"]
-        except KeyError:
-            if hasattr(self, "albumartist"):
-                self.artist = self.albumartist
+    def _mod_title(self, version, work):
+        if version is not None:
+            logger.debug("Version found: %s", version)
+            self.title = f"{self.title} (version)"
+        if work is not None:
+            logger.debug("Work found: %s", work)
+            self.title = f"{work}: {self.title}"
 
     @property
     def artist(self) -> Union[str, None]:
@@ -148,6 +176,9 @@ class TrackMetadata:
 
         :rtype: str
         """
+        if not hasattr(self, "_genres"):
+            return None
+
         genres = re.findall(r"([^\u2192\/]+)", "/".join(self._genres))
         no_repeats = []
         [no_repeats.append(g) for g in genres if g not in no_repeats]
@@ -253,7 +284,7 @@ class TrackMetadata:
         """
         for k, v in FLAC_KEY.items():
             tag = getattr(self, k)
-            if tag is not None:
+            if tag is not None and not k.startswith("_"):
                 yield (v, tag)
 
     def __gen_mp3_tags(self) -> Tuple[str, str]:
@@ -269,7 +300,7 @@ class TrackMetadata:
             else:
                 text = getattr(self, k)
 
-            if text is not None:
+            if text is not None and not k.startswith("_"):
                 yield (v.__name__, v(encoding=3, text=text))
 
     def __mp4_tags(self) -> Tuple[str, str]: