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

Deezer album download working

Browse Source 
The file comes pre-tagged, so tagging is not implemented yet.
This commit is contained in:
nathannathant
2021-03-16 14:50:01 -07:00
parent 376854809e
commit 4cfb8e452e
4 changed files with 97 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
qobuz_dl_rewrite/clients.py
+18 -6
View File
@@ -34,7 +34,7 @@ TIDAL_Q_IDS = {
4: "LOW", # AAC
5: "HIGH", # AAC
6: "LOSSLESS", # Lossless, but it also could be MQA
7: "HI_RES",
7: "HI_RES", # not available for download
}
TIDAL_MAX_Q = max(TIDAL_Q_IDS.keys())
@@ -43,6 +43,7 @@ TIDAL_MAX_Q = max(TIDAL_Q_IDS.keys())
DEEZER_BASE = "https://api.deezer.com"
DEEZER_DL = "http://dz.loaderapp.info/deezer"
DEEZER_Q_IDS = {4: 128, 5: 320, 6: 1411}
DEEZER_MAX_Q = max(DEEZER_Q_IDS.keys())
# ----------- Abstract Classes -----------------
@@ -362,7 +363,7 @@ class DeezerClient(ClientInterface):
return response.json()
def get(self, meta_id: Union[str, int], type_: str = "album"):
def get(self, meta_id: Union[str, int], media_type: str = "album"):
"""Get metadata.
:param meta_id:
@@ -370,14 +371,25 @@ class DeezerClient(ClientInterface):
:param type_:
:type type_: str
"""
response = self.session.get(f"{DEEZER_BASE}/{type_}/{meta_id}")
response.raise_for_status()
url = f"{DEEZER_BASE}/{media_type}/{meta_id}"
item = self.session.get(url).json()
if media_type in ("album", "playlist"):
tracks = self.session.get(f"{url}/tracks").json()
item["tracks"] = tracks["data"]
item["track_total"] = len(tracks["data"])
return response.json()
return item
@staticmethod
def get_file_url(meta_id: Union[str, int], quality: int = 6):
return f'{DEEZER_DL}/{DEEZER_Q_IDS[quality]}/"{DEEZER_BASE}/track/{meta_id}"'
quality = min(DEEZER_MAX_Q, quality)
url = f'{DEEZER_DL}/{DEEZER_Q_IDS[quality]}/{DEEZER_BASE}/track/{meta_id}'
logger.debug(f"Download url {url}")
return url
@property
def source(self):
return "deezer"
class TidalClient(SecureClientInterface):
qobuz_dl_rewrite/downloader.py
+42 -24
View File
@@ -1,10 +1,10 @@
import logging
import os
import shutil
from pprint import pformat
from abc import ABC, abstractmethod
from pprint import pformat, pprint
from tempfile import gettempdir
from typing import Any, Optional, Union
from abc import ABC, abstractmethod
import requests
from mutagen.flac import FLAC, Picture
@@ -31,6 +31,8 @@ TIDAL_Q_MAP = {
"HI_RES": 7,
}
COVER_SIZES = ("thumbnail", "small", "large")
class Track:
"""Represents a downloadable track.
@@ -96,8 +98,9 @@ class Track:
def _get_tracklist(resp, source):
if source in ("qobuz", "tidal"):
return resp["tracks"]["items"]
elif source == "deezer":
return resp["tracks"]
else:
# TODO: implement deezer
raise NotImplementedError
def download(
@@ -124,20 +127,25 @@ class Track:
return
dl_info = self.client.get_file_url(self.id, quality) # dict
if not (dl_info.get("url") or dl_info.get("sampling_rate")) or dl_info.get(
"sample"
):
logger.debug("Track is a sample: %s", dl_info)
return
logger.debug("Downloadable URL found: %s", dl_info["url"])
self.temp_file = os.path.join(gettempdir(), "~qdl_track.tmp")
logger.debug("Temporary file path: %s", self.temp_file)
tqdm_download(dl_info["url"], self.temp_file) # downloads file
if self.client.source == 'qobuz':
if not (dl_info.get("url") or dl_info.get("sampling_rate")) or dl_info.get(
"sample"
):
logger.debug("Track is a sample: %s", dl_info)
return
if self.client.source in ('qobuz', 'tidal'):
logger.debug("Downloadable URL found: %s", dl_info.get("url"))
tqdm_download(dl_info["url"], self.temp_file) # downloads file
elif self.client.source == 'deezer':
logger.debug("Downloadable URL found: %s", dl_info)
tqdm_download(dl_info, self.temp_file) # downloads file
else:
raise InvalidSourceError(self.client.source)
shutil.move(self.temp_file, self.final_path)
self.__is_downloaded = True
@@ -435,8 +443,10 @@ class Album(Tracklist):
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")
pprint(self.meta)
# update attributes based on response
self.__dict__.update(self._parse_get_resp(self.meta, self.client))
for k, v in self._parse_get_resp(self.meta, self.client).items():
setattr(self, k, v) # prefer to __dict__.update for properties
if not self.get("streamable", False):
raise NonStreamable(f"This album is not streamable ({self.id} ID)")
@@ -476,9 +486,7 @@ class Album(Tracklist):
size: TIDAL_COVER_URL.format(
uuid=resp.get("cover").replace("-", "/"), height=x, width=x
)
for size, x in zip(
("thumbnail", "small", "large"), (160, 320, 1280)
)
for size, x in zip(COVER_SIZES, (160, 320, 1280))
},
"streamable": resp.get("allowStreaming"),
"quality": TIDAL_Q_MAP[resp.get("audioQuality")],
@@ -486,12 +494,21 @@ class Album(Tracklist):
}
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"),
"title": resp.get("title"),
"_artist": safe_get(resp, "artist", "name"),
"albumartist": safe_get(resp, "artist", "name"),
# version not given by API
"cover_urls": {
sk: resp.get(rk) # size key, resp key
for sk, rk in zip(
COVER_SIZES, ("cover", "cover_medium", "cover_xl")
)
},
"url": resp.get("link"),
"streamable": True, # api only returns streamables
"quality": 6, # all tracks are 16/44.1 streamable
"tracktotal": resp.get("track_total"),
}
else:
raise NotImplementedError
@@ -577,14 +594,15 @@ class Album(Tracklist):
# create a single cover object and use them for all tracks
# TODO: avoid this method if embeded covers are not requested
cover = self.get_cover_obj(cover_path, quality)
if self.client.source != 'deezer':
cover = self.get_cover_obj(cover_path, quality)
for track in self:
logger.debug("Downloading track to %s with quality %s", folder, quality)
track.download(quality, folder, progress_bar)
if tag_tracks:
if tag_tracks and self.client.source != 'deezer':
logger.debug("Tagging track")
track.tag(album_meta=self.meta, cover=cover)
track.tag(cover=cover)
def __repr__(self) -> str:
"""Return a string representation of this Album object.
qobuz_dl_rewrite/metadata.py
+30 -10
View File
@@ -90,19 +90,28 @@ class TrackMetadata:
self.date = resp.get("release_date_original") or resp.get("release_date")
self.copyright = resp.get("copyright")
self.albumartist = resp.get("artist", {}).get("name")
self.label = resp.get("label")
if isinstance(self.label, dict):
self.label = self.label.get("name")
elif self.__source == "tidal":
self.album = resp.get("title")
self.tracktotal = resp.get("numberOfTracks")
# genre not returned by API
self.date = resp.get("releaseDate")
self.copyright = resp.get("copyright")
self.albumartist = resp.get("artist", {}).get("name")
# label not returned by API
elif self.__source == "deezer":
raise NotImplementedError
self.album = resp.get("title")
self.tracktotal = resp.get("track_total")
self.genre = resp.get("genres", {}).get("data")
self.date = resp.get("release_date")
self.albumartist = resp.get("artist", {}).get("name")
self.label = resp.get("label")
else:
raise ValueError
@@ -124,14 +133,21 @@ class TrackMetadata:
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 = str(track.get("trackNumber"))
self.discnumber = str(track.get("volumeNumber"))
self.artist = track.get("artist", {}).get("name")
elif self.__source == "deezer":
raise NotImplementedError
self.title = track.get("title").strip()
self._mod_title(track.get("version"), None)
self.tracknumber = track.get("track_position")
self.discnumber = track.get("disk_number")
self.artist = track.get("artist", {}).get("name")
else:
raise ValueError
@@ -176,20 +192,24 @@ class TrackMetadata:
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]
return ", ".join(no_repeats)
if isinstance(self._genres, list):
genres = re.findall(r"([^\u2192\/]+)", "/".join(self._genres))
no_repeats = []
[no_repeats.append(g) for g in genres if g not in no_repeats]
return ", ".join(no_repeats)
elif isinstance(self._genres, str):
return self._genres
else:
raise TypeError(f"Genre must be list or str, not {type(self._genre)}")
@genre.setter
def genre(self, val: list): # Is the assert necessary?
def genre(self, val: Union[str, list]):
"""Sets the internal `genre` field to the given list.
It is not formatted until it is requested with `meta.genre`.
:param val:
:type val: list
:type val: Union[str, list]
"""
assert type(val) == list
self._genres = val
@property
qobuz_dl_rewrite/util.py
+7 -2
View File
@@ -1,3 +1,4 @@
import sys
import logging
import logging.handlers as handlers
import os
@@ -8,6 +9,8 @@ from tqdm import tqdm
from .constants import LOG_DIR
logger = logging.getLogger(__name__)
def safe_get(d: dict, *keys, default=None):
"""A replacement for chained `get()` statements on dicts:
@@ -52,13 +55,15 @@ def tqdm_download(url: str, filepath: str):
"""Downloads a file with a progress bar.
:param url: url to direct download
:type url: str
:param filepath: file to write
:type url: str
:type filepath: str
"""
# Fixme: add the conditional to the progress_bar bool
# FIXME: add the conditional to the progress_bar bool
logger.debug(f"Downloading {url} to {filepath}")
r = requests.get(url, allow_redirects=True, stream=True)
total = int(r.headers.get("content-length", 0))
logger.debug(f"File size = {total}")
with open(filepath, "wb") as file, tqdm(
total=total, unit="iB", unit_scale=True, unit_divisor=1024
) as bar: