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

Formatting

Browse source 
Signed-off-by: nathom <nathanthomas707@gmail.com>
This commit is contained in:
nathom 2021-06-16 14:26:02 -07:00
parent 1e1e3c7062
commit 76ba2d413b
6 changed files with 42 additions and 111 deletions
Download patch file Download diff file Expand all files Collapse all files

53
streamrip/bases.py
View file

@ -45,9 +45,7 @@ logger = logging.getLogger("streamrip")
TYPE_REGEXES = {
"remaster": re.compile(r"(?i)(re)?master(ed)?"),
"extra": re.compile(
r"(?i)(anniversary|deluxe|live|collector|demo|expanded)"
),
"extra": re.compile(r"(?i)(anniversary|deluxe|live|collector|demo|expanded)"),
}
@ -120,15 +118,12 @@ class Track:
if self.client.source == "qobuz":
self.cover_url = self.resp["album"]["image"]["large"]
elif self.client.source == "tidal":
self.cover_url = tidal_cover_url(
self.resp["album"]["cover"], 320
)
self.cover_url = tidal_cover_url(self.resp["album"]["cover"], 320)
elif self.client.source == "deezer":
self.cover_url = self.resp["album"]["cover_medium"]
elif self.client.source == "soundcloud":
self.cover_url = (
self.resp["artwork_url"]
or self.resp["user"].get("avatar_url")
self.resp["artwork_url"] or self.resp["user"].get("avatar_url")
).replace("large", "t500x500")
else:
raise InvalidSourceError(self.client.source)
@ -175,9 +170,7 @@ class Track:
return False
self.download_cover() # only downloads for playlists and singles
self.path = os.path.join(
gettempdir(), f"{hash(self.id)}_{self.quality}.tmp"
)
self.path = os.path.join(gettempdir(), f"{hash(self.id)}_{self.quality}.tmp")
return True
def download(
@ -352,9 +345,7 @@ class Track:
if not hasattr(self, "cover_url"):
return False
self.cover_path = os.path.join(
gettempdir(), f"cover{hash(self.cover_url)}.jpg"
)
self.cover_path = os.path.join(gettempdir(), f"cover{hash(self.cover_url)}.jpg")
logger.debug(f"Downloading cover from {self.cover_url}")
# click.secho(f"\nDownloading cover art for {self!s}", fg="blue")
@ -377,18 +368,16 @@ class Track:
formatter = self.meta.get_formatter(max_quality=self.quality)
logger.debug("Track meta formatter %s", formatter)
filename = clean_format(self.file_format, formatter)
self.final_path = os.path.join(self.folder, filename)[
:250
].strip() + ext(self.quality, self.client.source)
self.final_path = os.path.join(self.folder, filename)[:250].strip() + ext(
self.quality, self.client.source
)
logger.debug("Formatted path: %s", self.final_path)
return self.final_path
@classmethod
def from_album_meta(
cls, album: TrackMetadata, track: dict, client: Client
):
def from_album_meta(cls, album: TrackMetadata, track: dict, client: Client):
"""Return a new Track object initialized with info.
:param album: album metadata returned by API
@ -456,9 +445,7 @@ class Track:
:param embed_cover: Embed cover art into file
:type embed_cover: bool
"""
assert isinstance(
self.meta, TrackMetadata
), "meta must be TrackMetadata"
assert isinstance(self.meta, TrackMetadata), "meta must be TrackMetadata"
if not self.downloaded:
logger.info(
"Track %s not tagged because it was not downloaded",
@ -554,9 +541,7 @@ class Track:
"""
if not self.downloaded:
logger.debug("Track not downloaded, skipping conversion")
click.secho(
"Track not downloaded, skipping conversion", fg="magenta"
)
click.secho("Track not downloaded, skipping conversion", fg="magenta")
return
CONV_CLASS = {
@ -573,21 +558,15 @@ class Track:
try:
self.container = codec.upper()
except AttributeError:
click.secho(
"Error: No audio codec chosen to convert to.", fg="red"
)
click.secho("Error: No audio codec chosen to convert to.", fg="red")
raise click.Abort
if not hasattr(self, "final_path"):
self.format_final_path()
if not os.path.isfile(self.path):
logger.info(
"File %s does not exist. Skipping conversion.", self.path
)
click.secho(
f"{self!s} does not exist. Skipping conversion.", fg="red"
)
logger.info("File %s does not exist. Skipping conversion.", self.path)
click.secho(f"{self!s} does not exist. Skipping conversion.", fg="red")
return
assert (
@ -854,9 +833,7 @@ class Tracklist(list):
# Tidal errors out with unlimited concurrency
# max_workers = 15 if self.client.source == "tidal" else 90
with concurrent.futures.ThreadPoolExecutor(15) as executor:
futures = [
executor.submit(target, item, **kwargs) for item in self
]
futures = [executor.submit(target, item, **kwargs) for item in self]
try:
concurrent.futures.wait(futures)
except (KeyboardInterrupt, SystemExit):

4
streamrip/constants.py
View file

@ -144,7 +144,9 @@ ALBUM_KEYS = (
"composer",
)
# TODO: rename these to DEFAULT_FOLDER_FORMAT etc
FOLDER_FORMAT = "{albumartist} - {title} ({year}) [{container}] [{bit_depth}B-{sampling_rate}kHz]"
FOLDER_FORMAT = (
"{albumartist} - {title} ({year}) [{container}] [{bit_depth}B-{sampling_rate}kHz]"
)
TRACK_FORMAT = "{tracknumber}. {artist} - {title}"

20
streamrip/converter.py
View file

@ -52,9 +52,7 @@ class Converter:
self.filename = filename
self.final_fn = f"{os.path.splitext(filename)[0]}.{self.container}"
self.tempfile = os.path.join(
gettempdir(), os.path.basename(self.final_fn)
)
self.tempfile = os.path.join(gettempdir(), os.path.basename(self.final_fn))
self.remove_source = remove_source
self.sampling_rate = sampling_rate
self.bit_depth = bit_depth
@ -117,13 +115,9 @@ class Converter:
if self.lossless:
if isinstance(self.sampling_rate, int):
sampling_rates = "|".join(
str(rate)
for rate in SAMPLING_RATES
if rate <= self.sampling_rate
)
command.extend(
["-af", f"aformat=sample_rates={sampling_rates}"]
str(rate) for rate in SAMPLING_RATES if rate <= self.sampling_rate
)
command.extend(["-af", f"aformat=sample_rates={sampling_rates}"])
elif self.sampling_rate is not None:
raise TypeError(
@ -138,9 +132,7 @@ class Converter:
else:
raise ValueError("Bit depth must be 16, 24, or 32")
elif self.bit_depth is not None:
raise TypeError(
f"Bit depth must be int, not {type(self.bit_depth)}"
)
raise TypeError(f"Bit depth must be int, not {type(self.bit_depth)}")
# automatically overwrite
command.extend(["-y", self.tempfile])
@ -203,9 +195,7 @@ class Vorbis(Converter):
codec_name = "vorbis"
codec_lib = "libvorbis"
container = "ogg"
default_ffmpeg_arg = (
"-q:a 6" # 160, aka the "high" quality profile from Spotify
)
default_ffmpeg_arg = "-q:a 6" # 160, aka the "high" quality profile from Spotify
class OPUS(Converter):

38
streamrip/metadata.py
View file

@ -130,9 +130,7 @@ class TrackMetadata:
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"
)
self.date = resp.get("release_date_original") or resp.get("release_date")
self.copyright = resp.get("copyright")
self.albumartist = safe_get(resp, "artist", "name")
self.composer = safe_get(resp, "composer", "name")
@ -141,9 +139,7 @@ class TrackMetadata:
self.disctotal = (
max(
track.get("media_number", 1)
for track in safe_get(
resp, "tracks", "items", default=[{}]
)
for track in safe_get(resp, "tracks", "items", default=[{}])
)
or 1
)
@ -155,9 +151,7 @@ class TrackMetadata:
# Non-embedded information
self.version = resp.get("version")
self.cover_urls = OrderedDict(resp["image"])
self.cover_urls["original"] = self.cover_urls["large"].replace(
"600", "org"
)
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")
self.sampling_rate = resp.get("maximum_sampling_rate")
@ -191,22 +185,14 @@ class TrackMetadata:
)
self.streamable = resp.get("allowStreaming", False)
if q := resp.get(
"audioQuality"
): # for album entries in single tracks
if q := resp.get("audioQuality"): # for album entries in single tracks
self._get_tidal_quality(q)
elif self.__source == "deezer":
self.album = resp.get("title", "Unknown Album")
self.tracktotal = resp.get("track_total", 0) or resp.get(
"nb_tracks", 0
)
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
max(track.get("disk_number") for track in resp.get("tracks", [{}])) or 1
)
self.genre = safe_get(resp, "genres", "data")
self.date = resp.get("release_date")
@ -370,9 +356,7 @@ class TrackMetadata:
if isinstance(self._genres, list):
if self.__source == "qobuz":
genres: Iterable = 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)
@ -404,9 +388,7 @@ class TrackMetadata:
if hasattr(self, "_copyright"):
if self._copyright is None:
return None
copyright: str = 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
@ -598,9 +580,7 @@ class TrackMetadata:
:rtype: int
"""
return sum(
hash(v) for v in self.asdict().values() if isinstance(v, Hashable)
)
return sum(hash(v) for v in self.asdict().values() if isinstance(v, Hashable))
def __repr__(self) -> str:
"""Return the string representation of the metadata object.

37
streamrip/tracklists.py
View file

@ -131,9 +131,7 @@ class Album(Tracklist):
tqdm_download(cover_url, cover_path)
hires_cov_path = os.path.join(self.folder, "cover.jpg")
if kwargs.get("keep_hires_cover", True) and not os.path.exists(
hires_cov_path
):
if kwargs.get("keep_hires_cover", True) and not os.path.exists(hires_cov_path):
tqdm_download(self.cover_urls["original"], hires_cov_path)
cover_size = os.path.getsize(cover_path)
@ -184,9 +182,7 @@ class Album(Tracklist):
"""
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}"
)
disc_folder = os.path.join(self.folder, f"Disc {track.meta.discnumber}")
kwargs["parent_folder"] = disc_folder
else:
kwargs["parent_folder"] = self.folder
@ -273,9 +269,7 @@ class Album(Tracklist):
# lossy codecs don't have these metrics
self.bit_depth = self.sampling_rate = None
formatted_folder = clean_format(
self.folder_format, self._get_formatter()
)
formatted_folder = clean_format(self.folder_format, self._get_formatter())
return os.path.join(parent_folder, formatted_folder)
@ -388,9 +382,7 @@ class Playlist(Tracklist):
if self.client.source == "qobuz":
self.name = self.meta["name"]
self.image = self.meta["images"]
self.creator = safe_get(
self.meta, "owner", "name", default="Qobuz"
)
self.creator = safe_get(self.meta, "owner", "name", default="Qobuz")
tracklist = self.meta["tracks"]["items"]
@ -403,9 +395,7 @@ class Playlist(Tracklist):
elif self.client.source == "tidal":
self.name = self.meta["title"]
self.image = tidal_cover_url(self.meta["image"], 640)
self.creator = safe_get(
self.meta, "creator", "name", default="TIDAL"
)
self.creator = safe_get(self.meta, "creator", "name", default="TIDAL")
tracklist = self.meta["tracks"]
@ -422,9 +412,7 @@ class Playlist(Tracklist):
elif self.client.source == "deezer":
self.name = self.meta["title"]
self.image = self.meta["picture_big"]
self.creator = safe_get(
self.meta, "creator", "name", default="Deezer"
)
self.creator = safe_get(self.meta, "creator", "name", default="Deezer")
tracklist = self.meta["tracks"]
@ -467,9 +455,7 @@ class Playlist(Tracklist):
logger.debug(f"Loaded {len(self)} tracks from playlist {self.name}")
def _prepare_download(
self, parent_folder: str = "StreamripDownloads", **kwargs
):
def _prepare_download(self, parent_folder: str = "StreamripDownloads", **kwargs):
fname = sanitize_filename(self.name)
self.folder = os.path.join(parent_folder, fname)
@ -667,9 +653,7 @@ class Artist(Tracklist):
final = self
if isinstance(filters, tuple) and self.client.source == "qobuz":
filter_funcs = (
getattr(self, f"_{filter_}") for filter_ in filters
)
filter_funcs = (getattr(self, f"_{filter_}") for filter_ in filters)
for func in filter_funcs:
final = filter(func, final)
@ -788,10 +772,7 @@ class Artist(Tracklist):
best_bd = bit_depth(a["bit_depth"] for a in group)
best_sr = sampling_rate(a["sampling_rate"] for a in group)
for album in group:
if (
album["bit_depth"] == best_bd
and album["sampling_rate"] == best_sr
):
if album["bit_depth"] == best_bd and album["sampling_rate"] == best_sr:
yield album
break

1
streamrip/utils.py
View file

@ -1,5 +1,6 @@
"""Miscellaneous utility functions."""
from __future__ import annotations
import base64
import logging
import os