Added support for SoundCloud downloads

Move soundcloud album parsing to Playlist

stash

Soundcloud downloads working
This commit is contained in:
nathom 2021-04-05 17:43:23 -07:00
parent 7f413c8290
commit 9d0a735cf5
11 changed files with 259 additions and 86 deletions

3
.gitignore vendored
View file

@ -7,3 +7,6 @@ test.py
/urls.txt /urls.txt
*.flac *.flac
/Downloads /Downloads
*.mp3
StreamripDownloads
*.wav

View file

@ -206,7 +206,7 @@ def config(ctx, **kwargs):
config.reset() config.reset()
if kwargs["open"]: if kwargs["open"]:
click.secho(f"Opening {CONFIG_PATH}", fg='green') click.secho(f"Opening {CONFIG_PATH}", fg="green")
click.launch(CONFIG_PATH) click.launch(CONFIG_PATH)
if kwargs["qobuz"]: if kwargs["qobuz"]:

View file

@ -4,7 +4,7 @@ import json
import logging import logging
import time import time
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from pprint import pformat # , pprint from pprint import pformat, pprint
from typing import Generator, Sequence, Tuple, Union from typing import Generator, Sequence, Tuple, Union
import click import click
@ -16,6 +16,7 @@ from .constants import (
AVAILABLE_QUALITY_IDS, AVAILABLE_QUALITY_IDS,
DEEZER_MAX_Q, DEEZER_MAX_Q,
QOBUZ_FEATURED_KEYS, QOBUZ_FEATURED_KEYS,
SOUNDCLOUD_CLIENT_ID,
TIDAL_MAX_Q, TIDAL_MAX_Q,
) )
from .exceptions import ( from .exceptions import (
@ -50,6 +51,9 @@ QOBUZ_BASE = "https://www.qobuz.com/api.json/0.2"
DEEZER_BASE = "https://api.deezer.com" DEEZER_BASE = "https://api.deezer.com"
DEEZER_DL = "http://dz.loaderapp.info/deezer" DEEZER_DL = "http://dz.loaderapp.info/deezer"
# SoundCloud
SOUNDCLOUD_BASE = "https://api-v2.soundcloud.com"
# ----------- Abstract Classes ----------------- # ----------- Abstract Classes -----------------
@ -101,12 +105,18 @@ class ClientInterface(ABC):
def source(self): def source(self):
pass pass
@property
@abstractmethod
def max_quality(self):
pass
# ------------- Clients ----------------- # ------------- Clients -----------------
class QobuzClient(ClientInterface): class QobuzClient(ClientInterface):
source = "qobuz" source = "qobuz"
max_quality = 4
# ------- Public Methods ------------- # ------- Public Methods -------------
def __init__(self): def __init__(self):
@ -361,6 +371,7 @@ class QobuzClient(ClientInterface):
class DeezerClient(ClientInterface): class DeezerClient(ClientInterface):
source = "deezer" source = "deezer"
max_quality = 2
def __init__(self): def __init__(self):
self.session = requests.Session() self.session = requests.Session()
@ -421,6 +432,7 @@ class DeezerClient(ClientInterface):
class TidalClient(ClientInterface): class TidalClient(ClientInterface):
source = "tidal" source = "tidal"
max_quality = 3
def __init__(self): def __init__(self):
self.logged_in = False self.logged_in = False
@ -639,3 +651,66 @@ class TidalClient(ClientInterface):
def _api_post(self, url, data, auth=None): def _api_post(self, url, data, auth=None):
r = requests.post(url, data=data, auth=auth, verify=False).json() r = requests.post(url, data=data, auth=auth, verify=False).json()
return r return r
class SoundCloudClient(ClientInterface):
source = "soundcloud"
max_quality = 0
logged_in = True
def login(self):
raise NotImplementedError
def get(self, id, media_type="track"):
assert media_type in ("track", "playlist"), f"{media_type} not supported"
if "http" in str(id):
resp, _ = self._get(f"resolve?url={id}")
elif media_type == "track":
resp, _ = self._get(f"{media_type}s/{id}")
else:
raise Exception(id)
return resp
def get_file_url(self, track: dict, quality) -> dict:
if not track["streamable"] or track["policy"] == "BLOCK":
raise Exception
if track["downloadable"] and track["has_downloads_left"]:
r = self._get(f"tracks/{track['id']}/download", resp_obj=True)
return {"url": r.json()["redirectUri"], "type": "original"}
else:
url = None
for tc in track["media"]["transcodings"]:
fmt = tc["format"]
if fmt["protocol"] == "hls" and fmt["mime_type"] == "audio/mpeg":
url = tc["url"]
break
assert url is not None
resp, _ = self._get(url, no_base=True)
return {"url": resp["url"], "type": "mp3"}
def search(self, query: str, media_type="album"):
params = {"q": query}
resp, _ = self._get(f"search/{media_type}s", params=params)
return resp
def _get(self, path, params=None, no_base=False, resp_obj=False):
if params is None:
params = {}
params["client_id"] = SOUNDCLOUD_CLIENT_ID
if no_base:
url = path
else:
url = f"{SOUNDCLOUD_BASE}/{path}"
logger.debug(f"Fetching url {url}")
r = requests.get(url, params=params)
if resp_obj:
return r
return r.json(), r.status_code

View file

@ -54,6 +54,9 @@ class Config:
"deezer": { "deezer": {
"quality": 2, "quality": 2,
}, },
"soundcloud": {
"quality": 0,
},
"database": {"enabled": True, "path": None}, "database": {"enabled": True, "path": None},
"conversion": { "conversion": {
"enabled": False, "enabled": False,

View file

@ -19,6 +19,7 @@ AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:83.0) Gecko/20100101 Firef
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"
EXT = { EXT = {
0: ".mp3",
1: ".mp3", 1: ".mp3",
2: ".flac", 2: ".flac",
3: ".flac", 3: ".flac",
@ -134,11 +135,14 @@ FOLDER_FORMAT = (
TRACK_FORMAT = "{tracknumber}. {artist} - {title}" TRACK_FORMAT = "{tracknumber}. {artist} - {title}"
URL_REGEX = ( URL_REGEX = (
r"https:\/\/(?:www|open|play|listen)?\.?(\w+)\.com(?:(?:\/(track|playlist|album|" r"https:\/\/(?:www|open|play|listen)?\.?(qobuz|tidal|deezer)\.com(?:(?:\/(track|playlist|album|"
r"artist|label))|(?:\/[-\w]+?))+\/([-\w]+)" r"artist|label))|(?:\/[-\w]+?))+\/([-\w]+)"
) )
SOUNDCLOUD_URL_REGEX = r"https://soundcloud.com/[-\w:/]+"
SOUNDCLOUD_CLIENT_ID = "a3e059563d7fd3372b49b37f00a00bcf"
TIDAL_MAX_Q = 7 TIDAL_MAX_Q = 7
DEEZER_MAX_Q = 6 DEEZER_MAX_Q = 6
AVAILABLE_QUALITY_IDS = (0, 1, 2, 3, 4) AVAILABLE_QUALITY_IDS = (0, 1, 2, 3, 4)
MEDIA_TYPES = ("track", "album", "artist", "label", "playlist")

View file

@ -97,7 +97,7 @@ class Converter:
"-i", "-i",
self.filename, self.filename,
"-loglevel", "-loglevel",
"warning", "panic",
"-c:a", "-c:a",
self.codec_lib, self.codec_lib,
] ]

View file

@ -1,4 +1,5 @@
import logging import logging
from pprint import pprint
import os import os
import re import re
import sys import sys
@ -9,9 +10,9 @@ from typing import Generator, Optional, Tuple, Union
import click import click
from .clients import DeezerClient, QobuzClient, TidalClient from .clients import DeezerClient, QobuzClient, SoundCloudClient, TidalClient
from .config import Config from .config import Config
from .constants import CONFIG_PATH, DB_PATH, URL_REGEX from .constants import (CONFIG_PATH, DB_PATH, SOUNDCLOUD_URL_REGEX, URL_REGEX, MEDIA_TYPES)
from .db import MusicDB from .db import MusicDB
from .downloader import Album, Artist, Label, Playlist, Track from .downloader import Album, Artist, Label, Playlist, Track
from .exceptions import AuthenticationError, ParsingError from .exceptions import AuthenticationError, ParsingError
@ -27,7 +28,6 @@ MEDIA_CLASS = {
"track": Track, "track": Track,
"label": Label, "label": Label,
} }
CLIENTS = {"qobuz": QobuzClient, "tidal": TidalClient, "deezer": DeezerClient}
Media = Union[Album, Playlist, Artist, Track] Media = Union[Album, Playlist, Artist, Track]
@ -38,6 +38,7 @@ class MusicDL(list):
): ):
self.url_parse = re.compile(URL_REGEX) self.url_parse = re.compile(URL_REGEX)
self.soundcloud_url_parse = re.compile(SOUNDCLOUD_URL_REGEX)
self.config = config self.config = config
if self.config is None: if self.config is None:
self.config = Config(CONFIG_PATH) self.config = Config(CONFIG_PATH)
@ -46,6 +47,7 @@ class MusicDL(list):
"qobuz": QobuzClient(), "qobuz": QobuzClient(),
"tidal": TidalClient(), "tidal": TidalClient(),
"deezer": DeezerClient(), "deezer": DeezerClient(),
"soundcloud": SoundCloudClient(),
} }
if config.session["database"]["enabled"]: if config.session["database"]["enabled"]:
@ -71,9 +73,9 @@ class MusicDL(list):
f"Enter {capitalize(source)} password (will not show on screen):", f"Enter {capitalize(source)} password (will not show on screen):",
fg="green", fg="green",
) )
self.config.file[source]["password"] = md5(getpass( self.config.file[source]["password"] = md5(
prompt="" getpass(prompt="").encode("utf-8")
).encode('utf-8')).hexdigest() ).hexdigest()
self.config.save() self.config.save()
click.secho(f'Credentials saved to config file at "{self.config._path}"') click.secho(f'Credentials saved to config file at "{self.config._path}"')
@ -81,11 +83,19 @@ class MusicDL(list):
raise Exception raise Exception
def assert_creds(self, source: str): def assert_creds(self, source: str):
assert source in ("qobuz", "tidal", "deezer"), f"Invalid source {source}" assert source in (
"qobuz",
"tidal",
"deezer",
"soundcloud",
), f"Invalid source {source}"
if source == "deezer": if source == "deezer":
# no login for deezer # no login for deezer
return return
if source == "soundcloud":
return
if source == "qobuz" and ( if source == "qobuz" and (
self.config.file[source]["email"] is None self.config.file[source]["email"] is None
or self.config.file[source]["password"] is None or self.config.file[source]["password"] is None
@ -118,6 +128,11 @@ class MusicDL(list):
client = self.get_client(source) client = self.get_client(source)
if media_type not in MEDIA_TYPES:
if 'playlist' in media_type: # for SoundCloud
media_type = 'playlist'
assert media_type in MEDIA_TYPES, media_type
item = MEDIA_CLASS[media_type](client=client, id=item_id) item = MEDIA_CLASS[media_type](client=client, id=item_id)
self.append(item) self.append(item)
@ -200,7 +215,15 @@ class MusicDL(list):
:raises exceptions.ParsingError :raises exceptions.ParsingError
""" """
parsed = self.url_parse.findall(url) parsed = self.url_parse.findall(url) # Qobuz, Tidal, Dezer
soundcloud_urls = self.soundcloud_url_parse.findall(url)
soundcloud_items = [self.clients["soundcloud"].get(u) for u in soundcloud_urls]
parsed.extend(
("soundcloud", item["kind"], url)
for item, url in zip(soundcloud_items, soundcloud_urls)
)
logger.debug(f"Parsed urls: {parsed}") logger.debug(f"Parsed urls: {parsed}")
if parsed != []: if parsed != []:

View file

@ -61,5 +61,5 @@ class MusicDB:
) )
conn.commit() conn.commit()
except sqlite3.Error as e: except sqlite3.Error as e:
if 'UNIQUE' not in str(e): if "UNIQUE" not in str(e):
raise raise

View file

@ -2,11 +2,14 @@ import logging
import os import os
import re import re
import shutil import shutil
from pprint import pformat import subprocess
import sys
from pprint import pformat, pprint
from tempfile import gettempdir from tempfile import gettempdir
from typing import Any, Callable, Optional, Tuple, Union from typing import Any, Callable, Optional, Tuple, Union
import click import click
import requests
from mutagen.flac import FLAC, Picture from mutagen.flac import FLAC, Picture
from mutagen.id3 import APIC, ID3, ID3NoHeaderError from mutagen.id3 import APIC, ID3, ID3NoHeaderError
from pathvalidate import sanitize_filename, sanitize_filepath from pathvalidate import sanitize_filename, sanitize_filepath
@ -18,6 +21,7 @@ from .constants import (
EXT, EXT,
FLAC_MAX_BLOCKSIZE, FLAC_MAX_BLOCKSIZE,
FOLDER_FORMAT, FOLDER_FORMAT,
SOUNDCLOUD_CLIENT_ID,
TRACK_FORMAT, TRACK_FORMAT,
) )
from .db import MusicDB from .db import MusicDB
@ -116,17 +120,19 @@ class Track:
assert hasattr(self, "id"), "id must be set before loading metadata" assert hasattr(self, "id"), "id must be set before loading metadata"
track_meta = self.client.get(self.id, media_type="track") self.resp = self.client.get(self.id, media_type="track")
self.meta = TrackMetadata( self.meta = TrackMetadata(
track=track_meta, source=self.client.source track=self.resp, source=self.client.source
) # meta dict -> TrackMetadata object ) # meta dict -> TrackMetadata object
try: try:
if self.client.source == "qobuz": if self.client.source == "qobuz":
self.cover_url = track_meta["album"]["image"]["small"] self.cover_url = self.resp["album"]["image"]["small"]
elif self.client.source == "tidal": elif self.client.source == "tidal":
self.cover_url = tidal_cover_url(track_meta["album"]["cover"], 320) self.cover_url = tidal_cover_url(self.resp["album"]["cover"], 320)
elif self.client.source == "deezer": elif self.client.source == "deezer":
self.cover_url = track_meta["album"]["cover_medium"] 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")).replace("large", "t500x500")
else: else:
raise InvalidSourceError(self.client.source) raise InvalidSourceError(self.client.source)
except KeyError: except KeyError:
@ -144,7 +150,7 @@ class Track:
def download( def download(
self, self,
quality: int = 7, quality: int = 3,
parent_folder: str = "StreamripDownloads", parent_folder: str = "StreamripDownloads",
progress_bar: bool = True, progress_bar: bool = True,
database: MusicDB = None, database: MusicDB = None,
@ -162,10 +168,8 @@ class Track:
:type progress_bar: bool :type progress_bar: bool
""" """
# args override attributes # args override attributes
self.quality, self.folder = ( self.quality = min(quality, self.client.max_quality)
quality or self.quality, self.folder = parent_folder or self.folder
parent_folder or self.folder,
)
self.file_format = kwargs.get("track_format", TRACK_FORMAT) self.file_format = kwargs.get("track_format", TRACK_FORMAT)
self.folder = sanitize_filepath(self.folder, platform="auto") self.folder = sanitize_filepath(self.folder, platform="auto")
@ -189,11 +193,17 @@ class Track:
return False return False
if hasattr(self, "cover_url"): # only for playlists and singles if hasattr(self, "cover_url"): # only for playlists and singles
logger.debug("Downloading cover")
self.download_cover() self.download_cover()
dl_info = self.client.get_file_url(self.id, quality) if self.client.source == "soundcloud":
url_id = self.resp
else:
url_id = self.id
temp_file = os.path.join(gettempdir(), f"~{self.id}_{quality}.tmp") dl_info = self.client.get_file_url(url_id, self.quality)
temp_file = os.path.join(gettempdir(), f"~{hash(self.id)}_{quality}.tmp")
logger.debug("Temporary file path: %s", temp_file) logger.debug("Temporary file path: %s", temp_file)
if self.client.source == "qobuz": if self.client.source == "qobuz":
@ -212,7 +222,8 @@ class Track:
if self.client.source in ("qobuz", "tidal"): if self.client.source in ("qobuz", "tidal"):
logger.debug("Downloadable URL found: %s", dl_info.get("url")) logger.debug("Downloadable URL found: %s", dl_info.get("url"))
tqdm_download(dl_info["url"], temp_file) # downloads file tqdm_download(dl_info["url"], temp_file) # downloads file
elif isinstance(dl_info, str): # Deezer
elif self.client.source == "deezer": # Deezer
logger.debug("Downloadable URL found: %s", dl_info) logger.debug("Downloadable URL found: %s", dl_info)
try: try:
tqdm_download(dl_info, temp_file) # downloads file tqdm_download(dl_info, temp_file) # downloads file
@ -220,6 +231,34 @@ class Track:
logger.debug(f"Track is not downloadable {dl_info}") logger.debug(f"Track is not downloadable {dl_info}")
click.secho("Track is not available for download", fg="red") click.secho("Track is not available for download", fg="red")
return False return False
elif self.client.source == "soundcloud":
if dl_info["type"] == "mp3":
temp_file += ".mp3"
# convert hls stream to mp3
subprocess.call(
[
"ffmpeg",
"-i",
dl_info['url'],
"-c",
"copy",
"-y",
temp_file,
"-loglevel",
"fatal",
]
)
elif dl_info["type"] == "original":
tqdm_download(dl_info["url"], temp_file)
# if a wav is returned, convert to flac
engine = converter.FLAC(temp_file)
temp_file = f"{temp_file}.flac"
engine.convert(custom_fn=temp_file)
self.final_path = self.final_path.replace(".mp3", ".flac")
self.quality = 2
else: else:
raise InvalidSourceError(self.client.source) raise InvalidSourceError(self.client.source)
@ -249,18 +288,15 @@ class Track:
assert hasattr(self, "cover_url"), "must set cover_url attribute" assert hasattr(self, "cover_url"), "must set cover_url attribute"
self.cover_path = os.path.join(self.folder, f"cover{hash(self.meta.album)}.jpg") self.cover_path = os.path.join(self.folder, f"cover{hash(self.cover_url)}.jpg")
logger.debug(f"Downloading cover from {self.cover_url}") logger.debug(f"Downloading cover from {self.cover_url}")
click.secho(f"\nDownloading cover art for {self!s}", fg='blue') click.secho(f"\nDownloading cover art for {self!s}", fg="blue")
if not os.path.exists(self.cover_path): if not os.path.exists(self.cover_path):
tqdm_download(self.cover_url, self.cover_path) tqdm_download(self.cover_url, self.cover_path)
else: else:
logger.debug("Cover already exists, skipping download") logger.debug("Cover already exists, skipping download")
self.cover = Tracklist.get_cover_obj(self.cover_path, self.quality)
logger.debug(f"Cover obj: {self.cover}")
def format_final_path(self) -> str: def format_final_path(self) -> str:
"""Return the final filepath of the downloaded file. """Return the final filepath of the downloaded file.
@ -359,16 +395,13 @@ class Track:
self.container = "FLAC" self.container = "FLAC"
logger.debug("Tagging file with %s container", self.container) logger.debug("Tagging file with %s container", self.container)
audio = FLAC(self.final_path) audio = FLAC(self.final_path)
elif self.quality == 1: elif self.quality <= 1:
self.container = "MP3" self.container = "MP3"
logger.debug("Tagging file with %s container", self.container) logger.debug("Tagging file with %s container", self.container)
try: try:
audio = ID3(self.final_path) audio = ID3(self.final_path)
except ID3NoHeaderError: except ID3NoHeaderError:
audio = ID3() audio = ID3()
elif self.quality == 0: # tidal and deezer
# TODO: add compatibility with MP4 container
raise NotImplementedError("Qualities < 320kbps not implemented")
else: else:
raise InvalidQuality(f'Invalid quality: "{self.quality}"') raise InvalidQuality(f'Invalid quality: "{self.quality}"')
@ -377,9 +410,9 @@ class Track:
audio[k] = v audio[k] = v
if embed_cover and cover is None: if embed_cover and cover is None:
assert hasattr(self, "cover") assert hasattr(self, "cover_path")
cover = self.cover
cover = Tracklist.get_cover_obj(self.cover_path, self.quality)
if isinstance(audio, FLAC): if isinstance(audio, FLAC):
if embed_cover: if embed_cover:
audio.add_picture(cover) audio.add_picture(cover)
@ -573,7 +606,7 @@ class Tracklist(list):
:type quality: int :type quality: int
:rtype: Union[Picture, APIC] :rtype: Union[Picture, APIC]
""" """
cover_type = {1: APIC, 2: Picture, 3: Picture, 4: Picture} cover_type = {0: APIC, 1: APIC, 2: Picture, 3: Picture, 4: Picture}
cover = cover_type.get(quality) cover = cover_type.get(quality)
if cover is Picture: if cover is Picture:
@ -731,7 +764,6 @@ class Album(Tracklist):
"tracktotal": resp.get("numberOfTracks"), "tracktotal": resp.get("numberOfTracks"),
} }
elif client.source == "deezer": elif client.source == "deezer":
logger.debug(pformat(resp))
return { return {
"id": resp.get("id"), "id": resp.get("id"),
"title": resp.get("title"), "title": resp.get("title"),
@ -794,7 +826,7 @@ class Album(Tracklist):
def download( def download(
self, self,
quality: int = 7, quality: int = 3,
parent_folder: Union[str, os.PathLike] = "StreamripDownloads", parent_folder: Union[str, os.PathLike] = "StreamripDownloads",
database: MusicDB = None, database: MusicDB = None,
**kwargs, **kwargs,
@ -829,7 +861,7 @@ class Album(Tracklist):
logger.debug("Cover already downloaded: %s. Skipping", cover_path) logger.debug("Cover already downloaded: %s. Skipping", cover_path)
else: else:
click.secho("Downloading cover art", fg="magenta") click.secho("Downloading cover art", fg="magenta")
if kwargs.get("large_cover", False): if kwargs.get("large_cover", True):
cover_url = self.cover_urls.get("large") cover_url = self.cover_urls.get("large")
if self.client.source == "qobuz": if self.client.source == "qobuz":
tqdm_download(cover_url.replace("600", "org"), cover_path) tqdm_download(cover_url.replace("600", "org"), cover_path)
@ -847,7 +879,7 @@ class Album(Tracklist):
else: else:
tqdm_download(self.cover_urls["small"], cover_path) tqdm_download(self.cover_urls["small"], cover_path)
embed_cover = kwargs.get('embed_cover', True) # embed by default embed_cover = kwargs.get("embed_cover", True) # embed by default
if self.client.source != "deezer" and embed_cover: if self.client.source != "deezer" and embed_cover:
cover = self.get_cover_obj(cover_path, quality) cover = self.get_cover_obj(cover_path, quality)
@ -881,17 +913,18 @@ class Album(Tracklist):
else: else:
fmt[key] = None fmt[key] = None
fmt["sampling_rate"] /= 1000 if fmt.get("sampling_rate", False):
# 48.0kHz -> 48kHz, 44.1kHz -> 44.1kHz fmt["sampling_rate"] /= 1000
if fmt["sampling_rate"] % 1 == 0.0: # 48.0kHz -> 48kHz, 44.1kHz -> 44.1kHz
fmt["sampling_rate"] = int(fmt["sampling_rate"]) if fmt["sampling_rate"] % 1 == 0.0:
fmt["sampling_rate"] = int(fmt["sampling_rate"])
return fmt return fmt
def _get_formatted_folder(self, parent_folder: str) -> str: def _get_formatted_folder(self, parent_folder: str) -> str:
if self.bit_depth is not None and self.sampling_rate is not None: if self.bit_depth is not None and self.sampling_rate is not None:
self.container = "FLAC" self.container = "FLAC"
elif self.client.source in ("qobuz", "deezer"): elif self.client.source in ("qobuz", "deezer", "soundcloud"):
self.container = "MP3" self.container = "MP3"
elif self.client.source == "tidal": elif self.client.source == "tidal":
self.container = "AAC" self.container = "AAC"
@ -930,7 +963,7 @@ class Playlist(Tracklist):
"""Represents a downloadable playlist. """Represents a downloadable playlist.
Usage: Usage:
>>> resp = client.get('hip hop', 'playlist') >>> resp = client.search('hip hop', 'playlist')
>>> pl = Playlist.from_api(resp['items'][0], client) >>> pl = Playlist.from_api(resp['items'][0], client)
>>> pl.load_meta() >>> pl.load_meta()
>>> pl.download() >>> pl.download()
@ -973,7 +1006,7 @@ class Playlist(Tracklist):
:type new_tracknumbers: bool :type new_tracknumbers: bool
:param kwargs: :param kwargs:
""" """
self.meta = self.client.get(self.id, "playlist") self.meta = self.client.get(id=self.id, media_type="playlist")
self._load_tracks(**kwargs) self._load_tracks(**kwargs)
def _load_tracks(self, new_tracknumbers: bool = True): def _load_tracks(self, new_tracknumbers: bool = True):
@ -983,17 +1016,17 @@ class Playlist(Tracklist):
:type new_tracknumbers: bool :type new_tracknumbers: bool
""" """
if self.client.source == "qobuz": if self.client.source == "qobuz":
self.name = self.meta['name'] self.name = self.meta["name"]
tracklist = self.meta["tracks"]["items"] tracklist = self.meta["tracks"]["items"]
def gen_cover(track): # ? def gen_cover(track):
return track["album"]["image"]["small"] return track["album"]["image"]["small"]
def meta_args(track): def meta_args(track):
return {"track": track, "album": track["album"]} return {"track": track, "album": track["album"]}
elif self.client.source == "tidal": elif self.client.source == "tidal":
self.name = self.meta['title'] self.name = self.meta["title"]
tracklist = self.meta["tracks"] tracklist = self.meta["tracks"]
def gen_cover(track): def gen_cover(track):
@ -1007,41 +1040,49 @@ class Playlist(Tracklist):
} }
elif self.client.source == "deezer": elif self.client.source == "deezer":
self.name = self.meta['title'] self.name = self.meta["title"]
tracklist = self.meta["tracks"] tracklist = self.meta["tracks"]
def gen_cover(track): def gen_cover(track):
return track["album"]["cover_medium"] return track["album"]["cover_medium"]
def meta_args(track): elif self.client.source == "soundcloud":
return {"track": track, "source": self.client.source} self.name = self.meta["title"]
tracklist = self.meta["tracks"]
def gen_cover(track):
return track["artwork_url"].replace("large", "t500x500")
else: else:
raise NotImplementedError raise NotImplementedError
for i, track in enumerate(tracklist): if self.client.source == "soundcloud":
# TODO: This should be managed with .m3u files and alike. Arbitrary # No meta is included in soundcloud playlist
# tracknumber tags might cause conflicts if the playlist files are # response, so it is loaded at download time
# inside of a library folder for track in tracklist:
meta = TrackMetadata(**meta_args(track)) self.append(Track(self.client, id=track["id"]))
if new_tracknumbers: else:
meta["tracknumber"] = str(i + 1) for track in tracklist:
# TODO: This should be managed with .m3u files and alike. Arbitrary
# tracknumber tags might cause conflicts if the playlist files are
# inside of a library folder
meta = TrackMetadata(track=track, source=self.client.source)
self.append( self.append(
Track( Track(
self.client, self.client,
id=track.get("id"), id=track.get("id"),
meta=meta, meta=meta,
cover_url=gen_cover(track), cover_url=gen_cover(track),
)
) )
)
logger.debug(f"Loaded {len(self)} tracks from playlist {self.name}") logger.debug(f"Loaded {len(self)} tracks from playlist {self.name}")
def download( def download(
self, self,
parent_folder: str = "Downloads", parent_folder: str = "StreamripDownloads",
quality: int = 6, quality: int = 3,
filters: Callable = None, filters: Callable = None,
database: MusicDB = None, database: MusicDB = None,
**kwargs, **kwargs,
@ -1060,10 +1101,19 @@ class Playlist(Tracklist):
logger.debug(f"Parent folder {folder}") logger.debug(f"Parent folder {folder}")
self.download_message() self.download_message()
for track in self: for i, track in enumerate(self):
track.download(parent_folder=folder, quality=quality, database=database) if self.client.source == "soundcloud":
if self.client.source != "deezer": track.load_meta()
track.tag(embed_cover=kwargs.get('embed_cover', True))
if kwargs.get("new_tracknumbers", True):
track.meta["tracknumber"] = str(i + 1)
if (
track.download(parent_folder=folder, quality=quality, database=database)
and self.client.source != "deezer"
):
track.tag(embed_cover=kwargs.get("embed_cover", True))
@staticmethod @staticmethod
def _parse_get_resp(item: dict, client: ClientInterface): def _parse_get_resp(item: dict, client: ClientInterface):
@ -1075,11 +1125,10 @@ class Playlist(Tracklist):
:param client: :param client:
:type client: ClientInterface :type client: ClientInterface
""" """
print(item.keys())
if client.source == "qobuz": if client.source == "qobuz":
return { return {
"name": item["name"], "name": item["name"],
"id": item['id'], "id": item["id"],
} }
elif client.source == "tidal": elif client.source == "tidal":
return { return {
@ -1172,7 +1221,7 @@ class Artist(Tracklist):
def download( def download(
self, self,
parent_folder: str = "Downloads", parent_folder: str = "StreamripDownloads",
filters: Optional[Tuple] = None, filters: Optional[Tuple] = None,
no_repeats: bool = False, no_repeats: bool = False,
quality: int = 6, quality: int = 6,

View file

@ -2,6 +2,7 @@ import json
import logging import logging
import re import re
import sys import sys
from pprint import pprint
from typing import Generator, Optional, Tuple, Union from typing import Generator, Optional, Tuple, Union
from .constants import ( from .constants import (
@ -113,9 +114,10 @@ class TrackMetadata:
self.date = resp.get("release_date") self.date = resp.get("release_date")
self.albumartist = resp.get("artist", {}).get("name") self.albumartist = resp.get("artist", {}).get("name")
self.label = resp.get("label") self.label = resp.get("label")
elif self.__source == "soundcloud":
raise Exception
else: else:
raise ValueError raise ValueError(self.__source)
def add_track_meta(self, track: dict): def add_track_meta(self, track: dict):
"""Parse the metadata from a track dict returned by the """Parse the metadata from a track dict returned by the
@ -150,8 +152,19 @@ class TrackMetadata:
self.discnumber = track.get("disk_number") self.discnumber = track.get("disk_number")
self.artist = track.get("artist", {}).get("name") 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.year = track["created_at"][:4]
self.label = track["label_name"]
self.description = track["description"]
self.tracknumber = 0
self.tracktotal = 0
else: else:
raise ValueError raise ValueError(self.__source)
if track.get("album"): if track.get("album"):
self.add_album_meta(track["album"]) self.add_album_meta(track["album"])

View file

@ -96,7 +96,7 @@ def get_quality_id(bit_depth: Optional[int], sampling_rate: Optional[int]):
return 4 return 4
def tqdm_download(url: str, filepath: str): def tqdm_download(url: str, filepath: str, params: dict = None):
"""Downloads a file with a progress bar. """Downloads a file with a progress bar.
:param url: url to direct download :param url: url to direct download
@ -104,11 +104,14 @@ def tqdm_download(url: str, filepath: str):
:type url: str :type url: str
:type filepath: str :type filepath: str
""" """
logger.debug(f"Downloading {url} to {filepath}") logger.debug(f"Downloading {url} to {filepath} with params {params}")
r = requests.get(url, allow_redirects=True, stream=True) if params is None:
params = {}
r = requests.get(url, allow_redirects=True, stream=True, params=params)
total = int(r.headers.get("content-length", 0)) total = int(r.headers.get("content-length", 0))
logger.debug(f"File size = {total}") logger.debug(f"File size = {total}")
if total < 1000: if total < 1000 and not url.endswith("jpg"):
raise NonStreamable raise NonStreamable
try: try: