mirrors/streamrip
1
0
Fork 0
mirror of https://github.com/nathom/streamrip.git synced 2025-05-20 02:05:41 -04:00
Code Issues Projects Releases Packages Wiki Activity

Switch config to TOML

Browse source 
Signed-off-by: nathom <nathanthomas707@gmail.com>
This commit is contained in:
nathom 2021-06-19 18:58:31 -07:00
parent 7698ad7a2e
commit 5a5a199be2
9 changed files with 188 additions and 368 deletions
Download patch file Download diff file Expand all files Collapse all files

2
requirements.txt
View file

@ -1,6 +1,6 @@
click
ruamel.yaml
pathvalidate
requests
mutagen>=1.45.1
tqdm
tomlkit

4
streamrip/cli.py
View file

@ -67,7 +67,7 @@ def cli(ctx, **kwargs):
if ctx.invoked_subcommand == "config":
return
if config.session["check_for_updates"]:
if config.session["misc"]["check_for_updates"]:
r = requests.get("https://pypi.org/pypi/streamrip/json").json()
newest = r["info"]["version"]
if __version__ != newest:
@ -285,7 +285,7 @@ def config(ctx, **kwargs):
config.update()
if kwargs["path"]:
print(CONFIG_PATH)
click.echo(CONFIG_PATH)
if kwargs["open"]:
click.secho(f"Opening {CONFIG_PATH}", fg="green")

6
streamrip/clients.py
View file

@ -31,6 +31,7 @@ from .exceptions import (
IneligibleError,
InvalidAppIdError,
InvalidAppSecretError,
MissingCredentials,
InvalidQuality,
)
from .spoofbuz import Spoofer
@ -113,6 +114,9 @@ class QobuzClient(Client):
click.secho(f"Logging into {self.source}", fg="green")
email: str = kwargs["email"]
pwd: str = kwargs["pwd"]
if not email or not pwd:
raise MissingCredentials
if self.logged_in:
logger.debug("Already logged in")
return
@ -542,7 +546,7 @@ class TidalClient(Client):
:param refresh_token:
"""
if access_token is not None:
self.token_expiry = token_expiry
self.token_expiry = float(token_expiry)
self.refresh_token = refresh_token
if self.token_expiry - time.time() < 86400: # 1 day

355
streamrip/config.py
View file

@ -2,26 +2,18 @@
import copy
import logging
import click
import os
import re
from collections import OrderedDict
import shutil
from pprint import pformat
from typing import Any, Dict, List
from typing import Any, Dict
from ruamel.yaml import YAML
import tomlkit
from .constants import (
CONFIG_DIR,
CONFIG_PATH,
DOWNLOADS_DIR,
FOLDER_FORMAT,
TRACK_FORMAT,
)
from .constants import CONFIG_DIR, CONFIG_PATH
from . import __version__
from .exceptions import InvalidSourceError
yaml = YAML()
logger = logging.getLogger("streamrip")
@ -30,7 +22,7 @@ class Config:
Usage:
>>> config = Config('test_config.yaml')
>>> config = Config('test_config.toml')
>>> config.defaults['qobuz']['quality']
3
@ -39,75 +31,20 @@ class Config:
values.
"""
defaults: Dict[str, Any] = {
"qobuz": {
"quality": 3,
"download_booklets": True,
"email": None,
"password": None,
"app_id": "",
"secrets": [],
},
"tidal": {
"quality": 3,
"download_videos": True,
"user_id": None,
"country_code": None,
"access_token": None,
"refresh_token": None,
"token_expiry": 0,
},
"deezer": {
"quality": 2,
},
"soundcloud": {
"quality": 0,
},
"youtube": {
"quality": 0,
"download_videos": False,
"video_downloads_folder": DOWNLOADS_DIR,
},
"database": {"enabled": True, "path": None},
"conversion": {
"enabled": False,
"codec": None,
"sampling_rate": None,
"bit_depth": None,
},
"filters": {
"extras": False,
"repeats": False,
"non_albums": False,
"features": False,
"non_studio_albums": False,
"non_remaster": False,
},
"downloads": {"folder": DOWNLOADS_DIR, "source_subdirectories": False},
"artwork": {
"embed": True,
"size": "large",
"keep_hires_cover": True,
},
"metadata": {
"set_playlist_to_album": False,
"new_playlist_tracknumbers": True,
},
"path_format": {"folder": FOLDER_FORMAT, "track": TRACK_FORMAT},
"check_for_updates": True,
"lastfm": {"source": "qobuz", "fallback_source": "deezer"},
"concurrent_downloads": False,
}
default_config_path = os.path.join(os.path.dirname(__file__), "config.toml")
with open(default_config_path) as cfg:
defaults: Dict[str, Any] = tomlkit.parse(cfg.read())
def __init__(self, path: str = None):
"""Create a Config object with state.
A YAML file is created at `path` if there is none.
A TOML file is created at `path` if there is none.
:param path:
:type path: str
"""
# to access settings loaded from yaml file
# to access settings loaded from toml file
self.file: Dict[str, Any] = copy.deepcopy(self.defaults)
self.session: Dict[str, Any] = copy.deepcopy(self.defaults)
@ -117,10 +54,14 @@ class Config:
self._path = path
if not os.path.isfile(self._path):
logger.debug("Creating yaml config file at '%s'", self._path)
self.dump(self.defaults)
logger.debug("Creating toml config file at '%s'", self._path)
shutil.copy(self.default_config_path, CONFIG_PATH)
else:
self.load()
if self.file["misc"]["version"] != __version__:
click.secho("Updating config file to new version...", fg="green")
self.reset()
self.load()
def update(self):
"""Reset the config file except for credentials."""
@ -129,6 +70,7 @@ class Config:
temp["qobuz"].update(self.file["qobuz"])
temp["tidal"].update(self.file["tidal"])
self.dump(temp)
del temp
def save(self):
"""Save the config state to file."""
@ -139,12 +81,12 @@ class Config:
if not os.path.isdir(CONFIG_DIR):
os.makedirs(CONFIG_DIR, exist_ok=True)
self.dump(self.defaults)
shutil.copy(self.default_config_path, self._path)
def load(self):
"""Load infomation from the config files, making a deepcopy."""
with open(self._path) as cfg:
for k, v in yaml.load(cfg).items():
for k, v in tomlkit.loads(cfg.read()).items():
self.file[k] = v
if hasattr(v, "copy"):
self.session[k] = v.copy()
@ -161,10 +103,7 @@ class Config:
"""
with open(self._path, "w") as cfg:
logger.debug("Config saved: %s", self._path)
yaml.dump(info, cfg)
docs = ConfigDocumentation()
docs.dump(self._path)
cfg.write(tomlkit.dumps(info))
@property
def tidal_creds(self):
@ -203,251 +142,3 @@ class Config:
def __repr__(self):
"""Return a string representation of the config."""
return f"Config({pformat(self.session)})"
class ConfigDocumentation:
"""Documentation is stored in this docstring.
qobuz:
quality: 1: 320kbps MP3, 2: 16/44.1, 3: 24/<=96, 4: 24/>=96
download_booklets: This will download booklet pdfs that are included with some albums
password: This is an md5 hash of the plaintext password
app_id: Do not change
secrets: Do not change
tidal:
quality: 0: 256kbps AAC, 1: 320kbps AAC, 2: 16/44.1 "HiFi" FLAC, 3: 24/44.1 "MQA" FLAC
download_videos: This will download videos included in Video Albums.
user_id: Do not change any of the fields below
token_expiry: Tokens last 1 week after refresh. This is the Unix timestamp of the expiration time.
deezer: Deezer doesn't require login
quality: 0, 1, or 2
soundcloud:
quality: Only 0 is available
youtube:
quality: Only 0 is available for now
download_videos: Download the video along with the audio
video_downloads_folder: The path to download the videos to
database: This stores a list of item IDs so that repeats are not downloaded.
conversion: Convert tracks to a codec after downloading them.
codec: FLAC, ALAC, OPUS, MP3, VORBIS, or AAC
sampling_rate: In Hz. Tracks are downsampled if their sampling rate is greater than this. Values greater than 48000 are only recommended if the audio will be processed. It is otherwise a waste of space as the human ear cannot discern higher frequencies.
bit_depth: Only 16 and 24 are available. It is only applied when the bit depth is higher than this value.
filters: Filter a Qobuz artist's discography. Set to 'true' to turn on a filter.
extras: Remove Collectors Editions, live recordings, etc.
repeats: Picks the highest quality out of albums with identical titles.
non_albums: Remove EPs and Singles
features: Remove albums whose artist is not the one requested
non_remaster: Only download remastered albums
downloads:
folder: Folder where tracks are downloaded to
source_subdirectories: Put Qobuz albums in a 'Qobuz' folder, Tidal albums in 'Tidal' etc.
artwork:
embed: Write the image to the audio file
size: The size of the artwork to embed. Options: thumbnail, small, large, original. 'original' images can be up to 30MB, and may fail embedding. Using 'large' is recommended.
keep_hires_cover: Save the cover image at the highest quality as a seperate jpg file
metadata: Only applicable for playlist downloads.
set_playlist_to_album: Sets the value of the 'ALBUM' field in the metadata to the playlist's name. This is useful if your music library software organizes tracks based on album name.
new_playlist_tracknumbers: Replaces the original track's tracknumber with it's position in the playlist
path_format: Changes the folder and file names generated by streamrip.
folder: Available keys: "albumartist", "title", "year", "bit_depth", "sampling_rate", and "container"
track: Available keys: "tracknumber", "artist", "albumartist", "composer", and "title"
lastfm: Last.fm playlists are downloaded by searching for the titles of the tracks
source: The source on which to search for the tracks.
fallback_source: If no results were found with the primary source, the item is searched for on this one.
concurrent_downloads: Download (and convert) tracks all at once, instead of sequentially. If you are converting the tracks, and/or have fast internet, this will substantially improve processing speed.
"""
def __init__(self):
"""Create a new ConfigDocumentation object."""
# not using ruamel because its super slow
self.docs = []
doctext = self.__doc__
# get indent level, key, and documentation
keyval = re.compile(r"( *)([\w_]+):\s*(.*)")
lines = (line[4:] for line in doctext.split("\n")[2:-1])
for line in lines:
info = list(keyval.match(line).groups())
if len(info) == 3:
info[0] = len(info[0]) // 4 # here use standard 4 spaces/tab
else: # line doesn't start with spaces
info.insert(0, 0)
self.docs.append(info)
def dump(self, path: str):
"""Write comments to an uncommented YAML file.
:param path:
:type path: str
"""
is_comment = re.compile(r"^\s*#.*")
with open(path) as f:
# includes newline at the end
lines = f.readlines()
with open(path, "w") as f:
while lines != []:
line = lines.pop(0)
found = False
to_remove = None
for level, key, doc in self.docs:
# using 1 indent = 2 spaces like ruamel.yaml
spaces = level * " "
comment = f"{spaces}# {doc}"
if is_comment.match(line):
# update comment
found = True
break
re_obj = self._get_key_regex(spaces, key)
match = re_obj.match(line)
if match is not None: # line contains the key
if doc != "":
f.write(f"{comment}\n{line}")
found = True
to_remove = [level, key, doc]
break
if not found: # field with no comment
f.write(line)
if to_remove is not None:
# key, doc pairs are unique
self.docs.remove(to_remove)
def _get_key_regex(self, spaces: str, key: str) -> re.Pattern:
"""Get a regex that matches a key in YAML.
:param spaces: a string spaces that represent the indent level.
:type spaces: str
:param key: the key to match.
:type key: str
:rtype: re.Pattern
"""
regex = rf"{spaces}{key}:(?:$|\s+?(.+))"
return re.compile(regex)
def strip_comments(self, path: str):
"""Remove single-line comments from a file.
:param path:
:type path: str
"""
with open(path, "r") as f:
lines = [
line
for line in f.readlines()
if not line.strip().startswith("#")
]
with open(path, "w") as f:
f.write("".join(lines))
# ------------- ~~ Experimental ~~ ----------------- #
def load_yaml(path: str):
"""Load a streamrip config YAML file.
Warning: this is not fully compliant with YAML. It was made for use
with streamrip.
:param path:
:type path: str
"""
with open(path) as f:
lines = f.readlines()
settings = OrderedDict()
type_dict = {t.__name__: t for t in (list, dict, str)}
for line in lines:
key_l: List[str] = []
val_l: List[str] = []
chars = StringWalker(line)
level = 0
# get indent level of line
while next(chars).isspace():
level += 1
chars.prev()
if (c := next(chars)) == "#":
# is a comment
continue
elif c == "-":
# is an item in a list
next(chars)
val_l = list(chars)
level += 2 # it is a child of the previous key
item_type = "list"
else:
# undo char read
chars.prev()
if not val_l:
while (c := next(chars)) != ":":
key_l.append(c)
val_l = list("".join(chars).strip())
if val_l:
val = "".join(val_l)
else:
# start of a section
item_type = "dict"
val = type_dict[item_type]()
key = "".join(key_l)
if level == 0:
settings[key] = val
elif level == 2:
parent = settings[tuple(settings.keys())[-1]]
if isinstance(parent, dict):
parent[key] = val
elif isinstance(parent, list):
parent.append(val)
else:
raise Exception(f"level too high: {level}")
return settings
class StringWalker:
"""A fancier str iterator."""
def __init__(self, s: str):
"""Create a StringWalker object.
:param s:
:type s: str
"""
self.__val = s.replace("\n", "")
self.__pos = 0
def __next__(self) -> str:
"""Get the next char.
:rtype: str
"""
try:
c = self.__val[self.__pos]
self.__pos += 1
return c
except IndexError:
raise StopIteration
def __iter__(self):
"""Get an iterator."""
return self
def prev(self, step: int = 1):
"""Un-read a character.
:param step: The number of steps backward to take.
:type step: int
"""
self.__pos -= step

128
streamrip/config.toml Normal file
View file

@ -0,0 +1,128 @@
[downloads]
# Folder where tracks are downloaded to
folder = "/Users/nathan/StreamripDownloads"
# Put Qobuz albums in a 'Qobuz' folder, Tidal albums in 'Tidal' etc.
source_subdirectories = false
# Download (and convert) tracks all at once, instead of sequentially.
# If you are converting the tracks, or have fast internet, this will
# substantially improve processing speed.
concurrent = false
[qobuz]
# 1: 320kbps MP3, 2: 16/44.1, 3: 24/<=96, 4: 24/>=96
quality = 3
# This will download booklet pdfs that are included with some albums
download_booklets = true
email = ""
# This is an md5 hash of the plaintext password
password = ""
# Do not change
app_id = ""
# Do not change
secrets = []
[tidal]
# 0: 256kbps AAC, 1: 320kbps AAC, 2: 16/44.1 "HiFi" FLAC, 3: 24/44.1 "MQA" FLAC
quality = 3
# This will download videos included in Video Albums.
download_videos = true
# Do not change any of the fields below
user_id = ""
country_code = ""
access_token = ""
refresh_token = ""
# Tokens last 1 week after refresh. This is the Unix timestamp of the expiration
# time. If you haven't used streamrip in more than a week, you may have to log
# in again using `rip config --tidal`
token_expiry = ""
# Doesn't require login
[deezer]
# 0, 1, or 2
quality = 2
[soundcloud]
# Only 0 is available for now
quality = 0
[youtube]
# Only 0 is available for now
quality = 0
# Download the video along with the audio
download_videos = false
# The path to download the videos to
video_downloads_folder = ""
# This stores a list of item IDs so that repeats are not downloaded.
[database]
enabled = true
path = ""
# Convert tracks to a codec after downloading them.
[conversion]
enabled = false
# FLAC, ALAC, OPUS, MP3, VORBIS, or AAC
codec = "ALAC"
# In Hz. Tracks are downsampled if their sampling rate is greater than this.
# Value of 48000 is recommended to maximize quality and minimize space
sampling_rate = 48000
# Only 16 and 24 are available. It is only applied when the bit depth is higher
# than this value.
bit_depth = 24
# Filter a Qobuz artist's discography. Set to 'true' to turn on a filter.
[filters]
# Remove Collectors Editions, live recordings, etc.
extras = false
# Picks the highest quality out of albums with identical titles.
repeats = false
# Remove EPs and Singles
non_albums = false
# Remove albums whose artist is not the one requested
features = false
# Skip non studio albums
non_studio_albums = false
# Only download remastered albums
non_remaster = false
[artwork]
# Write the image to the audio file
embed = true
# The size of the artwork to embed. Options: thumbnail, small, large, original.
# "original" images can be up to 30MB, and may fail embedding.
# Using "large" is recommended.
size = "large"
# Save the cover image at the highest quality as a seperate jpg file
keep_hires_cover = true
[metadata]
# Sets the value of the 'ALBUM' field in the metadata to the playlist's name.
# This is useful if your music library software organizes tracks based on album name.
set_playlist_to_album = true
# Replaces the original track's tracknumber with it's position in the playlist
new_playlist_tracknumbers = true
# Changes the folder and file names generated by streamrip.
[path_format]
# Available keys: "albumartist", "title", "year", "bit_depth", "sampling_rate",
# and "container"
folder = "{albumartist} - {title} ({year}) [{container}] [{bit_depth}B-{sampling_rate}kHz]"
# Available keys: "tracknumber", "artist", "albumartist", "composer", and "title"
track = "{tracknumber}. {artist} - {title}"
# Last.fm playlists are downloaded by searching for the titles of the tracks
[lastfm]
# The source on which to search for the tracks.
source = "qobuz"
# If no results were found with the primary source, the item is searched for
# on this one.
fallback_source = "deezer"
[misc]
# Check whether a newer version of streamrip is available when starting up
check_for_updates = true
# Metadata to identify this config file. Do not change.
version = "0.5.5"

7
streamrip/constants.py
View file

@ -10,7 +10,7 @@ APPNAME = "streamrip"
CACHE_DIR = click.get_app_dir(APPNAME)
CONFIG_DIR = click.get_app_dir(APPNAME)
CONFIG_PATH = os.path.join(CONFIG_DIR, "config.yaml")
CONFIG_PATH = os.path.join(CONFIG_DIR, "config.toml")
LOG_DIR = click.get_app_dir(APPNAME)
DB_PATH = os.path.join(LOG_DIR, "downloads.db")
@ -19,9 +19,7 @@ DOWNLOADS_DIR = os.path.join(HOME, "StreamripDownloads")
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"
)
TIDAL_COVER_URL = "https://resources.tidal.com/images/{uuid}/{width}x{height}.jpg"
QUALITY_DESC = {
@ -32,7 +30,6 @@ QUALITY_DESC = {
4: "24bit/192kHz",
}
QOBUZ_FEATURED_KEYS = (
"most-streamed",
"recent-releases",

48
streamrip/core.py
View file

@ -36,6 +36,7 @@ from .constants import (
from .db import MusicDB
from .exceptions import (
AuthenticationError,
MissingCredentials,
NonStreamable,
NoResultsFound,
ParsingError,
@ -96,9 +97,11 @@ class MusicDL(list):
}
self.db: Union[MusicDB, list]
if self.config.session["database"]["enabled"]:
if self.config.session["database"]["path"] is not None:
self.db = MusicDB(self.config.session["database"]["path"])
db_settings = self.config.session["database"]
if db_settings["enabled"]:
path = db_settings["path"]
if path:
self.db = MusicDB(path)
else:
self.db = MusicDB(DB_PATH)
self.config.file["database"]["path"] = DB_PATH
@ -172,6 +175,7 @@ class MusicDL(list):
:rtype: dict
"""
logger.debug(self.config.session)
return {
"database": self.db,
"parent_folder": self.config.session["downloads"]["folder"],
@ -179,24 +183,18 @@ class MusicDL(list):
"track_format": self.config.session["path_format"]["track"],
"embed_cover": self.config.session["artwork"]["embed"],
"embed_cover_size": self.config.session["artwork"]["size"],
"keep_hires_cover": self.config.session["artwork"][
"keep_hires_cover"
],
"keep_hires_cover": self.config.session["artwork"]["keep_hires_cover"],
"set_playlist_to_album": self.config.session["metadata"][
"set_playlist_to_album"
],
"stay_temp": self.config.session["conversion"]["enabled"],
"conversion": self.config.session["conversion"],
"concurrent_downloads": self.config.session[
"concurrent_downloads"
],
"concurrent_downloads": self.config.session["downloads"]["concurrent"],
"new_tracknumbers": self.config.session["metadata"][
"new_playlist_tracknumbers"
],
"download_videos": self.config.session["tidal"]["download_videos"],
"download_booklets": self.config.session["qobuz"][
"download_booklets"
],
"download_booklets": self.config.session["qobuz"]["download_booklets"],
"download_youtube_videos": self.config.session["youtube"][
"download_videos"
],
@ -209,9 +207,10 @@ class MusicDL(list):
"""Download all the items in self."""
try:
arguments = self._get_download_args()
except KeyError:
except KeyError as e:
self._config_updating_message()
self.config.update()
logger.debug("Config update error: %s", e)
exit()
except Exception as err:
self._config_corrupted_message(err)
@ -219,9 +218,7 @@ class MusicDL(list):
logger.debug("Arguments from config: %s", arguments)
source_subdirs = self.config.session["downloads"][
"source_subdirectories"
]
source_subdirs = self.config.session["downloads"]["source_subdirectories"]
for item in self:
if source_subdirs:
arguments["parent_folder"] = self.__get_source_subdir(
@ -232,26 +229,20 @@ class MusicDL(list):
item.download(**arguments)
continue
arguments["quality"] = self.config.session[item.client.source][
"quality"
]
arguments["quality"] = self.config.session[item.client.source]["quality"]
if isinstance(item, Artist):
filters_ = tuple(
k for k, v in self.config.session["filters"].items() if v
)
arguments["filters"] = filters_
logger.debug(
"Added filter argument for artist/label: %s", filters_
)
logger.debug("Added filter argument for artist/label: %s", filters_)
if not (isinstance(item, Tracklist) and item.loaded):
logger.debug("Loading metadata")
try:
item.load_meta()
except NonStreamable:
click.secho(
f"{item!s} is not available, skipping.", fg="red"
)
click.secho(f"{item!s} is not available, skipping.", fg="red")
continue
item.download(**arguments)
@ -290,6 +281,12 @@ class MusicDL(list):
except AuthenticationError:
click.secho("Invalid credentials, try again.")
self.prompt_creds(client.source)
creds = self.config.creds(client.source)
except MissingCredentials:
logger.debug("Credentials are missing. Prompting..")
self.prompt_creds(client.source)
creds = self.config.creds(client.source)
if (
client.source == "qobuz"
and not creds.get("secrets")
@ -311,7 +308,6 @@ class MusicDL(list):
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}
https://www.deezer.com/us/{type}/{id}
https://tidal.com/browse/{type}/{id}

4
streamrip/exceptions.py
View file

@ -2,6 +2,10 @@ class AuthenticationError(Exception):
pass
class MissingCredentials(Exception):
pass
class IneligibleError(Exception):
pass

2
streamrip/tracklists.py
View file

@ -613,7 +613,7 @@ class Artist(Tracklist):
albums = self.meta["albums"]
elif self.client.source == "deezer":
# TODO: load artist name
self.name = self.meta["name"]
albums = self.meta["albums"]
else: