diff --git a/streamrip/constants.py b/streamrip/constants.py index 1535f9a..ba673f6 100644 --- a/streamrip/constants.py +++ b/streamrip/constants.py @@ -133,14 +133,18 @@ FOLDER_FORMAT = ( ) TRACK_FORMAT = "{tracknumber}. {artist} - {title}" + +# ------------------ Regexes ------------------- # URL_REGEX = ( r"https?://(?:www|open|play|listen)?\.?(qobuz|tidal|deezer)\.com(?:(?:/" r"(track|playlist|album|artist|label))|(?:\/[-\w]+?))+\/([-\w]+)" ) - SOUNDCLOUD_URL_REGEX = r"https://soundcloud.com/[-\w:/]+" SOUNDCLOUD_CLIENT_ID = "a3e059563d7fd3372b49b37f00a00bcf" LASTFM_URL_REGEX = r"https://www.last.fm/user/\w+/playlists/\w+" +QOBUZ_INTERPRETER_URL_REGEX = ( + r"https?://www\.qobuz\.com/\w\w-\w\w/interpreter/[-\w]+/[-\w]+" +) TIDAL_MAX_Q = 7 diff --git a/streamrip/core.py b/streamrip/core.py index 0bb66c9..df06106 100644 --- a/streamrip/core.py +++ b/streamrip/core.py @@ -19,6 +19,7 @@ from .constants import ( DB_PATH, LASTFM_URL_REGEX, MEDIA_TYPES, + QOBUZ_INTERPRETER_URL_REGEX, SOUNDCLOUD_URL_REGEX, URL_REGEX, ) @@ -30,7 +31,7 @@ from .exceptions import ( NoResultsFound, ParsingError, ) -from .utils import capitalize +from .utils import capitalize, extract_interpreter_url logger = logging.getLogger(__name__) @@ -54,6 +55,7 @@ class MusicDL(list): self.url_parse = re.compile(URL_REGEX) self.soundcloud_url_parse = re.compile(SOUNDCLOUD_URL_REGEX) self.lastfm_url_parse = re.compile(LASTFM_URL_REGEX) + self.interpreter_url_parse = re.compile(QOBUZ_INTERPRETER_URL_REGEX) self.config = config if self.config is None: @@ -76,48 +78,6 @@ class MusicDL(list): else: self.db = [] - def prompt_creds(self, source: str): - """Prompt the user for credentials. - - :param source: - :type source: str - """ - if source == "qobuz": - click.secho(f"Enter {capitalize(source)} email:", fg="green") - self.config.file[source]["email"] = input() - click.secho( - f"Enter {capitalize(source)} password (will not show on screen):", - fg="green", - ) - self.config.file[source]["password"] = md5( - getpass(prompt="").encode("utf-8") - ).hexdigest() - - self.config.save() - click.secho(f'Credentials saved to config file at "{self.config._path}"') - else: - raise Exception - - def assert_creds(self, source: str): - assert source in ( - "qobuz", - "tidal", - "deezer", - "soundcloud", - ), f"Invalid source {source}" - if source == "deezer": - # no login for deezer - return - - if source == "soundcloud": - return - - if source == "qobuz" and ( - self.config.file[source]["email"] is None - or self.config.file[source]["password"] is None - ): - self.prompt_creds(source) - def handle_urls(self, url: str): """Download a url @@ -217,9 +177,6 @@ class MusicDL(list): if self.db != [] and hasattr(item, "id"): self.db.add(item.id) - # if self.config.session["conversion"]["enabled"]: - # item.convert(**self.config.session["conversion"]) - def get_client(self, source: str): client = self.clients[source] if not client.logged_in: @@ -265,7 +222,23 @@ class MusicDL(list): :raises exceptions.ParsingError """ - parsed = self.url_parse.findall(url) # Qobuz, Tidal, Dezer + + parsed = [] + + interpreter_urls = self.interpreter_url_parse.findall(url) + if interpreter_urls: + click.secho( + "Extracting IDs from Qobuz interpreter urls. Use urls " + "that include the artist ID for faster preprocessing.", + fg="yellow", + ) + parsed.extend( + ("qobuz", "artist", extract_interpreter_url(u)) + for u in interpreter_urls + ) + url = self.interpreter_url_parse.sub("", url) + + parsed.extend(self.url_parse.findall(url)) # Qobuz, Tidal, Dezer soundcloud_urls = self.soundcloud_url_parse.findall(url) soundcloud_items = [self.clients["soundcloud"].get(u) for u in soundcloud_urls] @@ -503,3 +476,46 @@ class MusicDL(list): def __get_source_subdir(self, source: str) -> str: path = self.config.session["downloads"]["folder"] return os.path.join(path, capitalize(source)) + + def prompt_creds(self, source: str): + """Prompt the user for credentials. + + :param source: + :type source: str + """ + if source == "qobuz": + click.secho(f"Enter {capitalize(source)} email:", fg="green") + self.config.file[source]["email"] = input() + click.secho( + f"Enter {capitalize(source)} password (will not show on screen):", + fg="green", + ) + self.config.file[source]["password"] = md5( + getpass(prompt="").encode("utf-8") + ).hexdigest() + + self.config.save() + click.secho(f'Credentials saved to config file at "{self.config._path}"') + else: + raise Exception + + def assert_creds(self, source: str): + assert source in ( + "qobuz", + "tidal", + "deezer", + "soundcloud", + ), f"Invalid source {source}" + if source == "deezer": + # no login for deezer + return + + if source == "soundcloud": + return + + if source == "qobuz" and ( + self.config.file[source]["email"] is None + or self.config.file[source]["password"] is None + ): + self.prompt_creds(source) + diff --git a/streamrip/downloader.py b/streamrip/downloader.py index 27e1131..eed1452 100644 --- a/streamrip/downloader.py +++ b/streamrip/downloader.py @@ -16,7 +16,7 @@ from mutagen.flac import FLAC, Picture from mutagen.id3 import APIC, ID3, ID3NoHeaderError from mutagen.mp4 import MP4, MP4Cover from pathvalidate import sanitize_filename, sanitize_filepath -from requests.packages import urllib3 +from tqdm import tqdm from . import converter from .clients import Client @@ -45,8 +45,6 @@ from .utils import ( ) logger = logging.getLogger(__name__) -urllib3.disable_warnings() - TYPE_REGEXES = { "remaster": re.compile(r"(?i)(re)?master(ed)?"), @@ -90,22 +88,19 @@ class Track: self.id = None self.__dict__.update(kwargs) - # adjustments after blind attribute sets + # TODO: remove these self.container = "FLAC" self.sampling_rate = 44100 self.bit_depth = 16 self.downloaded = False self.tagged = False + # TODO: find better solution for attr in ("quality", "folder", "meta"): setattr(self, attr, None) if isinstance(kwargs.get("meta"), TrackMetadata): self.meta = kwargs["meta"] - else: - self.meta = None - # `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 url: {u}") @@ -195,7 +190,12 @@ class Track: :param progress_bar: turn on/off progress bar :type progress_bar: bool """ - if not self._prepare_download(quality=quality, parent_folder=parent_folder, progress_bar=progress_bar, **kwargs): + if not self._prepare_download( + quality=quality, + parent_folder=parent_folder, + progress_bar=progress_bar, + **kwargs, + ): return False if self.client.source == "soundcloud": @@ -617,6 +617,7 @@ class Tracklist(list): concurrent.futures.wait(futures) except (KeyboardInterrupt, SystemExit): executor.shutdown() + tqdm.write("Aborted! May take some time to shutdown.") exit("Aborted!") else: @@ -807,8 +808,8 @@ class Album(Tracklist): self.meta = self.client.get(self.id, media_type="album") # update attributes based on response - for k, v in self._parse_get_resp(self.meta, self.client).items(): - setattr(self, k, v) # prefer to __dict__.update for properties + info = self._parse_get_resp(self.meta, self.client).items() + self.__dict__.update(info) if not self.get("streamable", False): raise NonStreamable(f"This album is not streamable ({self.id} ID)") diff --git a/streamrip/utils.py b/streamrip/utils.py index 93d1fd4..f114cad 100644 --- a/streamrip/utils.py +++ b/streamrip/utils.py @@ -1,8 +1,9 @@ import base64 import contextlib -import sys import logging import os +import re +import sys from string import Formatter from typing import Hashable, Optional, Union @@ -15,7 +16,7 @@ from requests.packages import urllib3 from tqdm import tqdm from tqdm.contrib import DummyTqdmFile -from .constants import LOG_DIR, TIDAL_COVER_URL +from .constants import LOG_DIR, TIDAL_COVER_URL, AGENT from .exceptions import InvalidSourceError, NonStreamable urllib3.disable_warnings() @@ -267,3 +268,16 @@ def decho(message, fg=None): """ click.secho(message, fg=fg) logger.debug(message) + + +def extract_interpreter_url(url: str) -> str: + """Extract artist ID from a Qobuz interpreter url. + + :param url: Urls of the form "https://www.qobuz.com/us-en/interpreter/{artist}/download-streaming-albums" + :type url: str + :rtype: str + """ + session = gen_threadsafe_session({'User-Agent': AGENT}) + r = session.get(url) + artist_id = re.search(r"getSimilarArtist\(\s*'(\w+)'", r.text).group(1) + return artist_id