raphaeltempest/qobuz-dl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Deezer and Tidal artist working

Browse Source
This commit is contained in:
nathannathant
2021-03-18 16:54:10 -07:00
parent 2e5ca57e16
commit 71e9bda493
6 changed files with 164 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
qobuz_dl_rewrite/clients.py
+32 -27
View File
@@ -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:
qobuz_dl_rewrite/config.py
+39 -14
View File
@@ -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)
qobuz_dl_rewrite/constants.py
+4
View File
@@ -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"
qobuz_dl_rewrite/core.py
+33 -9
View File
@@ -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"]:
qobuz_dl_rewrite/downloader.py
+54 -25
View File
@@ -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
qobuz_dl_rewrite/metadata.py
+2 -2
View File
@@ -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