Update README.md

Sanitize album and track attributes

.gitignore

@@ -1,155 +1,2 @@
 Qobuz Downloads
-*__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
+__pycache__

.isort.cfg

@@ -1,3 +0,0 @@
-[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/).
+Search, explore and download Lossless and Hi-Res music from [Qobuz](https://www.qobuz.com/). It *just works*™ (2025).
 [![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 .cli import main
 from .qopy import Client
+from .cli import main

qobuz_dl/bundle.py

@@ -0,0 +1,79 @@
+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,14 +1,15 @@
 import configparser
-import glob
 import hashlib
 import logging
+import glob
 import os
 import sys
 
-import qobuz_dl.spoofbuz as spoofbuz
+from qobuz_dl.bundle import Bundle
 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,
@@ -25,7 +26,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- ")
@@ -52,12 +53,11 @@ def reset_config(config_file):
     config["DEFAULT"]["no_cover"] = "false"
     config["DEFAULT"]["no_database"] = "false"
     logging.info(f"{YELLOW}Getting tokens. Please wait...")
-    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}"
+    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
     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,14 +77,41 @@ def remove_leftovers(directory):
             pass
 
 
-def main():
+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():
     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)
 
@@ -102,22 +129,6 @@ 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"]
@@ -128,15 +139,22 @@ def main():
         arguments = qobuz_dl_args(
             default_quality, default_limit, default_folder
         ).parse_args()
-    except (KeyError, UnicodeDecodeError, configparser.Error):
+    except (KeyError, UnicodeDecodeError, configparser.Error) as error:
         arguments = qobuz_dl_args().parse_args()
         if not arguments.reset:
             sys.exit(
-                f"{RED}Your config file is corrupted! Run 'qobuz-dl -r' to fix this."
+                f"{RED}Your config file is corrupted: {error}! "
+                "Run 'qobuz-dl -r' to fix this."
             )
 
     if arguments.reset:
-        sys.exit(reset_config(CONFIG_FILE))
+        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()
 
     if arguments.purge:
         try:
@@ -161,26 +179,7 @@ def main():
     )
     qobuz.initialize_client(email, password, app_id, secrets)
 
-    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)
+    _handle_commands(qobuz, arguments)
 
 
 if __name__ == "__main__":

qobuz_dl/color.py

@@ -1,4 +1,4 @@
-from colorama import Fore, Style, init
+from colorama import Style, Fore, 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_rate, tracktitle, version.
+        albumartist, album, year, sampling_rate, bit_depth, tracktitle, version.
         cannot contain characters used by the system, which includes /:<>""",
     )
     custom_parser.add_argument(
@@ -149,6 +149,12 @@ 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,27 +1,28 @@
 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
 
-import qobuz_dl.spoofbuz as spoofbuz
+from qobuz_dl.bundle import Bundle
 from qobuz_dl import downloader, qopy
-from qobuz_dl.color import CYAN, DF, OFF, RED, RESET, YELLOW
-from qobuz_dl.db import create_db, handle_download_id
+from qobuz_dl.color import CYAN, OFF, RED, YELLOW, DF, RESET
 from qobuz_dl.exceptions import NonStreamable
+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,
+)
 
 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",
@@ -32,28 +33,6 @@ 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,
@@ -74,7 +53,7 @@ class QobuzDL:
         track_format="{tracknumber}. {tracktitle}",
         smart_discography=False,
     ):
-        self.directory = self.create_dir(directory)
+        self.directory = create_and_return_dir(directory)
         self.quality = quality
         self.embed_art = embed_art
         self.lucky_limit = lucky_limit
@@ -95,34 +74,12 @@ class QobuzDL:
         logger.info(f"{YELLOW}Set max quality: {QUALITIES[int(self.quality)]}\n")
 
     def get_tokens(self):
-        spoofer = spoofbuz.Spoofer()
-        self.app_id = spoofer.getAppId()
+        bundle = Bundle()
+        self.app_id = bundle.get_app_id()
         self.secrets = [
-            secret for secret in spoofer.getSecrets().values() if secret
+            secret for secret in bundle.get_secrets().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(
@@ -132,20 +89,20 @@ class QobuzDL:
             )
             return
         try:
-            downloader.download_id_by_type(
+            dloader = downloader.Download(
                 self.client,
                 item_id,
                 alt_path or self.directory,
-                str(self.quality),
-                album,
+                int(self.quality),
                 self.embed_art,
                 self.ignore_singles_eps,
                 self.quality_fallback,
                 self.cover_og_quality,
                 self.no_cover,
-                folder_format=self.folder_format,
-                track_format=self.track_format,
+                self.folder_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...")
@@ -168,7 +125,7 @@ class QobuzDL:
             "track": {"album": False, "func": None, "iterable_key": None},
         }
         try:
-            url_type, item_id = self.get_url_info(url)
+            url_type, item_id = get_url_info(url)
             type_dict = possibles[url_type]
         except (KeyError, IndexError):
             logger.info(
@@ -182,13 +139,13 @@ class QobuzDL:
                 f"{YELLOW}Downloading all the music from {content_name} "
                 f"({url_type})!"
             )
-            new_path = self.create_dir(
+            new_path = create_and_return_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 = self._smart_discography_filter(
+                items = smart_discography_filter(
                     content,
                     save_space=True,
                     skip_extras=True,
@@ -205,8 +162,8 @@ class QobuzDL:
                     True if type_dict["iterable_key"] == "albums" else False,
                     new_path,
                 )
-            if url_type == "playlist":
-                self.make_m3u(new_path)
+            if url_type == "playlist" and not self.no_m3u_for_playlists:
+                make_m3u(new_path)
         else:
             self.download_from_id(item_id, type_dict["album"])
 
@@ -256,9 +213,6 @@ 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")
@@ -307,7 +261,7 @@ class QobuzDL:
 
                     text = "{} - {} [{}]".format(
                         text,
-                        self.format_duration(i["duration"]),
+                        format_duration(i["duration"]),
                         "HI-RES" if i["hires_streamable"] else "LOSSLESS",
                     )
 
@@ -435,141 +389,11 @@ class QobuzDL:
         )
 
         for i in track_list:
-            track_id = self.get_url_info(
-                self.search_by_type(i, "track", 1, lucky=True)[0]
-            )[1]
+            track_id = 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)
 
-        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
+        if not self.no_m3u_for_playlists:
+            make_m3u(pl_directory)

qobuz_dl/db.py

@@ -1,7 +1,7 @@
 import logging
 import sqlite3
 
-from qobuz_dl.color import RED, YELLOW
+from qobuz_dl.color import YELLOW, RED
 
 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
+from pathvalidate import sanitize_filename, sanitize_filepath
 from tqdm import tqdm
 
 import qobuz_dl.metadata as metadata
-from qobuz_dl.color import CYAN, GREEN, OFF, RED, YELLOW
+from qobuz_dl.color import OFF, GREEN, RED, YELLOW, CYAN
 from qobuz_dl.exceptions import NonStreamable
 
 QL_DOWNGRADE = "FormatRestrictedByFormatAvailability"
@@ -23,67 +23,319 @@ DEFAULT_FORMATS = {
     ],
 }
 
+DEFAULT_FOLDER = "{artist} - {album} ({year}) [{bit_depth}B-{sampling_rate}kHz]"
+DEFAULT_TRACK = "{tracknumber}. {tracktitle}"
+
 logger = logging.getLogger(__name__)
 
 
-def tqdm_download(url, fname, track_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):
     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=track_name,
+        desc=desc,
         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(u: dict, track_title, multiple=None):
+def _get_description(item: dict, track_title, multiple=None):
     downloading_title = f"{track_title} "
-    f'[{u["bit_depth"]}/{u["sampling_rate"]}]'
+    f'[{item["bit_depth"]}/{item["sampling_rate"]}]'
     if multiple:
         downloading_title = f"[Disc {multiple}] {downloading_title}"
     return downloading_title
 
 
-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):
+def _get_title(item_dict):
     album_title = item_dict["title"]
     version = item_dict.get("version")
     if version:
@@ -95,265 +347,18 @@ def get_title(item_dict):
     return album_title
 
 
-def get_extra(i, dirn, extra="cover.jpg", og_quality=False):
+def _get_extra(item, 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(
-        i.replace("_600.", "_org.") if og_quality else i,
+        item.replace("_600.", "_org.") if og_quality else item,
         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 logging
-import os
 import re
+import os
+import logging
 
-import mutagen.id3 as id3
 from mutagen.flac import FLAC, Picture
+import mutagen.id3 as id3
 from mutagen.id3 import ID3NoHeaderError
 
 logger = logging.getLogger(__name__)
@@ -15,22 +15,39 @@ 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 = "{}: {}".format(track_dict["work"], title)
+        title = f"{track_dict['work']}: {title}"
 
     return title
 
 
 def _format_copyright(s: str) -> str:
-    s = s.replace("(P)", PHON_COPYRIGHT)
-    s = s.replace("(C)", COPYRIGHT)
+    if s:
+        s = s.replace("(P)", PHON_COPYRIGHT)
+        s = s.replace("(C)", COPYRIGHT)
     return s
 
 
@@ -46,8 +63,54 @@ 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, album, istrack=True, em_image=False):
+def tag_flac(
+    filename, root_dir, final_name, d: dict, album, istrack=True, em_image=False
+):
     """
     Tag a FLAC file
 
@@ -61,7 +124,7 @@ def tag_flac(filename, root_dir, final_name, d, album, istrack=True, em_image=Fa
     """
     audio = FLAC(filename)
 
-    audio["TITLE"] = get_title(d)
+    audio["TITLE"] = _get_title(d)
 
     audio["TRACKNUMBER"] = str(d["track_number"])  # TRACK NUMBER
 
@@ -73,18 +136,13 @@ def tag_flac(filename, root_dir, final_name, d, album, istrack=True, em_image=Fa
     except KeyError:
         pass
 
-    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"]
+    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["LABEL"] = album["label"]["name"]
-    except KeyError:
-        pass
+    audio["LABEL"] = album.get("label", {}).get("name", "n/a")
 
     if istrack:
         audio["GENRE"] = _format_genres(d["album"]["genres_list"])
@@ -92,43 +150,17 @@ def tag_flac(filename, root_dir, final_name, d, album, istrack=True, em_image=Fa
         audio["TRACKTOTAL"] = str(d["album"]["tracks_count"])
         audio["ALBUM"] = d["album"]["title"]
         audio["DATE"] = d["album"]["release_date_original"]
-        audio["COPYRIGHT"] = _format_copyright(d["copyright"])
+        audio["COPYRIGHT"] = _format_copyright(d.get("copyright") or "n/a")
     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["copyright"])
+        audio["COPYRIGHT"] = _format_copyright(album.get("copyright") or "n/a")
 
     if em_image:
-        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)
+        _embed_flac_img(root_dir, audio)
 
     audio.save()
     os.rename(filename, final_name)
@@ -146,21 +178,6 @@ 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:
@@ -168,19 +185,17 @@ 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
 
-    try:
-        tags["artist"] = d["performer"]["name"]
-    except KeyError:
-        if istrack:
-            tags["artist"] = d["album"]["artist"]["name"]
-        else:
-            tags["artist"] = album["artist"]["name"]
+    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"]
 
     if istrack:
         tags["genre"] = _format_genres(d["album"]["genres_list"])
@@ -204,21 +219,11 @@ 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:
-        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()))
+        _embed_id3_img(root_dir, audio)
 
     audio.save(filename, "v2_version=3")
     os.rename(filename, final_name)

qobuz_dl/qopy.py

@@ -8,7 +8,6 @@ import time
 
 import requests
 
-from qobuz_dl.color import GREEN, YELLOW
 from qobuz_dl.exceptions import (
     AuthenticationError,
     IneligibleError,
@@ -16,6 +15,7 @@ from qobuz_dl.exceptions import (
     InvalidAppSecretError,
     InvalidQuality,
 )
+from qobuz_dl.color import GREEN, YELLOW
 
 RESET = "Reset your credentials with 'qobuz-dl -r'"
 
@@ -26,15 +26,18 @@ class Client:
     def __init__(self, email, pwd, app_id, secrets):
         logger.info(f"{YELLOW}Logging...")
         self.secrets = secrets
-        self.id = app_id
+        self.id = str(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()
 
@@ -71,13 +74,15 @@ class Client:
                 "offset": kwargs["offset"],
                 "extra": "albums",
             }
-        elif epoint == "userLibrary/getAlbumsList":
+        elif epoint == "favorite/getUserFavorites":
             unix = time.time()
-            r_sig = "userLibrarygetAlbumsList" + str(unix) + kwargs["sec"]
+            # r_sig = "userLibrarygetAlbumsList" + str(unix) + kwargs["sec"]
+            r_sig = "favoritegetUserFavorites" + 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,
             }
@@ -88,7 +93,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, self.sec
+                fmt_id, track_id, unix, kwargs.get("sec", self.sec)
             )
             r_sig_hashed = hashlib.md5(r_sig.encode("utf-8")).hexdigest()
             params = {
@@ -108,9 +113,12 @@ class Client:
                 raise InvalidAppIdError("Invalid app id.\n" + RESET)
             else:
                 logger.info(f"{GREEN}Logged: OK")
-        elif epoint in ["track/getFileUrl", "userLibrary/getAlbumsList"]:
-            if r.status_code == 400:
-                raise InvalidAppSecretError("Invalid app secret.\n" + RESET)
+        elif (
+            epoint in ["track/getFileUrl", "favorite/getUserFavorites"]
+            and r.status_code == 400
+        ):
+            raise InvalidAppSecretError(f"Invalid app secret: {r.json()}.\n" + RESET)
+
         r.raise_for_status()
         return r.json()
 
@@ -189,16 +197,20 @@ class Client:
 
     def test_secret(self, sec):
         try:
-            r = self.api_call("userLibrary/getAlbumsList", sec=sec)
+            self.api_call("track/getFileUrl", id=5966783, fmt_id=5, 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 not hasattr(self, "sec"):
-            raise InvalidAppSecretError("Invalid app secret.\n" + RESET)
+
+        if self.sec is None:
+            raise InvalidAppSecretError("Can't find any valid app secret.\n" + RESET)

qobuz_dl/spoofbuz.py

@@ -1,51 +0,0 @@
-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

@@ -0,0 +1,190 @@
+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

@@ -1,107 +0,0 @@
-# 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

@@ -1,503 +0,0 @@
-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

@@ -1,159 +0,0 @@
-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

@@ -1,146 +0,0 @@
-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

@@ -1,213 +0,0 @@
-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

@@ -1,186 +0,0 @@
-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

@@ -1,62 +0,0 @@
-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

@@ -1,46 +0,0 @@
-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

@@ -1,391 +0,0 @@
-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

@@ -1,56 +0,0 @@
-# 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

@@ -1,154 +0,0 @@
-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
+pick==1.6.0
 beautifulsoup4
 colorama

setup.py

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