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

Qobuz Playlist download working

Browse Source
This commit is contained in:
nathannathant
2021-03-17 22:13:11 -07:00
parent 16d262f3d3
commit 528e38d1a3
2 changed files with 67 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
qobuz_dl_rewrite/downloader.py
+55 -9
View File
@@ -2,6 +2,7 @@ import logging
import os
import re
import shutil
import sys
from abc import ABC, abstractmethod
from pprint import pprint
from tempfile import gettempdir
@@ -91,6 +92,10 @@ class Track:
# `load_meta` must be called at some point
logger.debug("Track: meta not provided")
if (u := kwargs.get("cover_url")) is not None:
logger.debug(f"Cover {u}")
self.cover_url = u
def load_meta(self):
"""Send a request to the client to get metadata for this Track."""
assert hasattr(self, "id"), "id must be set before loading metadata"
@@ -112,7 +117,7 @@ class Track:
def download(
self,
quality: int = 7,
folder: Optional[Union[str, os.PathLike]] = None,
folder: Optional[Union[str, os.PathLike]] = "Downloads",
progress_bar: bool = True,
):
"""Download the track
@@ -127,11 +132,17 @@ class Track:
assert not self.__is_downloaded
self.quality, self.folder = quality or self.quality, folder or self.folder
if not os.path.isdir(self.folder):
os.makedirs(folder, exist_ok=True)
if os.path.isfile(self.format_final_path()):
self.__is_downloaded = True
logger.debug("File already exists: %s", self.final_path)
return
if hasattr(self, "cover_url"):
self.download_cover()
dl_info = self.client.get_file_url(self.id, quality) # dict
self.temp_file = os.path.join(gettempdir(), "~qdl_track.tmp")
logger.debug("Temporary file path: %s", self.temp_file)
@@ -156,6 +167,14 @@ class Track:
self.__is_downloaded = True
def download_cover(self):
self.cover_path = os.path.join(self.folder, f"cover{hash(self.meta.title)}.jpg")
logger.debug(f"Downloading cover from {self.cover_url}")
tqdm_download(self.cover_url, self.cover_path)
self.cover = Tracklist.get_cover_obj(self.cover_path, self.quality)
logger.debug(f"Cover obj: {self.cover}")
def format_final_path(self) -> str:
"""Return the final filepath of the downloaded file.
@@ -165,9 +184,12 @@ class Track:
"""
if not hasattr(self, "final_path"):
formatter = self.meta.get_formatter()
print(formatter)
filename = self.track_file_format.format(**formatter)
print(filename)
print(self.folder, filename, EXT[self.quality])
self.final_path = (
os.path.join(self.folder, sanitize_filename(filename))[:250]
os.path.join(self.folder, sanitize_filename(filename))[:250].strip()
+ EXT[self.quality] # file extension dict
)
@@ -240,7 +262,10 @@ class Track:
for k, v in self.meta.tags(container):
audio[k] = v
assert cover is not None # remove this later with no_embed option
if cover is None:
assert hasattr(self, "cover")
cover = self.cover
if container == "flac":
audio.add_picture(cover)
audio.save()
@@ -657,21 +682,30 @@ class Playlist(Tracklist):
if kwargs.get("load_on_init"):
self.load_meta()
@classmethod
def from_api(cls, resp: dict, client: ClientInterface):
info = cls._parse_get_resp(resp, client)
return cls(client, **info)
def load_meta(self):
self.meta = list(self.client.get(self.id, media_type="playlist"))[
0
] # generator
self.name = self.meta.get("name")
self.meta = self.client.get(self.id, "playlist")
self._load_tracks()
def _load_tracks(self):
for track in self.meta.get("tracks", {}).get("items", []):
logger.debug("Appending track: %s", track.get("title"))
self.append(Track(self.client, id=track.get("id")))
meta = TrackMetadata(track=track, album=track["album"])
self.append(Track(self.client, id=track.get("id"), meta=meta, cover_url=track['album']['image']['small']))
logger.debug(f"Loaded {len(self)} tracks from playlist {self.name}")
def download(self, filters=None):
for track in self:
logger.debug(track.meta)
track.download()
track.tag()
@staticmethod
def _parse_get_resp(item, client):
if client.source in ("qobuz", "deezer"):
@@ -805,17 +839,29 @@ class Artist(Tracklist):
yield album
break
# ----------- Filters --------------
@staticmethod
def studio_albums(artist, album):
return (
album["albumartist"] != "Various Artists"
and TYPE_REGEXES['extra'].search(album.title) is None
and TYPE_REGEXES["extra"].search(album.title) is None
)
@staticmethod
def no_features(artist, album):
return artist["name"] == album["albumartist"]
@staticmethod
def no_extras(artist, album):
return TYPE_REGEXES["extra"].search(album.title) is None
@staticmethod
def remaster_only(artist, album):
return TYPE_REGEXES["remaster"].search(album.title) is not None
# --------- Magic Methods --------
def __repr__(self) -> str:
"""Return a string representation of this Artist object.
Useful for pprint and json.dumps.
qobuz_dl_rewrite/metadata.py
+12 -3
View File
@@ -1,6 +1,8 @@
import json
import logging
import re
import sys
from pprint import pprint
from typing import Generator, Optional, Tuple, Union
from .constants import (
@@ -68,13 +70,17 @@ class TrackMetadata:
self.__source = source # not included in tags
if (track and album) is None:
logger.debug(f"{track and album}")
if track is None and album is None:
logger.debug("No params passed, returning")
return
if track is not None:
logger.debug("Parsing track metadata")
self.add_track_meta(track)
if album is not None:
logger.debug("Parsing album metadata")
self.add_album_meta(album)
def add_album_meta(self, resp: dict):
@@ -154,7 +160,7 @@ class TrackMetadata:
def _mod_title(self, version, work):
if version is not None:
logger.debug("Version found: %s", version)
self.title = f"{self.title} (version)"
self.title = f"{self.title} ({version})"
if work is not None:
logger.debug("Work found: %s", work)
self.title = f"{work}: {self.title}"
@@ -220,6 +226,8 @@ class TrackMetadata:
:rtype: str, None
"""
if hasattr(self, "_copyright"):
if self._copyright is None:
return None
cr = self._copyright.replace("(P)", PHON_COPYRIGHT)
cr = cr.replace("(C)", COPYRIGHT)
return cr
@@ -301,7 +309,8 @@ class TrackMetadata:
"""
for k, v in FLAC_KEY.items():
tag = getattr(self, k)
if tag is not None:
if tag:
logger.debug(f"Adding tag {v}: {repr(tag)}")
yield (v, tag)
def __gen_mp3_tags(self) -> Tuple[str, str]: