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

Merge dev

Browse source
This commit is contained in:
nathom 2021-05-06 22:03:55 -07:00
commit d4c31122fa
15 changed files with 909 additions and 384 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -19,3 +19,4 @@ StreamripDownloads
*.pyc
*test.py
/.mypy_cache
/streamrip/test.yaml

20
.mypy.ini Normal file
View file

@ -0,0 +1,20 @@
[mypy-mutagen.*]
ignore_missing_imports = True
[mypy-tqdm.*]
ignore_missing_imports = True
[mypy-pathvalidate.*]
ignore_missing_imports = True
[mypy-packaging.*]
ignore_missing_imports = True
[mypy-ruamel.yaml.*]
ignore_missing_imports = True
[mypy-pick.*]
ignore_missing_imports = True
[mypy-simple_term_menu.*]
ignore_missing_imports = True

6
README.md
View file

@ -1,7 +1,11 @@
# streamrip
[![Downloads](https://static.pepy.tech/personalized-badge/streamrip?period=total&units=international_system&left_color=black&right_color=green&left_text=Downloads)](https://pepy.tech/project/streamrip)
A scriptable stream downloader for Qobuz, Tidal, Deezer and SoundCloud.
## Features
- Super fast, as it utilizes concurrent downloads and conversion
@ -31,8 +35,8 @@ pip3 install streamrip windows-curses --upgrade
```
If you would like to use `streamrip`'s conversion capabilities, download TIDAL videos, or download music from SoundCloud, install [ffmpeg](https://ffmpeg.org/download.html).
If you would like to use `streamrip`'s conversion capabilities, or download music from SoundCloud, install [ffmpeg](https://ffmpeg.org/download.html). To download streams from YouTube, install `youtube-dl`.
## Example Usage

185
streamrip/bases.py
View file

@ -1,4 +1,6 @@
"""These are the lower level classes that are handled by Album, Playlist,
"""Bases that handle parsing and downloading media.
These are the lower level classes that are handled by Album, Playlist,
and the other objects. They can also be downloaded individually, for example,
as a single track.
"""
@ -86,6 +88,10 @@ class Track:
self.downloaded = False
self.tagged = False
self.converted = False
self.final_path: str
self.container: str
# TODO: find better solution
for attr in ("quality", "folder", "meta"):
setattr(self, attr, None)
@ -99,7 +105,6 @@ class Track:
def load_meta(self):
"""Send a request to the client to get metadata for this Track."""
assert self.id is not None, "id must be set before loading metadata"
self.resp = self.client.get(self.id, media_type="track")
@ -124,7 +129,8 @@ class Track:
self.cover_url = None
def _prepare_download(self, **kwargs):
"""This function does preprocessing to prepare for downloading tracks.
"""Do preprocessing before downloading items.
It creates the directories, downloads cover art, and (optionally)
downloads booklets.
@ -198,6 +204,7 @@ class Track:
return False
if self.client.source == "qobuz":
assert isinstance(dl_info, dict) # for typing
if not self.__validate_qobuz_dl_info(dl_info):
click.secho("Track is not available for download", fg="red")
return False
@ -207,6 +214,7 @@ class Track:
# --------- Download Track ----------
if self.client.source in ("qobuz", "tidal", "deezer"):
assert isinstance(dl_info, dict)
logger.debug("Downloadable URL found: %s", dl_info.get("url"))
try:
tqdm_download(
@ -214,11 +222,12 @@ class Track:
) # downloads file
except NonStreamable:
click.secho(
"Track {self!s} is not available for download, skipping.", fg="red"
f"Track {self!s} is not available for download, skipping.", fg="red"
)
return False
elif self.client.source == "soundcloud":
assert isinstance(dl_info, dict)
self._soundcloud_download(dl_info)
else:
@ -236,12 +245,10 @@ class Track:
if not kwargs.get("stay_temp", False):
self.move(self.final_path)
try:
database = kwargs.get("database")
database = kwargs.get("database")
if database:
database.add(self.id)
logger.debug(f"{self.id} added to database")
except AttributeError: # assume database=None was passed
pass
logger.debug("Downloaded: %s -> %s", self.path, self.final_path)
@ -264,7 +271,7 @@ class Track:
)
def move(self, path: str):
"""Moves the Track and sets self.path to the new path.
"""Move the Track and set self.path to the new path.
:param path:
:type path: str
@ -273,9 +280,11 @@ class Track:
shutil.move(self.path, path)
self.path = path
def _soundcloud_download(self, dl_info: dict) -> str:
"""Downloads a soundcloud track. This requires a seperate function
because there are three methods that can be used to download a track:
def _soundcloud_download(self, dl_info: dict):
"""Download a soundcloud track.
This requires a seperate function because there are three methods that
can be used to download a track:
* original file downloads
* direct mp3 downloads
* hls stream ripping
@ -314,15 +323,14 @@ class Track:
@property
def _progress_desc(self) -> str:
"""The description that is used on the progress bar.
"""Get the description that is used on the progress bar.
:rtype: str
"""
return click.style(f"Track {int(self.meta.tracknumber):02}", fg="blue")
def download_cover(self):
"""Downloads the cover art, if cover_url is given."""
"""Download the cover art, if cover_url is given."""
if not hasattr(self, "cover_url"):
return False
@ -357,8 +365,7 @@ class Track:
@classmethod
def from_album_meta(cls, album: TrackMetadata, track: dict, client: Client):
"""Return a new Track object initialized with info from the album dicts
returned by client.get calls.
"""Return a new Track object initialized with info.
:param album: album metadata returned by API
:param pos: index of the track
@ -366,14 +373,12 @@ class Track:
:type client: Client
:raises IndexError
"""
meta = TrackMetadata(album=album, track=track, source=client.source)
return cls(client=client, meta=meta, id=track["id"])
@classmethod
def from_api(cls, item: dict, client: Client):
"""Given a track dict from an API, return a new Track object
initialized with the proper values.
"""Return a new Track initialized from search result.
:param item:
:type item: dict
@ -401,7 +406,7 @@ class Track:
cover_url=cover_url,
)
def tag(
def tag( # noqa
self,
album_meta: dict = None,
cover: Union[Picture, APIC, MP4Cover] = None,
@ -496,7 +501,7 @@ class Track:
self.tagged = True
def convert(self, codec: str = "ALAC", **kwargs):
"""Converts the track to another codec.
"""Convert the track to another codec.
Valid values for codec:
* FLAC
@ -560,7 +565,7 @@ class Track:
@property
def title(self) -> str:
"""The title of the track.
"""Get the title of the track.
:rtype: str
"""
@ -581,8 +586,9 @@ class Track:
return safe_get(self.meta, *keys, default=default)
def set(self, key, val):
"""Equivalent to __setitem__. Implemented only for
consistency.
"""Set attribute `key` to `val`.
Equivalent to __setitem__. Implemented only for consistency.
:param key:
:param val:
@ -612,8 +618,7 @@ class Track:
return f"<Track - {self['title']}>"
def __str__(self) -> str:
"""Return a readable string representation of
this track.
"""Return a readable string representation of this track.
:rtype: str
"""
@ -624,6 +629,14 @@ class Video:
"""Only for Tidal."""
def __init__(self, client: Client, id: str, **kwargs):
"""Initialize a Video object.
:param client:
:type client: Client
:param id: The TIDAL Video ID
:type id: str
:param kwargs: title, explicit, and tracknumber
"""
self.id = id
self.client = client
self.title = kwargs.get("title", "MusicVideo")
@ -654,12 +667,21 @@ class Video:
return False # so that it is not tagged
def tag(self, *args, **kwargs):
"""Return False.
This is a dummy method.
:param args:
:param kwargs:
"""
return False
@classmethod
def from_album_meta(cls, track: dict, client: Client):
"""Given an video response dict from an album, return a new
Video object from the information.
"""Return a new Video object given an album API response.
:param track:
:param track: track dict from album
:type track: dict
:param client:
:type client: Client
@ -674,7 +696,7 @@ class Video:
@property
def path(self) -> str:
"""The path to download the mp4 file.
"""Get path to download the mp4 file.
:rtype: str
"""
@ -688,9 +710,17 @@ class Video:
return os.path.join(self.parent_folder, f"{fname}.mp4")
def __str__(self) -> str:
"""Return the title.
:rtype: str
"""
return self.title
def __repr__(self) -> str:
"""Return a string representation of self.
:rtype: str
"""
return f"<Video - {self.title}>"
@ -698,9 +728,9 @@ class Booklet:
"""Only for Qobuz."""
def __init__(self, resp: dict):
"""Initialized from the `goodies` field of the Qobuz API
response.
"""Initialize from the `goodies` field of the Qobuz API response.
Usage:
>>> album_meta = client.get('v4m7e0qiorycb', 'album')
>>> booklet = Booklet(album_meta['goodies'][0])
>>> booklet.download()
@ -708,6 +738,9 @@ class Booklet:
:param resp:
:type resp: dict
"""
self.url: str
self.description: str
self.__dict__.update(resp)
def download(self, parent_folder: str, **kwargs):
@ -734,8 +767,7 @@ class Tracklist(list):
essence_regex = re.compile(r"([^\(]+)(?:\s*[\(\[][^\)][\)\]])*")
def download(self, **kwargs):
"""Uses the _prepare_download and _download_item methods to download
all of the tracks contained in the Tracklist.
"""Download all of the items in the tracklist.
:param kwargs:
"""
@ -774,7 +806,7 @@ class Tracklist(list):
self.downloaded = True
def _download_and_convert_item(self, item, **kwargs):
"""Downloads and converts an item.
"""Download and convert an item.
:param item:
:param kwargs: should contain a `conversion` dict.
@ -782,7 +814,7 @@ class Tracklist(list):
if self._download_item(item, **kwargs):
item.convert(**kwargs["conversion"])
def _download_item(item, **kwargs):
def _download_item(item, *args: Any, **kwargs: Any) -> bool:
"""Abstract method.
:param item:
@ -798,7 +830,7 @@ class Tracklist(list):
raise NotImplementedError
def get(self, key: Union[str, int], default=None):
"""A safe `get` method similar to `dict.get`.
"""Get an item if key is int, otherwise get an attr.
:param key: If it is a str, get an attribute. If an int, get the item
at the index.
@ -826,13 +858,14 @@ class Tracklist(list):
self.__setitem__(key, val)
def convert(self, codec="ALAC", **kwargs):
"""Converts every item in `self`.
"""Convert every item in `self`.
Deprecated. Use _download_and_convert_item instead.
:param codec:
:param kwargs:
"""
if (sr := kwargs.get("sampling_rate")) :
if sr := kwargs.get("sampling_rate"):
if sr < 44100:
logger.warning(
"Sampling rate %d is lower than 44.1kHz."
@ -847,8 +880,7 @@ class Tracklist(list):
@classmethod
def from_api(cls, item: dict, client: Client):
"""Create an Album object from the api response of Qobuz, Tidal,
or Deezer.
"""Create an Album object from an API response.
:param resp: response dict
:type resp: dict
@ -858,18 +890,15 @@ class Tracklist(list):
info = cls._parse_get_resp(item, client=client)
# equivalent to Album(client=client, **info)
return cls(client=client, **info)
return cls(client=client, **info) # type: ignore
@staticmethod
def get_cover_obj(
cover_path: str, container: str, source: str
) -> Union[Picture, APIC]:
"""Given the path to an image and a quality id, return an initialized
cover object that can be used for every track in the album.
def get_cover_obj(cover_path: str, container: str, source: str):
"""Return an initialized cover object that is reused for every track.
:param cover_path:
:param cover_path: Path to the image, must be a JPEG.
:type cover_path: str
:param quality:
:param quality: quality ID
:type quality: int
:rtype: Union[Picture, APIC]
"""
@ -907,8 +936,8 @@ class Tracklist(list):
with open(cover_path, "rb") as img:
return cover(img.read(), imageformat=MP4Cover.FORMAT_JPEG)
def download_message(self) -> str:
"""The message to display after calling `Tracklist.download`.
def download_message(self):
"""Get the message to display after calling `Tracklist.download`.
:rtype: str
"""
@ -929,6 +958,7 @@ class Tracklist(list):
@staticmethod
def essence(album: str) -> str:
"""Ignore text in parens/brackets, return all lowercase.
Used to group two albums that may be named similarly, but not exactly
the same.
"""
@ -938,14 +968,23 @@ class Tracklist(list):
return album
def __getitem__(self, key: Union[str, int]):
def __getitem__(self, key):
"""Get an item if key is int, otherwise get an attr.
:param key:
"""
if isinstance(key, str):
return getattr(self, key)
if isinstance(key, int):
return super().__getitem__(key)
def __setitem__(self, key: Union[str, int], val: Any):
def __setitem__(self, key, val):
"""Set an item if key is int, otherwise set an attr.
:param key:
:param val:
"""
if isinstance(key, str):
setattr(self, key, val)
@ -957,19 +996,37 @@ class YoutubeVideo:
"""Dummy class implemented for consistency with the Media API."""
class DummyClient:
"""Used because YouTube downloads use youtube-dl, not a client."""
source = "youtube"
def __init__(self, url: str):
"""Create a YoutubeVideo object.
:param url: URL to the youtube video.
:type url: str
"""
self.url = url
self.client = self.DummyClient()
def download(
self,
parent_folder="StreamripDownloads",
download_youtube_videos=False,
youtube_video_downloads_folder="StreamripDownloads",
parent_folder: str = "StreamripDownloads",
download_youtube_videos: bool = False,
youtube_video_downloads_folder: str = "StreamripDownloads",
**kwargs,
):
"""Download the video using 'youtube-dl'.
:param parent_folder:
:type parent_folder: str
:param download_youtube_videos: True if the video should be downloaded.
:type download_youtube_videos: bool
:param youtube_video_downloads_folder: Folder to put videos if
downloaded.
:type youtube_video_downloads_folder: str
:param kwargs:
"""
click.secho(f"Downloading url {self.url}", fg="blue")
filename_formatter = "%(track_number)s.%(track)s.%(container)s"
filename = os.path.join(parent_folder, filename_formatter)
@ -1006,7 +1063,21 @@ class YoutubeVideo:
p.wait()
def load_meta(self, *args, **kwargs):
"""Return None.
Dummy method.
:param args:
:param kwargs:
"""
pass
def tag(self, *args, **kwargs):
"""Return None.
Dummy method.
:param args:
:param kwargs:
"""
pass

11
streamrip/cli.py
View file

@ -1,3 +1,5 @@
"""The streamrip command line interface."""
import logging
import os
from getpass import getpass
@ -34,6 +36,7 @@ if not os.path.isdir(CACHE_DIR):
@click.option("-t", "--text", metavar="PATH")
@click.option("-nd", "--no-db", is_flag=True)
@click.option("--debug", is_flag=True)
@click.version_option(prog_name="streamrip")
@click.pass_context
def cli(ctx, **kwargs):
"""Streamrip: The all-in-one Qobuz, Tidal, SoundCloud, and Deezer music downloader.
@ -124,7 +127,6 @@ def filter_discography(ctx, **kwargs):
For basic filtering, use the `--repeats` and `--features` filters.
"""
filters = kwargs.copy()
filters.pop("urls")
config.session["filters"] = filters
@ -178,7 +180,7 @@ def search(ctx, **kwargs):
@click.option("-l", "--list", default="ideal-discography")
@click.pass_context
def discover(ctx, **kwargs):
"""Searches for albums in Qobuz's featured lists.
"""Search for albums in Qobuz's featured lists.
Avaiable options for `--list`:
@ -229,7 +231,7 @@ def discover(ctx, **kwargs):
@click.argument("URL")
@click.pass_context
def lastfm(ctx, source, url):
"""Searches for tracks from a last.fm playlist on a given source.
"""Search for tracks from a last.fm playlist on a given source.
Examples:
@ -241,7 +243,6 @@ def lastfm(ctx, source, url):
Download a playlist using Tidal as the source
"""
if source is not None:
config.session["lastfm"]["source"] = source
@ -290,8 +291,10 @@ def config(ctx, **kwargs):
def none_chosen():
"""Print message if nothing was chosen."""
click.secho("No items chosen, exiting.", fg="bright_red")
def main():
"""Run the main program."""
cli(obj={})

137
streamrip/clients.py
View file

@ -1,3 +1,5 @@
"""The clients that interact with the service APIs."""
import base64
import hashlib
import json
@ -44,6 +46,10 @@ class Client(ABC):
it is merely a template.
"""
source: str
max_quality: int
logged_in: bool
@abstractmethod
def login(self, **kwargs):
"""Authenticate the client.
@ -72,35 +78,26 @@ class Client(ABC):
pass
@abstractmethod
def get_file_url(self, track_id, quality=3) -> Union[dict]:
def get_file_url(self, track_id, quality=3) -> Union[dict, str]:
"""Get the direct download url dict for a file.
:param track_id: id of the track
"""
pass
@property
@abstractmethod
def source(self):
"""Source from which the Client retrieves data."""
pass
@property
@abstractmethod
def max_quality(self):
"""The maximum quality that the Client supports."""
pass
class QobuzClient(Client):
"""QobuzClient."""
source = "qobuz"
max_quality = 4
# ------- Public Methods -------------
def __init__(self):
"""Create a QobuzClient object."""
self.logged_in = False
def login(self, email: str, pwd: str, **kwargs):
def login(self, **kwargs):
"""Authenticate the QobuzClient. Must have a paid membership.
If `app_id` and `secrets` are not provided, this will run the
@ -114,6 +111,8 @@ class QobuzClient(Client):
:param kwargs: app_id: str, secrets: list, return_secrets: bool
"""
click.secho(f"Logging into {self.source}", fg="green")
email: str = kwargs["email"]
pwd: str = kwargs["pwd"]
if self.logged_in:
logger.debug("Already logged in")
return
@ -140,6 +139,12 @@ class QobuzClient(Client):
self.logged_in = True
def get_tokens(self) -> Tuple[str, Sequence[str]]:
"""Return app id and secrets.
These can be saved and reused.
:rtype: Tuple[str, Sequence[str]]
"""
return self.app_id, self.secrets
def search(
@ -178,18 +183,31 @@ class QobuzClient(Client):
return self._api_search(query, media_type, limit)
def get(self, item_id: Union[str, int], media_type: str = "album") -> dict:
"""Get an item from the API.
:param item_id:
:type item_id: Union[str, int]
:param media_type:
:type media_type: str
:rtype: dict
"""
resp = self._api_get(media_type, item_id=item_id)
logger.debug(resp)
return resp
def get_file_url(self, item_id, quality=3) -> dict:
"""Get the downloadble file url for a track.
:param item_id:
:param quality:
:rtype: dict
"""
return self._api_get_file_url(item_id, quality=quality)
# ---------- Private Methods ---------------
def _gen_pages(self, epoint: str, params: dict) -> dict:
"""When there are multiple pages of results, this lazily
yields them.
def _gen_pages(self, epoint: str, params: dict) -> Generator:
"""When there are multiple pages of results, this yields them.
:param epoint:
:type epoint: str
@ -218,7 +236,7 @@ class QobuzClient(Client):
yield page
def _validate_secrets(self):
"""Checks if the secrets are usable."""
"""Check if the secrets are usable."""
for secret in self.secrets:
if self._test_secret(secret):
self.sec = secret
@ -228,8 +246,7 @@ class QobuzClient(Client):
raise InvalidAppSecretError(f"Invalid secrets: {self.secrets}")
def _api_get(self, media_type: str, **kwargs) -> dict:
"""Internal function that sends the request for metadata to the
Qobuz API.
"""Request metadata from the Qobuz API.
:param media_type:
:type media_type: str
@ -262,7 +279,7 @@ class QobuzClient(Client):
return response
def _api_search(self, query: str, media_type: str, limit: int = 500) -> Generator:
"""Internal function that sends a search request to the API.
"""Send a search request to the API.
:param query:
:type query: str
@ -297,8 +314,7 @@ class QobuzClient(Client):
return self._gen_pages(epoint, params)
def _api_login(self, email: str, pwd: str):
"""Internal function that logs into the api to get the user
authentication token.
"""Log into the api to get the user authentication token.
:param email:
:type email: str
@ -330,7 +346,7 @@ class QobuzClient(Client):
def _api_get_file_url(
self, track_id: Union[str, int], quality: int = 3, sec: str = None
) -> dict:
"""Internal function that gets the file url given an id.
"""Get the file url given a track id.
:param track_id:
:type track_id: Union[str, int]
@ -355,7 +371,7 @@ class QobuzClient(Client):
else:
raise InvalidAppSecretError("Cannot find app secret")
quality = get_quality(quality, self.source)
quality = int(get_quality(quality, self.source))
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()
@ -375,7 +391,7 @@ class QobuzClient(Client):
return response
def _api_request(self, epoint: str, params: dict) -> Tuple[dict, int]:
"""The function that handles all requests to the API.
"""Send a request to the API.
:param epoint:
:type epoint: str
@ -392,7 +408,7 @@ class QobuzClient(Client):
raise
def _test_secret(self, secret: str) -> bool:
"""Tests a secret.
"""Test the authenticity of a secret.
:param secret:
:type secret: str
@ -407,10 +423,13 @@ class QobuzClient(Client):
class DeezerClient(Client):
"""DeezerClient."""
source = "deezer"
max_quality = 2
def __init__(self):
"""Create a DeezerClient."""
self.session = gen_threadsafe_session()
# no login required
@ -426,16 +445,21 @@ class DeezerClient(Client):
:param limit:
:type limit: int
"""
# TODO: more robust url sanitize
query = query.replace(" ", "+")
# TODO: use limit parameter
response = self.session.get(f"{DEEZER_BASE}/search/{media_type}?q={query}")
response = self.session.get(
f"{DEEZER_BASE}/search/{media_type}", params={"q": query}
)
response.raise_for_status()
return response.json()
def login(self, **kwargs):
"""Return None.
Dummy method.
:param kwargs:
"""
logger.debug("Deezer does not require login call, returning")
def get(self, meta_id: Union[str, int], media_type: str = "album"):
@ -449,7 +473,7 @@ class DeezerClient(Client):
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()
tracks = self.session.get(f"{url}/tracks", params={"limit": 1000}).json()
item["tracks"] = tracks["data"]
item["track_total"] = len(tracks["data"])
elif media_type == "artist":
@ -461,6 +485,13 @@ class DeezerClient(Client):
@staticmethod
def get_file_url(meta_id: Union[str, int], quality: int = 6):
"""Get downloadable url for a track.
:param meta_id: The track ID.
:type meta_id: Union[str, int]
:param quality:
:type quality: int
"""
quality = min(DEEZER_MAX_Q, quality)
url = f"{DEEZER_DL}/{get_quality(quality, 'deezer')}/{DEEZER_BASE}/track/{meta_id}"
logger.debug(f"Download url {url}")
@ -468,12 +499,15 @@ class DeezerClient(Client):
class TidalClient(Client):
"""TidalClient."""
source = "tidal"
max_quality = 3
# ----------- Public Methods --------------
def __init__(self):
"""Create a TidalClient."""
self.logged_in = False
self.device_code = None
@ -582,7 +616,7 @@ class TidalClient(Client):
}
def get_tokens(self) -> dict:
"""Used for saving them for later use.
"""Return tokens to save for later use.
:rtype: dict
"""
@ -599,10 +633,11 @@ class TidalClient(Client):
# ------------ Utilities to login -------------
def _login_new_user(self, launch=True):
"""This will launch the browser and ask the user to log into tidal.
def _login_new_user(self, launch: bool = True):
"""Create app url where the user can log in.
:param launch:
:param launch: Launch the browser.
:type launch: bool
"""
login_link = f"https://{self._get_device_code()}"
@ -613,7 +648,7 @@ class TidalClient(Client):
click.launch(login_link)
start = time.time()
elapsed = 0
elapsed = 0.0
while elapsed < 600: # 5 mins to login
elapsed = time.time() - start
status = self._check_auth_status()
@ -694,7 +729,9 @@ class TidalClient(Client):
return True
def _refresh_access_token(self):
"""The access token expires in a week, so it must be refreshed.
"""Refresh the access token given a refresh token.
The access token expires in a week, so it must be refreshed.
Requires a refresh token.
"""
data = {
@ -719,7 +756,9 @@ class TidalClient(Client):
self._update_authorization()
def _login_by_access_token(self, token, user_id=None):
"""This is the method used to login after the access token has been saved.
"""Login using the access token.
Used after the initial authorization.
:param token:
:param user_id: Not necessary.
@ -745,7 +784,7 @@ class TidalClient(Client):
@property
def authorization(self):
"""The auth header."""
"""Get the auth header."""
return {"authorization": f"Bearer {self.access_token}"}
# ------------- Fetch data ------------------
@ -781,7 +820,7 @@ class TidalClient(Client):
return item
def _api_request(self, path: str, params=None) -> dict:
"""The function that handles all tidal API requests.
"""Handle Tidal API requests.
:param path:
:type path: str
@ -797,8 +836,7 @@ class TidalClient(Client):
return r
def _get_video_stream_url(self, video_id: str) -> str:
"""Videos have to be ripped from an hls stream, so they require
seperate processing.
"""Get the HLS video stream url.
:param video_id:
:type video_id: str
@ -824,7 +862,7 @@ class TidalClient(Client):
return url_info[-1]
def _api_post(self, url, data, auth=None):
"""Function used for posting to tidal API.
"""Post to the Tidal API.
:param url:
:param data:
@ -835,11 +873,14 @@ class TidalClient(Client):
class SoundCloudClient(Client):
"""SoundCloudClient."""
source = "soundcloud"
max_quality = 0
logged_in = True
def __init__(self):
"""Create a SoundCloudClient."""
self.session = gen_threadsafe_session(headers={"User-Agent": AGENT})
def login(self):
@ -864,7 +905,7 @@ class SoundCloudClient(Client):
logger.debug(resp)
return resp
def get_file_url(self, track: dict, quality) -> dict:
def get_file_url(self, track, quality):
"""Get the streamable file url from soundcloud.
It will most likely be an hls stream, which will have to be manually
@ -875,6 +916,9 @@ class SoundCloudClient(Client):
:param quality:
:rtype: dict
"""
# TODO: find better solution for typing
assert isinstance(track, dict)
if not track["streamable"] or track["policy"] == "BLOCK":
raise Exception
@ -908,8 +952,7 @@ class SoundCloudClient(Client):
return resp
def _get(self, path, params=None, no_base=False, resp_obj=False):
"""The lower level of `SoundCloudClient.get` that handles request
parameters and other options.
"""Send a request to the SoundCloud API.
:param path:
:param params:

174
streamrip/config.py
View file

@ -1,9 +1,12 @@
"""A config class that manages arguments between the config file and CLI."""
import copy
from collections import OrderedDict
import logging
import os
import re
from pprint import pformat
from typing import Any, Dict, List
from ruamel.yaml import YAML
@ -22,29 +25,21 @@ yaml = YAML()
logger = logging.getLogger(__name__)
# ---------- Utilities -------------
def _set_to_none(d: dict):
for k, v in d.items():
if isinstance(v, dict):
_set_to_none(v)
else:
d[k] = None
class Config:
"""Config class that handles command line args and config files.
Usage:
>>> config = Config('test_config.yaml')
>>> config.defaults['qobuz']['quality']
3
>>> config = Config('test_config.yaml')
>>> config.defaults['qobuz']['quality']
3
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.
"""
defaults = {
defaults: Dict[str, Any] = {
"qobuz": {
"quality": 3,
"download_booklets": True,
@ -105,9 +100,16 @@ class Config:
}
def __init__(self, path: str = None):
"""Create a Config object with state.
A YAML file is created at `path` if there is none.
:param path:
:type path: str
"""
# to access settings loaded from yaml file
self.file = copy.deepcopy(self.defaults)
self.session = copy.deepcopy(self.defaults)
self.file: Dict[str, Any] = copy.deepcopy(self.defaults)
self.session: Dict[str, Any] = copy.deepcopy(self.defaults)
if path is None:
self._path = CONFIG_PATH
@ -121,7 +123,7 @@ class Config:
self.load()
def update(self):
"""Resets the config file except for credentials."""
"""Reset the config file except for credentials."""
self.reset()
temp = copy.deepcopy(self.defaults)
temp["qobuz"].update(self.file["qobuz"])
@ -130,12 +132,10 @@ class Config:
def save(self):
"""Save the config state to file."""
self.dump(self.file)
def reset(self):
"""Reset the config file."""
if not os.path.isdir(CONFIG_DIR):
os.makedirs(CONFIG_DIR, exist_ok=True)
@ -143,7 +143,6 @@ class Config:
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():
self.file[k] = v
@ -197,24 +196,18 @@ class Config:
if source == "tidal":
return self.tidal_creds
if source == "deezer" or source == "soundcloud":
return dict()
return {}
raise InvalidSourceError(source)
def __getitem__(self, key):
assert key in ("file", "defaults", "session")
return getattr(self, key)
def __setitem__(self, key, val):
assert key in ("file", "session")
setattr(self, key, val)
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
@ -260,12 +253,13 @@ class ConfigDocumentation:
"""
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")[1:-1])
lines = (line[4:] for line in doctext.split("\n")[2:-1])
for line in lines:
info = list(keyval.match(line).groups())
@ -318,13 +312,133 @@ class ConfigDocumentation:
# key, doc pairs are unique
self.docs.remove(to_remove)
def _get_key_regex(self, spaces, key):
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

5
streamrip/constants.py
View file

@ -1,3 +1,5 @@
"""Constants that are kept in one place."""
import os
from pathlib import Path
@ -68,6 +70,7 @@ __MP4_KEYS = (
"disk",
None,
None,
None,
)
__MP3_KEYS = (
@ -91,6 +94,7 @@ __MP3_KEYS = (
id3.TPOS,
None,
None,
None,
)
__METADATA_TYPES = (
@ -114,6 +118,7 @@ __METADATA_TYPES = (
"discnumber",
"tracktotal",
"disctotal",
"date",
)

43
streamrip/converter.py
View file

@ -1,3 +1,5 @@
"""Wrapper classes over FFMPEG."""
import logging
import os
import shutil
@ -15,11 +17,11 @@ SAMPLING_RATES = (44100, 48000, 88200, 96000, 176400, 192000)
class Converter:
"""Base class for audio codecs."""
codec_name = None
codec_lib = None
container = None
lossless = False
default_ffmpeg_arg = ""
codec_name: str
codec_lib: str
container: str
lossless: bool = False
default_ffmpeg_arg: str = ""
def __init__(
self,
@ -31,7 +33,8 @@ class Converter:
remove_source: bool = False,
show_progress: bool = False,
):
"""
"""Create a Converter object.
:param filename:
:type filename: str
:param ffmpeg_arg: The codec ffmpeg argument (defaults to an "optimal value")
@ -42,7 +45,7 @@ class Converter:
: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:
:param remove_source: Remove the source file after conversion.
:type remove_source: bool
"""
logger.debug(locals())
@ -148,7 +151,8 @@ class Converter:
class FLAC(Converter):
" Class for FLAC converter. "
"""Class for FLAC converter."""
codec_name = "flac"
codec_lib = "flac"
container = "flac"
@ -156,8 +160,9 @@ class FLAC(Converter):
class LAME(Converter):
"""
Class for libmp3lame converter. Defaul ffmpeg_arg: `-q:a 0`.
"""Class for libmp3lame converter.
Default ffmpeg_arg: `-q:a 0`.
See available options:
https://trac.ffmpeg.org/wiki/Encode/MP3
@ -170,7 +175,8 @@ class LAME(Converter):
class ALAC(Converter):
" Class for ALAC converter. "
"""Class for ALAC converter."""
codec_name = "alac"
codec_lib = "alac"
container = "m4a"
@ -178,8 +184,9 @@ class ALAC(Converter):
class Vorbis(Converter):
"""
Class for libvorbis converter. Default ffmpeg_arg: `-q:a 6`.
"""Class for libvorbis converter.
Default ffmpeg_arg: `-q:a 6`.
See available options:
https://trac.ffmpeg.org/wiki/TheoraVorbisEncodingGuide
@ -192,8 +199,9 @@ class Vorbis(Converter):
class OPUS(Converter):
"""
Class for libopus. Default ffmpeg_arg: `-b:a 128 -vbr on`.
"""Class for libopus.
Default ffmpeg_arg: `-b:a 128 -vbr on`.
See more:
http://ffmpeg.org/ffmpeg-codecs.html#libopus-1
@ -206,8 +214,9 @@ class OPUS(Converter):
class AAC(Converter):
"""
Class for libfdk_aac converter. Default ffmpeg_arg: `-b:a 256k`.
"""Class for libfdk_aac converter.
Default ffmpeg_arg: `-b:a 256k`.
See available options:
https://trac.ffmpeg.org/wiki/Encode/AAC

175
streamrip/core.py
View file

@ -1,4 +1,7 @@
"""The stuff that ties everything together for the CLI to use."""
import concurrent.futures
import html
import logging
import os
import re
@ -6,14 +9,14 @@ import sys
from getpass import getpass
from hashlib import md5
from string import Formatter
from typing import Generator, Optional, Tuple, Union
from typing import Dict, Generator, List, Optional, Tuple, Type, Union
import click
import requests
from tqdm import tqdm
from .bases import Track, Video, YoutubeVideo
from .clients import DeezerClient, QobuzClient, SoundCloudClient, TidalClient
from .clients import DeezerClient, QobuzClient, SoundCloudClient, TidalClient, Client
from .config import Config
from .constants import (
CONFIG_PATH,
@ -38,7 +41,10 @@ from .utils import extract_interpreter_url
logger = logging.getLogger(__name__)
MEDIA_CLASS = {
Media = Union[
Type[Album], Type[Playlist], Type[Artist], Type[Track], Type[Label], Type[Video]
]
MEDIA_CLASS: Dict[str, Media] = {
"album": Album,
"playlist": Playlist,
"artist": Artist,
@ -46,24 +52,31 @@ MEDIA_CLASS = {
"label": Label,
"video": Video,
}
Media = Union[Album, Playlist, Artist, Track]
class MusicDL(list):
"""MusicDL."""
def __init__(
self,
config: Optional[Config] = None,
):
"""Create a MusicDL object.
:param config:
:type config: Optional[Config]
"""
self.url_parse = re.compile(URL_REGEX)
self.soundcloud_url_parse = re.compile(SOUNDCLOUD_URL_REGEX)
self.lastfm_url_parse = re.compile(LASTFM_URL_REGEX)
self.interpreter_url_parse = re.compile(QOBUZ_INTERPRETER_URL_REGEX)
self.youtube_url_parse = re.compile(YOUTUBE_URL_REGEX)
self.config = config
if self.config is None:
self.config: Config
if config is None:
self.config = Config(CONFIG_PATH)
else:
self.config = config
self.clients = {
"qobuz": QobuzClient(),
@ -72,25 +85,25 @@ class MusicDL(list):
"soundcloud": SoundCloudClient(),
}
if config.session["database"]["enabled"]:
if config.session["database"]["path"] is not None:
self.db = MusicDB(config.session["database"]["path"])
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"])
else:
self.db = MusicDB(DB_PATH)
config.file["database"]["path"] = DB_PATH
config.save()
self.config.file["database"]["path"] = DB_PATH
self.config.save()
else:
self.db = []
def handle_urls(self, url: str):
"""Download a url
"""Download a url.
:param url:
:type url: str
:raises InvalidSourceError
:raises ParsingError
"""
# youtube is handled by youtube-dl, so much of the
# processing is not necessary
youtube_urls = self.youtube_url_parse.findall(url)
@ -115,6 +128,15 @@ class MusicDL(list):
self.handle_item(source, url_type, item_id)
def handle_item(self, source: str, media_type: str, item_id: str):
"""Get info and parse into a Media object.
:param source:
:type source: str
:param media_type:
:type media_type: str
:param item_id:
:type item_id: str
"""
self.assert_creds(source)
client = self.get_client(source)
@ -128,6 +150,10 @@ class MusicDL(list):
self.append(item)
def _get_download_args(self) -> dict:
"""Get the arguments to pass to Media.download.
:rtype: dict
"""
return {
"database": self.db,
"parent_folder": self.config.session["downloads"]["folder"],
@ -156,6 +182,7 @@ class MusicDL(list):
}
def download(self):
"""Download all the items in self."""
try:
arguments = self._get_download_args()
except KeyError:
@ -216,7 +243,13 @@ class MusicDL(list):
if self.db != [] and hasattr(item, "id"):
self.db.add(item.id)
def get_client(self, source: str):
def get_client(self, source: str) -> Client:
"""Get a client given the source and log in.
:param source:
:type source: str
:rtype: Client
"""
client = self.clients[source]
if not client.logged_in:
self.assert_creds(source)
@ -224,6 +257,10 @@ class MusicDL(list):
return client
def login(self, client):
"""Log into a client, if applicable.
:param client:
"""
creds = self.config.creds(client.source)
if not client.logged_in:
while True:
@ -247,8 +284,8 @@ class MusicDL(list):
self.config.file["tidal"].update(client.get_tokens())
self.config.save()
def parse_urls(self, url: str) -> Tuple[str, str]:
"""Returns the type of the url and the id.
def parse_urls(self, url: str) -> List[Tuple[str, str, str]]:
"""Return the type of the url and the id.
Compatible with urls of the form:
https://www.qobuz.com/us-en/{type}/{name}/{id}
@ -261,8 +298,7 @@ class MusicDL(list):
:raises exceptions.ParsingError
"""
parsed = []
parsed: List[Tuple[str, str, str]] = []
interpreter_urls = self.interpreter_url_parse.findall(url)
if interpreter_urls:
@ -290,15 +326,31 @@ class MusicDL(list):
return parsed
def handle_lastfm_urls(self, urls):
def handle_lastfm_urls(self, urls: str):
"""Get info from lastfm url, and parse into Media objects.
This works by scraping the last.fm page and using a regex to
find the track titles and artists. The information is queried
in a Client.search(query, 'track') call and the first result is
used.
:param urls:
"""
# For testing:
# https://www.last.fm/user/nathan3895/playlists/12058911
user_regex = re.compile(r"https://www\.last\.fm/user/([^/]+)/playlists/\d+")
lastfm_urls = self.lastfm_url_parse.findall(urls)
lastfm_source = self.config.session["lastfm"]["source"]
tracks_not_found = 0
def search_query(query: str, playlist: Playlist):
global tracks_not_found
def search_query(query: str, playlist: Playlist) -> bool:
"""Search for a query and add the first result to playlist.
:param query:
:type query: str
:param playlist:
:type playlist: Playlist
:rtype: bool
"""
try:
track = next(self.search(lastfm_source, query, media_type="track"))
if self.config.session["metadata"]["set_playlist_to_album"]:
@ -307,29 +359,33 @@ class MusicDL(list):
track.meta.version = track.meta.work = None
playlist.append(track)
return True
except NoResultsFound:
tracks_not_found += 1
return
return False
for purl in lastfm_urls:
click.secho(f"Fetching playlist at {purl}", fg="blue")
title, queries = self.get_lastfm_playlist(purl)
pl = Playlist(client=self.get_client(lastfm_source), name=title)
pl.creator = user_regex.search(purl).group(1)
creator_match = user_regex.search(purl)
if creator_match is not None:
pl.creator = creator_match.group(1)
tracks_not_found: int = 0
with concurrent.futures.ThreadPoolExecutor(max_workers=15) as executor:
futures = [
executor.submit(search_query, f"{title} {artist}", pl)
for title, artist in queries
]
# only for the progress bar
for f in tqdm(
for search_attempt in tqdm(
concurrent.futures.as_completed(futures),
total=len(futures),
desc="Searching",
):
pass
if not search_attempt.result():
tracks_not_found += 1
pl.loaded = True
click.secho(f"{tracks_not_found} tracks not found.", fg="yellow")
@ -350,6 +406,18 @@ class MusicDL(list):
def search(
self, source: str, query: str, media_type: str = "album", limit: int = 200
) -> Generator:
"""Universal search.
:param source:
:type source: str
:param query:
:type query: str
:param media_type:
:type media_type: str
:param limit:
:type limit: int
:rtype: Generator
"""
client = self.get_client(source)
results = client.search(query, media_type)
@ -362,7 +430,7 @@ class MusicDL(list):
else page["albums"]["items"]
)
for item in tracklist:
yield MEDIA_CLASS[
yield MEDIA_CLASS[ # type: ignore
media_type if media_type != "featured" else "album"
].from_api(item, client)
i += 1
@ -376,12 +444,16 @@ class MusicDL(list):
raise NoResultsFound(query)
for item in items:
yield MEDIA_CLASS[media_type].from_api(item, client)
yield MEDIA_CLASS[media_type].from_api(item, client) # type: ignore
i += 1
if i > limit:
return
def preview_media(self, media):
def preview_media(self, media) -> str:
"""Return a preview string of a Media object.
:param media:
"""
if isinstance(media, Album):
fmt = (
"{albumartist} - {album}\n"
@ -408,9 +480,18 @@ class MusicDL(list):
ret = fmt.format(**{k: media.get(k, default="Unknown") for k in fields})
return ret
def interactive_search(
def interactive_search( # noqa
self, query: str, source: str = "qobuz", media_type: str = "album"
):
"""Show an interactive menu that contains search results.
:param query:
:type query: str
:param source:
:type source: str
:param media_type:
:type media_type: str
"""
results = tuple(self.search(source, query, media_type, limit=50))
def title(res):
@ -491,6 +572,15 @@ class MusicDL(list):
return True
def get_lastfm_playlist(self, url: str) -> Tuple[str, list]:
"""From a last.fm url, find the playlist title and tracks.
Each page contains 50 results, so `num_tracks // 50 + 1` requests
are sent per playlist.
:param url:
:type url: str
:rtype: Tuple[str, list]
"""
info = []
words = re.compile(r"[\w\s]+")
title_tags = re.compile('title="([^"]+)"')
@ -506,13 +596,21 @@ class MusicDL(list):
r = requests.get(url)
get_titles(r.text)
remaining_tracks = (
int(re.search(r'data-playlisting-entry-count="(\d+)"', r.text).group(1))
- 50
remaining_tracks_match = re.search(
r'data-playlisting-entry-count="(\d+)"', r.text
)
playlist_title = re.search(
if remaining_tracks_match is not None:
remaining_tracks = int(remaining_tracks_match.group(1)) - 50
else:
raise Exception("Error parsing lastfm page")
playlist_title_match = re.search(
r'<h1 class="playlisting-playlist-header-title">([^<]+)</h1>', r.text
).group(1)
)
if playlist_title_match is not None:
playlist_title = html.unescape(playlist_title_match.group(1))
else:
raise Exception("Error finding title from response")
page = 1
while remaining_tracks > 0:
@ -550,6 +648,11 @@ class MusicDL(list):
raise Exception
def assert_creds(self, source: str):
"""Ensure that the credentials for `source` are valid.
:param source:
:type source: str
"""
assert source in (
"qobuz",
"tidal",

12
streamrip/db.py
View file

@ -1,6 +1,4 @@
"""A simple wrapper over an sqlite database that stores
the downloaded media IDs.
"""
"""Wrapper over a database that stores item IDs."""
import logging
import os
@ -14,7 +12,7 @@ class MusicDB:
"""Simple interface for the downloaded track database."""
def __init__(self, db_path: Union[str, os.PathLike]):
"""Create a MusicDB object
"""Create a MusicDB object.
:param db_path: filepath of the database
:type db_path: Union[str, os.PathLike]
@ -24,7 +22,7 @@ class MusicDB:
self.create()
def create(self):
"""Create a database at `self.path`"""
"""Create a database at `self.path`."""
with sqlite3.connect(self.path) as conn:
try:
conn.execute("CREATE TABLE downloads (id TEXT UNIQUE NOT NULL);")
@ -35,7 +33,7 @@ class MusicDB:
return self.path
def __contains__(self, item_id: Union[str, int]) -> bool:
"""Checks whether the database contains an id.
"""Check whether the database contains an id.
:param item_id: the id to check
:type item_id: str
@ -51,7 +49,7 @@ class MusicDB:
)
def add(self, item_id: str):
"""Adds an id to the database.
"""Add an id to the database.
:param item_id:
:type item_id: str

218
streamrip/metadata.py
View file

@ -1,8 +1,11 @@
"""Manages the information that will be embeded in the audio file. """
"""Manages the information that will be embeded in the audio file."""
from __future__ import annotations
import logging
import re
from collections import OrderedDict
from typing import Generator, Hashable, Optional, Tuple, Union
from typing import Generator, Hashable, Iterable, Optional, Union
from .constants import (
COPYRIGHT,
@ -22,8 +25,8 @@ logger = logging.getLogger(__name__)
class TrackMetadata:
"""Contains all of the metadata needed to tag the file.
Tags contained:
Tags contained:
* title
* artist
* album
@ -44,14 +47,15 @@ class TrackMetadata:
* discnumber
* tracktotal
* disctotal
"""
def __init__(
self, track: Optional[dict] = None, album: Optional[dict] = None, source="qobuz"
self,
track: Optional[Union[TrackMetadata, dict]] = None,
album: Optional[Union[TrackMetadata, dict]] = None,
source="qobuz",
):
"""Creates a TrackMetadata object optionally initialized with
dicts returned by the Qobuz API.
"""Create a TrackMetadata object.
:param track: track dict from API
:type track: Optional[dict]
@ -59,34 +63,37 @@ class TrackMetadata:
:type album: Optional[dict]
"""
# embedded information
self.title = None
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.tracktotal = None
self.tracknumber = None
self.discnumber = None
self.disctotal = None
self.title: str
self.album: str
self.albumartist: str
self.composer: Optional[str] = None
self.comment: Optional[str] = None
self.description: Optional[str] = None
self.purchase_date: Optional[str] = None
self.grouping: Optional[str] = None
self.lyrics: Optional[str] = None
self.encoder: Optional[str] = None
self.compilation: Optional[str] = None
self.cover: Optional[str] = None
self.tracktotal: int
self.tracknumber: int
self.discnumber: int
self.disctotal: int
# not included in tags
self.explicit = False
self.quality = None
self.sampling_rate = None
self.bit_depth = None
self.explicit: Optional[bool] = False
self.quality: Optional[int] = None
self.sampling_rate: Optional[int] = None
self.bit_depth: Optional[int] = None
self.booklets = None
self.cover_urls = Optional[OrderedDict]
self.work: Optional[str]
self.id: Optional[str]
# Internals
self._artist = None
self._copyright = None
self._genres = None
self._artist: Optional[str] = None
self._copyright: Optional[str] = None
self._genres: Optional[Iterable] = None
self.__source = source
@ -100,9 +107,8 @@ class TrackMetadata:
elif album is not None:
self.add_album_meta(album)
def update(self, meta):
"""Given a TrackMetadata object (usually from an album), the fields
of the current object are updated.
def update(self, meta: TrackMetadata):
"""Update the attributes from another TrackMetadata object.
:param meta:
:type meta: TrackMetadata
@ -114,14 +120,13 @@ class TrackMetadata:
setattr(self, k, v)
def add_album_meta(self, resp: dict):
"""Parse the metadata from an resp dict returned by the
API.
"""Parse the metadata from an resp dict returned by the API.
:param dict resp: from API
"""
if self.__source == "qobuz":
# Tags
self.album = resp.get("title")
self.album = resp.get("title", "Unknown Album")
self.tracktotal = resp.get("tracks_count", 1)
self.genre = resp.get("genres_list") or resp.get("genre")
self.date = resp.get("release_date_original") or resp.get("release_date")
@ -144,7 +149,7 @@ class TrackMetadata:
# Non-embedded information
self.version = resp.get("version")
self.cover_urls = OrderedDict(resp.get("image"))
self.cover_urls = OrderedDict(resp["image"])
self.cover_urls["original"] = self.cover_urls["large"].replace("600", "org")
self.streamable = resp.get("streamable", False)
self.bit_depth = resp.get("maximum_bit_depth")
@ -156,14 +161,14 @@ class TrackMetadata:
self.sampling_rate *= 1000
elif self.__source == "tidal":
self.album = resp.get("title")
self.album = resp.get("title", "Unknown Album")
self.tracktotal = resp.get("numberOfTracks", 1)
# genre not returned by API
self.date = resp.get("releaseDate")
self.copyright = resp.get("copyright")
self.albumartist = safe_get(resp, "artist", "name")
self.disctotal = resp.get("numberOfVolumes")
self.disctotal = resp.get("numberOfVolumes", 1)
self.isrc = resp.get("isrc")
# label not returned by API
@ -185,8 +190,8 @@ class TrackMetadata:
self.sampling_rate = 44100
elif self.__source == "deezer":
self.album = resp.get("title")
self.tracktotal = resp.get("track_total") or resp.get("nb_tracks")
self.album = resp.get("title", "Unknown Album")
self.tracktotal = resp.get("track_total", 0) or resp.get("nb_tracks", 0)
self.disctotal = (
max(track.get("disk_number") for track in resp.get("tracks", [{}])) or 1
)
@ -218,41 +223,37 @@ class TrackMetadata:
raise InvalidSourceError(self.__source)
def add_track_meta(self, track: dict):
"""Parse the metadata from a track dict returned by an
API.
"""Parse the metadata from a track dict returned by an API.
:param track:
"""
if self.__source == "qobuz":
self.title = track.get("title").strip()
self.title = track["title"].strip()
self._mod_title(track.get("version"), track.get("work"))
self.composer = track.get("composer", {}).get("name")
self.tracknumber = track.get("track_number", 1)
self.discnumber = track.get("media_number", 1)
self.artist = safe_get(track, "performer", "name")
if self.artist is None:
self.artist = self.get("albumartist")
elif self.__source == "tidal":
self.title = track.get("title").strip()
self.title = track["title"].strip()
self._mod_title(track.get("version"), None)
self.tracknumber = track.get("trackNumber", 1)
self.discnumber = track.get("volumeNumber")
self.discnumber = track.get("volumeNumber", 1)
self.artist = track.get("artist", {}).get("name")
elif self.__source == "deezer":
self.title = track.get("title").strip()
self.title = track["title"].strip()
self._mod_title(track.get("version"), None)
self.tracknumber = track.get("track_position", 1)
self.discnumber = track.get("disk_number")
self.discnumber = track.get("disk_number", 1)
self.artist = track.get("artist", {}).get("name")
elif self.__source == "soundcloud":
self.title = track["title"].strip()
self.genre = track["genre"]
self.artist = track["user"]["username"]
self.albumartist = self.artist
self.artist = self.albumartist = track["user"]["username"]
self.year = track["created_at"][:4]
self.label = track["label_name"]
self.description = track["description"]
@ -265,7 +266,14 @@ class TrackMetadata:
if track.get("album"):
self.add_album_meta(track["album"])
def _mod_title(self, version, work):
def _mod_title(self, version: Optional[str], work: Optional[str]):
"""Modify title using the version and work.
:param version:
:type version: str
:param work:
:type work: str
"""
if version is not None:
self.title = f"{self.title} ({version})"
if work is not None:
@ -274,6 +282,10 @@ class TrackMetadata:
@property
def album(self) -> str:
"""Return the album of the track.
:rtype: str
"""
assert hasattr(self, "_album"), "Must set album before accessing"
album = self._album
@ -287,19 +299,21 @@ class TrackMetadata:
return album
@album.setter
def album(self, val) -> str:
def album(self, val):
"""Set the value of the album.
:param val:
"""
self._album = val
@property
def artist(self) -> Optional[str]:
"""Returns the value to set for the artist tag. Defaults to
`self.albumartist` if there is no track artist.
"""Return 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
@ -307,7 +321,7 @@ class TrackMetadata:
@artist.setter
def artist(self, val: str):
"""Sets the internal artist variable to val.
"""Set the internal artist variable to val.
:param val:
:type val: str
@ -316,10 +330,12 @@ class TrackMetadata:
@property
def genre(self) -> Optional[str]:
"""Formats the genre list returned by the Qobuz API.
>>> meta.genre = ['Pop/Rock', 'Pop/Rock→Rock', 'Pop/Rock→Rock→Alternatif et Indé']
>>> meta.genre
'Pop, Rock, Alternatif et Indé'
"""Format the genre list returned by an API.
It cleans up the Qobuz Response:
>>> meta.genre = ['Pop/Rock', 'Pop/Rock→Rock', 'Pop/Rock→Rock→Alternatif et Indé']
>>> meta.genre
'Pop, Rock, Alternatif et Indé'
:rtype: str
"""
@ -331,7 +347,7 @@ class TrackMetadata:
if isinstance(self._genres, list):
if self.__source == "qobuz":
genres = re.findall(r"([^\u2192\/]+)", "/".join(self._genres))
genres: Iterable = re.findall(r"([^\u2192\/]+)", "/".join(self._genres))
genres = set(genres)
elif self.__source == "deezer":
genres = ", ".join(g["name"] for g in self._genres)
@ -344,8 +360,9 @@ class TrackMetadata:
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.
def genre(self, val: Union[Iterable, dict]):
"""Set the internal `genre` field to the given list.
It is not formatted until it is requested with `meta.genre`.
:param val:
@ -354,25 +371,25 @@ class TrackMetadata:
self._genres = val
@property
def copyright(self) -> Union[str, None]:
"""Formats the copyright string to use nice-looking unicode
characters.
def copyright(self) -> Optional[str]:
"""Format the copyright string to use unicode characters.
:rtype: str, None
"""
if hasattr(self, "_copyright"):
if self._copyright is None:
return None
copyright = re.sub(r"(?i)\(P\)", PHON_COPYRIGHT, self._copyright)
copyright: str = re.sub(r"(?i)\(P\)", PHON_COPYRIGHT, self._copyright)
copyright = re.sub(r"(?i)\(C\)", COPYRIGHT, copyright)
return copyright
logger.debug("Accessed copyright tag before setting, return None")
logger.debug("Accessed copyright tag before setting, returning None")
return None
@copyright.setter
def copyright(self, val: str):
"""Sets the internal copyright variable to the given value.
"""Set the internal copyright variable to the given value.
Only formatted when requested.
:param val:
@ -382,7 +399,7 @@ class TrackMetadata:
@property
def year(self) -> Optional[str]:
"""Returns the year published of the track.
"""Return the year published of the track.
:rtype: str
"""
@ -397,14 +414,14 @@ class TrackMetadata:
@year.setter
def year(self, val):
"""Sets the internal year variable to val.
"""Set 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.
"""Return a dict that is used to apply values to file format strings.
:rtype: dict
"""
@ -412,21 +429,22 @@ class TrackMetadata:
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
"""Create a generator of key, value pairs for use with mutagen.
>>> {attribute_name: key_to_use_for_metadata}
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}
>>> {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()
>>> audio = MP4(path)
>>> for k, v in meta.tags(container='MP4'):
... audio[k] = v
>>> audio.save()
:param container: the container format
:type container: str
@ -442,7 +460,7 @@ class TrackMetadata:
raise InvalidContainerError(f"Invalid container {container}")
def __gen_flac_tags(self) -> Tuple[str, str]:
def __gen_flac_tags(self) -> Generator:
"""Generate key, value pairs to tag FLAC files.
:rtype: Tuple[str, str]
@ -456,7 +474,7 @@ class TrackMetadata:
logger.debug("Adding tag %s: %s", v, tag)
yield (v, str(tag))
def __gen_mp3_tags(self) -> Tuple[str, str]:
def __gen_mp3_tags(self) -> Generator:
"""Generate key, value pairs to tag MP3 files.
:rtype: Tuple[str, str]
@ -472,9 +490,8 @@ class TrackMetadata:
if text is not None and v is not None:
yield (v.__name__, v(encoding=3, text=text))
def __gen_mp4_tags(self) -> Tuple[str, Union[str, int, tuple]]:
"""Generate key, value pairs to tag ALAC or AAC files in
an MP4 container.
def __gen_mp4_tags(self) -> Generator:
"""Generate key, value pairs to tag ALAC or AAC files.
:rtype: Tuple[str, str]
"""
@ -490,6 +507,10 @@ class TrackMetadata:
yield (v, text)
def asdict(self) -> dict:
"""Return a dict representation of self.
:rtype: dict
"""
ret = {}
for attr in dir(self):
if not attr.startswith("_") and not callable(getattr(self, attr)):
@ -512,9 +533,8 @@ class TrackMetadata:
"""
return getattr(self, key)
def get(self, key, default=None) -> str:
"""Returns the requested attribute of the object, with
a default value.
def get(self, key, default=None):
"""Return the requested attribute of the object, with a default value.
:param key:
:param default:
@ -529,8 +549,10 @@ class TrackMetadata:
return default
def set(self, key, val) -> str:
"""Equivalent to
>>> meta[key] = val
"""Set an attribute.
Equivalent to:
>>> meta[key] = val
:param key:
:param val:
@ -539,10 +561,16 @@ class TrackMetadata:
return self.__setitem__(key, val)
def __hash__(self) -> int:
"""Get a hash of this.
Warning: slow.
:rtype: int
"""
return sum(hash(v) for v in self.asdict().values() if isinstance(v, Hashable))
def __repr__(self) -> str:
"""Returns the string representation of the metadata object.
"""Return the string representation of the metadata object.
:rtype: str
"""

22
streamrip/spoofbuz.py
View file

@ -1,4 +1,7 @@
# Credits to Dash for this tool.
"""Get app id and secrets for Qobuz.
Credits to Dash for this tool.
"""
import base64
import re
@ -8,7 +11,10 @@ import requests
class Spoofer:
"""Spoofs the information required to stream tracks from Qobuz."""
def __init__(self):
"""Create a Spoofer."""
self.seed_timezone_regex = (
r'[a-z]\.initialSeed\("(?P<seed>[\w=]+)",window\.ut'
r"imezone\.(?P<timezone>[a-z]+)\)"
@ -33,11 +39,19 @@ class Spoofer:
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_app_id(self) -> str:
"""Get the app id.
:rtype: str
"""
match = re.search(self.app_id_regex, self.bundle)
if match is not None:
return str(match.group("app_id"))
raise Exception("Could not find app id.")
def get_secrets(self):
"""Get secrets."""
seed_matches = re.finditer(self.seed_timezone_regex, self.bundle)
secrets = OrderedDict()
for match in seed_matches:

172
streamrip/tracklists.py
View file

@ -1,13 +1,13 @@
"""These classes parse information from Clients into a universal,
downloadable form.
"""
"""These classes parse information from Clients into a universal, downloadable form."""
from __future__ import annotations
import functools
import logging
import os
import re
from tempfile import gettempdir
from typing import Dict, Generator, Iterable, Union
from typing import Dict, Generator, Iterable, Optional, Union
import click
from pathvalidate import sanitize_filename
@ -52,7 +52,11 @@ class Album(Tracklist):
self.sampling_rate = None
self.bit_depth = None
self.container = None
self.container: Optional[str] = None
self.disctotal: int
self.tracktotal: int
self.albumartist: str
# usually an unpacked TrackMetadata.asdict()
self.__dict__.update(kwargs)
@ -66,7 +70,6 @@ class Album(Tracklist):
def load_meta(self):
"""Load detailed metadata from API using the id."""
assert hasattr(self, "id"), "id must be set to load metadata"
resp = self.client.get(self.id, media_type="album")
@ -82,6 +85,13 @@ class Album(Tracklist):
@classmethod
def from_api(cls, resp: dict, client: Client):
"""Create an Album object from an API response.
:param resp:
:type resp: dict
:param client:
:type client: Client
"""
if client.source == "soundcloud":
return Playlist.from_api(resp, client)
@ -89,6 +99,10 @@ class Album(Tracklist):
return cls(client, **info.asdict())
def _prepare_download(self, **kwargs):
"""Prepare the download of the album.
:param kwargs:
"""
# Generate the folder name
self.folder_format = kwargs.get("folder_format", FOLDER_FORMAT)
self.quality = min(kwargs.get("quality", 3), self.client.max_quality)
@ -146,13 +160,24 @@ class Album(Tracklist):
for item in self.booklets:
Booklet(item).download(parent_folder=self.folder)
def _download_item(
def _download_item( # type: ignore
self,
track: Union[Track, Video],
quality: int = 3,
database: MusicDB = None,
**kwargs,
) -> bool:
"""Download an item.
:param track: The item.
:type track: Union[Track, Video]
:param quality:
:type quality: int
:param database:
:type database: MusicDB
:param kwargs:
:rtype: bool
"""
logger.debug("Downloading track to %s", self.folder)
if self.disctotal > 1 and isinstance(track, Track):
disc_folder = os.path.join(self.folder, f"Disc {track.meta.discnumber}")
@ -171,7 +196,7 @@ class Album(Tracklist):
return True
@staticmethod
def _parse_get_resp(resp: dict, client: Client) -> dict:
def _parse_get_resp(resp: dict, client: Client) -> TrackMetadata:
"""Parse information from a client.get(query, 'album') call.
:param resp:
@ -183,8 +208,7 @@ class Album(Tracklist):
return meta
def _load_tracks(self, resp):
"""Given an album metadata dict returned by the API, append all of its
tracks to `self`.
"""Load the tracks into self from an API response.
This uses a classmethod to convert an item into a Track object, which
stores the metadata inside a TrackMetadata object.
@ -208,6 +232,10 @@ class Album(Tracklist):
)
def _get_formatter(self) -> dict:
"""Get a formatter that is used for previews in core.py.
:rtype: dict
"""
fmt = dict()
for key in ALBUM_KEYS:
# default to None
@ -222,6 +250,14 @@ class Album(Tracklist):
return fmt
def _get_formatted_folder(self, parent_folder: str, quality: int) -> str:
"""Generate the folder name for this album.
:param parent_folder:
:type parent_folder: str
:param quality:
:type quality: int
:rtype: str
"""
# necessary to format the folder
self.container = get_container(quality, self.client.source)
if self.container in ("AAC", "MP3"):
@ -234,10 +270,19 @@ class Album(Tracklist):
@property
def title(self) -> str:
"""Get the title of the album.
:rtype: str
"""
return self.album
@title.setter
def title(self, val: str):
"""Set the title of the Album.
:param val:
:type val: str
"""
self.album = val
def __repr__(self) -> str:
@ -252,17 +297,21 @@ class Album(Tracklist):
return f"<Album: V/A - {self.title}>"
def __str__(self) -> str:
"""Return a readable string representation of
this album.
"""Return a readable string representation of this album.
:rtype: str
"""
return f"{self['albumartist']} - {self['title']}"
def __len__(self) -> int:
"""Get the length of the album.
:rtype: int
"""
return self.tracktotal
def __hash__(self):
"""Hash the album."""
return hash(self.id)
@ -297,8 +346,7 @@ class Playlist(Tracklist):
@classmethod
def from_api(cls, resp: dict, client: Client):
"""Return a Playlist object initialized with information from
a search result returned by the API.
"""Return a Playlist object from an API response.
:param resp: a single search result entry of a playlist
:type resp: dict
@ -321,7 +369,7 @@ class Playlist(Tracklist):
self.loaded = True
def _load_tracks(self, new_tracknumbers: bool = True):
"""Parses the tracklist returned by the API.
"""Parse the tracklist returned by the API.
:param new_tracknumbers: replace tracknumber tag with playlist position
:type new_tracknumbers: bool
@ -409,7 +457,7 @@ class Playlist(Tracklist):
self.__download_index = 1 # used for tracknumbers
self.download_message()
def _download_item(self, item: Track, **kwargs):
def _download_item(self, item: Track, **kwargs) -> bool: # type: ignore
kwargs["parent_folder"] = self.folder
if self.client.source == "soundcloud":
item.load_meta()
@ -433,8 +481,7 @@ class Playlist(Tracklist):
@staticmethod
def _parse_get_resp(item: dict, client: Client) -> dict:
"""Parses information from a search result returned
by a client.search call.
"""Parse information from a search result returned by a client.search call.
:param item:
:type item: dict
@ -469,6 +516,10 @@ class Playlist(Tracklist):
@property
def title(self) -> str:
"""Get the title.
:rtype: str
"""
return self.name
def __repr__(self) -> str:
@ -479,8 +530,7 @@ class Playlist(Tracklist):
return f"<Playlist: {self.name}>"
def __str__(self) -> str:
"""Return a readable string representation of
this track.
"""Return a readable string representation of this track.
:rtype: str
"""
@ -524,13 +574,18 @@ class Artist(Tracklist):
# override
def download(self, **kwargs):
"""Download all items in self.
:param kwargs:
"""
iterator = self._prepare_download(**kwargs)
for item in iterator:
self._download_item(item, **kwargs)
def _load_albums(self):
"""From the discography returned by client.get(query, 'artist'),
generate album objects and append them to self.
"""Load Album objects to self.
This parses the response of client.get(query, 'artist') responses.
"""
if self.client.source == "qobuz":
self.name = self.meta["name"]
@ -554,13 +609,23 @@ class Artist(Tracklist):
def _prepare_download(
self, parent_folder: str = "StreamripDownloads", filters: tuple = (), **kwargs
) -> Iterable:
"""Prepare the download.
:param parent_folder:
:type parent_folder: str
:param filters:
:type filters: tuple
:param kwargs:
:rtype: Iterable
"""
folder = sanitize_filename(self.name)
folder = os.path.join(parent_folder, folder)
self.folder = os.path.join(parent_folder, folder)
logger.debug("Artist folder: %s", folder)
logger.debug(f"Length of tracklist {len(self)}")
logger.debug(f"Filters: {filters}")
final: Iterable
if "repeats" in filters:
final = self._remove_repeats(bit_depth=max, sampling_rate=min)
filters = tuple(f for f in filters if f != "repeats")
@ -575,7 +640,7 @@ class Artist(Tracklist):
self.download_message()
return final
def _download_item(
def _download_item( # type: ignore
self,
item,
parent_folder: str = "StreamripDownloads",
@ -583,15 +648,27 @@ class Artist(Tracklist):
database: MusicDB = None,
**kwargs,
) -> bool:
"""Download an item.
:param item:
:param parent_folder:
:type parent_folder: str
:param quality:
:type quality: int
:param database:
:type database: MusicDB
:param kwargs:
:rtype: bool
"""
try:
item.load_meta()
except NonStreamable:
logger.info("Skipping album, not available to stream.")
return
return False
# always an Album
status = item.download(
parent_folder=parent_folder,
parent_folder=self.folder,
quality=quality,
database=database,
**kwargs,
@ -600,12 +677,17 @@ class Artist(Tracklist):
@property
def title(self) -> str:
"""Get the artist name.
Implemented for consistency.
:rtype: str
"""
return self.name
@classmethod
def from_api(cls, item: dict, client: Client, source: str = "qobuz"):
"""Create an Artist object from the api response of Qobuz, Tidal,
or Deezer.
"""Create an Artist object from the api response of Qobuz, Tidal, or Deezer.
:param resp: response dict
:type resp: dict
@ -652,8 +734,9 @@ class Artist(Tracklist):
}
def _remove_repeats(self, bit_depth=max, sampling_rate=max) -> Generator:
"""Remove the repeated albums from self. May remove different
versions of the same album.
"""Remove the repeated albums from self.
May remove different versions of the same album.
:param bit_depth: either max or min functions
:param sampling_rate: either max or min functions
@ -674,9 +757,7 @@ class Artist(Tracklist):
break
def _non_studio_albums(self, album: Album) -> bool:
"""Passed as a parameter by the user.
This will download only studio albums.
"""Filter non-studio-albums.
:param artist: usually self
:param album: the album to check
@ -689,7 +770,7 @@ class Artist(Tracklist):
)
def _features(self, album: Album) -> bool:
"""Passed as a parameter by the user.
"""Filter features.
This will download only albums where the requested
artist is the album artist.
@ -702,9 +783,7 @@ class Artist(Tracklist):
return self["name"] == album["albumartist"]
def _extras(self, album: Album) -> bool:
"""Passed as a parameter by the user.
This will skip any extras.
"""Filter extras.
:param artist: usually self
:param album: the album to check
@ -714,9 +793,7 @@ class Artist(Tracklist):
return self.TYPE_REGEXES["extra"].search(album.title) is None
def _non_remasters(self, album: Album) -> bool:
"""Passed as a parameter by the user.
This will download only remasterd albums.
"""Filter non remasters.
:param artist: usually self
:param album: the album to check
@ -726,7 +803,7 @@ class Artist(Tracklist):
return self.TYPE_REGEXES["remaster"].search(album.title) is not None
def _non_albums(self, album: Album) -> bool:
"""This will ignore non-album releases.
"""Filter releases that are not albums.
:param artist: usually self
:param album: the album to check
@ -745,19 +822,22 @@ class Artist(Tracklist):
return f"<Artist: {self.name}>"
def __str__(self) -> str:
"""Return a readable string representation of
this Artist.
"""Return a readable string representation of this Artist.
:rtype: str
"""
return self.name
def __hash__(self):
"""Hash self."""
return hash(self.id)
class Label(Artist):
"""Represents a downloadable Label."""
def load_meta(self):
"""Load metadata given an id."""
assert self.client.source == "qobuz", "Label source must be qobuz"
resp = self.client.get(self.id, "label")
@ -768,11 +848,11 @@ class Label(Artist):
self.loaded = True
def __repr__(self):
"""Return a string representation of the Label."""
return f"<Label - {self.name}>"
def __str__(self) -> str:
"""Return a readable string representation of
this track.
"""Return the name of the Label.
:rtype: str
"""
@ -783,7 +863,7 @@ class Label(Artist):
def _get_tracklist(resp: dict, source: str) -> list:
"""Returns the tracklist from an API response.
"""Return the tracklist from an API response.
:param resp:
:type resp: dict

112
streamrip/utils.py
View file

@ -1,3 +1,5 @@
"""Miscellaneous utility functions."""
import base64
import contextlib
import logging
@ -5,7 +7,7 @@ import os
import re
import sys
from string import Formatter
from typing import Hashable, Optional, Union
from typing import Dict, Hashable, Optional, Union
import click
import requests
@ -24,12 +26,20 @@ logger = logging.getLogger(__name__)
def safe_get(d: dict, *keys: Hashable, 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'
"""Traverse dict layers safely.
Usage:
>>> d = {'foo': {'bar': 'baz'}}
>>> _safe_get(d, 'baz')
None
>>> _safe_get(d, 'foo', 'bar')
'baz'
:param d:
:type d: dict
:param keys:
:type keys: Hashable
:param default: the default value to use if a key isn't found
"""
curr = d
res = default
@ -43,15 +53,15 @@ def safe_get(d: dict, *keys: Hashable, default=None):
def get_quality(quality_id: int, source: str) -> Union[str, int]:
"""Given the quality id in (0, 1, 2, 3, 4), return the streaming quality
value to send to the api for a given source.
"""Get the source-specific quality id.
:param quality_id: the quality id
:param quality_id: the universal quality id (0, 1, 2, 4)
:type quality_id: int
:param source: qobuz, tidal, or deezer
:type source: str
:rtype: Union[str, int]
"""
q_map: Dict[int, Union[int, str]]
if source == "qobuz":
q_map = {
1: 5,
@ -81,15 +91,15 @@ def get_quality(quality_id: int, source: str) -> Union[str, int]:
def get_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.
"""Get the universal quality id from bit depth and sampling rate.
: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
# XXX: Should `0` quality be supported?
if bit_depth is None or sampling_rate is None: # is lossy
return 1
if bit_depth == 16:
@ -102,22 +112,8 @@ def get_quality_id(bit_depth: Optional[int], sampling_rate: Optional[int]):
return 4
@contextlib.contextmanager
def std_out_err_redirect_tqdm():
orig_out_err = sys.stdout, sys.stderr
try:
sys.stdout, sys.stderr = map(DummyTqdmFile, orig_out_err)
yield orig_out_err[0]
# Relay exceptions
except Exception as exc:
raise exc
# Always restore sys.stdout/err if necessary
finally:
sys.stdout, sys.stderr = orig_out_err
def tqdm_download(url: str, filepath: str, params: dict = None, desc: str = None):
"""Downloads a file with a progress bar.
"""Download a file with a progress bar.
:param url: url to direct download
:param filepath: file to write
@ -157,7 +153,7 @@ def tqdm_download(url: str, filepath: str, params: dict = None, desc: str = None
def clean_format(formatter: str, format_info):
"""Formats track or folder names sanitizing every formatter key.
"""Format track or folder names sanitizing every formatter key.
:param formatter:
:type formatter: str
@ -180,6 +176,11 @@ def clean_format(formatter: str, format_info):
def tidal_cover_url(uuid, size):
"""Generate a tidal cover url.
:param uuid:
:param size:
"""
possibles = (80, 160, 320, 640, 1280)
assert size in possibles, f"size must be in {possibles}"
@ -202,6 +203,12 @@ def init_log(path: Optional[str] = None, level: str = "DEBUG"):
def decrypt_mqa_file(in_path, out_path, encryption_key):
"""Decrypt an MQA file.
:param in_path:
:param out_path:
:param encryption_key:
"""
# Do not change this
master_key = "UIlTTEMmmLfGowo/UC60x2H45W6MdGgTRfo/umg4754="
@ -233,6 +240,13 @@ def decrypt_mqa_file(in_path, out_path, encryption_key):
def ext(quality: int, source: str):
"""Get the extension of an audio file.
:param quality:
:type quality: int
:param source:
:type source: str
"""
if quality <= 1:
if source == "tidal":
return ".m4a"
@ -245,6 +259,16 @@ def ext(quality: int, source: str):
def gen_threadsafe_session(
headers: dict = None, pool_connections: int = 100, pool_maxsize: int = 100
) -> requests.Session:
"""Create a new Requests session with a large poolsize.
:param headers:
:type headers: dict
:param pool_connections:
:type pool_connections: int
:param pool_maxsize:
:type pool_maxsize: int
:rtype: requests.Session
"""
if headers is None:
headers = {}
@ -266,6 +290,9 @@ def decho(message, fg=None):
logger.debug(message)
interpreter_artist_regex = re.compile(r"getSimilarArtist\(\s*'(\w+)'")
def extract_interpreter_url(url: str) -> str:
"""Extract artist ID from a Qobuz interpreter url.
@ -275,13 +302,20 @@ def extract_interpreter_url(url: str) -> str:
"""
session = gen_threadsafe_session({"User-Agent": AGENT})
r = session.get(url)
artist_id = re.search(r"getSimilarArtist\(\s*'(\w+)'", r.text).group(1)
return artist_id
match = interpreter_artist_regex.search(r.text)
if match:
return match.group(1)
raise Exception(
"Unable to extract artist id from interpreter url. Use a "
"url that contains an artist id."
)
def get_container(quality: int, source: str) -> str:
"""Get the "container" given the quality. `container` can also be the
the codec; both work.
"""Get the file container given the quality.
`container` can also be the the codec; both work.
:param quality: quality id
:type quality: int
@ -290,11 +324,9 @@ def get_container(quality: int, source: str) -> str:
:rtype: str
"""
if quality >= 2:
container = "FLAC"
else:
if source == "tidal":
container = "AAC"
else:
container = "MP3"
return "FLAC"
return container
if source == "tidal":
return "AAC"
return "MP3"