diff --git a/qobuz_dl_rewrite/downloader.py b/qobuz_dl_rewrite/downloader.py index 69fb05c..7ab0814 100644 --- a/qobuz_dl_rewrite/downloader.py +++ b/qobuz_dl_rewrite/downloader.py @@ -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. diff --git a/qobuz_dl_rewrite/metadata.py b/qobuz_dl_rewrite/metadata.py index 6cc0955..7330e14 100644 --- a/qobuz_dl_rewrite/metadata.py +++ b/qobuz_dl_rewrite/metadata.py @@ -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]: