Added support for urls from multiple sources

Also added database integration

.gitignore

@@ -1,2 +1,155 @@
 Qobuz Downloads
-__pycache__
+*__pycache*
+.env
+__pycache__/
+*.py[cod]
+*$py.class
+.bumpversion.cfg
+/*.py
+!/setup.py
+
+# C extensions
+*.so
+
+*.json
+*.txt
+*.db
+*.sh
+
+*.txt
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+*.yaml
+*.flac
+*.m4a
+*.ogg
+*.mp3

.isort.cfg

@@ -0,0 +1,3 @@
+[settings]
+multi_line_output=3
+include_trailing_comma=True

README.md

@@ -1,5 +1,5 @@
 # qobuz-dl
-Search, explore and download Lossless and Hi-Res music from [Qobuz](https://www.qobuz.com/). It *just works*™ (2025).
+Search, explore and download Lossless and Hi-Res music from [Qobuz](https://www.qobuz.com/).
 [![Donate](https://img.shields.io/badge/Donate-PayPal-green.svg)](https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=VZWSWVGZGJRMU&source=url)
 
 ## Features

qobuz_dl/__init__.py

@@ -1,2 +1,2 @@
-from .qopy import Client
 from .cli import main
+from .qopy import Client

qobuz_dl/bundle.py

@@ -1,79 +0,0 @@
-import base64
-import logging
-import re
-from collections import OrderedDict
-
-from requests import Session
-
-# Modified code based on DashLt's spoofbuz
-
-logger = logging.getLogger(__name__)
-
-_SEED_TIMEZONE_REGEX = re.compile(
-    r'[a-z]\.initialSeed\("(?P<seed>[\w=]+)",window\.utimezone\.(?P<timezone>[a-z]+)\)'
-)
-_INFO_EXTRAS_REGEX = r'name:"\w+/(?P<timezone>{timezones})",info:"(?P<info>[\w=]+)",extras:"(?P<extras>[\w=]+)"'
-_APP_ID_REGEX = re.compile(
-    r'production:{api:{appId:"(?P<app_id>\d{9})",appSecret:"\w{32}"'
-)
-
-_BUNDLE_URL_REGEX = re.compile(
-    r'<script src="(/resources/\d+\.\d+\.\d+-[a-z]\d{3}/bundle\.js)"></script>'
-)
-
-_BASE_URL = "https://play.qobuz.com"
-_BUNDLE_URL_REGEX = re.compile(
-    r'<script src="(/resources/\d+\.\d+\.\d+-[a-z]\d{3}/bundle\.js)"></script>'
-)
-
-
-class Bundle:
-    def __init__(self):
-        self._session = Session()
-
-        logger.debug("Getting logging page")
-        response = self._session.get(f"{_BASE_URL}/login")
-        response.raise_for_status()
-
-        bundle_url_match = _BUNDLE_URL_REGEX.search(response.text)
-        if not bundle_url_match:
-            raise NotImplementedError("Bundle URL found")
-
-        bundle_url = bundle_url_match.group(1)
-
-        logger.debug("Getting bundle")
-        response = self._session.get(_BASE_URL + bundle_url)
-        response.raise_for_status()
-
-        self._bundle = response.text
-
-    def get_app_id(self):
-        match = _APP_ID_REGEX.search(self._bundle)
-        if not match:
-            raise NotImplementedError("Failed to match APP ID")
-
-        return match.group("app_id")
-
-    def get_secrets(self):
-        logger.debug("Getting secrets")
-        seed_matches = _SEED_TIMEZONE_REGEX.finditer(self._bundle)
-        secrets = OrderedDict()
-
-        for match in seed_matches:
-            seed, timezone = match.group("seed", "timezone")
-            secrets[timezone] = [seed]
-
-        keypairs = list(secrets.items())
-        secrets.move_to_end(keypairs[1][0], last=False)
-        info_extras_regex = _INFO_EXTRAS_REGEX.format(
-            timezones="|".join([timezone.capitalize() for timezone in secrets])
-        )
-        info_extras_matches = re.finditer(info_extras_regex, self._bundle)
-        for match in info_extras_matches:
-            timezone, info, extras = match.group("timezone", "info", "extras")
-            secrets[timezone.lower()] += [info, extras]
-        for secret_pair in secrets:
-            secrets[secret_pair] = base64.standard_b64decode(
-                "".join(secrets[secret_pair])[:-44]
-            ).decode("utf-8")
-        return secrets

qobuz_dl/cli.py

@@ -1,15 +1,14 @@
 import configparser
+import glob
 import hashlib
 import logging
-import glob
 import os
 import sys
 
-from qobuz_dl.bundle import Bundle
+import qobuz_dl.spoofbuz as spoofbuz
 from qobuz_dl.color import GREEN, RED, YELLOW
 from qobuz_dl.commands import qobuz_dl_args
 from qobuz_dl.core import QobuzDL
-from qobuz_dl.downloader import DEFAULT_FOLDER, DEFAULT_TRACK
 
 logging.basicConfig(
     level=logging.INFO,
@@ -26,7 +25,7 @@ CONFIG_FILE = os.path.join(CONFIG_PATH, "config.ini")
 QOBUZ_DB = os.path.join(CONFIG_PATH, "qobuz_dl.db")
 
 
-def _reset_config(config_file):
+def reset_config(config_file):
     logging.info(f"{YELLOW}Creating config file: {config_file}")
     config = configparser.ConfigParser()
     config["DEFAULT"]["email"] = input("Enter your email:\n- ")
@@ -53,11 +52,12 @@ def _reset_config(config_file):
     config["DEFAULT"]["no_cover"] = "false"
     config["DEFAULT"]["no_database"] = "false"
     logging.info(f"{YELLOW}Getting tokens. Please wait...")
-    bundle = Bundle()
-    config["DEFAULT"]["app_id"] = str(bundle.get_app_id())
-    config["DEFAULT"]["secrets"] = ",".join(bundle.get_secrets().values())
-    config["DEFAULT"]["folder_format"] = DEFAULT_FOLDER
-    config["DEFAULT"]["track_format"] = DEFAULT_TRACK
+    spoofer = spoofbuz.Spoofer()
+    config["DEFAULT"]["app_id"] = str(spoofer.getAppId())
+    config["DEFAULT"]["secrets"] = ",".join(spoofer.getSecrets().values())
+    config["DEFAULT"]["folder_format"] = "{artist} - {album} ({year}) "
+    "[{bit_depth}B-{sampling_rate}kHz]"
+    config["DEFAULT"]["track_format"] = "{tracknumber}. {tracktitle}"
     config["DEFAULT"]["smart_discography"] = "false"
     with open(config_file, "w") as configfile:
         config.write(configfile)
@@ -68,7 +68,7 @@ def _reset_config(config_file):
     )
 
 
-def _remove_leftovers(directory):
+def remove_leftovers(directory):
     directory = os.path.join(directory, "**", ".*.tmp")
     for i in glob.glob(directory, recursive=True):
         try:
@@ -77,41 +77,14 @@ def _remove_leftovers(directory):
             pass
 
 
-def _handle_commands(qobuz, arguments):
-    try:
-        if arguments.command == "dl":
-            qobuz.download_list_of_urls(arguments.SOURCE)
-        elif arguments.command == "lucky":
-            query = " ".join(arguments.QUERY)
-            qobuz.lucky_type = arguments.type
-            qobuz.lucky_limit = arguments.number
-            qobuz.lucky_mode(query)
-        else:
-            qobuz.interactive_limit = arguments.limit
-            qobuz.interactive()
-
-    except KeyboardInterrupt:
-        logging.info(
-            f"{RED}Interrupted by user\n{YELLOW}Already downloaded items will "
-            "be skipped if you try to download the same releases again."
-        )
-
-    finally:
-        _remove_leftovers(qobuz.directory)
-
-
-def _initial_checks():
+def main():
     if not os.path.isdir(CONFIG_PATH) or not os.path.isfile(CONFIG_FILE):
         os.makedirs(CONFIG_PATH, exist_ok=True)
-        _reset_config(CONFIG_FILE)
+        reset_config(CONFIG_FILE)
 
     if len(sys.argv) < 2:
         sys.exit(qobuz_dl_args().print_help())
 
-
-def main():
-    _initial_checks()
-
     config = configparser.ConfigParser()
     config.read(CONFIG_FILE)
 
@@ -129,6 +102,22 @@ def main():
         no_cover = config.getboolean("DEFAULT", "no_cover")
         no_database = config.getboolean("DEFAULT", "no_database")
         app_id = config["DEFAULT"]["app_id"]
+
+        if (
+            "folder_format" not in config["DEFAULT"]
+            or "track_format" not in config["DEFAULT"]
+            or "smart_discography" not in config["DEFAULT"]
+        ):
+            logging.info(
+                f"{YELLOW}Config file does not include some settings, updating..."
+            )
+            config["DEFAULT"]["folder_format"] = "{artist} - {album} ({year}) "
+            "[{bit_depth}B-{sampling_rate}kHz]"
+            config["DEFAULT"]["track_format"] = "{tracknumber}. {tracktitle}"
+            config["DEFAULT"]["smart_discography"] = "false"
+            with open(CONFIG_FILE, "w") as cf:
+                config.write(cf)
+
         smart_discography = config.getboolean("DEFAULT", "smart_discography")
         folder_format = config["DEFAULT"]["folder_format"]
         track_format = config["DEFAULT"]["track_format"]
@@ -139,22 +128,15 @@ def main():
         arguments = qobuz_dl_args(
             default_quality, default_limit, default_folder
         ).parse_args()
-    except (KeyError, UnicodeDecodeError, configparser.Error) as error:
+    except (KeyError, UnicodeDecodeError, configparser.Error):
         arguments = qobuz_dl_args().parse_args()
         if not arguments.reset:
             sys.exit(
-                f"{RED}Your config file is corrupted: {error}! "
-                "Run 'qobuz-dl -r' to fix this."
+                f"{RED}Your config file is corrupted! Run 'qobuz-dl -r' to fix this."
             )
 
     if arguments.reset:
-        sys.exit(_reset_config(CONFIG_FILE))
-
-    if arguments.show_config:
-        print(f"Configuation: {CONFIG_FILE}\nDatabase: {QOBUZ_DB}\n---")
-        with open(CONFIG_FILE, "r") as f:
-            print(f.read())
-        sys.exit()
+        sys.exit(reset_config(CONFIG_FILE))
 
     if arguments.purge:
         try:
@@ -179,7 +161,26 @@ def main():
     )
     qobuz.initialize_client(email, password, app_id, secrets)
 
-    _handle_commands(qobuz, arguments)
+    try:
+        if arguments.command == "dl":
+            qobuz.download_list_of_urls(arguments.SOURCE)
+        elif arguments.command == "lucky":
+            query = " ".join(arguments.QUERY)
+            qobuz.lucky_type = arguments.type
+            qobuz.lucky_limit = arguments.number
+            qobuz.lucky_mode(query)
+        else:
+            qobuz.interactive_limit = arguments.limit
+            qobuz.interactive()
+
+    except KeyboardInterrupt:
+        logging.info(
+            f"{RED}Interrupted by user\n{YELLOW}Already downloaded items will "
+            "be skipped if you try to download the same releases again."
+        )
+
+    finally:
+        remove_leftovers(qobuz.directory)
 
 
 if __name__ == "__main__":

qobuz_dl/color.py

@@ -1,4 +1,4 @@
-from colorama import Style, Fore, init
+from colorama import Fore, Style, init
 
 init(autoreset=True)
 

qobuz_dl/commands.py

@@ -108,7 +108,7 @@ def add_common_arg(custom_parser, default_folder, default_quality):
         metavar="PATTERN",
         help="""pattern for formatting folder names, e.g
         "{artist} - {album} ({year})". available keys: artist,
-        albumartist, album, year, sampling_rate, bit_depth, tracktitle, version.
+        albumartist, album, year, sampling_rate, bit_rate, tracktitle, version.
         cannot contain characters used by the system, which includes /:<>""",
     )
     custom_parser.add_argument(
@@ -149,12 +149,6 @@ def qobuz_dl_args(
         action="store_true",
         help="purge/delete downloaded-IDs database",
     )
-    parser.add_argument(
-        "-sc",
-        "--show-config",
-        action="store_true",
-        help="show configuration",
-    )
 
     subparsers = parser.add_subparsers(
         title="commands",

qobuz_dl/core.py

@@ -1,28 +1,27 @@
 import logging
 import os
+import re
+import string
 import sys
+import time
+from typing import Tuple
 
 import requests
 from bs4 import BeautifulSoup as bso
+from mutagen.flac import FLAC
+from mutagen.mp3 import EasyMP3
 from pathvalidate import sanitize_filename
 
-from qobuz_dl.bundle import Bundle
+import qobuz_dl.spoofbuz as spoofbuz
 from qobuz_dl import downloader, qopy
-from qobuz_dl.color import CYAN, OFF, RED, YELLOW, DF, RESET
-from qobuz_dl.exceptions import NonStreamable
+from qobuz_dl.color import CYAN, DF, OFF, RED, RESET, YELLOW
 from qobuz_dl.db import create_db, handle_download_id
-from qobuz_dl.utils import (
-    get_url_info,
-    make_m3u,
-    smart_discography_filter,
-    format_duration,
-    create_and_return_dir,
-    PartialFormatter,
-)
+from qobuz_dl.exceptions import NonStreamable
 
 WEB_URL = "https://play.qobuz.com/"
 ARTISTS_SELECTOR = "td.chartlist-artist > a"
 TITLE_SELECTOR = "td.chartlist-name > a"
+EXTENSIONS = (".mp3", ".flac")
 QUALITIES = {
     5: "5 - MP3",
     6: "6 - 16 bit, 44.1kHz",
@@ -33,6 +32,28 @@ QUALITIES = {
 logger = logging.getLogger(__name__)
 
 
+class PartialFormatter(string.Formatter):
+    def __init__(self, missing="n/a", bad_fmt="n/a"):
+        self.missing, self.bad_fmt = missing, bad_fmt
+
+    def get_field(self, field_name, args, kwargs):
+        try:
+            val = super(PartialFormatter, self).get_field(field_name, args, kwargs)
+        except (KeyError, AttributeError):
+            val = None, field_name
+        return val
+
+    def format_field(self, value, spec):
+        if not value:
+            return self.missing
+        try:
+            return super(PartialFormatter, self).format_field(value, spec)
+        except ValueError:
+            if self.bad_fmt:
+                return self.bad_fmt
+            raise
+
+
 class QobuzDL:
     def __init__(
         self,
@@ -53,7 +74,7 @@ class QobuzDL:
         track_format="{tracknumber}. {tracktitle}",
         smart_discography=False,
     ):
-        self.directory = create_and_return_dir(directory)
+        self.directory = self.create_dir(directory)
         self.quality = quality
         self.embed_art = embed_art
         self.lucky_limit = lucky_limit
@@ -74,12 +95,34 @@ class QobuzDL:
         logger.info(f"{YELLOW}Set max quality: {QUALITIES[int(self.quality)]}\n")
 
     def get_tokens(self):
-        bundle = Bundle()
-        self.app_id = bundle.get_app_id()
+        spoofer = spoofbuz.Spoofer()
+        self.app_id = spoofer.getAppId()
         self.secrets = [
-            secret for secret in bundle.get_secrets().values() if secret
+            secret for secret in spoofer.getSecrets().values() if secret
         ]  # avoid empty fields
 
+    def create_dir(self, directory=None):
+        fix = os.path.normpath(directory)
+        os.makedirs(fix, exist_ok=True)
+        return fix
+
+    def get_url_info(self, url: str) -> Tuple[str, str]:
+        """Returns the type of the url and the id.
+
+        Compatible with urls of the form:
+            https://www.qobuz.com/us-en/{type}/{name}/{id}
+            https://open.qobuz.com/{type}/{id}
+            https://play.qobuz.com/{type}/{id}
+            /us-en/{type}/-/{id}
+        """
+
+        r = re.search(
+            r"(?:https:\/\/(?:w{3}|open|play)\.qobuz\.com)?(?:\/[a-z]{2}-[a-z]{2})"
+            r"?\/(album|artist|track|playlist|label)(?:\/[-\w\d]+)?\/([\w\d]+)",
+            url,
+        )
+        return r.groups()
+
     def download_from_id(self, item_id, album=True, alt_path=None):
         if handle_download_id(self.downloads_db, item_id, add_id=False):
             logger.info(
@@ -89,20 +132,20 @@ class QobuzDL:
             )
             return
         try:
-            dloader = downloader.Download(
+            downloader.download_id_by_type(
                 self.client,
                 item_id,
                 alt_path or self.directory,
-                int(self.quality),
+                str(self.quality),
+                album,
                 self.embed_art,
                 self.ignore_singles_eps,
                 self.quality_fallback,
                 self.cover_og_quality,
                 self.no_cover,
-                self.folder_format,
-                self.track_format,
+                folder_format=self.folder_format,
+                track_format=self.track_format,
             )
-            dloader.download_id_by_type(not album)
             handle_download_id(self.downloads_db, item_id, add_id=True)
         except (requests.exceptions.RequestException, NonStreamable) as e:
             logger.error(f"{RED}Error getting release: {e}. Skipping...")
@@ -125,7 +168,7 @@ class QobuzDL:
             "track": {"album": False, "func": None, "iterable_key": None},
         }
         try:
-            url_type, item_id = get_url_info(url)
+            url_type, item_id = self.get_url_info(url)
             type_dict = possibles[url_type]
         except (KeyError, IndexError):
             logger.info(
@@ -139,13 +182,13 @@ class QobuzDL:
                 f"{YELLOW}Downloading all the music from {content_name} "
                 f"({url_type})!"
             )
-            new_path = create_and_return_dir(
+            new_path = self.create_dir(
                 os.path.join(self.directory, sanitize_filename(content_name))
             )
 
             if self.smart_discography and url_type == "artist":
                 # change `save_space` and `skip_extras` for customization
-                items = smart_discography_filter(
+                items = self._smart_discography_filter(
                     content,
                     save_space=True,
                     skip_extras=True,
@@ -162,8 +205,8 @@ class QobuzDL:
                     True if type_dict["iterable_key"] == "albums" else False,
                     new_path,
                 )
-            if url_type == "playlist" and not self.no_m3u_for_playlists:
-                make_m3u(new_path)
+            if url_type == "playlist":
+                self.make_m3u(new_path)
         else:
             self.download_from_id(item_id, type_dict["album"])
 
@@ -213,6 +256,9 @@ class QobuzDL:
 
         return results
 
+    def format_duration(self, duration):
+        return time.strftime("%H:%M:%S", time.gmtime(duration))
+
     def search_by_type(self, query, item_type, limit=10, lucky=False):
         if len(query) < 3:
             logger.info("{RED}Your search query is too short or invalid")
@@ -261,7 +307,7 @@ class QobuzDL:
 
                     text = "{} - {} [{}]".format(
                         text,
-                        format_duration(i["duration"]),
+                        self.format_duration(i["duration"]),
                         "HI-RES" if i["hires_streamable"] else "LOSSLESS",
                     )
 
@@ -389,11 +435,141 @@ class QobuzDL:
         )
 
         for i in track_list:
-            track_id = get_url_info(self.search_by_type(i, "track", 1, lucky=True)[0])[
-                1
-            ]
+            track_id = self.get_url_info(
+                self.search_by_type(i, "track", 1, lucky=True)[0]
+            )[1]
             if track_id:
                 self.download_from_id(track_id, False, pl_directory)
 
-        if not self.no_m3u_for_playlists:
-            make_m3u(pl_directory)
+        self.make_m3u(pl_directory)
+
+    def make_m3u(self, pl_directory):
+        if self.no_m3u_for_playlists:
+            return
+
+        track_list = ["#EXTM3U"]
+        rel_folder = os.path.basename(os.path.normpath(pl_directory))
+        pl_name = rel_folder + ".m3u"
+        for local, dirs, files in os.walk(pl_directory):
+            dirs.sort()
+            audio_rel_files = [
+                # os.path.abspath(os.path.join(local, file_))
+                # os.path.join(rel_folder,
+                #              os.path.basename(os.path.normpath(local)),
+                #              file_)
+                os.path.join(os.path.basename(os.path.normpath(local)), file_)
+                for file_ in files
+                if os.path.splitext(file_)[-1] in EXTENSIONS
+            ]
+            audio_files = [
+                os.path.abspath(os.path.join(local, file_))
+                for file_ in files
+                if os.path.splitext(file_)[-1] in EXTENSIONS
+            ]
+            if not audio_files or len(audio_files) != len(audio_rel_files):
+                continue
+
+            for audio_rel_file, audio_file in zip(audio_rel_files, audio_files):
+                try:
+                    pl_item = (
+                        EasyMP3(audio_file)
+                        if ".mp3" in audio_file
+                        else FLAC(audio_file)
+                    )
+                    title = pl_item["TITLE"][0]
+                    artist = pl_item["ARTIST"][0]
+                    length = int(pl_item.info.length)
+                    index = "#EXTINF:{}, {} - {}\n{}".format(
+                        length, artist, title, audio_rel_file
+                    )
+                except:  # noqa
+                    continue
+                track_list.append(index)
+
+        if len(track_list) > 1:
+            with open(os.path.join(pl_directory, pl_name), "w") as pl:
+                pl.write("\n\n".join(track_list))
+
+    def _smart_discography_filter(
+        self, contents: list, save_space: bool = False, skip_extras: bool = False
+    ) -> list:
+        """When downloading some artists' discography, many random and spam-like
+        albums can get downloaded. This helps filter those out to just get the good stuff.
+
+        This function removes:
+            * albums by other artists, which may contain a feature from the requested artist
+            * duplicate albums in different qualities
+            * (optionally) removes collector's, deluxe, live albums
+
+        :param list contents: contents returned by qobuz API
+        :param bool save_space: choose highest bit depth, lowest sampling rate
+        :param bool remove_extras: remove albums with extra material (i.e. live, deluxe,...)
+        :returns: filtered items list
+        """
+
+        # for debugging
+        def print_album(album: dict) -> None:
+            logger.debug(
+                f"{album['title']} - {album.get('version', '~~')} ({album['maximum_bit_depth']}/{album['maximum_sampling_rate']} by {album['artist']['name']}) {album['id']}"
+            )
+
+        TYPE_REGEXES = {
+            "remaster": r"(?i)(re)?master(ed)?",
+            "extra": r"(?i)(anniversary|deluxe|live|collector|demo|expanded)",
+        }
+
+        def is_type(album_t: str, album: dict) -> bool:
+            """Check if album is of type `album_t`"""
+            version = album.get("version", "")
+            title = album.get("title", "")
+            regex = TYPE_REGEXES[album_t]
+            return re.search(regex, f"{title} {version}") is not None
+
+        def essence(album: dict) -> str:
+            """Ignore text in parens/brackets, return all lowercase.
+            Used to group two albums that may be named similarly, but not exactly
+            the same.
+            """
+            r = re.match(r"([^\(]+)(?:\s*[\(\[][^\)][\)\]])*", album)
+            return r.group(1).strip().lower()
+
+        requested_artist = contents[0]["name"]
+        items = [item["albums"]["items"] for item in contents][0]
+
+        # use dicts to group duplicate albums together by title
+        title_grouped = dict()
+        for item in items:
+            if (t := essence(item["title"])) not in title_grouped:
+                title_grouped[t] = []
+            title_grouped[t].append(item)
+
+        items = []
+        for albums in title_grouped.values():
+            best_bit_depth = max(a["maximum_bit_depth"] for a in albums)
+            get_best = min if save_space else max
+            best_sampling_rate = get_best(
+                a["maximum_sampling_rate"]
+                for a in albums
+                if a["maximum_bit_depth"] == best_bit_depth
+            )
+            remaster_exists = any(is_type("remaster", a) for a in albums)
+
+            def is_valid(album: dict) -> bool:
+                return (
+                    album["maximum_bit_depth"] == best_bit_depth
+                    and album["maximum_sampling_rate"] == best_sampling_rate
+                    and album["artist"]["name"] == requested_artist
+                    and not (  # states that are not allowed
+                        (remaster_exists and not is_type("remaster", album))
+                        or (skip_extras and is_type("extra", album))
+                    )
+                )
+
+            filtered = tuple(filter(is_valid, albums))
+            # most of the time, len is 0 or 1.
+            # if greater, it is a complete duplicate,
+            # so it doesn't matter which is chosen
+            if len(filtered) >= 1:
+                items.append(filtered[0])
+
+        return items

qobuz_dl/db.py

@@ -1,7 +1,7 @@
 import logging
 import sqlite3
 
-from qobuz_dl.color import YELLOW, RED
+from qobuz_dl.color import RED, YELLOW
 
 logger = logging.getLogger(__name__)
 

qobuz_dl/downloader.py

@@ -3,11 +3,11 @@ import os
 from typing import Tuple
 
 import requests
-from pathvalidate import sanitize_filename, sanitize_filepath
+from pathvalidate import sanitize_filename
 from tqdm import tqdm
 
 import qobuz_dl.metadata as metadata
-from qobuz_dl.color import OFF, GREEN, RED, YELLOW, CYAN
+from qobuz_dl.color import CYAN, GREEN, OFF, RED, YELLOW
 from qobuz_dl.exceptions import NonStreamable
 
 QL_DOWNGRADE = "FormatRestrictedByFormatAvailability"
@@ -23,319 +23,67 @@ DEFAULT_FORMATS = {
     ],
 }
 
-DEFAULT_FOLDER = "{artist} - {album} ({year}) [{bit_depth}B-{sampling_rate}kHz]"
-DEFAULT_TRACK = "{tracknumber}. {tracktitle}"
-
 logger = logging.getLogger(__name__)
 
 
-class Download:
-    def __init__(
-        self,
-        client,
-        item_id: str,
-        path: str,
-        quality: int,
-        embed_art: bool = False,
-        albums_only: bool = False,
-        downgrade_quality: bool = False,
-        cover_og_quality: bool = False,
-        no_cover: bool = False,
-        folder_format=None,
-        track_format=None,
-    ):
-        self.client = client
-        self.item_id = item_id
-        self.path = path
-        self.quality = quality
-        self.albums_only = albums_only
-        self.embed_art = embed_art
-        self.downgrade_quality = downgrade_quality
-        self.cover_og_quality = cover_og_quality
-        self.no_cover = no_cover
-        self.folder_format = folder_format or DEFAULT_FOLDER
-        self.track_format = track_format or DEFAULT_TRACK
-
-    def download_id_by_type(self, track=True):
-        if not track:
-            self.download_release()
-        else:
-            self.download_track()
-
-    def download_release(self):
-        count = 0
-        meta = self.client.get_album_meta(self.item_id)
-
-        if not meta.get("streamable"):
-            raise NonStreamable("This release is not streamable")
-
-        if self.albums_only and (
-            meta.get("release_type") != "album"
-            or meta.get("artist").get("name") == "Various Artists"
-        ):
-            logger.info(f'{OFF}Ignoring Single/EP/VA: {meta.get("title", "n/a")}')
-            return
-
-        album_title = _get_title(meta)
-
-        format_info = self._get_format(meta)
-        file_format, quality_met, bit_depth, sampling_rate = format_info
-
-        if not self.downgrade_quality and not quality_met:
-            logger.info(
-                f"{OFF}Skipping {album_title} as it doesn't meet quality requirement"
-            )
-            return
-
-        logger.info(
-            f"\n{YELLOW}Downloading: {album_title}\nQuality: {file_format}"
-            f" ({bit_depth}/{sampling_rate})\n"
-        )
-        album_attr = self._get_album_attr(
-            meta, album_title, file_format, bit_depth, sampling_rate
-        )
-        folder_format, track_format = _clean_format_str(
-            self.folder_format, self.track_format, file_format
-        )
-        sanitized_title = sanitize_filepath(folder_format.format(**album_attr))
-        dirn = os.path.join(self.path, sanitized_title)
-        os.makedirs(dirn, exist_ok=True)
-
-        if self.no_cover:
-            logger.info(f"{OFF}Skipping cover")
-        else:
-            _get_extra(meta["image"]["large"], dirn, og_quality=self.cover_og_quality)
-
-        if "goodies" in meta:
-            try:
-                _get_extra(meta["goodies"][0]["url"], dirn, "booklet.pdf")
-            except:  # noqa
-                pass
-        media_numbers = [track["media_number"] for track in meta["tracks"]["items"]]
-        is_multiple = True if len([*{*media_numbers}]) > 1 else False
-        for i in meta["tracks"]["items"]:
-            parse = self.client.get_track_url(i["id"], fmt_id=self.quality)
-            if "sample" not in parse and parse["sampling_rate"]:
-                is_mp3 = True if int(self.quality) == 5 else False
-                self._download_and_tag(
-                    dirn,
-                    count,
-                    parse,
-                    i,
-                    meta,
-                    False,
-                    is_mp3,
-                    i["media_number"] if is_multiple else None,
-                )
-            else:
-                logger.info(f"{OFF}Demo. Skipping")
-            count = count + 1
-        logger.info(f"{GREEN}Completed")
-
-    def download_track(self):
-        parse = self.client.get_track_url(self.item_id, self.quality)
-
-        if "sample" not in parse and parse["sampling_rate"]:
-            meta = self.client.get_track_meta(self.item_id)
-            track_title = _get_title(meta)
-            artist = _safe_get(meta, "performer", "name")
-            logger.info(f"\n{YELLOW}Downloading: {artist} - {track_title}")
-            format_info = self._get_format(meta, is_track_id=True, track_url_dict=parse)
-            file_format, quality_met, bit_depth, sampling_rate = format_info
-
-            folder_format, track_format = _clean_format_str(
-                self.folder_format, self.track_format, str(bit_depth)
-            )
-
-            if not self.downgrade_quality and not quality_met:
-                logger.info(
-                    f"{OFF}Skipping {track_title} as it doesn't "
-                    "meet quality requirement"
-                )
-                return
-            track_attr = self._get_track_attr(
-                meta, track_title, bit_depth, sampling_rate
-            )
-            sanitized_title = sanitize_filepath(folder_format.format(**track_attr))
-
-            dirn = os.path.join(self.path, sanitized_title)
-            os.makedirs(dirn, exist_ok=True)
-            if self.no_cover:
-                logger.info(f"{OFF}Skipping cover")
-            else:
-                _get_extra(
-                    meta["album"]["image"]["large"],
-                    dirn,
-                    og_quality=self.cover_og_quality,
-                )
-            is_mp3 = True if int(self.quality) == 5 else False
-            self._download_and_tag(
-                dirn,
-                1,
-                parse,
-                meta,
-                meta,
-                True,
-                is_mp3,
-                False,
-            )
-        else:
-            logger.info(f"{OFF}Demo. Skipping")
-        logger.info(f"{GREEN}Completed")
-
-    def _download_and_tag(
-        self,
-        root_dir,
-        tmp_count,
-        track_url_dict,
-        track_metadata,
-        album_or_track_metadata,
-        is_track,
-        is_mp3,
-        multiple=None,
-    ):
-        extension = ".mp3" if is_mp3 else ".flac"
-
-        try:
-            url = track_url_dict["url"]
-        except KeyError:
-            logger.info(f"{OFF}Track not available for download")
-            return
-
-        if multiple:
-            root_dir = os.path.join(root_dir, f"Disc {multiple}")
-            os.makedirs(root_dir, exist_ok=True)
-
-        filename = os.path.join(root_dir, f".{tmp_count:02}.tmp")
-
-        # Determine the filename
-        track_title = track_metadata.get("title")
-        artist = _safe_get(track_metadata, "performer", "name")
-        filename_attr = self._get_filename_attr(artist, track_metadata, track_title)
-
-        # track_format is a format string
-        # e.g. '{tracknumber}. {artist} - {tracktitle}'
-        formatted_path = sanitize_filename(self.track_format.format(**filename_attr))
-        final_file = os.path.join(root_dir, formatted_path)[:250] + extension
-
-        if os.path.isfile(final_file):
-            logger.info(f"{OFF}{track_title} was already downloaded")
-            return
-
-        tqdm_download(url, filename, filename)
-        tag_function = metadata.tag_mp3 if is_mp3 else metadata.tag_flac
-        try:
-            tag_function(
-                filename,
-                root_dir,
-                final_file,
-                track_metadata,
-                album_or_track_metadata,
-                is_track,
-                self.embed_art,
-            )
-        except Exception as e:
-            logger.error(f"{RED}Error tagging the file: {e}", exc_info=True)
-
-    @staticmethod
-    def _get_filename_attr(artist, track_metadata, track_title):
-        return {
-            "artist": artist,
-            "albumartist": _safe_get(
-                track_metadata, "album", "artist", "name", default=artist
-            ),
-            "bit_depth": track_metadata["maximum_bit_depth"],
-            "sampling_rate": track_metadata["maximum_sampling_rate"],
-            "tracktitle": track_title,
-            "version": track_metadata.get("version"),
-            "tracknumber": f"{track_metadata['track_number']:02}",
-        }
-
-    @staticmethod
-    def _get_track_attr(meta, track_title, bit_depth, sampling_rate):
-        return {
-            "album": sanitize_filename(meta["album"]["title"]),
-            "artist": sanitize_filename(meta["album"]["artist"]["name"]),
-            "tracktitle": track_title,
-            "year": meta["album"]["release_date_original"].split("-")[0],
-            "bit_depth": bit_depth,
-            "sampling_rate": sampling_rate,
-        }
-
-    @staticmethod
-    def _get_album_attr(meta, album_title, file_format, bit_depth, sampling_rate):
-        return {
-            "artist": sanitize_filename(meta["artist"]["name"]),
-            "album": sanitize_filename(album_title),
-            "year": meta["release_date_original"].split("-")[0],
-            "format": file_format,
-            "bit_depth": bit_depth,
-            "sampling_rate": sampling_rate,
-        }
-
-    def _get_format(self, item_dict, is_track_id=False, track_url_dict=None):
-        quality_met = True
-        if int(self.quality) == 5:
-            return ("MP3", quality_met, None, None)
-        track_dict = item_dict
-        if not is_track_id:
-            track_dict = item_dict["tracks"]["items"][0]
-
-        try:
-            new_track_dict = (
-                self.client.get_track_url(track_dict["id"], fmt_id=self.quality)
-                if not track_url_dict
-                else track_url_dict
-            )
-            restrictions = new_track_dict.get("restrictions")
-            if isinstance(restrictions, list):
-                if any(
-                    restriction.get("code") == QL_DOWNGRADE
-                    for restriction in restrictions
-                ):
-                    quality_met = False
-
-            return (
-                "FLAC",
-                quality_met,
-                new_track_dict["bit_depth"],
-                new_track_dict["sampling_rate"],
-            )
-        except (KeyError, requests.exceptions.HTTPError):
-            return ("Unknown", quality_met, None, None)
-
-
-def tqdm_download(url, fname, desc):
+def tqdm_download(url, fname, track_name):
     r = requests.get(url, allow_redirects=True, stream=True)
     total = int(r.headers.get("content-length", 0))
-    download_size = 0
     with open(fname, "wb") as file, tqdm(
         total=total,
         unit="iB",
         unit_scale=True,
         unit_divisor=1024,
-        desc=desc,
+        desc=track_name,
         bar_format=CYAN + "{n_fmt}/{total_fmt} /// {desc}",
     ) as bar:
         for data in r.iter_content(chunk_size=1024):
             size = file.write(data)
             bar.update(size)
-            download_size += size
-
-    if total != download_size:
-        # https://stackoverflow.com/questions/69919912/requests-iter-content-thinks-file-is-complete-but-its-not
-        raise ConnectionError("File download was interrupted for " + fname)
 
 
-def _get_description(item: dict, track_title, multiple=None):
+def get_description(u: dict, track_title, multiple=None):
     downloading_title = f"{track_title} "
-    f'[{item["bit_depth"]}/{item["sampling_rate"]}]'
+    f'[{u["bit_depth"]}/{u["sampling_rate"]}]'
     if multiple:
         downloading_title = f"[Disc {multiple}] {downloading_title}"
     return downloading_title
 
 
-def _get_title(item_dict):
+def get_format(
+    client, item_dict, quality, is_track_id=False, track_url_dict=None
+) -> Tuple[str, bool, int, int]:
+    quality_met = True
+    if int(quality) == 5:
+        return ("MP3", quality_met, None, None)
+    track_dict = item_dict
+    if not is_track_id:
+        track_dict = item_dict["tracks"]["items"][0]
+
+    try:
+        new_track_dict = (
+            client.get_track_url(track_dict["id"], quality)
+            if not track_url_dict
+            else track_url_dict
+        )
+        restrictions = new_track_dict.get("restrictions")
+        if isinstance(restrictions, list):
+            if any(
+                restriction.get("code") == QL_DOWNGRADE for restriction in restrictions
+            ):
+                quality_met = False
+
+        return (
+            "FLAC",
+            quality_met,
+            new_track_dict["bit_depth"],
+            new_track_dict["sampling_rate"],
+        )
+    except (KeyError, requests.exceptions.HTTPError):
+        return ("Unknown", quality_met, None, None)
+
+
+def get_title(item_dict):
     album_title = item_dict["title"]
     version = item_dict.get("version")
     if version:
@@ -347,18 +95,265 @@ def _get_title(item_dict):
     return album_title
 
 
-def _get_extra(item, dirn, extra="cover.jpg", og_quality=False):
+def get_extra(i, dirn, extra="cover.jpg", og_quality=False):
     extra_file = os.path.join(dirn, extra)
     if os.path.isfile(extra_file):
         logger.info(f"{OFF}{extra} was already downloaded")
         return
     tqdm_download(
-        item.replace("_600.", "_org.") if og_quality else item,
+        i.replace("_600.", "_org.") if og_quality else i,
         extra_file,
         extra,
     )
 
 
+# Download and tag a file
+def download_and_tag(
+    root_dir,
+    tmp_count,
+    track_url_dict,
+    track_metadata,
+    album_or_track_metadata,
+    is_track,
+    is_mp3,
+    embed_art=False,
+    multiple=None,
+    track_format="{tracknumber}. {tracktitle}",
+):
+    """
+    Download and tag a file
+
+    :param str root_dir: Root directory where the track will be stored
+    :param int tmp_count: Temporal download file number
+    :param dict track_url_dict: get_track_url dictionary from Qobuz client
+    :param dict track_metadata: Track item dictionary from Qobuz client
+    :param dict album_or_track_metadata: Album/track dict from Qobuz client
+    :param bool is_track
+    :param bool is_mp3
+    :param bool embed_art: Embed cover art into file (FLAC-only)
+    :param str track_format format-string that determines file naming
+    :param multiple: Multiple disc integer
+    :type multiple: integer or None
+    """
+
+    extension = ".mp3" if is_mp3 else ".flac"
+
+    try:
+        url = track_url_dict["url"]
+    except KeyError:
+        logger.info(f"{OFF}Track not available for download")
+        return
+
+    if multiple:
+        root_dir = os.path.join(root_dir, f"Disc {multiple}")
+        os.makedirs(root_dir, exist_ok=True)
+
+    filename = os.path.join(root_dir, f".{tmp_count:02}.tmp")
+
+    # Determine the filename
+    track_title = track_metadata.get("title")
+    artist = _safe_get(track_metadata, "performer", "name")
+    filename_attr = {
+        "artist": artist,
+        "albumartist": _safe_get(
+            track_metadata, "album", "artist", "name", default=artist
+        ),
+        "bit_depth": track_metadata["maximum_bit_depth"],
+        "sampling_rate": track_metadata["maximum_sampling_rate"],
+        "tracktitle": track_title,
+        "version": track_metadata.get("version"),
+        "tracknumber": f"{track_metadata['track_number']:02}",
+    }
+    # track_format is a format string
+    # e.g. '{tracknumber}. {artist} - {tracktitle}'
+    formatted_path = sanitize_filename(track_format.format(**filename_attr))
+    final_file = os.path.join(root_dir, formatted_path)[:250] + extension
+
+    if os.path.isfile(final_file):
+        logger.info(f"{OFF}{track_title} was already downloaded")
+        return
+
+    desc = get_description(track_url_dict, track_title, multiple)
+    tqdm_download(url, filename, desc)
+    tag_function = metadata.tag_mp3 if is_mp3 else metadata.tag_flac
+    try:
+        tag_function(
+            filename,
+            root_dir,
+            final_file,
+            track_metadata,
+            album_or_track_metadata,
+            is_track,
+            embed_art,
+        )
+    except Exception as e:
+        logger.error(f"{RED}Error tagging the file: {e}", exc_info=True)
+
+
+def download_id_by_type(
+    client,
+    item_id,
+    path,
+    quality,
+    album=False,
+    embed_art=False,
+    albums_only=False,
+    downgrade_quality=True,
+    cover_og_quality=False,
+    no_cover=False,
+    folder_format="{artist} - {album} ({year}) " "[{bit_depth}B-{sampling_rate}kHz]",
+    track_format="{tracknumber}. {tracktitle}",
+):
+    """
+    Download and get metadata by ID and type (album or track)
+
+    :param Qopy client: qopy Client
+    :param int item_id: Qobuz item id
+    :param str path: The root directory where the item will be downloaded
+    :param int quality: Audio quality (5, 6, 7, 27)
+    :param bool album: album type or not
+    :param embed_art album: Embed cover art into files
+    :param bool albums_only: Ignore Singles, EPs and VA releases
+    :param bool downgrade: Skip releases not available in set quality
+    :param bool cover_og_quality: Download cover in its original quality
+    :param bool no_cover: Don't download cover art
+    :param str folder_format: format string that determines folder naming
+    :param str track_format: format string that determines track naming
+    """
+    count = 0
+
+    if album:
+        meta = client.get_album_meta(item_id)
+
+        if not meta.get("streamable"):
+            raise NonStreamable("This release is not streamable")
+
+        if albums_only and (
+            meta.get("release_type") != "album"
+            or meta.get("artist").get("name") == "Various Artists"
+        ):
+            logger.info(f'{OFF}Ignoring Single/EP/VA: {meta.get("title", "")}')
+            return
+
+        album_title = get_title(meta)
+
+        format_info = get_format(client, meta, quality)
+        file_format, quality_met, bit_depth, sampling_rate = format_info
+
+        if not downgrade_quality and not quality_met:
+            logger.info(
+                f"{OFF}Skipping {album_title} as it doesn't " "meet quality requirement"
+            )
+            return
+
+        logger.info(
+            f"\n{YELLOW}Downloading: {album_title}\nQuality: {file_format} ({bit_depth}/{sampling_rate})\n"
+        )
+        album_attr = {
+            "artist": meta["artist"]["name"],
+            "album": album_title,
+            "year": meta["release_date_original"].split("-")[0],
+            "format": file_format,
+            "bit_depth": bit_depth,
+            "sampling_rate": sampling_rate,
+        }
+        folder_format, track_format = _clean_format_str(
+            folder_format, track_format, file_format
+        )
+        sanitized_title = sanitize_filename(folder_format.format(**album_attr))
+        dirn = os.path.join(path, sanitized_title)
+        os.makedirs(dirn, exist_ok=True)
+
+        if no_cover:
+            logger.info(f"{OFF}Skipping cover")
+        else:
+            get_extra(meta["image"]["large"], dirn, og_quality=cover_og_quality)
+
+        if "goodies" in meta:
+            try:
+                get_extra(meta["goodies"][0]["url"], dirn, "booklet.pdf")
+            except:  # noqa
+                pass
+        media_numbers = [track["media_number"] for track in meta["tracks"]["items"]]
+        is_multiple = True if len([*{*media_numbers}]) > 1 else False
+        for i in meta["tracks"]["items"]:
+            parse = client.get_track_url(i["id"], quality)
+            if "sample" not in parse and parse["sampling_rate"]:
+                is_mp3 = True if int(quality) == 5 else False
+                download_and_tag(
+                    dirn,
+                    count,
+                    parse,
+                    i,
+                    meta,
+                    False,
+                    is_mp3,
+                    embed_art,
+                    i["media_number"] if is_multiple else None,
+                    track_format=track_format,
+                )
+            else:
+                logger.info(f"{OFF}Demo. Skipping")
+            count = count + 1
+    else:
+        parse = client.get_track_url(item_id, quality)
+
+        if "sample" not in parse and parse["sampling_rate"]:
+            meta = client.get_track_meta(item_id)
+            track_title = get_title(meta)
+            logger.info(f"\n{YELLOW}Downloading: {track_title}")
+            format_info = get_format(
+                client, meta, quality, is_track_id=True, track_url_dict=parse
+            )
+            file_format, quality_met, bit_depth, sampling_rate = format_info
+
+            folder_format, track_format = _clean_format_str(
+                folder_format, track_format, bit_depth
+            )
+
+            if not downgrade_quality and not quality_met:
+                logger.info(
+                    f"{OFF}Skipping {track_title} as it doesn't "
+                    "meet quality requirement"
+                )
+                return
+            track_attr = {
+                "artist": meta["album"]["artist"]["name"],
+                "tracktitle": track_title,
+                "year": meta["album"]["release_date_original"].split("-")[0],
+                "bit_depth": bit_depth,
+                "sampling_rate": sampling_rate,
+            }
+            sanitized_title = sanitize_filename(folder_format.format(**track_attr))
+
+            dirn = os.path.join(path, sanitized_title)
+            os.makedirs(dirn, exist_ok=True)
+            if no_cover:
+                logger.info(f"{OFF}Skipping cover")
+            else:
+                get_extra(
+                    meta["album"]["image"]["large"], dirn, og_quality=cover_og_quality
+                )
+            is_mp3 = True if int(quality) == 5 else False
+            download_and_tag(
+                dirn,
+                count,
+                parse,
+                meta,
+                meta,
+                True,
+                is_mp3,
+                embed_art,
+                track_format=track_format,
+            )
+        else:
+            logger.info(f"{OFF}Demo. Skipping")
+    logger.info(f"{GREEN}Completed")
+
+
+# ----------- Utilities -----------
+
+
 def _clean_format_str(folder: str, track: str, file_format: str) -> Tuple[str, str]:
     """Cleans up the format strings, avoids errors
     with MP3 files.

qobuz_dl/metadata.py

@@ -1,9 +1,9 @@
-import re
-import os
 import logging
+import os
+import re
 
-from mutagen.flac import FLAC, Picture
 import mutagen.id3 as id3
+from mutagen.flac import FLAC, Picture
 from mutagen.id3 import ID3NoHeaderError
 
 logger = logging.getLogger(__name__)
@@ -15,39 +15,22 @@ COPYRIGHT, PHON_COPYRIGHT = "\u2117", "\u00a9"
 # and the file won't be tagged
 FLAC_MAX_BLOCKSIZE = 16777215
 
-ID3_LEGEND = {
-    "album": id3.TALB,
-    "albumartist": id3.TPE2,
-    "artist": id3.TPE1,
-    "comment": id3.COMM,
-    "composer": id3.TCOM,
-    "copyright": id3.TCOP,
-    "date": id3.TDAT,
-    "genre": id3.TCON,
-    "isrc": id3.TSRC,
-    "label": id3.TPUB,
-    "performer": id3.TOPE,
-    "title": id3.TIT2,
-    "year": id3.TYER,
-}
 
-
-def _get_title(track_dict):
+def get_title(track_dict):
     title = track_dict["title"]
     version = track_dict.get("version")
     if version:
         title = f"{title} ({version})"
     # for classical works
     if track_dict.get("work"):
-        title = f"{track_dict['work']}: {title}"
+        title = "{}: {}".format(track_dict["work"], title)
 
     return title
 
 
 def _format_copyright(s: str) -> str:
-    if s:
-        s = s.replace("(P)", PHON_COPYRIGHT)
-        s = s.replace("(C)", COPYRIGHT)
+    s = s.replace("(P)", PHON_COPYRIGHT)
+    s = s.replace("(C)", COPYRIGHT)
     return s
 
 
@@ -63,54 +46,8 @@ def _format_genres(genres: list) -> str:
     return ", ".join(no_repeats)
 
 
-def _embed_flac_img(root_dir, audio: FLAC):
-    emb_image = os.path.join(root_dir, "cover.jpg")
-    multi_emb_image = os.path.join(
-        os.path.abspath(os.path.join(root_dir, os.pardir)), "cover.jpg"
-    )
-    if os.path.isfile(emb_image):
-        cover_image = emb_image
-    else:
-        cover_image = multi_emb_image
-
-    try:
-        # rest of the metadata still gets embedded
-        # when the image size is too big
-        if os.path.getsize(cover_image) > FLAC_MAX_BLOCKSIZE:
-            raise Exception(
-                "downloaded cover size too large to embed. "
-                "turn off `og_cover` to avoid error"
-            )
-
-        image = Picture()
-        image.type = 3
-        image.mime = "image/jpeg"
-        image.desc = "cover"
-        with open(cover_image, "rb") as img:
-            image.data = img.read()
-        audio.add_picture(image)
-    except Exception as e:
-        logger.error(f"Error embedding image: {e}", exc_info=True)
-
-
-def _embed_id3_img(root_dir, audio: id3.ID3):
-    emb_image = os.path.join(root_dir, "cover.jpg")
-    multi_emb_image = os.path.join(
-        os.path.abspath(os.path.join(root_dir, os.pardir)), "cover.jpg"
-    )
-    if os.path.isfile(emb_image):
-        cover_image = emb_image
-    else:
-        cover_image = multi_emb_image
-
-    with open(cover_image, "rb") as cover:
-        audio.add(id3.APIC(3, "image/jpeg", 3, "", cover.read()))
-
-
 # Use KeyError catching instead of dict.get to avoid empty tags
-def tag_flac(
-    filename, root_dir, final_name, d: dict, album, istrack=True, em_image=False
-):
+def tag_flac(filename, root_dir, final_name, d, album, istrack=True, em_image=False):
     """
     Tag a FLAC file
 
@@ -124,7 +61,7 @@ def tag_flac(
     """
     audio = FLAC(filename)
 
-    audio["TITLE"] = _get_title(d)
+    audio["TITLE"] = get_title(d)
 
     audio["TRACKNUMBER"] = str(d["track_number"])  # TRACK NUMBER
 
@@ -136,13 +73,18 @@ def tag_flac(
     except KeyError:
         pass
 
-    artist_ = d.get("performer", {}).get("name")  # TRACK ARTIST
-    if istrack:
-        audio["ARTIST"] = artist_ or d["album"]["artist"]["name"]  # TRACK ARTIST
-    else:
-        audio["ARTIST"] = artist_ or album["artist"]["name"]
+    try:
+        audio["ARTIST"] = d["performer"]["name"]  # TRACK ARTIST
+    except KeyError:
+        if istrack:
+            audio["ARTIST"] = d["album"]["artist"]["name"]  # TRACK ARTIST
+        else:
+            audio["ARTIST"] = album["artist"]["name"]
 
-    audio["LABEL"] = album.get("label", {}).get("name", "n/a")
+    try:
+        audio["LABEL"] = album["label"]["name"]
+    except KeyError:
+        pass
 
     if istrack:
         audio["GENRE"] = _format_genres(d["album"]["genres_list"])
@@ -150,17 +92,43 @@ def tag_flac(
         audio["TRACKTOTAL"] = str(d["album"]["tracks_count"])
         audio["ALBUM"] = d["album"]["title"]
         audio["DATE"] = d["album"]["release_date_original"]
-        audio["COPYRIGHT"] = _format_copyright(d.get("copyright") or "n/a")
+        audio["COPYRIGHT"] = _format_copyright(d["copyright"])
     else:
         audio["GENRE"] = _format_genres(album["genres_list"])
         audio["ALBUMARTIST"] = album["artist"]["name"]
         audio["TRACKTOTAL"] = str(album["tracks_count"])
         audio["ALBUM"] = album["title"]
         audio["DATE"] = album["release_date_original"]
-        audio["COPYRIGHT"] = _format_copyright(album.get("copyright") or "n/a")
+        audio["COPYRIGHT"] = _format_copyright(album["copyright"])
 
     if em_image:
-        _embed_flac_img(root_dir, audio)
+        emb_image = os.path.join(root_dir, "cover.jpg")
+        multi_emb_image = os.path.join(
+            os.path.abspath(os.path.join(root_dir, os.pardir)), "cover.jpg"
+        )
+        if os.path.isfile(emb_image):
+            cover_image = emb_image
+        else:
+            cover_image = multi_emb_image
+
+        try:
+            # rest of the metadata still gets embedded
+            # when the image size is too big
+            if os.path.getsize(cover_image) > FLAC_MAX_BLOCKSIZE:
+                raise Exception(
+                    "downloaded cover size too large to embed. "
+                    "turn off `og_cover` to avoid error"
+                )
+
+            image = Picture()
+            image.type = 3
+            image.mime = "image/jpeg"
+            image.desc = "cover"
+            with open(cover_image, "rb") as img:
+                image.data = img.read()
+            audio.add_picture(image)
+        except Exception as e:
+            logger.error(f"Error embedding image: {e}", exc_info=True)
 
     audio.save()
     os.rename(filename, final_name)
@@ -178,6 +146,21 @@ def tag_mp3(filename, root_dir, final_name, d, album, istrack=True, em_image=Fal
     :param bool em_image: Embed cover art into file
     """
 
+    id3_legend = {
+        "album": id3.TALB,
+        "albumartist": id3.TPE2,
+        "artist": id3.TPE1,
+        "comment": id3.COMM,
+        "composer": id3.TCOM,
+        "copyright": id3.TCOP,
+        "date": id3.TDAT,
+        "genre": id3.TCON,
+        "isrc": id3.TSRC,
+        "label": id3.TPUB,
+        "performer": id3.TOPE,
+        "title": id3.TIT2,
+        "year": id3.TYER,
+    }
     try:
         audio = id3.ID3(filename)
     except ID3NoHeaderError:
@@ -185,17 +168,19 @@ def tag_mp3(filename, root_dir, final_name, d, album, istrack=True, em_image=Fal
 
     # temporarily holds metadata
     tags = dict()
-    tags["title"] = _get_title(d)
+    tags["title"] = get_title(d)
     try:
         tags["label"] = album["label"]["name"]
     except KeyError:
         pass
 
-    artist_ = d.get("performer", {}).get("name")  # TRACK ARTIST
-    if istrack:
-        tags["artist"] = artist_ or d["album"]["artist"]["name"]  # TRACK ARTIST
-    else:
-        tags["artist"] = artist_ or album["artist"]["name"]
+    try:
+        tags["artist"] = d["performer"]["name"]
+    except KeyError:
+        if istrack:
+            tags["artist"] = d["album"]["artist"]["name"]
+        else:
+            tags["artist"] = album["artist"]["name"]
 
     if istrack:
         tags["genre"] = _format_genres(d["album"]["genres_list"])
@@ -219,11 +204,21 @@ def tag_mp3(filename, root_dir, final_name, d, album, istrack=True, em_image=Fal
 
     # write metadata in `tags` to file
     for k, v in tags.items():
-        id3tag = ID3_LEGEND[k]
+        id3tag = id3_legend[k]
         audio[id3tag.__name__] = id3tag(encoding=3, text=v)
 
     if em_image:
-        _embed_id3_img(root_dir, audio)
+        emb_image = os.path.join(root_dir, "cover.jpg")
+        multi_emb_image = os.path.join(
+            os.path.abspath(os.path.join(root_dir, os.pardir)), "cover.jpg"
+        )
+        if os.path.isfile(emb_image):
+            cover_image = emb_image
+        else:
+            cover_image = multi_emb_image
+
+        with open(cover_image, "rb") as cover:
+            audio.add(id3.APIC(3, "image/jpeg", 3, "", cover.read()))
 
     audio.save(filename, "v2_version=3")
     os.rename(filename, final_name)

qobuz_dl/qopy.py

@@ -8,6 +8,7 @@ import time
 
 import requests
 
+from qobuz_dl.color import GREEN, YELLOW
 from qobuz_dl.exceptions import (
     AuthenticationError,
     IneligibleError,
@@ -15,7 +16,6 @@ from qobuz_dl.exceptions import (
     InvalidAppSecretError,
     InvalidQuality,
 )
-from qobuz_dl.color import GREEN, YELLOW
 
 RESET = "Reset your credentials with 'qobuz-dl -r'"
 
@@ -26,18 +26,15 @@ class Client:
     def __init__(self, email, pwd, app_id, secrets):
         logger.info(f"{YELLOW}Logging...")
         self.secrets = secrets
-        self.id = str(app_id)
+        self.id = app_id
         self.session = requests.Session()
         self.session.headers.update(
             {
                 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:83.0) Gecko/20100101 Firefox/83.0",
                 "X-App-Id": self.id,
-                "Content-Type": "application/json;charset=UTF-8"
-
             }
         )
         self.base = "https://www.qobuz.com/api.json/0.2/"
-        self.sec = None
         self.auth(email, pwd)
         self.cfg_setup()
 
@@ -74,15 +71,13 @@ class Client:
                 "offset": kwargs["offset"],
                 "extra": "albums",
             }
-        elif epoint == "favorite/getUserFavorites":
+        elif epoint == "userLibrary/getAlbumsList":
             unix = time.time()
-            # r_sig = "userLibrarygetAlbumsList" + str(unix) + kwargs["sec"]
-            r_sig = "favoritegetUserFavorites" + str(unix) + kwargs["sec"]
+            r_sig = "userLibrarygetAlbumsList" + str(unix) + kwargs["sec"]
             r_sig_hashed = hashlib.md5(r_sig.encode("utf-8")).hexdigest()
             params = {
                 "app_id": self.id,
                 "user_auth_token": self.uat,
-                "type": "albums",
                 "request_ts": unix,
                 "request_sig": r_sig_hashed,
             }
@@ -93,7 +88,7 @@ class Client:
             if int(fmt_id) not in (5, 6, 7, 27):
                 raise InvalidQuality("Invalid quality id: choose between 5, 6, 7 or 27")
             r_sig = "trackgetFileUrlformat_id{}intentstreamtrack_id{}{}{}".format(
-                fmt_id, track_id, unix, kwargs.get("sec", self.sec)
+                fmt_id, track_id, unix, self.sec
             )
             r_sig_hashed = hashlib.md5(r_sig.encode("utf-8")).hexdigest()
             params = {
@@ -113,12 +108,9 @@ class Client:
                 raise InvalidAppIdError("Invalid app id.\n" + RESET)
             else:
                 logger.info(f"{GREEN}Logged: OK")
-        elif (
-            epoint in ["track/getFileUrl", "favorite/getUserFavorites"]
-            and r.status_code == 400
-        ):
-            raise InvalidAppSecretError(f"Invalid app secret: {r.json()}.\n" + RESET)
-
+        elif epoint in ["track/getFileUrl", "userLibrary/getAlbumsList"]:
+            if r.status_code == 400:
+                raise InvalidAppSecretError("Invalid app secret.\n" + RESET)
         r.raise_for_status()
         return r.json()
 
@@ -197,20 +189,16 @@ class Client:
 
     def test_secret(self, sec):
         try:
-            self.api_call("track/getFileUrl", id=5966783, fmt_id=5, sec=sec)
+            r = self.api_call("userLibrary/getAlbumsList", sec=sec)
             return True
         except InvalidAppSecretError:
             return False
 
     def cfg_setup(self):
+        logging.debug(self.secrets)
         for secret in self.secrets:
-            # Falsy secrets
-            if not secret:
-                continue
-
             if self.test_secret(secret):
                 self.sec = secret
                 break
-
-        if self.sec is None:
-            raise InvalidAppSecretError("Can't find any valid app secret.\n" + RESET)
+        if not hasattr(self, "sec"):
+            raise InvalidAppSecretError("Invalid app secret.\n" + RESET)

qobuz_dl/spoofbuz.py

@@ -0,0 +1,51 @@
+import base64
+import re
+from collections import OrderedDict
+
+import requests
+
+
+class Spoofer:
+    def __init__(self):
+        self.seed_timezone_regex = r'[a-z]\.initialSeed\("(?P<seed>[\w=]+)",window\.utimezone\.(?P<timezone>[a-z]+)\)'
+        # note: {timezones} should be replaced with every capitalized timezone joined by a |
+        self.info_extras_regex = r'name:"\w+/(?P<timezone>{timezones})",info:"(?P<info>[\w=]+)",extras:"(?P<extras>[\w=]+)"'
+        self.appId_regex = r'{app_id:"(?P<app_id>\d{9})",app_secret:"\w{32}",base_port:"80",base_url:"https://www\.qobuz\.com",base_method:"/api\.json/0\.2/"},n\.base_url="https://play\.qobuz\.com"'
+        login_page_request = requests.get("https://play.qobuz.com/login")
+        login_page = login_page_request.text
+        bundle_url_match = re.search(
+            r'<script src="(/resources/\d+\.\d+\.\d+-[a-z]\d{3}/bundle\.js)"></script>',
+            login_page,
+        )
+        bundle_url = bundle_url_match.group(1)
+        bundle_req = requests.get("https://play.qobuz.com" + bundle_url)
+        self.bundle = bundle_req.text
+
+    def getAppId(self):
+        return re.search(self.appId_regex, self.bundle).group("app_id")
+
+    def getSecrets(self):
+        seed_matches = re.finditer(self.seed_timezone_regex, self.bundle)
+        secrets = OrderedDict()
+        for match in seed_matches:
+            seed, timezone = match.group("seed", "timezone")
+            secrets[timezone] = [seed]
+        """The code that follows switches around the first and second timezone. Why? Read on:
+            Qobuz uses two ternary (a shortened if statement) conditions that should always return false.
+            The way Javascript's ternary syntax works, the second option listed is what runs if the condition returns false.
+            Because of this, we must prioritize the *second* seed/timezone pair captured, not the first.
+        """
+        keypairs = list(secrets.items())
+        secrets.move_to_end(keypairs[1][0], last=False)
+        info_extras_regex = self.info_extras_regex.format(
+            timezones="|".join([timezone.capitalize() for timezone in secrets])
+        )
+        info_extras_matches = re.finditer(info_extras_regex, self.bundle)
+        for match in info_extras_matches:
+            timezone, info, extras = match.group("timezone", "info", "extras")
+            secrets[timezone.lower()] += [info, extras]
+        for secret_pair in secrets:
+            secrets[secret_pair] = base64.standard_b64decode(
+                "".join(secrets[secret_pair])[:-44]
+            ).decode("utf-8")
+        return secrets

qobuz_dl/utils.py

@@ -1,190 +0,0 @@
-import re
-import string
-import os
-import logging
-import time
-
-from mutagen.mp3 import EasyMP3
-from mutagen.flac import FLAC
-
-logger = logging.getLogger(__name__)
-
-EXTENSIONS = (".mp3", ".flac")
-
-
-class PartialFormatter(string.Formatter):
-    def __init__(self, missing="n/a", bad_fmt="n/a"):
-        self.missing, self.bad_fmt = missing, bad_fmt
-
-    def get_field(self, field_name, args, kwargs):
-        try:
-            val = super(PartialFormatter, self).get_field(field_name, args, kwargs)
-        except (KeyError, AttributeError):
-            val = None, field_name
-        return val
-
-    def format_field(self, value, spec):
-        if not value:
-            return self.missing
-        try:
-            return super(PartialFormatter, self).format_field(value, spec)
-        except ValueError:
-            if self.bad_fmt:
-                return self.bad_fmt
-            raise
-
-
-def make_m3u(pl_directory):
-    track_list = ["#EXTM3U"]
-    rel_folder = os.path.basename(os.path.normpath(pl_directory))
-    pl_name = rel_folder + ".m3u"
-    for local, dirs, files in os.walk(pl_directory):
-        dirs.sort()
-        audio_rel_files = [
-            os.path.join(os.path.basename(os.path.normpath(local)), file_)
-            for file_ in files
-            if os.path.splitext(file_)[-1] in EXTENSIONS
-        ]
-        audio_files = [
-            os.path.abspath(os.path.join(local, file_))
-            for file_ in files
-            if os.path.splitext(file_)[-1] in EXTENSIONS
-        ]
-        if not audio_files or len(audio_files) != len(audio_rel_files):
-            continue
-
-        for audio_rel_file, audio_file in zip(audio_rel_files, audio_files):
-            try:
-                pl_item = (
-                    EasyMP3(audio_file) if ".mp3" in audio_file else FLAC(audio_file)
-                )
-                title = pl_item["TITLE"][0]
-                artist = pl_item["ARTIST"][0]
-                length = int(pl_item.info.length)
-                index = "#EXTINF:{}, {} - {}\n{}".format(
-                    length, artist, title, audio_rel_file
-                )
-            except:  # noqa
-                continue
-            track_list.append(index)
-
-    if len(track_list) > 1:
-        with open(os.path.join(pl_directory, pl_name), "w") as pl:
-            pl.write("\n\n".join(track_list))
-
-
-def smart_discography_filter(
-    contents: list, save_space: bool = False, skip_extras: bool = False
-) -> list:
-    """When downloading some artists' discography, many random and spam-like
-    albums can get downloaded. This helps filter those out to just get the good stuff.
-
-    This function removes:
-        * albums by other artists, which may contain a feature from the requested artist
-        * duplicate albums in different qualities
-        * (optionally) removes collector's, deluxe, live albums
-
-    :param list contents: contents returned by qobuz API
-    :param bool save_space: choose highest bit depth, lowest sampling rate
-    :param bool remove_extras: remove albums with extra material (i.e. live, deluxe,...)
-    :returns: filtered items list
-    """
-
-    # for debugging
-    def print_album(album: dict) -> None:
-        logger.debug(
-            f"{album['title']} - {album.get('version', '~~')} "
-            "({album['maximum_bit_depth']}/{album['maximum_sampling_rate']}"
-            " by {album['artist']['name']}) {album['id']}"
-        )
-
-    TYPE_REGEXES = {
-        "remaster": r"(?i)(re)?master(ed)?",
-        "extra": r"(?i)(anniversary|deluxe|live|collector|demo|expanded)",
-    }
-
-    def is_type(album_t: str, album: dict) -> bool:
-        """Check if album is of type `album_t`"""
-        version = album.get("version", "")
-        title = album.get("title", "")
-        regex = TYPE_REGEXES[album_t]
-        return re.search(regex, f"{title} {version}") is not None
-
-    def essence(album: dict) -> str:
-        """Ignore text in parens/brackets, return all lowercase.
-        Used to group two albums that may be named similarly, but not exactly
-        the same.
-        """
-        r = re.match(r"([^\(]+)(?:\s*[\(\[][^\)][\)\]])*", album)
-        return r.group(1).strip().lower()
-
-    requested_artist = contents[0]["name"]
-    items = [item["albums"]["items"] for item in contents][0]
-
-    # use dicts to group duplicate albums together by title
-    title_grouped = dict()
-    for item in items:
-        title_ = essence(item["title"])
-        if title_ not in title_grouped:  # ?
-            #            if (t := essence(item["title"])) not in title_grouped:
-            title_grouped[title_] = []
-        title_grouped[title_].append(item)
-
-    items = []
-    for albums in title_grouped.values():
-        best_bit_depth = max(a["maximum_bit_depth"] for a in albums)
-        get_best = min if save_space else max
-        best_sampling_rate = get_best(
-            a["maximum_sampling_rate"]
-            for a in albums
-            if a["maximum_bit_depth"] == best_bit_depth
-        )
-        remaster_exists = any(is_type("remaster", a) for a in albums)
-
-        def is_valid(album: dict) -> bool:
-            return (
-                album["maximum_bit_depth"] == best_bit_depth
-                and album["maximum_sampling_rate"] == best_sampling_rate
-                and album["artist"]["name"] == requested_artist
-                and not (  # states that are not allowed
-                    (remaster_exists and not is_type("remaster", album))
-                    or (skip_extras and is_type("extra", album))
-                )
-            )
-
-        filtered = tuple(filter(is_valid, albums))
-        # most of the time, len is 0 or 1.
-        # if greater, it is a complete duplicate,
-        # so it doesn't matter which is chosen
-        if len(filtered) >= 1:
-            items.append(filtered[0])
-
-    return items
-
-
-def format_duration(duration):
-    return time.strftime("%H:%M:%S", time.gmtime(duration))
-
-
-def create_and_return_dir(directory):
-    fix = os.path.normpath(directory)
-    os.makedirs(fix, exist_ok=True)
-    return fix
-
-
-def get_url_info(url):
-    """Returns the type of the url and the id.
-
-    Compatible with urls of the form:
-        https://www.qobuz.com/us-en/{type}/{name}/{id}
-        https://open.qobuz.com/{type}/{id}
-        https://play.qobuz.com/{type}/{id}
-        /us-en/{type}/-/{id}
-    """
-
-    r = re.search(
-        r"(?:https:\/\/(?:w{3}|open|play)\.qobuz\.com)?(?:\/[a-z]{2}-[a-z]{2})"
-        r"?\/(album|artist|track|playlist|label)(?:\/[-\w\d]+)?\/([\w\d]+)",
-        url,
-    )
-    return r.groups()

qobuz_dl_rewrite/cli.py

@@ -0,0 +1,107 @@
+# For tests
+
+import logging
+import os
+
+import click
+
+from qobuz_dl_rewrite.config import Config
+from qobuz_dl_rewrite.constants import CACHE_DIR, CONFIG_DIR
+from qobuz_dl_rewrite.core import QobuzDL
+from qobuz_dl_rewrite.utils import init_log
+
+logger = logging.getLogger(__name__)
+
+
+def _get_config(ctx):
+    if not os.path.isdir(CONFIG_DIR):
+        os.makedirs(CONFIG_DIR)
+    if not os.path.isdir(CACHE_DIR):
+        os.makedirs(CONFIG_DIR)
+
+    config = Config(ctx.obj.get("config"))
+    config.update_from_cli(**ctx.obj)
+    return config
+
+
+# fmt: off
+@click.group()
+@click.option("--disable", metavar="PROVIDER,...", help="Disable following providers (comma separated)")
+@click.option("-q", "--quality", metavar="INT", help="Quality integer ID (5, 6, 7, 27)")
+@click.option("--embed-cover", is_flag=True, help="Embed cover art into files")
+@click.option("--no-extras", is_flag=True, help="Ignore extras")
+@click.option("--no-features", is_flag=True, help="Ignore features")
+@click.option("--studio-albums", is_flag=True, help="Ignore non-studio albums")
+@click.option("--remaster-only", is_flag=True, help="Ignore non-remastered albums")
+@click.option("--albums-only", is_flag=True, help="Ignore non-album downloads")
+@click.option("--large-cover", is_flag=True, help="Download large covers (it might fail with embed)")
+@click.option("--remove-extra-tags", default=False, is_flag=True, help="Remove extra metadata from tags and files")
+@click.option("--debug", default=False, is_flag=True, help="Enable debug logging")
+@click.option("-f", "--folder", metavar="PATH", help="Custom download folder")
+@click.option("--default-comment", metavar="COMMENT", help="Custom comment tag for audio files")
+@click.option("-c", "--config", metavar="PATH", help="Custom config file")
+@click.option("--db-file", metavar="PATH", help="Custom database file")
+@click.option("--log-file", metavar="PATH", help="Custom logfile")
+@click.option("--flush-cache", metavar="PATH", help="Flush the cache before running (only for extreme cases)")
+# TODO: add options for conversion
+@click.pass_context
+# fmt: on
+def cli(ctx, **kwargs):
+    ctx.ensure_object(dict)
+
+    for key in kwargs.keys():
+        ctx.obj[key] = kwargs.get(key)
+
+    if ctx.obj["debug"]:
+        init_log(path=ctx.obj.get("log_file"))
+    else:
+        click.secho("Debug is not enabled", fg="yellow")
+
+
+@click.command(name="dl")
+@click.argument("items", nargs=-1)
+@click.pass_context
+def download(ctx, items):
+    """
+    Download an URL, space separated URLs or a text file with URLs.
+    Mixed arguments are also supported.
+
+    Examples:
+
+        * `qobuz-dl dl https://some.url/some_type/some_id`
+
+        * `qobuz-dl dl file_with_urls.txt`
+
+        * `qobuz-dl dl URL URL URL`
+
+    Supported sources and their types:
+
+        * Deezer (album, artist, track, playlist)
+
+        * Qobuz (album, artist, label, track, playlist)
+
+        * Tidal (album, artist, track, playlist)
+    """
+    config = _get_config(ctx)
+    core = QobuzDL(config)
+    for item in items:
+        try:
+            if os.path.isfile(item):
+                core.from_txt(item)
+                click.secho(f"File input found: {item}", fg="yellow")
+            else:
+                core.handle_url(item)
+        except Exception as error:
+            logger.error(error, exc_info=True)
+            click.secho(
+                f"{type(error).__name__} raised processing {item}: {error}", fg="red"
+            )
+
+
+def main():
+    cli.add_command(download)
+    cli(obj={})
+
+
+if __name__ == "__main__":
+    main()

qobuz_dl_rewrite/clients.py

@@ -0,0 +1,503 @@
+import datetime
+import hashlib
+import logging
+import os
+import time
+from abc import ABC, abstractmethod
+from typing import Generator, Sequence, Tuple, Union
+
+import requests
+import tidalapi
+from dogpile.cache import make_region
+
+from .constants import (
+    AGENT,
+    CACHE_DIR,
+    DEEZER_MAX_Q,
+    DEEZER_Q_IDS,
+    QOBUZ_FEATURED_KEYS,
+    TIDAL_MAX_Q,
+    TIDAL_Q_IDS,
+)
+from .exceptions import (
+    AuthenticationError,
+    IneligibleError,
+    InvalidAppIdError,
+    InvalidAppSecretError,
+    InvalidQuality,
+)
+from .spoofbuz import Spoofer
+
+os.makedirs(CACHE_DIR, exist_ok=True)
+region = make_region().configure(
+    "dogpile.cache.dbm",
+    arguments={"filename": os.path.join(CACHE_DIR, "clients.db")},
+)
+
+logger = logging.getLogger(__name__)
+
+TRACK_CACHE_TIME = datetime.timedelta(weeks=2).total_seconds()
+RELEASE_CACHE_TIME = datetime.timedelta(days=1).total_seconds()
+
+# Qobuz
+QOBUZ_BASE = "https://www.qobuz.com/api.json/0.2"
+
+
+# Deezer
+DEEZER_BASE = "https://api.deezer.com"
+DEEZER_DL = "http://dz.loaderapp.info/deezer"
+
+
+# ----------- Abstract Classes -----------------
+
+
+class ClientInterface(ABC):
+    """Common API for clients of all platforms.
+
+    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
+
+    @abstractmethod
+    def search(self, query: str, media_type="album"):
+        """Search API for query.
+
+        :param query:
+        :type query: str
+        :param type_:
+        """
+        pass
+
+    @abstractmethod
+    def get(self, item_id, media_type="album"):
+        """Get metadata.
+
+        :param meta_id:
+        :param type_:
+        """
+        pass
+
+    @abstractmethod
+    def get_file_url(self, track_id, quality=6) -> Union[dict]:
+        """Get the direct download url dict for a file.
+
+        :param track_id: id of the track
+        """
+        pass
+
+    @property
+    @abstractmethod
+    def source(self):
+        pass
+
+
+# ------------- Clients -----------------
+
+
+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.
+
+        If `app_id` and `secrets` are not provided, this will run the
+        Spoofer script, which retrieves them. This will take some time,
+        so it is recommended to cache them somewhere for reuse.
+
+        :param email: email for the qobuz account
+        :type email: str
+        :param pwd: password for the qobuz account
+        :type pwd: str
+        :param kwargs: app_id: str, secrets: list, return_secrets: bool
+        """
+        if self.logged_in:
+            logger.debug("Already logged in")
+            return
+
+        if (kwargs.get("app_id") or kwargs.get("secrets")) in (None, [], ""):
+            logger.info("Fetching tokens from Qobuz")
+            spoofer = Spoofer()
+            kwargs["app_id"] = spoofer.get_app_id()
+            kwargs["secrets"] = spoofer.get_secrets()
+
+        self.app_id = str(kwargs["app_id"])  # Ensure it is a string
+        self.secrets = kwargs["secrets"]
+
+        self.session = requests.Session()
+        self.session.headers.update(
+            {
+                "User-Agent": AGENT,
+                "X-App-Id": self.app_id,
+            }
+        )
+
+        self._api_login(email, pwd)
+        logger.debug("Logged into Qobuz")
+        self._validate_secrets()
+        logger.debug("Qobuz client is ready to use")
+
+        self.logged_in = True
+
+    def get_tokens(self) -> Tuple[str, Sequence[str]]:
+        return self.app_id, self.secrets
+
+    def search(
+        self, query: str, media_type: str = "album", limit: int = 500
+    ) -> Generator:
+        """Search the qobuz API.
+
+        If 'featured' is given as media type, this will retrieve results
+        from the featured albums in qobuz. The queries available with this type
+        are:
+
+            * most-streamed
+            * recent-releases
+            * best-sellers
+            * press-awards
+            * ideal-discography
+            * editor-picks
+            * most-featured
+            * qobuzissims
+            * new-releases
+            * new-releases-full
+            * harmonia-mundi
+            * universal-classic
+            * universal-jazz
+            * universal-jeunesse
+            * universal-chanson
+
+        :param query:
+        :type query: str
+        :param media_type:
+        :type media_type: str
+        :param limit:
+        :type limit: int
+        :rtype: Generator
+        """
+        return self._api_search(query, media_type, limit)
+
+    @region.cache_on_arguments(expiration_time=RELEASE_CACHE_TIME)
+    def get(self, item_id: Union[str, int], media_type: str = "album") -> dict:
+        return self._api_get(media_type, item_id=item_id)
+
+    def get_file_url(self, item_id, quality=6) -> dict:
+        return self._api_get_file_url(item_id, quality=quality)
+
+    @property
+    def source(self):
+        return "qobuz"
+
+    # ---------- Private Methods ---------------
+
+    # Credit to Sorrow446 for the original methods
+
+    def _gen_pages(self, epoint: str, params: dict) -> dict:
+        page, status_code = self._api_request(epoint, params)
+        logger.debug("Keys returned from _gen_pages: %s", ", ".join(page.keys()))
+        key = epoint.split("/")[0] + "s"
+        total = page.get(key, {})
+        total = total.get("total") or total.get("items")
+
+        if not total:
+            logger.debug("Nothing found from %s epoint", epoint)
+            return
+
+        limit = page.get(key, {}).get("limit", 500)
+        offset = page.get(key, {}).get("offset", 0)
+        params.update({"limit": limit})
+        yield page
+        while (offset + limit) < total:
+            offset += limit
+            params.update({"offset": offset})
+            page, status_code = self._api_request(epoint, params)
+            yield page
+
+    def _validate_secrets(self):
+        for secret in self.secrets:
+            if self._test_secret(secret):
+                self.sec = secret
+                logger.debug("Working secret and app_id: %s - %s", secret, self.app_id)
+                break
+        if not hasattr(self, "sec"):
+            raise InvalidAppSecretError(f"Invalid secrets: {self.secrets}")
+
+    def _api_get(self, media_type: str, **kwargs) -> dict:
+        item_id = kwargs.get("item_id")
+
+        params = {
+            "app_id": self.app_id,
+            f"{media_type}_id": item_id,
+            "limit": kwargs.get("limit", 500),
+            "offset": kwargs.get("offset", 0),
+        }
+        extras = {
+            "artist": "albums",
+            "playlist": "tracks",
+            "label": "albums",  # not tested
+        }
+
+        if media_type in extras:
+            params.update({"extra": extras[media_type]})
+
+        epoint = f"{media_type}/get"
+
+        response, status_code = self._api_request(epoint, params)
+        return response
+
+    def _api_search(self, query, media_type, limit=500) -> Generator:
+        params = {
+            "query": query,
+            "limit": limit,
+        }
+        # TODO: move featured, favorites, and playlists into _api_get later
+        if media_type == "featured":
+            assert query in QOBUZ_FEATURED_KEYS, f'query "{query}" is invalid.'
+            params.update({"type": query})
+            del params["query"]
+            epoint = "album/getFeatured"
+
+        elif query == "user-favorites":
+            assert query in ("track", "artist", "album")
+            params.update({"type": f"{media_type}s"})
+            epoint = "favorite/getUserFavorites"
+
+        elif query == "user-playlists":
+            epoint = "playlist/getUserPlaylists"
+
+        else:
+            epoint = f"{media_type}/search"
+
+        return self._gen_pages(epoint, params)
+
+    def _api_login(self, email: str, pwd: str):
+        # usr_info = self._api_call("user/login", email=email, pwd=pwd)
+        params = {
+            "email": email,
+            "password": pwd,
+            "app_id": self.app_id,
+        }
+        epoint = "user/login"
+        resp, status_code = self._api_request(epoint, params)
+
+        if status_code == 401:
+            raise AuthenticationError(f"Invalid credentials from params {params}")
+        elif status_code == 400:
+            raise InvalidAppIdError(f"Invalid app id from params {params}")
+        else:
+            logger.info("Logged in to Qobuz")
+
+        if not resp["user"]["credential"]["parameters"]:
+            raise IneligibleError("Free accounts are not eligible to download tracks.")
+
+        self.uat = resp["user_auth_token"]
+        self.session.headers.update({"X-User-Auth-Token": self.uat})
+        self.label = resp["user"]["credential"]["parameters"]["short_label"]
+
+    def _api_get_file_url(
+        self, track_id: Union[str, int], quality: int = 6, sec: str = None
+    ) -> dict:
+        unix_ts = time.time()
+
+        if int(quality) not in (5, 6, 7, 27):  # Needed?
+            raise InvalidQuality(f"Invalid quality id {quality}. Choose 5, 6, 7 or 27")
+
+        if sec is not None:
+            secret = sec
+        elif hasattr(self, "sec"):
+            secret = self.sec
+        else:
+            raise InvalidAppSecretError("Cannot find app secret")
+
+        r_sig = f"trackgetFileUrlformat_id{quality}intentstreamtrack_id{track_id}{unix_ts}{secret}"
+        logger.debug("Raw request signature: %s", r_sig)
+        r_sig_hashed = hashlib.md5(r_sig.encode("utf-8")).hexdigest()
+        logger.debug("Hashed request signature: %s", r_sig_hashed)
+
+        params = {
+            "request_ts": unix_ts,
+            "request_sig": r_sig_hashed,
+            "track_id": track_id,
+            "format_id": quality,
+            "intent": "stream",
+        }
+        response, status_code = self._api_request("track/getFileUrl", params)
+        if status_code == 400:
+            raise InvalidAppSecretError("Invalid app secret from params %s" % params)
+
+        return response
+
+    def _api_request(self, epoint: str, params: dict) -> Tuple[dict, int]:
+        logging.debug(f"Calling API with endpoint {epoint} params {params}")
+        r = self.session.get(f"{QOBUZ_BASE}/{epoint}", params=params)
+        try:
+            return r.json(), r.status_code
+        except Exception:
+            logger.error("Problem getting JSON. Status code: %s", r.status_code)
+            raise
+
+    def _test_secret(self, secret: str) -> bool:
+        try:
+            self._api_get_file_url("19512574", sec=secret)
+            return True
+        except InvalidAppSecretError as error:
+            logger.debug("Test for %s secret didn't work: %s", secret, error)
+            return False
+
+
+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.
+
+        :param query:
+        :type query: str
+        :param media_type:
+        :type media_type: str
+        :param limit:
+        :type limit: int
+        """
+        # TODO: more robust url sanitize
+        query = query.replace(" ", "+")
+
+        if media_type.endswith("s"):
+            media_type = media_type[:-1]
+
+        # TODO: use limit parameter
+        response = self.session.get(f"{DEEZER_BASE}/search/{media_type}?q={query}")
+        response.raise_for_status()
+
+        return response.json()
+
+    def login(self, **kwargs):
+        logger.debug("Deezer does not require login call, returning")
+
+    @region.cache_on_arguments(expiration_time=RELEASE_CACHE_TIME)
+    def get(self, meta_id: Union[str, int], media_type: str = "album"):
+        """Get metadata.
+
+        :param meta_id:
+        :type meta_id: Union[str, int]
+        :param type_:
+        :type type_: str
+        """
+        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"])
+        elif media_type == "artist":
+            albums = self.session.get(f"{url}/albums").json()
+            item["albums"] = albums["data"]
+
+        return item
+
+    @staticmethod
+    def get_file_url(meta_id: Union[str, int], quality: int = 6):
+        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(ClientInterface):
+    def __init__(self):
+        self.logged_in = False
+
+    def login(self, email: str, pwd: str):
+        if self.logged_in:
+            return
+
+        config = tidalapi.Config()
+
+        self.session = tidalapi.Session(config=config)
+        self.session.login(email, pwd)
+        logger.info("Logged into Tidal")
+
+        self.logged_in = True
+
+    @region.cache_on_arguments(expiration_time=RELEASE_CACHE_TIME)
+    def search(self, query: str, media_type: str = "album", limit: int = 50):
+        """
+        :param query:
+        :type query: str
+        :param media_type: artist, album, playlist, or track
+        :type media_type: str
+        :param limit:
+        :type limit: int
+        :raises ValueError: if field value is invalid
+        """
+
+        return self._search(query, media_type, limit=limit)
+
+    @region.cache_on_arguments(expiration_time=RELEASE_CACHE_TIME)
+    def get(self, meta_id: Union[str, int], media_type: str = "album"):
+        """Get metadata.
+
+        :param meta_id:
+        :type meta_id: Union[str, int]
+        :param media_type:
+        :type media_type: str
+        """
+        return self._get(meta_id, media_type)
+
+    def get_file_url(self, meta_id: Union[str, int], quality: int = 6):
+        """
+        :param meta_id:
+        :type meta_id: Union[str, int]
+        :param quality:
+        :type quality: int
+        """
+        logger.debug(f"Fetching file url with quality {quality}")
+        return self._get_file_url(meta_id, quality=min(TIDAL_MAX_Q, quality))
+
+    @property
+    def source(self):
+        return "tidal"
+
+    def _search(self, query, media_type="album", **kwargs):
+        params = {
+            "query": query,
+            "limit": kwargs.get("limit", 50),
+        }
+        return self.session.request("GET", f"search/{media_type}s", params).json()
+
+    def _get(self, media_id, media_type="album"):
+        if media_type == "album":
+            info = self.session.request("GET", f"albums/{media_id}")
+            tracklist = self.session.request("GET", f"albums/{media_id}/tracks")
+            album = info.json()
+            album["tracks"] = tracklist.json()
+            return album
+
+        elif media_type == "track":
+            return self.session.request("GET", f"tracks/{media_id}").json()
+        elif media_type == "playlist":
+            return self.session.request("GET", f"playlists/{media_id}/tracks").json()
+        elif media_type == "artist":
+            return self.session.request("GET", f"artists/{media_id}/albums").json()
+        else:
+            raise ValueError
+
+    def _get_file_url(self, track_id, quality=6):
+        params = {"soundQuality": TIDAL_Q_IDS[quality]}
+        resp = self.session.request("GET", f"tracks/{track_id}/streamUrl", params)
+        resp.raise_for_status()
+        return resp.json()

qobuz_dl_rewrite/config.py

@@ -0,0 +1,159 @@
+import logging
+import os
+from pprint import pformat
+
+from ruamel.yaml import YAML
+
+from .constants import CONFIG_PATH, FOLDER_FORMAT, TRACK_FORMAT
+from .exceptions import InvalidSourceError
+
+yaml = YAML()
+
+
+logger = logging.getLogger(__name__)
+
+
+class Config:
+    """Config class that handles command line args and config files.
+
+    Usage:
+    >>> config = Config('test_config.yaml')
+
+    If test_config was already initialized with values, this will load them
+    into `config`. Otherwise, a new config file is created with the default
+    values.
+
+    >>> config.update_from_cli(**args)
+
+    This will update the config values based on command line args.
+    """
+
+    def __init__(self, path: str):
+
+        # DEFAULTS
+        folder = "Downloads"
+        quality = 6
+        folder_format = FOLDER_FORMAT
+        track_format = TRACK_FORMAT
+
+        self.qobuz = {
+            "enabled": True,
+            "email": None,
+            "password": None,
+            "app_id": "",  # Avoid NoneType error
+            "secrets": [],
+        }
+        self.tidal = {"enabled": True, "email": None, "password": None}
+        self.deezer = {"enabled": True}
+        self.downloads_database = None
+        self.filters = {
+            "no_extras": False,
+            "albums_only": False,
+            "no_features": False,
+            "studio_albums": False,
+            "remaster_only": False,
+        }
+        self.downloads = {"folder": folder, "quality": quality}
+        self.metadata = {
+            "embed_cover": False,
+            "large_cover": False,
+            "default_comment": None,
+            "remove_extra_tags": False,
+        }
+        self.path_format = {"folder": folder_format, "track": track_format}
+
+        if path is None:
+            self._path = CONFIG_PATH
+        else:
+            self._path = path
+
+        if not os.path.exists(self._path):
+            logger.debug(f"Creating yaml config file at {self._path}")
+            self.dump(self.info)
+        else:
+            # sometimes the file gets erased, this will reset it
+            with open(self._path) as f:
+                if f.read().strip() == "":
+                    logger.debug(f"Config file {self._path} corrupted, resetting.")
+                    self.dump(self.info)
+                else:
+                    self.load()
+
+    def save(self):
+        self.dump(self.info)
+
+    def reset(self):
+        os.remove(self._path)
+        # re initialize with default info
+        self.__init__(self._path)
+
+    def load(self):
+        with open(self._path) as cfg:
+            for k, v in yaml.load(cfg).items():
+                setattr(self, k, v)
+
+        logger.debug("Config loaded")
+        self.__loaded = True
+
+    def update_from_cli(self, **kwargs):
+        for category in (self.downloads, self.metadata, self.filters):
+            for key in category.keys():
+                if kwargs[key] is None:
+                    continue
+
+                # For debugging's sake
+                og_value = category[key]
+                new_value = kwargs[key] or og_value
+                category[key] = new_value
+
+                if og_value != new_value:
+                    logger.debug("Updated %s config key from args: %s", key, new_value)
+
+    def dump(self, info):
+        with open(self._path, "w") as cfg:
+            logger.debug("Config saved: %s", self._path)
+            yaml.dump(info, cfg)
+
+    @property
+    def tidal_creds(self):
+        return {
+            "email": self.tidal["email"],
+            "pwd": self.tidal["password"],
+        }
+
+    @property
+    def qobuz_creds(self):
+        return {
+            "email": self.qobuz["email"],
+            "pwd": self.qobuz["password"],
+            "app_id": self.qobuz["app_id"],
+            "secrets": self.qobuz["secrets"],
+        }
+
+    def creds(self, source: str):
+        if source == "qobuz":
+            return self.qobuz_creds
+        elif source == "tidal":
+            return self.tidal_creds
+        elif source == "deezer":
+            return dict()
+        else:
+            raise InvalidSourceError(source)
+
+    @property
+    def info(self):
+        return {k: v for k, v in self.__dict__.items() if not k.startswith("_")}
+
+    @info.setter
+    def info(self, val):
+        for k, v in val.items():
+            setattr(self, k, v)
+
+    def __getitem__(self, key):
+        return getattr(self, key)
+
+    def __setitem__(self, key, val):
+        setattr(self, key, val)
+
+    def __repr__(self):
+        return f"Config({pformat(self.info)})"

qobuz_dl_rewrite/constants.py

@@ -0,0 +1,146 @@
+import os
+
+import appdirs
+import mutagen.id3 as id3
+
+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"
+
+TIDAL_COVER_URL = "https://resources.tidal.com/images/{uuid}/{width}x{height}.jpg"
+
+EXT = {
+    5: ".mp3",
+    6: ".flac",
+    7: ".flac",
+    27: ".flac",
+}
+
+QUALITY_DESC = {
+    4: "128kbps",
+    5: "320kbps",
+    6: "16bit/44.1kHz",
+    7: "24bit/96kHz",
+    27: "24bit/192kHz",
+}
+
+
+QOBUZ_FEATURED_KEYS = (
+    "most-streamed",
+    "recent-releases",
+    "best-sellers",
+    "press-awards",
+    "ideal-discography",
+    "editor-picks",
+    "most-featured",
+    "qobuzissims",
+    "new-releases",
+    "new-releases-full",
+    "harmonia-mundi",
+    "universal-classic",
+    "universal-jazz",
+    "universal-jeunesse",
+    "universal-chanson",
+)
+
+__MP4_KEYS = (
+    "\xa9nam",
+    "\xa9ART",
+    "\xa9alb",
+    r"aART",
+    "\xa9day",
+    "\xa9day",
+    "\xa9cmt",
+    "desc",
+    "purd",
+    "\xa9grp",
+    "\xa9gen",
+    "\xa9lyr",
+    "\xa9too",
+    "cprt",
+    "cpil",
+    "covr",
+    "trkn",
+    "disk",
+)
+
+__MP3_KEYS = (
+    id3.TIT2,
+    id3.TPE1,
+    id3.TALB,
+    id3.TPE2,
+    id3.TCOM,
+    id3.TYER,
+    id3.COMM,
+    id3.TT1,
+    id3.TT1,
+    id3.GP1,
+    id3.TCON,
+    id3.USLT,
+    id3.TEN,
+    id3.TCOP,
+    id3.TCMP,
+    None,
+    id3.TRCK,
+    id3.TPOS,
+)
+
+__METADATA_TYPES = (
+    "title",
+    "artist",
+    "album",
+    "albumartist",
+    "composer",
+    "year",
+    "comment",
+    "description",
+    "purchase_date",
+    "grouping",
+    "genre",
+    "lyrics",
+    "encoder",
+    "copyright",
+    "compilation",
+    "cover",
+    "tracknumber",
+    "discnumber",
+)
+
+
+FLAC_KEY = {v: v.upper() for v in __METADATA_TYPES}
+MP4_KEY = dict(zip(__METADATA_TYPES, __MP4_KEYS))
+MP3_KEY = dict(zip(__METADATA_TYPES, __MP3_KEYS))
+
+COPYRIGHT = "\u2117"
+PHON_COPYRIGHT = "\u00a9"
+FLAC_MAX_BLOCKSIZE = 16777215  # 16.7 MB
+
+TRACK_KEYS = ("tracknumber", "artist", "albumartist", "composer", "title")
+ALBUM_KEYS = ("albumartist", "title", "year", "bit_depth", "sampling_rate", "container")
+FOLDER_FORMAT = (
+    "{albumartist} - {title} ({year}) [{container}] [{bit_depth}B-{sampling_rate}kHz]"
+)
+TRACK_FORMAT = "{tracknumber}. {artist} - {title}"
+
+URL_REGEX = (
+    r"https:\/\/(?:www|open|play)?\.?(\w+)\.com(?:(?:\/(track|playlist|album|"
+    r"artist|label))|(?:\/[-\w]+?))+\/(\w+)"
+)
+
+
+TIDAL_Q_IDS = {
+    4: "LOW",  # AAC
+    5: "HIGH",  # AAC
+    6: "LOSSLESS",  # Lossless, but it also could be MQA
+    7: "HI_RES",  # not available for download
+}
+TIDAL_MAX_Q = 7
+
+DEEZER_Q_IDS = {4: 128, 5: 320, 6: 1411}
+DEEZER_MAX_Q = 6

qobuz_dl_rewrite/converter.py

@@ -0,0 +1,213 @@
+import logging
+import os
+import shutil
+import subprocess
+from tempfile import gettempdir
+from typing import Optional
+
+from .exceptions import ConversionError
+
+logger = logging.getLogger(__name__)
+
+
+class Converter:
+    """Base class for audio codecs."""
+
+    codec_name = None
+    codec_lib = None
+    container = None
+    lossless = False
+    default_ffmpeg_arg = ""
+
+    def __init__(
+        self,
+        filename: str,
+        ffmpeg_arg: Optional[str] = None,
+        sampling_rate: Optional[int] = None,
+        bit_depth: Optional[int] = None,
+        copy_art: bool = True,
+        remove_source: bool = False,
+        show_progress: bool = False,
+    ):
+        """
+        :param filename:
+        :type filename: str
+        :param ffmpeg_arg: The codec ffmpeg argument (defaults to an "optimal value")
+        :type ffmpeg_arg: Optional[str]
+        :param sampling_rate: This value is ignored if a lossy codec is detected
+        :type sampling_rate: Optional[int]
+        :param bit_depth: This value is ignored if a lossy codec is detected
+        :type bit_depth: Optional[int]
+        :param copy_art: Embed the cover art (if found) into the encoded file
+        :type copy_art: bool
+        :param remove_source:
+        :type remove_source: bool
+        """
+        logger.debug(locals())
+
+        self.filename = filename
+        self.final_fn = f"{os.path.splitext(filename)[0]}.{self.container}"
+        self.tempfile = os.path.join(gettempdir(), os.path.basename(self.final_fn))
+        self.remove_source = remove_source
+        self.sampling_rate = sampling_rate
+        self.bit_depth = bit_depth
+        self.copy_art = copy_art
+        self.show_progress = show_progress
+
+        if ffmpeg_arg is None:
+            logger.debug("No arguments provided. Codec defaults will be used")
+            self.ffmpeg_arg = self.default_ffmpeg_arg
+        else:
+            self.ffmpeg_arg = ffmpeg_arg
+            self._is_command_valid()
+
+        logger.debug("FFmpeg codec extra argument: %s", self.ffmpeg_arg)
+
+    def convert(self, custom_fn: Optional[str] = None):
+        """Convert the file.
+
+        :param custom_fn: Custom output filename (defaults to the original
+        name with a replaced container)
+        :type custom_fn: Optional[str]
+        """
+        if custom_fn:
+            self.final_fn = custom_fn
+
+        self.command = self._gen_command()
+        logger.debug("Generated conversion command: %s", self.command)
+
+        process = subprocess.Popen(self.command)
+        process.wait()
+        if os.path.isfile(self.tempfile):
+            if self.remove_source:
+                os.remove(self.filename)
+                logger.debug("Source removed: %s", self.filename)
+
+            shutil.move(self.tempfile, self.final_fn)
+            logger.debug("Moved: %s -> %s", self.tempfile, self.final_fn)
+            logger.debug("Converted: %s -> %s", self.filename, self.final_fn)
+        else:
+            raise ConversionError("No file was returned from conversion")
+
+    def _gen_command(self):
+        command = [
+            "ffmpeg",
+            "-i",
+            self.filename,
+            "-loglevel",
+            "warning",
+            "-c:a",
+            self.codec_lib,
+        ]
+        if self.show_progress:
+            command.append("-stats")
+
+        if self.copy_art:
+            command.extend(["-c:v", "copy"])
+
+        if self.ffmpeg_arg:
+            command.extend(self.ffmpeg_arg.split())
+
+        if self.lossless:
+            if isinstance(self.sampling_rate, int):
+                command.extend(["-ar", str(self.sampling_rate)])
+            elif self.sampling_rate is not None:
+                raise TypeError(
+                    f"Sampling rate must be int, not {type(self.sampling_rate)}"
+                )
+
+            if isinstance(self.bit_depth, int):
+                if int(self.bit_depth) == 16:
+                    command.extend(["-sample_fmt", "s16"])
+                elif int(self.bit_depth) in (24, 32):
+                    command.extend(["-sample_fmt", "s32"])
+                else:
+                    raise ValueError("Bit depth must be 16, 24, or 32")
+            elif self.bit_depth is not None:
+                raise TypeError(f"Bit depth must be int, not {type(self.bit_depth)}")
+
+        command.extend(["-y", self.tempfile])
+
+        return command
+
+    def _is_command_valid(self):
+        # TODO: add error handling for lossy codecs
+        if self.ffmpeg_arg is not None and self.lossless:
+            logger.debug(
+                "Lossless codecs don't support extra arguments; "
+                "the extra argument will be ignored"
+            )
+            self.ffmpeg_arg = self.default_ffmpeg_arg
+            return
+
+
+class FLAC(Converter):
+    " Class for FLAC converter. "
+    codec_name = "flac"
+    codec_lib = "flac"
+    container = "flac"
+    lossless = True
+
+
+class LAME(Converter):
+    """
+    Class for libmp3lame converter. Defaul ffmpeg_arg: `-q:a 0`.
+
+    See available options:
+    https://trac.ffmpeg.org/wiki/Encode/MP3
+    """
+
+    codec_name = "lame"
+    codec_lib = "libmp3lame"
+    container = "mp3"
+    default_ffmpeg_arg = "-q:a 0"  # V0
+
+
+class ALAC(Converter):
+    " Class for ALAC converter. "
+    codec_name = "alac"
+    codec_lib = "alac"
+    container = "m4a"
+    lossless = True
+
+
+class Vorbis(Converter):
+    """
+    Class for libvorbis converter. Default ffmpeg_arg: `-q:a 6`.
+
+    See available options:
+    https://trac.ffmpeg.org/wiki/TheoraVorbisEncodingGuide
+    """
+
+    codec_name = "vorbis"
+    codec_lib = "libvorbis"
+    container = "ogg"
+    default_ffmpeg_arg = "-q:a 6"  # 160, aka the "high" quality profile from Spotify
+
+
+class OPUS(Converter):
+    """
+    Class for libopus. Default ffmpeg_arg: `-b:a 128 -vbr on`.
+
+    See more:
+    http://ffmpeg.org/ffmpeg-codecs.html#libopus-1
+    """
+
+    codec_name = "opus"
+    codec_lib = "libopus"
+    container = "opus"
+    default_ffmpeg_arg = "-b:a 128k"  # Transparent
+
+
+class AAC(Converter):
+    """
+    Class for libfdk_aac converter. Default ffmpeg_arg: `-b:a 256k`.
+
+    See available options:
+    https://trac.ffmpeg.org/wiki/Encode/AAC
+    """
+
+    codec_name = "aac"
+    codec_lib = "libfdk_aac"
+    container = "m4a"
+    default_ffmpeg_arg = "-b:a 256k"

qobuz_dl_rewrite/core.py

@@ -0,0 +1,186 @@
+import hashlib
+import inspect
+import logging
+import os
+import re
+from getpass import getpass
+from typing import Generator, Optional, Tuple, Union
+
+import click
+
+from .clients import DeezerClient, QobuzClient, TidalClient
+from .config import Config
+from .constants import CONFIG_PATH, DB_PATH, URL_REGEX
+from .db import QobuzDB
+from .downloader import Album, Artist, Playlist, Track, Label
+from .exceptions import AuthenticationError, ParsingError
+from .utils import capitalize
+
+logger = logging.getLogger(__name__)
+
+
+MEDIA_CLASS = {"album": Album, "playlist": Playlist, "artist": Artist, "track": Track, "label": Label}
+CLIENTS = {"qobuz": QobuzClient, "tidal": TidalClient, "deezer": DeezerClient}
+Media = Union[Album, Playlist, Artist, Track]  # type hint
+
+# TODO: add support for database
+
+
+class QobuzDL:
+    def __init__(
+        self,
+        config: Optional[Config] = None,
+        database: Optional[str] = None,
+    ):
+        logger.debug(locals())
+
+        self.url_parse = re.compile(URL_REGEX)
+        self.config = config
+        if self.config is None:
+            self.config = Config(CONFIG_PATH)
+
+        self.clients = {
+            "qobuz": QobuzClient(),
+            "tidal": TidalClient(),
+            "deezer": DeezerClient(),
+        }
+
+        if database is None:
+            self.db = QobuzDB(DB_PATH)
+        else:
+            assert isinstance(database, QobuzDB)
+            self.db = database
+
+    def prompt_creds(self, source: str):
+        """Prompt the user for credentials.
+
+        :param source:
+        :type source: str
+        """
+        click.secho(f"Enter {capitalize(source)} email:", fg="green")
+        self.config[source]["email"] = input()
+        click.secho(
+            f"Enter {capitalize(source)} password (will not show on screen):",
+            fg="green",
+        )
+        self.config[source]["password"] = getpass(
+            prompt=""
+        )  # does hashing work for tidal?
+
+        self.config.save()
+        click.secho(f'Credentials saved to config file at "{self.config._path}"')
+
+    def assert_creds(self, source: str):
+        assert source in ("qobuz", "tidal", "deezer"), f"Invalid source {source}"
+        if source == "deezer":
+            # no login for deezer
+            return
+
+        if (
+            self.config[source]["email"] is None
+            or self.config[source]["password"] is None
+        ):
+            self.prompt_creds(source)
+
+    def handle_url(self, url: str):
+        """Download an url
+
+        :param url:
+        :type url: str
+        :raises InvalidSourceError
+        :raises ParsingError
+        """
+        source, url_type, item_id = self.parse_url(url)
+        if item_id in self.db:
+            logger.info(f"{url} already downloaded, use --no-db to override.")
+            return
+        self.handle_item(source, url_type, item_id)
+
+    def handle_item(self, source: str, media_type: str, item_id: str):
+        self.assert_creds(source)
+
+        arguments = {
+            "database": self.db,
+            "parent_folder": self.config.downloads["folder"],
+            "quality": self.config.downloads["quality"],
+            "embed_cover": self.config.metadata["embed_cover"],
+        }
+
+        client = self.clients[source]
+        if not client.logged_in:
+            while True:
+                try:
+                    client.login(**self.config.creds(source))
+                    break
+                except AuthenticationError:
+                    click.secho("Invalid credentials, try again.")
+                    self.prompt_creds(source)
+
+        item = MEDIA_CLASS[media_type](client=client, id=item_id)
+        if isinstance(item, Artist):
+            keys = self.config.filters.keys()
+            # TODO: move this to config.py
+            filters_ = tuple(key for key in keys if self.config.filters[key])
+            arguments["filters"] = filters_
+            logger.debug("Added filter argument for artist/label: %s", filters_)
+
+        logger.debug("Arguments from config: %s", arguments)
+
+        item.load_meta()
+        item.download(**arguments)
+
+    def parse_url(self, url: str) -> Tuple[str, str]:
+        """Returns the type of the url and the id.
+
+        Compatible with urls of the form:
+            https://www..com/us-en/{type}/{name}/{id}
+            https://open.qobuz.com/{type}/{id}
+            https://play.qobuz.com/{type}/{id}
+            /us-en/{type}/-/{id}
+
+            https://www.deezer.com/us/{type}/{id}
+            https://tidal.com/browse/{type}/{id}
+
+        :raises exceptions.ParsingError
+        """
+        parsed = self.url_parse.search(url)
+
+        if parsed is not None:
+            parsed = parsed.groups()
+
+            if len(parsed) == 3:
+                return tuple(parsed)  # Convert from Seq for the sake of typing
+
+        raise ParsingError(f"Error parsing URL: `{url}`")
+
+    def from_txt(self, filepath: Union[str, os.PathLike]):
+        """
+        Handle a text file containing URLs. Lines starting with `#` are ignored.
+
+        :param filepath:
+        :type filepath: Union[str, os.PathLike]
+        :raises OSError
+        :raises exceptions.ParsingError
+        """
+        with open(filepath) as txt:
+            lines = (
+                line for line in txt.readlines() if not line.strip().startswith("#")
+            )
+
+            click.secho(f"URLs found in text file: {len(lines)}")
+
+            for line in lines:
+                self.handle_url(line)
+
+    def search(
+        self, query: str, media_type: str = "album", limit: int = 200
+    ) -> Generator:
+        results = self.client.search(query, media_type, limit)
+
+        if isinstance(results, Generator):  # QobuzClient
+            for page in results:
+                for item in page[f"{media_type}s"]["items"]:
+                    yield MEDIA_CLASS[media_type].from_api(item, self.client)
+        else:
+            for item in results.get("data") or results.get("items"):
+                yield MEDIA_CLASS[media_type].from_api(item, self.client)

qobuz_dl_rewrite/db.py

@@ -0,0 +1,62 @@
+import logging
+import os
+import sqlite3
+from typing import Union
+
+logger = logging.getLogger(__name__)
+
+
+class QobuzDB:
+    """Simple interface for the downloaded track database."""
+
+    def __init__(self, db_path: Union[str, os.PathLike]):
+        """Create a QobuzDB object
+
+        :param db_path: filepath of the database
+        :type db_path: Union[str, os.PathLike]
+        """
+        self.path = db_path
+        if not os.path.exists(self.path):
+            self.create()
+
+    def create(self):
+        """Create a database at `self.path`"""
+        with sqlite3.connect(self.path) as conn:
+            try:
+                conn.execute("CREATE TABLE downloads (id TEXT UNIQUE NOT NULL);")
+                logger.debug("Download-IDs database created: %s", self.path)
+            except sqlite3.OperationalError:
+                pass
+
+            return self.path
+
+    def __contains__(self, item_id: Union[str, int]) -> bool:
+        """Checks whether the database contains an id.
+
+        :param item_id: the id to check
+        :type item_id: str
+        :rtype: bool
+        """
+        with sqlite3.connect(self.path) as conn:
+            return (
+                conn.execute(
+                    "SELECT id FROM downloads where id=?", (item_id,)
+                ).fetchone()
+                is not None
+            )
+
+    def add(self, item_id: str):
+        """Adds an id to the database.
+
+        :param item_id:
+        :type item_id: str
+        """
+        with sqlite3.connect(self.path) as conn:
+            try:
+                conn.execute(
+                    "INSERT INTO downloads (id) VALUES (?)",
+                    (item_id,),
+                )
+                conn.commit()
+            except sqlite3.Error as error:
+                logger.error("Unexpected DB error: %s", error)

qobuz_dl_rewrite/exceptions.py

@@ -0,0 +1,46 @@
+class AuthenticationError(Exception):
+    pass
+
+
+class IneligibleError(Exception):
+    pass
+
+
+class InvalidAppIdError(Exception):
+    pass
+
+
+class InvalidAppSecretError(Exception):
+    pass
+
+
+class InvalidQuality(Exception):
+    pass
+
+
+class NonStreamable(Exception):
+    pass
+
+
+class InvalidContainerError(Exception):
+    pass
+
+
+class InvalidSourceError(Exception):
+    pass
+
+
+class ParsingError(Exception):
+    pass
+
+
+class TooLargeCoverArt(Exception):
+    pass
+
+
+class BadEncoderOption(Exception):
+    pass
+
+
+class ConversionError(Exception):
+    pass

qobuz_dl_rewrite/metadata.py

@@ -0,0 +1,391 @@
+import json
+import logging
+import re
+import sys
+from typing import Generator, Optional, Tuple, Union
+
+from .constants import (
+    COPYRIGHT,
+    FLAC_KEY,
+    MP3_KEY,
+    MP4_KEY,
+    PHON_COPYRIGHT,
+    TRACK_KEYS,
+)
+from .exceptions import InvalidContainerError
+
+logger = logging.getLogger(__name__)
+
+
+class TrackMetadata:
+    """Contains all of the metadata needed to tag the file.
+    Available attributes:
+
+        * title
+        * artist
+        * album
+        * albumartist
+        * composer
+        * year
+        * comment
+        * description
+        * purchase_date
+        * grouping
+        * genre
+        * lyrics
+        * encoder
+        * copyright
+        * compilation
+        * cover
+        * tracknumber
+        * discnumber
+
+    """
+
+    def __init__(
+        self, track: Optional[dict] = None, album: Optional[dict] = None, source="qobuz"
+    ):
+        """Creates a TrackMetadata object optionally initialized with
+        dicts returned by the Qobuz API.
+
+        :param track: track dict from API
+        :type track: Optional[dict]
+        :param album: album dict from API
+        :type album: Optional[dict]
+        """
+        self.album = None
+        self.albumartist = None
+        self.composer = None
+        self.comment = None
+        self.description = None
+        self.purchase_date = None
+        self.grouping = None
+        self.lyrics = None
+        self.encoder = None
+        self.compilation = None
+        self.cover = None
+        self.tracknumber = None
+        self.discnumber = None
+
+        self.__source = source  # not included in tags
+
+        if track is None and album is None:
+            logger.debug("No params passed, returning")
+            return
+
+        if track is not None:
+            self.add_track_meta(track)
+
+        if album is not None:
+            self.add_album_meta(album)
+
+    def add_album_meta(self, resp: dict):
+        """Parse the metadata from an resp dict returned by the
+        Qobuz API.
+
+        :param dict resp: from the Qobuz API
+        """
+        if self.__source == "qobuz":
+            self.album = resp.get("title")
+            self.tracktotal = str(resp.get("tracks_count", 1))
+            self.genre = resp.get("genres_list", [])
+            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":
+            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
+
+    def add_track_meta(self, track: dict):
+        """Parse the metadata from a track dict returned by the
+        Qobuz API.
+
+        :param track:
+        """
+        if self.__source == "qobuz":
+            self.title = track.get("title").strip()
+            self._mod_title(track.get("version"), track.get("work"))
+            self.composer = track.get("composer", {}).get("name")
+
+            self.tracknumber = f"{int(track.get('track_number', 1)):02}"
+            self.discnumber = str(track.get("media_number", 1))
+            try:
+                self.artist = track["performer"]["name"]
+            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 = f"{int(track.get('trackNumber', 1)):02}"
+            self.discnumber = str(track.get("volumeNumber"))
+            self.artist = track.get("artist", {}).get("name")
+
+        elif self.__source == "deezer":
+            self.title = track.get("title").strip()
+            self._mod_title(track.get("version"), None)
+            self.tracknumber = f"{int(track.get('track_position', 1)):02}"
+            self.discnumber = track.get("disk_number")
+            self.artist = track.get("artist", {}).get("name")
+
+        else:
+            raise ValueError
+
+        if track.get("album"):
+            self.add_album_meta(track["album"])
+
+    def _mod_title(self, version, work):
+        if version is not None:
+            logger.debug("Version found: %s", version)
+            self.title = f"{self.title} ({version})"
+        if work is not None:
+            logger.debug("Work found: %s", work)
+            self.title = f"{work}: {self.title}"
+
+    @property
+    def artist(self) -> Union[str, None]:
+        """Returns the value to set for the artist tag. Defaults to
+        `self.albumartist` if there is no track artist.
+
+        :rtype: str
+        """
+        if self._artist is None and self.albumartist is not None:
+            return self.albumartist
+
+        if self._artist is not None:
+            return self._artist
+
+    @artist.setter
+    def artist(self, val: str):
+        """Sets the internal artist variable to val.
+
+        :param val:
+        :type val: str
+        """
+        self._artist = val
+
+    @property
+    def genre(self) -> Union[str, None]:
+        """Formats the genre list returned by the Qobuz API.
+        >>> g = ['Pop/Rock', 'Pop/Rock→Rock', 'Pop/Rock→Rock→Alternatif et Indé']
+        >>> _format_genres(g)
+        'Pop, Rock, Alternatif et Indé'
+
+        :rtype: str
+        """
+        if not self.get("_genres"):
+            return None
+
+        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
+
+        raise TypeError(f"Genre must be list or str, not {type(self._genres)}")
+
+    @genre.setter
+    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: Union[str, list]
+        """
+        self._genres = val
+
+    @property
+    def copyright(self) -> Union[str, None]:
+        """Formats the copyright string to use nice-looking unicode
+        characters.
+
+        :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
+
+        logger.debug("Accessed copyright tag before setting, return None")
+        return None
+
+    @copyright.setter
+    def copyright(self, val: str):
+        """Sets the internal copyright variable to the given value.
+        Only formatted when requested.
+
+        :param val:
+        :type val: str
+        """
+        self._copyright = val
+
+    @property
+    def year(self) -> Union[str, None]:
+        """Returns the year published of the track.
+
+        :rtype: str
+        """
+        if hasattr(self, "_year"):
+            return self._year
+
+        if hasattr(self, "date"):
+            if self.date is not None:
+                return self.date[:4]
+
+    @year.setter
+    def year(self, val):
+        """Sets the internal year variable to val.
+
+        :param val:
+        """
+        self._year = val
+
+    def get_formatter(self) -> dict:
+        """Returns a dict that is used to apply values to file format strings.
+
+        :rtype: dict
+        """
+        # the keys in the tuple are the possible keys for format strings
+        return {k: getattr(self, k) for k in TRACK_KEYS}
+
+    def tags(self, container: str = "flac") -> Generator:
+        """Return a generator of (key, value) pairs to use for tagging
+        files with mutagen. The *_KEY dicts are organized in the format
+
+        >>> {attribute_name: key_to_use_for_metadata}
+
+        They are then converted to the format
+
+        >>> {key_to_use_for_metadata: value_of_attribute}
+
+        so that they can be used like this:
+
+        >>> audio = MP4(path)
+        >>> for k, v in meta.tags(container='MP4'):
+        ...     audio[k] = v
+        >>> audio.save()
+
+        :param container: the container format
+        :type container: str
+        :rtype: Generator
+        """
+        container = container.lower()
+        if container in ("flac", "vorbis"):
+            return self.__gen_flac_tags()
+        elif container in ("mp3", "id3"):
+            return self.__gen_mp3_tags()
+        elif container in ("alac", "m4a", "mp4", "aac"):
+            return self.__gen_mp4_tags()
+        else:
+            raise InvalidContainerError(f"Invalid container {container}")
+
+    def __gen_flac_tags(self) -> Tuple[str, str]:
+        """Generate key, value pairs to tag FLAC files.
+
+        :rtype: Tuple[str, str]
+        """
+        for k, v in FLAC_KEY.items():
+            tag = getattr(self, k)
+            if tag:
+                logger.debug(f"Adding tag {v}: {repr(tag)}")
+                yield (v, tag)
+
+    def __gen_mp3_tags(self) -> Tuple[str, str]:
+        """Generate key, value pairs to tag MP3 files.
+
+        :rtype: Tuple[str, str]
+        """
+        for k, v in MP3_KEY.items():
+            if k == "tracknumber":
+                text = f"{self.tracknumber}/{self.tracktotal}"
+            elif k == "discnumber":
+                text = str(self.discnumber)
+            else:
+                text = getattr(self, k)
+
+            if text is not None:
+                yield (v.__name__, v(encoding=3, text=text))
+
+    def __mp4_tags(self) -> Tuple[str, str]:
+        """Generate key, value pairs to tag ALAC or AAC files in
+        an MP4 container.
+
+        :rtype: Tuple[str, str]
+        """
+        for k, v in MP4_KEY.items():
+            return (v, getattr(self, k))
+
+    def __setitem__(self, key, val):
+        """Dict-like access for tags.
+
+        :param key:
+        :param val:
+        """
+        setattr(self, key, val)
+
+    def __getitem__(self, key):
+        """Dict-like access for tags.
+
+        :param key:
+        """
+        return getattr(self, key)
+
+    def get(self, key, default=None) -> str:
+        """Returns the requested attribute of the object, with
+        a default value.
+
+        :param key:
+        :param default:
+        """
+        if hasattr(self, key):
+            res = self.__getitem__(key)
+            if res is not None:
+                return res
+
+            return default
+
+        return default
+
+    def set(self, key, val) -> str:
+        """Equivalent to
+        >>> meta[key] = val
+
+        :param key:
+        :param val:
+        :rtype: str
+        """
+        return self.__setitem__(key, val)
+
+    def __repr__(self) -> str:
+        """Returns the string representation of the metadata object.
+
+        :rtype: str
+        """
+        # TODO: make a more readable repr
+        return json.dumps(self.__dict__, indent=2)

qobuz_dl_rewrite/spoofbuz.py

@@ -0,0 +1,56 @@
+# Credits to Dash for this tool.
+
+import base64
+import re
+from collections import OrderedDict
+
+import requests
+
+
+class Spoofer:
+    def __init__(self):
+        self.seed_timezone_regex = r'[a-z]\.initialSeed\("(?P<seed>[\w=]+)",window\.utimezone\.(?P<timezone>[a-z]+)\)'
+        # note: {timezones} should be replaced with every capitalized timezone joined by a |
+        self.info_extras_regex = r'name:"\w+/(?P<timezone>{timezones})",info:"(?P<info>[\w=]+)",extras:"(?P<extras>[\w=]+)"'
+        self.app_id_regex = r'{app_id:"(?P<app_id>\d{9})",app_secret:"\w{32}",base_port:"80",base_url:"https://www\.qobuz\.com",base_method:"/api\.json/0\.2/"},n\.base_url="https://play\.qobuz\.com"'
+        login_page_request = requests.get("https://play.qobuz.com/login")
+        login_page = login_page_request.text
+        bundle_url_match = re.search(
+            r'<script src="(/resources/\d+\.\d+\.\d+-[a-z]\d{3}/bundle\.js)"></script>',
+            login_page,
+        )
+        bundle_url = bundle_url_match.group(1)
+        bundle_req = requests.get("https://play.qobuz.com" + bundle_url)
+        self.bundle = bundle_req.text
+
+    def get_app_id(self):
+        match = re.search(self.app_id_regex, self.bundle).group("app_id")
+        return str(match)
+
+    def get_secrets(self):
+        seed_matches = re.finditer(self.seed_timezone_regex, self.bundle)
+        secrets = OrderedDict()
+        for match in seed_matches:
+            seed, timezone = match.group("seed", "timezone")
+            secrets[timezone] = [seed]
+        """The code that follows switches around the first and second timezone. Why? Read on:
+            Qobuz uses two ternary (a shortened if statement) conditions that should always return false.
+            The way Javascript's ternary syntax works, the second option listed is what runs if the condition returns false.
+            Because of this, we must prioritize the *second* seed/timezone pair captured, not the first.
+        """
+        keypairs = list(secrets.items())
+        secrets.move_to_end(keypairs[1][0], last=False)
+        info_extras_regex = self.info_extras_regex.format(
+            timezones="|".join([timezone.capitalize() for timezone in secrets])
+        )
+        info_extras_matches = re.finditer(info_extras_regex, self.bundle)
+        for match in info_extras_matches:
+            timezone, info, extras = match.group("timezone", "info", "extras")
+            secrets[timezone.lower()] += [info, extras]
+        for secret_pair in secrets:
+            secrets[secret_pair] = base64.standard_b64decode(
+                "".join(secrets[secret_pair])[:-44]
+            ).decode("utf-8")
+        vals = list(secrets.values())
+        vals.remove("")
+        return vals

qobuz_dl_rewrite/utils.py

@@ -0,0 +1,154 @@
+import logging
+import logging.handlers as handlers
+import os
+from string import Formatter
+from typing import Optional
+
+import requests
+from pathvalidate import sanitize_filename
+from tqdm import tqdm
+
+from .constants import LOG_DIR, TIDAL_COVER_URL
+
+logger = logging.getLogger(__name__)
+
+
+def safe_get(d: dict, *keys, default=None):
+    """A replacement for chained `get()` statements on dicts:
+    >>> d = {'foo': {'bar': 'baz'}}
+    >>> _safe_get(d, 'baz')
+    None
+    >>> _safe_get(d, 'foo', 'bar')
+    'baz'
+    """
+    curr = d
+    res = default
+    for key in keys:
+        res = curr.get(key, default)
+        if res == default or not hasattr(res, "__getitem__"):
+            return res
+        else:
+            curr = res
+    return res
+
+
+def quality_id(bit_depth: Optional[int], sampling_rate: Optional[int]):
+    """Return a quality id in (5, 6, 7, 27) from bit depth and
+    sampling rate. If None is provided, mp3/lossy is assumed.
+
+    :param bit_depth:
+    :type bit_depth: Optional[int]
+    :param sampling_rate:
+    :type sampling_rate: Optional[int]
+    """
+    if not (bit_depth or sampling_rate):  # is lossy
+        return 5
+
+    if bit_depth == 16:
+        return 6
+
+    if bit_depth == 24:
+        if sampling_rate <= 96:
+            return 7
+
+        return 27
+
+
+def tqdm_download(url: str, filepath: str):
+    """Downloads a file with a progress bar.
+
+    :param url: url to direct download
+    :param filepath: file to write
+    :type url: str
+    :type filepath: str
+    """
+    # 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}")
+    try:
+        with open(filepath, "wb") as file, tqdm(
+            total=total, unit="iB", unit_scale=True, unit_divisor=1024
+        ) as bar:
+            for data in r.iter_content(chunk_size=1024):
+                size = file.write(data)
+                bar.update(size)
+    except Exception:
+        try:
+            os.remove(filepath)
+        except OSError:
+            pass
+        raise
+
+
+def clean_format(formatter: str, format_info):
+    """Formats track or folder names sanitizing every formatter key.
+
+    :param formatter:
+    :type formatter: str
+    :param kwargs:
+    """
+    fmt_keys = [i[1] for i in Formatter().parse(formatter) if i[1] is not None]
+
+    logger.debug("Formatter keys: %s", fmt_keys)
+
+    clean_dict = dict()
+    for key in fmt_keys:
+        if isinstance(format_info.get(key), (str, int, float)):  # int for track numbers
+            clean_dict[key] = sanitize_filename(str(format_info[key]))
+        else:
+            clean_dict[key] = "Unknown"
+
+    return formatter.format(**clean_dict)
+
+
+def tidal_cover_url(uuid, size):
+    possibles = (80, 160, 320, 640, 1280)
+    assert size in possibles, f"size must be in {possibles}"
+
+    return TIDAL_COVER_URL.format(uuid=uuid.replace("-", "/"), height=size, width=size)
+
+
+def init_log(
+    path: Optional[str] = None, level: str = "DEBUG", rotate: str = "midnight"
+):
+    """
+    Initialize a log instance with a stream handler and a rotating file handler.
+    If a path is not set, fallback to the default app log directory.
+
+    :param path:
+    :type path: Optional[str]
+    :param level:
+    :type level: str
+    :param rotate:
+    :type rotate: str
+    """
+    if not path:
+        os.makedirs(LOG_DIR, exist_ok=True)
+        path = os.path.join(LOG_DIR, "qobuz_dl.log")
+
+    logger = logging.getLogger()
+    level = logging.getLevelName(level)
+    logger.setLevel(level)
+
+    formatter = logging.Formatter(
+        fmt="%(asctime)s - %(module)s.%(funcName)s.%(levelname)s: %(message)s",
+        datefmt="%H:%M:%S",
+    )
+
+    rotable = handlers.TimedRotatingFileHandler(path, when=rotate)
+    printable = logging.StreamHandler()
+
+    rotable.setFormatter(formatter)
+    printable.setFormatter(formatter)
+
+    logger.addHandler(printable)
+    logger.addHandler(rotable)
+
+    logging.getLogger("urllib3").setLevel(logging.WARNING)
+    logging.getLogger("tidal_api").setLevel(logging.WARNING)
+
+
+def capitalize(s: str) -> str:
+    return s[0].upper() + s[1:]

requirements.txt

@@ -2,6 +2,6 @@ pathvalidate
 requests
 mutagen
 tqdm
-pick==1.6.0
+pick
 beautifulsoup4
 colorama

setup.py

@@ -1,4 +1,4 @@
-from setuptools import setup, find_packages
+from setuptools import find_packages, setup
 
 pkg_name = "qobuz-dl"
 
@@ -8,19 +8,12 @@ def read_file(fname):
         return f.read()
 
 
-requirements = [
-    "pathvalidate",
-    "requests",
-    "mutagen",
-    "tqdm",
-    "pick==1.6.0",
-    "beautifulsoup4",
-    "colorama",
-]
+requirements = read_file("requirements.txt").strip().split()
+
 
 setup(
     name=pkg_name,
-    version="0.9.9.10",
+    version="0.9.7",
     author="Vitiko",
     author_email="vhnz98@gmail.com",
     description="The complete Lossless and Hi-Res music downloader for Qobuz",