KevinMidboe/spotify-downloader
1
0
Fork 0
mirror of https://github.com/KevinMidboe/spotify-downloader.git synced 2025-10-29 18:00:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Link arguments to spotipy helpers

Browse Source
This commit is contained in:
Ritiek Malhotra
2020-05-05 02:25:05 +05:30
parent 5a75687722
commit c3e8a0f0db
4 changed files with 116 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spotdl/command_line/arguments.py
View File

@@ -70,7 +70,7 @@ def get_arguments(argv=None, base_config_file=spotdl.config.default_config_file)
"if `--write-to=<path/to/file.txt>` has been passed", "if `--write-to=<path/to/file.txt>` has been passed",
) )
group.add_argument( group.add_argument(
"-b" "-b",
"--album", "--album",
help="load tracks from album URL into <album_name>.txt or if " help="load tracks from album URL into <album_name>.txt or if "
"`--write-to=<path/to/file.txt>` has been passed" "`--write-to=<path/to/file.txt>` has been passed"

120
spotdl/command_line/lib.py
View File

@@ -21,6 +21,8 @@ import spotdl.config
from spotdl.command_line.exceptions import NoYouTubeVideoFoundError from spotdl.command_line.exceptions import NoYouTubeVideoFoundError
from spotdl.command_line.exceptions import NoYouTubeVideoMatchError from spotdl.command_line.exceptions import NoYouTubeVideoMatchError
from spotdl.helpers.spotify import SpotifyHelpers
import sys import sys
import os import os
import urllib.request import urllib.request
@@ -47,15 +49,20 @@ def search_metadata_on_spotify(query):
def prompt_for_youtube_search_result(videos): def prompt_for_youtube_search_result(videos):
print("0. Skip downloading this track", file=sys.stderr) max_index_length = len(str(len(videos)))
max_title_length = max(len(v["title"]) for v in videos)
print(" 0. Skip downloading this track", file=sys.stderr)
for index, video in enumerate(videos, 1): for index, video in enumerate(videos, 1):
video_repr = "{index}. {title} ({url}) [{duration}]".format( vid_details = "{index:>{max_index}}. {title:<{max_title}} {url} [{duration}]".format(
index=index, index=index,
max_index=max_index_length,
title=video["title"], title=video["title"],
max_title=max_title_length,
url=video["url"], url=video["url"],
duration=video["duration"], duration=video["duration"],
) )
print(video_repr, file=sys.stderr) print(vid_details, file=sys.stderr)
print("", file=sys.stderr)
selection = spotdl.util.prompt_user_for_selection(range(1, len(videos)+1)) selection = spotdl.util.prompt_user_for_selection(range(1, len(videos)+1))
@@ -121,7 +128,7 @@ def search_metadata(track, search_format="{artist} - {track-name} lyrics", manua
youtube_videos = youtube_searcher.search(search_query) youtube_videos = youtube_searcher.search(search_query)
if not youtube_videos: if not youtube_videos:
raise NoYouTubeVideoFoundError( raise NoYouTubeVideoFoundError(
'YouTube returned no videos for the search query "{}"'.format(search_query) 'YouTube returned no videos for the search query "{}".'.format(search_query)
) )
if manual: if manual:
youtube_video = prompt_for_youtube_search_result(youtube_videos) youtube_video = prompt_for_youtube_search_result(youtube_videos)
@@ -130,7 +137,7 @@ def search_metadata(track, search_format="{artist} - {track-name} lyrics", manua
if youtube_video is None: if youtube_video is None:
raise NoYouTubeVideoMatchError( raise NoYouTubeVideoMatchError(
'No matching videos found on YouTube for the search query "{}"'.format( 'No apparent matching videos found on YouTube for the search query "{}"'.format(
search_query search_query
) )
) )
@@ -177,9 +184,10 @@ class Spotdl:
client_id=self.arguments["spotify_client_id"], client_id=self.arguments["spotify_client_id"],
client_secret=self.arguments["spotify_client_secret"] client_secret=self.arguments["spotify_client_secret"]
) )
spotify_tools = SpotifyHelpers()
# youtube_tools.set_api_key()
logger.debug("Received arguments:\n{}".format(self.arguments)) logger.debug("Received arguments:\n{}".format(self.arguments))
# youtube_tools.set_api_key()
if self.arguments["song"]: if self.arguments["song"]:
for track in self.arguments["song"]: for track in self.arguments["song"]:
if track == "-": if track == "-":
@@ -205,27 +213,21 @@ class Spotdl:
self.arguments["list"], self.arguments["list"],
) )
elif self.arguments["playlist"]: elif self.arguments["playlist"]:
spotify_tools.write_playlist( playlist = spotify_tools.fetch_playlist(self.arguments["playlist"])
playlist_url=self.arguments["playlist"], text_file=self.arguments["write_to"] spotify_tools.write_playlist_tracks(playlist, self.arguments["write_to"])
)
elif self.arguments["album"]: elif self.arguments["album"]:
spotify_tools.write_album( album = spotify_tools.fetch_album(self.arguments["album"])
album_url=self.arguments["album"], text_file=self.arguments["write_to"] spotify_tools.write_album_tracks(album, self.arguments["write_to"])
)
elif self.arguments["all_albums"]: elif self.arguments["all_albums"]:
spotify_tools.write_all_albums_from_artist( albums = spotify_tools.fetch_albums_from_artist(self.arguments["all_albums"])
artist_url=self.arguments["all_albums"], text_file=self.arguments["write_to"] spotify_tools.write_all_albums(albums, self.arguments["write_to"])
)
elif self.arguments["username"]: elif self.arguments["username"]:
spotify_tools.write_user_playlist( playlist_url = spotify_tools.prompt_for_user_playlist(self.arguments["username"])
username=self.arguments["username"], text_file=self.arguments["write_to"] playlist = spotify_tools.fetch_playlist(playlist_url)
) spotify_tools.write_playlist_tracks(playlist, self.arguments["write_to"])
def download_track(self, track): def download_track(self, track):
logger.info('Downloading "{}"'.format(track)) logger.info('Downloading "{}"'.format(track))
track_splits = track.split(":")
if len(track_splits) == 2:
youtube_track, spotify_track = track_splits
try: try:
metadata = search_metadata( metadata = search_metadata(
track, track,
@@ -237,8 +239,19 @@ class Spotdl:
else: else:
self.download_track_from_metadata(metadata) self.download_track_from_metadata(metadata)
def should_we_overwrite_existing_file(self):
if self.arguments["overwrite"] == "force":
logger.info("Forcing overwrite on existing file.")
to_overwrite = True
elif self.arguments["overwrite"] == "prompt":
to_overwrite = input("Overwrite? (y/N): ").lower() == "y"
else:
logger.info("Not overwriting existing file.")
to_overwrite = False
return to_overwrite
def download_track_from_metadata(self, metadata): def download_track_from_metadata(self, metadata):
# TODO: Add `-m` flag
track = Track(metadata, cache_albumart=(not self.arguments["no_metadata"])) track = Track(metadata, cache_albumart=(not self.arguments["no_metadata"]))
stream = metadata["streams"].get( stream = metadata["streams"].get(
quality=self.arguments["quality"], quality=self.arguments["quality"],
@@ -264,28 +277,19 @@ class Spotdl:
) )
) )
to_skip = self.arguments["dry_run"] to_skip_download = self.arguments["dry_run"]
if not to_skip and os.path.isfile(filename): if os.path.isfile(filename):
msg_already_exists = 'A file with name "{filename}" already exists.'.format( logger.info('A file with name "{filename}" already exists.'.format(
filename=filename filename=filename
) ))
if self.arguments["overwrite"] == "force": to_skip_download = to_skip_download \
force_overwrite_msg = msg_already_exists + " Forcing overwrite." or not self.should_we_overwrite_existing_file()
to_skip = False
logger.info(force_overwrite_msg)
elif self.arguments["overwrite"] == "prompt":
prompt_overwrite_msg = msg_already_exists + " Overwrite? (y/N): "
to_skip = not input(prompt_overwrite_msg).lower() == "y"
else:
skip_overwrite_msg = msg_already_exists + " Skipping download."
to_skip = True
logger.info(skip_overwrite_msg)
if to_skip: if to_skip_download:
logger.debug("Skip track download.") logger.debug("Skip track download.")
return return
logger.info('Downloading to "{filename}".'.format(filename=filename)) logger.info('Downloading to "{filename}"'.format(filename=filename))
if Encoder is None: if Encoder is None:
track.download(stream, filename) track.download(stream, filename)
else: else:
@@ -297,12 +301,15 @@ class Spotdl:
) )
if not self.arguments["no_metadata"]: if not self.arguments["no_metadata"]:
track.metadata["lyrics"] = track.metadata["lyrics"].join() self.apply_metadata(track, filename, output_extension)
try:
logger.info("Applying metadata") def apply_metadata(self, track, filename, encoding):
track.apply_metadata(filename, encoding=output_extension) track.metadata["lyrics"] = track.metadata["lyrics"].join()
except TypeError: logger.info("Applying metadata")
logger.warning("Cannot apply metadata on provided output format.") try:
track.apply_metadata(filename, encoding=encoding)
except TypeError:
logger.warning("Cannot apply metadata on provided output format.")
def download_tracks_from_file(self, path): def download_tracks_from_file(self, path):
logger.info( logger.info(
@@ -324,19 +331,26 @@ class Spotdl:
with open(path, "w") as fout: with open(path, "w") as fout:
fout.writelines(tracks) fout.writelines(tracks)
print("", file=sys.stderr)
for number, track in enumerate(tracks, 1): for number, track in enumerate(tracks, 1):
try: try:
metadata = search_metadata(track, self.arguments["search_format"]) log_track_query = '{position}. Downloading "{track}"'.format(
log_track_query = str(number) + ". {artist} - {track-name}" position=number,
track=track
)
logger.info(log_track_query) logger.info(log_track_query)
metadata = search_metadata(track, self.arguments["search_format"])
self.download_track_from_metadata(metadata) self.download_track_from_metadata(metadata)
except (urllib.request.URLError, TypeError, IOError) as e: except (urllib.request.URLError, TypeError, IOError) as e:
logger.exception(e.args[0]) logger.exception(e.args[0])
logger.warning("Failed. Will retry after other songs\n") logger.warning(
"Failed to download current track due to possible network issue. "
"Will retry after other songs."
)
tracks.append(track) tracks.append(track)
except NoYouTubeVideoFoundError: except (NoYouTubeVideoFoundError, NoYouTubeVideoMatchError) as e:
logger.warning("Failed. No YouTube video found.\n") logger.error(e.args[0])
pass
else: else:
if self.arguments["write_successful"]: if self.arguments["write_successful"]:
with open(self.arguments["write_successful"], "a") as fout: with open(self.arguments["write_successful"], "a") as fout:
@@ -344,12 +358,13 @@ class Spotdl:
finally: finally:
with open(path, "w") as fout: with open(path, "w") as fout:
fout.writelines(tracks[number-1:]) fout.writelines(tracks[number-1:])
print("", file=sys.stderr)
def download_tracks_from_file_threaded(self, path): def download_tracks_from_file_threaded(self, path):
# FIXME: Can we make this function cleaner? # FIXME: Can we make this function cleaner?
logger.info( logger.info(
"Checking and removing any duplicate tracks in {}.".format(path) "Checking and removing any duplicate tracks in {}.\n".format(path)
) )
with open(path, "r") as fin: with open(path, "r") as fin:
# Read tracks into a list and remove any duplicates # Read tracks into a list and remove any duplicates
@@ -402,6 +417,7 @@ class Spotdl:
print(metadata["current_track"]["name"], file=sys.stderr) print(metadata["current_track"]["name"], file=sys.stderr)
# self.download_track_from_metadata(metadata["current_track"]) # self.download_track_from_metadata(metadata["current_track"])
except (urllib.request.URLError, TypeError, IOError) as e: except (urllib.request.URLError, TypeError, IOError) as e:
print("", file=sys.stderr)
logger.exception(e.args[0]) logger.exception(e.args[0])
logger.warning("Failed. Will retry after other songs\n") logger.warning("Failed. Will retry after other songs\n")
tracks.append(current_track) tracks.append(current_track)

77
spotdl/helpers/spotify.py
View File

@@ -4,26 +4,47 @@
# to `spotify._get_id` in below methods. # to `spotify._get_id` in below methods.
from spotdl.authorize.services import AuthorizeSpotify from spotdl.authorize.services import AuthorizeSpotify
import spotdl.util
import sys
import logging import logging
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
try:
from slugify import SLUG_OK, slugify
except ImportError:
logger.error("Oops! `unicode-slugify` was not found.")
logger.info("Please remove any other slugify library and install `unicode-slugify`.")
sys.exit(5)
ALBUM_BASE_URL = "https://open.spotify.com/album/"
class SpotifyHelpers: class SpotifyHelpers:
def __init__(self, spotify=None): def __init__(self, spotify=None):
self._ALBUM_BASE_URL = ALBUM_BASE_URL
if spotify is None: if spotify is None:
spotify = AuthorizeSpotify() spotify = AuthorizeSpotify()
self.spotify = spotify self.spotify = spotify
def prompt_for_user_playlist(self, username): def prompt_for_user_playlist(self, username):
""" Write user playlists to text_file """ """ Write user playlists to text_file """
links = fetch_user_playlist_urls(username) playlists = self.fetch_user_playlist_urls(username)
playlist = internals.input_link(links) for i, playlist in enumerate(playlists, 1):
return playlist playlist_details = "{0}. {1:<30} ({2} tracks)".format(
i, playlist["name"], playlist["tracks"]["total"]
)
print(playlist_details, file=sys.stderr)
print("", file=sys.stderr)
playlist = spotdl.util.prompt_user_for_selection(playlists)
return playlist["external_urls"]["spotify"]
def fetch_user_playlist_urls(self, username): def fetch_user_playlist_urls(self, username):
""" Fetch user playlists when using the -u option. """ """ Fetch user playlists when using the -u option. """
logger.debug('Fetching playlists for "{username}".'.format(username=username))
playlists = self.spotify.user_playlists(username) playlists = self.spotify.user_playlists(username)
links = [] collected_playlists = []
check = 1 check = 1
while True: while True:
@@ -31,56 +52,51 @@ class SpotifyHelpers:
# in rare cases, playlists may not be found, so playlists['next'] # in rare cases, playlists may not be found, so playlists['next']
# is None. Skip these. Also see Issue #91. # is None. Skip these. Also see Issue #91.
if playlist["name"] is not None: if playlist["name"] is not None:
logger.info( collected_playlists.append(playlist)
# u"{0:>5}. {1:<30} ({2} tracks)".format(
# check, playlist["name"], playlist["tracks"]["total"]
# )
# )
playlist_url = playlist["external_urls"]["spotify"]
# log.debug(playlist_url)
links.append(playlist_url)
check += 1 check += 1
if playlists["next"]: if playlists["next"]:
playlists = self.spotify.next(playlists) playlists = self.spotify.next(playlists)
else: else:
break break
return links return collected_playlists
def fetch_playlist(self, playlist_url): def fetch_playlist(self, playlist_url):
logger.debug('Fetching playlist "{playlist}".'.format(playlist=playlist_url))
try: try:
playlist_id = self.spotify._get_id("playlist", playlist_url) playlist_id = self.spotify._get_id("playlist", playlist_url)
except IndexError: except IndexError:
# Wrong format, in either case # Wrong format, in either case
# log.error("The provided playlist URL is not in a recognized format!") logger.error("The provided playlist URL is not in a recognized format!")
sys.exit(10) sys.exit(10)
try: try:
results = self.spotify.user_playlist( results = self.spotify.user_playlist(
user=None, playlist_id=playlist_id, fields="tracks,next,name" user=None, playlist_id=playlist_id, fields="tracks,next,name"
) )
except spotipy.client.SpotifyException: except spotipy.client.SpotifyException:
# log.error("Unable to find playlist") logger.error("Unable to find playlist")
logger.info("Make sure the playlist is set to publicly visible and then try again") logger.info("Make sure the playlist is set to publicly visible and then try again.")
sys.exit(11) sys.exit(11)
return results return results
def write_playlist(self, playlist, text_file=None): def write_playlist_tracks(self, playlist, text_file=None):
tracks = playlist["tracks"] tracks = playlist["tracks"]
if not text_file: if not text_file:
text_file = u"{0}.txt".format(slugify(playlist["name"], ok="-_()[]{}")) text_file = u"{0}.txt".format(slugify(playlist["name"], ok="-_()[]{}"))
return write_tracks(tracks, text_file) return self.write_tracks(tracks, text_file)
def fetch_album(self, album_url): def fetch_album(self, album_url):
logger.debug('Fetching album "{album}".'.format(album=album_url))
album_id = self.spotify._get_id("album", album_url) album_id = self.spotify._get_id("album", album_url)
album = self.spotify.album(album_id) album = self.spotify.album(album_id)
return album return album
def write_album(self, album, text_file=None): def write_album_tracks(self, album, text_file=None):
tracks = self.spotify.album_tracks(album["id"]) tracks = self.spotify.album_tracks(album["id"])
if not text_file: if not text_file:
text_file = u"{0}.txt".format(slugify(album["name"], ok="-_()[]{}")) text_file = u"{0}.txt".format(slugify(album["name"], ok="-_()[]{}"))
return write_tracks(tracks, text_file) return self.write_tracks(tracks, text_file)
def fetch_albums_from_artist(self, artist_url, album_type=None): def fetch_albums_from_artist(self, artist_url, album_type=None):
""" """
@@ -92,9 +108,10 @@ class SpotifyHelpers:
:param return - the album from the artist :param return - the album from the artist
""" """
logger.debug('Fetching all albums for "{artist}".'.format(artist=artist_url))
artist_id = self.spotify._get_id("artist", artist_url)
# fetching artist's albums limitting the results to the US to avoid duplicate # fetching artist's albums limitting the results to the US to avoid duplicate
# albums from multiple markets # albums from multiple markets
artist_id = self.spotify._get_id("artist", artist_url)
results = self.spotify.artist_albums(artist_id, album_type=album_type, country="US") results = self.spotify.artist_albums(artist_id, album_type=album_type, country="US")
albums = results["items"] albums = results["items"]
@@ -106,7 +123,7 @@ class SpotifyHelpers:
return albums return albums
def write_all_albums_from_artist(self, albums, text_file=None): def write_all_albums(self, albums, text_file=None):
""" """
This function gets all albums from an artist and writes it to a file in the This function gets all albums from an artist and writes it to a file in the
current working directory called [ARTIST].txt, where [ARTIST] is the artist current working directory called [ARTIST].txt, where [ARTIST] is the artist
@@ -115,20 +132,17 @@ class SpotifyHelpers:
:param text_file - file to write albums to :param text_file - file to write albums to
""" """
album_base_url = "https://open.spotify.com/album/"
# if no file if given, the default save file is in the current working # if no file if given, the default save file is in the current working
# directory with the name of the artist # directory with the name of the artist
if text_file is None: if text_file is None:
text_file = albums[0]["artists"][0]["name"] + ".txt" text_file = albums[0]["artists"][0]["name"] + ".txt"
for album in albums: for album in albums:
logging album name logger.info('Fetching album "{album}".'.format(album=album["name"]))
log.info("Fetching album: " + album["name"]) self.write_album_tracks(album, text_file=text_file)
write_album(album_base_url + album["id"], text_file=text_file)
def write_tracks(self, tracks, text_file): def write_tracks(self, tracks, text_file):
logger.info(u"Writing {0} tracks to {1}".format(tracks["total"], text_file)) logger.info(u"Writing {0} tracks to {1}.".format(tracks["total"], text_file))
track_urls = [] track_urls = []
with open(text_file, "a") as file_out: with open(text_file, "a") as file_out:
while True: while True:
@@ -139,12 +153,13 @@ class SpotifyHelpers:
track = item track = item
try: try:
track_url = track["external_urls"]["spotify"] track_url = track["external_urls"]["spotify"]
# log.debug(track_url)
file_out.write(track_url + "\n") file_out.write(track_url + "\n")
track_urls.append(track_url) track_urls.append(track_url)
except KeyError: except KeyError:
# log.warning( # FIXME: Write "{artist} - {name}" instead of Spotify URI for
u"Skipping track {0} by {1} (local only?)".format( # "local only" tracks.
logger.warning(
'Skipping track "{0}" by "{1}" (local only?)'.format(
track["name"], track["artists"][0]["name"] track["name"], track["artists"][0]["name"]
) )
) )

2
spotdl/util.py
View File

@@ -52,7 +52,7 @@ def merge(base, overrider):
def prompt_user_for_selection(items): def prompt_user_for_selection(items):
""" Let the user input a choice. """ """ Let the user input a choice. """
logger.info("Choose your number:") logger.info("Enter a number:")
while True: while True:
try: try:
the_chosen_one = int(input("> ")) the_chosen_one = int(input("> "))