diff --git a/qobuz_dl_rewrite/clients.py b/qobuz_dl_rewrite/clients.py index 92a9359..3055b11 100644 --- a/qobuz_dl_rewrite/clients.py +++ b/qobuz_dl_rewrite/clients.py @@ -1,6 +1,4 @@ -import base64 import hashlib -import json import logging import time from abc import ABC, abstractmethod @@ -9,11 +7,6 @@ from typing import Generator, Tuple, Union import requests import tidalapi -try: - from urlparse import urljoin -except ImportError: - from urllib.parse import urljoin - from .constants import AGENT, QOBUZ_FEATURED_KEYS from .exceptions import ( AuthenticationError, @@ -56,6 +49,14 @@ class ClientInterface(ABC): it is merely a template. """ + @abstractmethod + def login(self, **kwargs): + """Authenticate the client. + + :param kwargs: + """ + pass + @abstractmethod def search(self, query: str, media_type="album"): """Search API for query. @@ -88,29 +89,14 @@ class ClientInterface(ABC): def source(self): pass - -class SecureClientInterface(ClientInterface): - """Identical to a ClientInterface except for a login - method. - - This is an Abstract Base Class. It cannot be instantiated; - it is merely a template. - """ - - @abstractmethod - def login(self, **kwargs): - """Authenticate the client. - - :param kwargs: - """ - pass - - # ------------- Clients ----------------- -class QobuzClient(SecureClientInterface): +class QobuzClient(ClientInterface): # ------- Public Methods ------------- + def __init__(self): + self.logged_in = False + def login(self, email: str, pwd: str, **kwargs): """Authenticate the QobuzClient. Must have a paid membership. @@ -124,6 +110,10 @@ class QobuzClient(SecureClientInterface): :type pwd: str :param kwargs: app_id: str, secrets: list, return_secrets: bool """ + if self.logged_in: + logger.debug("Already logged in") + return + if not (kwargs.get("app_id") or kwargs.get("secrets")): logger.info("Fetching tokens from Qobuz") spoofer = Spoofer() @@ -149,6 +139,7 @@ class QobuzClient(SecureClientInterface): # used for caching app_id and secrets if kwargs.get("return_secrets"): return self.app_id, self.secrets + self.logged_in = True def search( self, query: str, media_type: str = "album", limit: int = 500 @@ -344,6 +335,7 @@ class QobuzClient(SecureClientInterface): class DeezerClient(ClientInterface): def __init__(self): self.session = requests.Session() + self.logged_in = True def search(self, query: str, media_type: str = "album", limit: int = 200) -> dict: """Search API for query. @@ -367,6 +359,9 @@ class DeezerClient(ClientInterface): return response.json() + def login(self, **kwargs): + logger.debug("Deezer does not require login call, returning") + def get(self, meta_id: Union[str, int], media_type: str = "album"): """Get metadata. @@ -381,6 +376,9 @@ class DeezerClient(ClientInterface): tracks = self.session.get(f"{url}/tracks").json() item["tracks"] = tracks["data"] item["track_total"] = len(tracks["data"]) + elif media_type == 'artist': + albums = self.session.get(f"{url}/albums").json() + item['albums'] = albums['data'] return item @@ -396,8 +394,13 @@ class DeezerClient(ClientInterface): return "deezer" -class TidalClient(SecureClientInterface): +class TidalClient(ClientInterface): + def __init__(self): + self.logged_in = False + def login(self, email: str, pwd: str): + if self.logged_in: + return config = tidalapi.Config() @@ -405,6 +408,8 @@ class TidalClient(SecureClientInterface): self.session.login(email, pwd) logger.info("Logged into Tidal") + self.logged_in = True + def search(self, query: str, media_type: str = "album", limit: int = 50): """ :param query: diff --git a/qobuz_dl_rewrite/config.py b/qobuz_dl_rewrite/config.py index 725eda8..93086d1 100644 --- a/qobuz_dl_rewrite/config.py +++ b/qobuz_dl_rewrite/config.py @@ -37,8 +37,9 @@ class Config: quality = 6 folder_format = "{artist} - {album} ({year}) [{bit_depth}B-{sampling_rate}kHz]" track_format = "{tracknumber}. {tracktitle}" - self.credentials = {"email": None, "password": None} - self.authentication = {"app_id": None, "secrets": None} + + self.Qobuz = {"email": None, "password": None, "app_id": None, "secrets": None} + self.Tidal = {"email": None, "password": None} self.downloads_database = None self.filters = {"smart_discography": False, "albums_only": False} self.downloads = {"folder": folder, "quality": quality} @@ -49,23 +50,47 @@ class Config: } self.path_format = {"folder": folder_format, "track": track_format} - if config_path is not None: - self.load(config_path) + self.__path = config_path + self.__loaded = False - def reset(self, path): - with open(path, "w") as cfg: - yaml.dump(self.__dict__, cfg) + def save(self): + if self.__loaded: + info = dict() + for k, v in self.__dict__.items(): + if not k.startswith("_"): + info[k] = v - def load(self, path): - with open(path) as cfg: + with open(self.__path, "w") as cfg: + yaml.dump(info, cfg) + + def load(self): + with open(self.__path) as cfg: self.__dict__.update(yaml.load(cfg)) + self.__loaded = True - def update(self, args): - """Update configuration based on args from CLI. + @property + def tidal_creds(self): + return { + "email": self.Tidal["email"], + "pwd": self.Tidal["password"], + } - :param args: - """ - self.__dict__.update(args) + @property + def qobuz_creds(self): + return { + "email": self.Qobuz["email"], + "pwd": self.Qobuz["password"], + "app_id": self.Qobuz["app_id"], + "secrets": self.Qobuz["secrets"].split(","), + } + + def creds(self, source: str): + if source == "qobuz": + return self.qobuz_creds + elif source == "tidal": + return self.tidal_creds + else: + raise NotImplementedError def __getitem__(self, key): return getattr(self, key) diff --git a/qobuz_dl_rewrite/constants.py b/qobuz_dl_rewrite/constants.py index 4200341..c4e11df 100644 --- a/qobuz_dl_rewrite/constants.py +++ b/qobuz_dl_rewrite/constants.py @@ -1,3 +1,5 @@ +import os + import appdirs import mutagen.id3 as id3 @@ -5,7 +7,9 @@ APPNAME = "qobuz-dl" CACHE_DIR = appdirs.user_cache_dir(APPNAME) CONFIG_DIR = appdirs.user_config_dir(APPNAME) +CONFIG_PATH = os.path.join(CONFIG_DIR, "config.yaml") LOG_DIR = appdirs.user_config_dir(APPNAME) +DB_PATH = os.path.join(LOG_DIR, 'qobuz-dl.db') AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:83.0) Gecko/20100101 Firefox/83.0" diff --git a/qobuz_dl_rewrite/core.py b/qobuz_dl_rewrite/core.py index ee0fbab..ac19f5e 100644 --- a/qobuz_dl_rewrite/core.py +++ b/qobuz_dl_rewrite/core.py @@ -4,8 +4,9 @@ import re # ------- Testing ---------- from typing import Generator, Sequence, Tuple, Union -from .clients import QobuzClient -from .constants import QOBUZ_URL_REGEX +from .clients import DeezerClient, QobuzClient, TidalClient +from .config import Config +from .constants import CONFIG_PATH, QOBUZ_URL_REGEX, DB_PATH from .db import QobuzDB from .downloader import Album, Artist, Playlist, Track from .exceptions import ParsingError @@ -16,22 +17,45 @@ logger = logging.getLogger(__name__) MEDIA_CLASS = {"album": Album, "playlist": Playlist, "artist": Artist, "track": Track} +CLIENTS = {"qobuz": QobuzClient, "tidal": TidalClient, "deezer": DeezerClient} Media = Union[Album, Playlist, Artist] # type hint class QobuzDL: def __init__( self, - creds: Tuple[str, str], - directory="Downloads", - quality=6, - downloads_db=None, config=None, + client=None, + source="qobuz", + database=None, **kwargs, ): - self.client = QobuzClient() - self.client.login(creds[0], creds[1]) self.url_parse = re.compile(QOBUZ_URL_REGEX) + if config is None: + self.config = Config(CONFIG_PATH) + self.config.load() + + if client is not None: + self.client = client + elif source is not None: + self.client = CLIENTS[source]() + else: + self.client = QobuzClient() + + if database is None: + self.db = QobuzDB(DB_PATH) + else: + assert isinstance(database, QobuzDB) + self.db = database + + creds = config.creds(self.client.source) + if creds.get("app_id") is None and self.client.source == "qobuz": + app_id, secrets = self.client.login(**creds, return_secrets=True) + config["Qobuz"]["app_id"] = app_id + config["Qobuz"]["secrets"] = secrets + config.save() + else: + self.client.login(**creds) def handle_url(self, url: str): url_type, item_id = self.parse_url(url) @@ -88,7 +112,7 @@ class QobuzDL: raise ParsingError("Error parsing URLs from file `{filepath}`") - def search(self, query: str, media_type: str, limit: int = 200) -> Media: + def search(self, query: str, media_type: str, limit: int = 200) -> Generator: results = self.client.search(query, media_type, limit) for page in results: for item in page[f"{media_type}s"]["items"]: diff --git a/qobuz_dl_rewrite/downloader.py b/qobuz_dl_rewrite/downloader.py index 8fd4e4b..e677a03 100644 --- a/qobuz_dl_rewrite/downloader.py +++ b/qobuz_dl_rewrite/downloader.py @@ -29,6 +29,8 @@ logger = logging.getLogger(__name__) # TODO: add the other quality options TIDAL_Q_MAP = { + "LOW": 4, + "HIGH": 5, "LOSSLESS": 6, "HI_RES": 7, } @@ -480,6 +482,8 @@ class Album(Tracklist): if kwargs.get("load_on_init"): self.load_meta() + self.downloaded = False + def load_meta(self): assert hasattr(self, "id"), "id must be set to load metadata" self.meta = self.client.get(self.id, media_type="album") @@ -493,6 +497,11 @@ class Album(Tracklist): self._load_tracks() + @classmethod + def from_api(cls, resp, client): + info = cls._parse_get_resp(resp, client) + return cls(client, **info) + @staticmethod def _parse_get_resp(resp: dict, client: ClientInterface) -> dict: """Parse information from a client.get(query, 'album') call. @@ -649,6 +658,8 @@ class Album(Tracklist): logger.debug("Tagging track") track.tag(cover=cover) + self.downloaded = True + def __repr__(self) -> str: """Return a string representation of this Album object. Useful for pprint and json.dumps. @@ -719,14 +730,14 @@ class Playlist(Tracklist): "source": self.client.source, } - elif self.client.source == 'deezer': - tracklist = self.meta['tracks'] + elif self.client.source == "deezer": + tracklist = self.meta["tracks"] def gen_cover(track): - return track['album']['cover_medium'] + return track["album"]["cover_medium"] def meta_args(track): - return {'track': track, 'source': self.client.source} + return {"track": track, "source": self.client.source} else: raise NotImplementedError @@ -734,7 +745,7 @@ class Playlist(Tracklist): for i, track in enumerate(tracklist): meta = TrackMetadata(**meta_args(track)) if new_tracknumbers: - meta['tracknumber'] = str(i+1) + meta["tracknumber"] = str(i + 1) self.append( Track( @@ -751,14 +762,14 @@ class Playlist(Tracklist): for track in self: logger.debug(track.meta) track.download() - if self.client.source != 'deezer': + if self.client.source != "deezer": # deezer comes pre-tagged logger.debug("Skipping tagging for deezer track") track.tag() @staticmethod def _parse_get_resp(item, client): - if client.source == 'qobuz': + if client.source == "qobuz": info = { "name": item.get("name"), "id": item.get("id"), @@ -768,10 +779,10 @@ class Playlist(Tracklist): "name": item["title"], "id": item["uuid"], } - elif client.source == 'deezer': + elif client.source == "deezer": info = { - 'name': item['title'], - 'id': item['id'], + "name": item["title"], + "id": item["id"], } else: raise InvalidSourceError(client.source) @@ -814,9 +825,21 @@ class Artist(Tracklist): self.name = self.meta.get("name") def _load_albums(self): - for album in self.meta.get("albums", {}).get("items", []): + if self.client.source == "qobuz": + albums = self.meta["albums"]["items"] + + elif self.client.source == "tidal": + albums = self.meta["items"] + + elif self.client.source == "deezer": + albums = self.meta["albums"] + + else: + raise InvalidSourceError(self.client.source) + + for album in albums: logger.debug("Appending album: %s", album.get("title")) - self.append(Album(self.client, **Album._parse_get_resp(album, self.client))) + self.append(Album.from_api(album, self.client)) def download(self, filters=None, no_repeats=False, quality=7): logger.debug(f"Length of tracklist {len(self)}") @@ -861,19 +884,7 @@ class Artist(Tracklist): :type source: str """ logging.debug("Loading item from API") - if source in ("qobuz", "deezer"): - info = { - "name": item.get("name"), - "id": item.get("id"), - } - elif source == "tidal": - info = { - "name": item.name, - "id": item.id, - } - else: - raise InvalidSourceError(source) - logging.debug(f"Loaded info {info}") + info = cls._parse_get_resp(item, client) # equivalent to Artist(client=client, **info) return cls(client=client, **info) @@ -894,6 +905,24 @@ class Artist(Tracklist): yield album break + @staticmethod + def _parse_get_resp(item, client): + if client.source in ("qobuz", "deezer"): + info = { + "name": item.get("name"), + "id": item.get("id"), + } + elif client.source == "tidal": + info = { + "name": item["name"], + "id": item["id"], + } + else: + raise InvalidSourceError(client.source) + + logging.debug(f"Loaded info {info}") + return info + # ----------- Filters -------------- @staticmethod diff --git a/qobuz_dl_rewrite/metadata.py b/qobuz_dl_rewrite/metadata.py index 6182b00..4746516 100644 --- a/qobuz_dl_rewrite/metadata.py +++ b/qobuz_dl_rewrite/metadata.py @@ -141,7 +141,7 @@ class TrackMetadata: self.artist = self.albumartist if track.get("album"): - self.add_album_meta(track['album']) + self.add_album_meta(track["album"]) elif self.__source == "tidal": self.title = track.get("title").strip() @@ -259,7 +259,7 @@ class TrackMetadata: """ if hasattr(self, "_year"): return self._year - elif hasattr(self, 'date'): + elif hasattr(self, "date"): return self.date[:4] else: return None