mirror of
https://github.com/KevinMidboe/spotify-downloader.git
synced 2025-10-29 09:50:16 +00:00
Create a class for command-line functions
This commit is contained in:
@@ -1,3 +1,4 @@
|
||||
from spotdl.version import __version__
|
||||
from spotdl.command_line import Spotdl
|
||||
from spotdl.track import Track
|
||||
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
from spotdl.command_line import helpers
|
||||
from spotdl.command_line import lib
|
||||
from spotdl.command_line.arguments import get_arguments
|
||||
from spotdl.command_line.lib import Spotdl
|
||||
|
||||
|
||||
@@ -1,76 +1,16 @@
|
||||
from spotdl.authorize.services import AuthorizeSpotify
|
||||
from spotdl import command_line
|
||||
|
||||
import logzero
|
||||
import sys
|
||||
|
||||
def match_arguments(arguments):
|
||||
if arguments["song"]:
|
||||
for track in arguments["song"]:
|
||||
if track == "-":
|
||||
for line in sys.stdin:
|
||||
command_line.helpers.download_track(
|
||||
line,
|
||||
arguments
|
||||
)
|
||||
else:
|
||||
command_line.helpers.download_track(track, arguments)
|
||||
elif arguments["list"]:
|
||||
if arguments["write_m3u"]:
|
||||
youtube_tools.generate_m3u(
|
||||
track_file=arguments["list"]
|
||||
)
|
||||
else:
|
||||
list_download = {
|
||||
"sequential": command_line.helpers.download_tracks_from_file,
|
||||
"threaded" : command_line.helpers.download_tracks_from_file_threaded,
|
||||
}[arguments["processor"]]
|
||||
|
||||
list_download(
|
||||
arguments["list"],
|
||||
arguments,
|
||||
)
|
||||
elif arguments["playlist"]:
|
||||
spotify_tools.write_playlist(
|
||||
playlist_url=arguments["playlist"], text_file=arguments["write_to"]
|
||||
)
|
||||
elif arguments["album"]:
|
||||
spotify_tools.write_album(
|
||||
album_url=arguments["album"], text_file=arguments["write_to"]
|
||||
)
|
||||
elif arguments.all_albums:
|
||||
spotify_tools.write_all_albums_from_artist(
|
||||
artist_url=arguments["all_albums"], text_file=arguments["write_to"]
|
||||
)
|
||||
elif arguments.username:
|
||||
spotify_tools.write_user_playlist(
|
||||
username=arguments["username"], text_file=arguments["write_to"]
|
||||
)
|
||||
|
||||
|
||||
def set_logger(level):
|
||||
fmt = "%(color)s%(levelname)s:%(end_color)s %(message)s"
|
||||
formatter = logzero.LogFormatter(fmt=fmt)
|
||||
logzero.formatter(formatter)
|
||||
logzero.loglevel(level)
|
||||
return logzero.logger
|
||||
from spotdl import command_line
|
||||
|
||||
|
||||
def main():
|
||||
arguments = command_line.get_arguments()
|
||||
logger = set_logger(arguments.log_level)
|
||||
logger.debug(arguments.__dict__)
|
||||
|
||||
AuthorizeSpotify(
|
||||
client_id=arguments.spotify_client_id,
|
||||
client_secret=arguments.spotify_client_secret
|
||||
)
|
||||
# youtube_tools.set_api_key()
|
||||
|
||||
spotdl = command_line.Spotdl(arguments.__dict__)
|
||||
try:
|
||||
match_arguments(arguments.__dict__)
|
||||
spotdl.match_arguments()
|
||||
except KeyboardInterrupt as e:
|
||||
logger.exception(e)
|
||||
logzero.logger.exception(e)
|
||||
sys.exit(2)
|
||||
|
||||
|
||||
|
||||
@@ -103,7 +103,7 @@ def get_arguments(argv=None, to_merge=True):
|
||||
parser.add_argument(
|
||||
"-nm",
|
||||
"--no-metadata",
|
||||
default=config["no-metadata"],
|
||||
default=config["no_metadata"],
|
||||
help="do not embed metadata in tracks",
|
||||
action="store_true",
|
||||
)
|
||||
@@ -130,40 +130,40 @@ def get_arguments(argv=None, to_merge=True):
|
||||
parser.add_argument(
|
||||
"-i",
|
||||
"--input-ext",
|
||||
default=config["input-ext"],
|
||||
default=config["input_ext"],
|
||||
choices={"automatic", "m4a", "opus"},
|
||||
help="preferred input format",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-o",
|
||||
"--output-ext",
|
||||
default=config["output-ext"],
|
||||
default=config["output_ext"],
|
||||
choices={"mp3", "m4a", "flac"},
|
||||
help="preferred output format",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--write-to",
|
||||
default=config["write-to"],
|
||||
default=config["write_to"],
|
||||
help="write tracks from Spotify playlist, album, etc. to this file",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-f",
|
||||
"--output-file",
|
||||
default=config["output-file"],
|
||||
default=config["output_file"],
|
||||
help="path where to write the downloaded track to, special tags "
|
||||
"are to be surrounded by curly braces. Possible tags: "
|
||||
# "{}".format([spotdl.util.formats[x] for x in spotdl.util.formats]),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--trim-silence",
|
||||
default=config["trim-silence"],
|
||||
default=config["trim_silence"],
|
||||
help="remove silence from the start of the audio",
|
||||
action="store_true",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-sf",
|
||||
"--search-format",
|
||||
default=config["search-format"],
|
||||
default=config["search_format"],
|
||||
help="search format to search for on YouTube, special tags "
|
||||
"are to be surrounded by curly braces. Possible tags: "
|
||||
# "{}".format([spotdl.util.formats[x] for x in spotdl.util.formats]),
|
||||
@@ -171,7 +171,7 @@ def get_arguments(argv=None, to_merge=True):
|
||||
parser.add_argument(
|
||||
"-d",
|
||||
"--dry-run",
|
||||
default=config["dry-run"],
|
||||
default=config["dry_run"],
|
||||
help="show only track title and YouTube URL, and then skip "
|
||||
"to the next track (if any)",
|
||||
action="store_true",
|
||||
@@ -179,7 +179,7 @@ def get_arguments(argv=None, to_merge=True):
|
||||
parser.add_argument(
|
||||
"-mo",
|
||||
"--music-videos-only",
|
||||
default=config["music-videos-only"],
|
||||
default=config["music_videos_only"],
|
||||
help="search only for music videos on Youtube (works only "
|
||||
"when YouTube API key is set)",
|
||||
action="store_true",
|
||||
@@ -196,14 +196,14 @@ def get_arguments(argv=None, to_merge=True):
|
||||
parser.add_argument(
|
||||
"-ns",
|
||||
"--no-spaces",
|
||||
default=config["no-spaces"],
|
||||
default=config["no_spaces"],
|
||||
help="replace spaces with underscores in file names",
|
||||
action="store_true",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-ll",
|
||||
"--log-level",
|
||||
default=config["log-level"],
|
||||
default=config["log_level"],
|
||||
choices=_LOG_LEVELS_STR,
|
||||
type=str.upper,
|
||||
help="set log verbosity",
|
||||
@@ -211,7 +211,7 @@ def get_arguments(argv=None, to_merge=True):
|
||||
parser.add_argument(
|
||||
"-yk",
|
||||
"--youtube-api-key",
|
||||
default=config["youtube-api-key"],
|
||||
default=config["youtube_api_key"],
|
||||
help=argparse.SUPPRESS,
|
||||
)
|
||||
parser.add_argument(
|
||||
@@ -223,19 +223,19 @@ def get_arguments(argv=None, to_merge=True):
|
||||
parser.add_argument(
|
||||
"-w",
|
||||
"--write-successful",
|
||||
default=config["write-successful"],
|
||||
default=config["write_successful"],
|
||||
help="path to file to write successful tracks to",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-sci",
|
||||
"--spotify-client-id",
|
||||
default=config["spotify-client-id"],
|
||||
default=config["spotify_client_id"],
|
||||
help=argparse.SUPPRESS,
|
||||
)
|
||||
parser.add_argument(
|
||||
"-scs",
|
||||
"--spotify-client-secret",
|
||||
default=config["spotify-client-secret"],
|
||||
default=config["spotify_client_secret"],
|
||||
help=argparse.SUPPRESS,
|
||||
)
|
||||
parser.add_argument(
|
||||
@@ -308,5 +308,8 @@ def run_errands(parser, parsed, config):
|
||||
|
||||
parsed.log_level = log_leveller(parsed.log_level)
|
||||
|
||||
# We're done dealing with configuration file here and don't need to use it later
|
||||
del parsed.config
|
||||
|
||||
return parsed
|
||||
|
||||
|
||||
@@ -1,320 +0,0 @@
|
||||
from spotdl.metadata.providers import ProviderSpotify
|
||||
from spotdl.metadata.providers import ProviderYouTube
|
||||
from spotdl.metadata.providers import YouTubeSearch
|
||||
from spotdl.metadata.embedders import EmbedderDefault
|
||||
from spotdl.metadata.exceptions import SpotifyMetadataNotFoundError
|
||||
import spotdl.metadata
|
||||
|
||||
from spotdl.encode.encoders import EncoderFFmpeg
|
||||
from spotdl.encode.encoders import EncoderAvconv
|
||||
|
||||
from spotdl.lyrics.providers import LyricWikia
|
||||
from spotdl.lyrics.providers import Genius
|
||||
from spotdl.lyrics.exceptions import LyricsNotFoundError
|
||||
|
||||
from spotdl.command_line.exceptions import NoYouTubeVideoError
|
||||
|
||||
from spotdl.track import Track
|
||||
|
||||
import spotdl.util
|
||||
|
||||
from logzero import logger
|
||||
import os
|
||||
import urllib.request
|
||||
|
||||
|
||||
def search_lyrics(query):
|
||||
provider = Genius()
|
||||
try:
|
||||
lyrics = provider.from_query(query)
|
||||
except LyricsNotFoundError:
|
||||
lyrics = None
|
||||
return lyrics
|
||||
|
||||
|
||||
def search_metadata_on_spotify(query):
|
||||
provider = ProviderSpotify()
|
||||
try:
|
||||
metadata = provider.from_query(query)
|
||||
except SpotifyMetadataNotFoundError:
|
||||
metadata = {}
|
||||
return metadata
|
||||
|
||||
|
||||
def prompt_for_youtube_search_result(videos):
|
||||
urls = []
|
||||
print("0. Skip downloading this track")
|
||||
for index, video in enumerate(videos, 1):
|
||||
video_repr = "{index}. {title} [{duration}] ({url})".format(
|
||||
index=index,
|
||||
title=video["title"],
|
||||
duration=video["duration"],
|
||||
url=video["url"]
|
||||
)
|
||||
print(video_repr)
|
||||
urls.append(video["url"])
|
||||
return spotdl.util.prompt_user_for_selection(urls)
|
||||
|
||||
|
||||
def search_metadata(track, search_format="{artist} - {track-name} lyrics", manual=False):
|
||||
# TODO: Clean this function
|
||||
youtube = ProviderYouTube()
|
||||
youtube_searcher = YouTubeSearch()
|
||||
|
||||
if spotdl.util.is_spotify(track):
|
||||
spotify = ProviderSpotify()
|
||||
spotify_metadata = spotify.from_url(track)
|
||||
lyric_query = spotdl.metadata.format_string(
|
||||
"{artist} - {track-name}",
|
||||
spotify_metadata,
|
||||
)
|
||||
search_query = spotdl.metadata.format_string(search_format, spotify_metadata)
|
||||
youtube_videos = youtube_searcher.search(search_query)
|
||||
if not youtube_videos:
|
||||
raise NoYouTubeVideoError(
|
||||
'No videos found for the search query: "{}"'.format(search_query)
|
||||
)
|
||||
if manual:
|
||||
youtube_video = prompt_for_youtube_search_result(youtube_videos)
|
||||
else:
|
||||
youtube_video = youtube_videos.bestmatch()["url"]
|
||||
youtube_metadata = youtube.from_url(youtube_video)
|
||||
metadata = spotdl.util.merge(
|
||||
youtube_metadata,
|
||||
spotify_metadata
|
||||
)
|
||||
|
||||
elif spotdl.util.is_youtube(track):
|
||||
metadata = youtube.from_url(track)
|
||||
lyric_query = spotdl.metadata.format_string(
|
||||
"{artist} - {track-name}",
|
||||
metadata,
|
||||
)
|
||||
|
||||
else:
|
||||
lyric_query = track
|
||||
spotify_metadata = spotdl.util.ThreadWithReturnValue(
|
||||
target=search_metadata_on_spotify,
|
||||
args=(track,)
|
||||
)
|
||||
spotify_metadata.start()
|
||||
youtube_videos = youtube_searcher.search(track)
|
||||
if not youtube_videos:
|
||||
raise NoYouTubeVideoError(
|
||||
'No videos found for the search query: "{}"'.format(track)
|
||||
)
|
||||
return
|
||||
if manual:
|
||||
youtube_video = prompt_for_youtube_search_result(youtube_videos)
|
||||
else:
|
||||
youtube_video = youtube_videos.bestmatch()["url"]
|
||||
youtube_metadata = youtube.from_url(youtube_video)
|
||||
metadata = spotdl.util.merge(
|
||||
youtube_metadata,
|
||||
spotify_metadata.join()
|
||||
)
|
||||
|
||||
metadata["lyrics"] = spotdl.util.ThreadWithReturnValue(
|
||||
target=search_lyrics,
|
||||
args=(lyric_query,)
|
||||
)
|
||||
|
||||
metadata["lyrics"].start()
|
||||
return metadata
|
||||
|
||||
|
||||
def download_track(track, arguments):
|
||||
track_splits = track.split(":")
|
||||
if len(track_splits) == 2:
|
||||
youtube_track, spotify_track = track_splits
|
||||
metadata = search_metadata(
|
||||
track,
|
||||
search_format=arguments["search_format"],
|
||||
manual=arguments["manual"],
|
||||
)
|
||||
log_fmt = spotdl.metadata.format_string(
|
||||
arguments["output_file"],
|
||||
metadata,
|
||||
output_extension=arguments["output_ext"],
|
||||
)
|
||||
# log.info(log_fmt)
|
||||
download_track_from_metadata(metadata, arguments)
|
||||
|
||||
|
||||
def download_track_from_metadata(metadata, arguments):
|
||||
# TODO: Add `-m` flag
|
||||
track = Track(metadata, cache_albumart=(not arguments["no_metadata"]))
|
||||
print(metadata["name"])
|
||||
|
||||
stream = metadata["streams"].get(
|
||||
quality=arguments["quality"],
|
||||
preftype=arguments["input_ext"],
|
||||
)
|
||||
# log.info(stream)
|
||||
|
||||
Encoder = {
|
||||
"ffmpeg": EncoderFFmpeg,
|
||||
"avconv": EncoderAvconv,
|
||||
}.get(arguments["encoder"])
|
||||
|
||||
if Encoder is None:
|
||||
output_extension = stream["encoding"]
|
||||
else:
|
||||
output_extension = arguments["output_ext"]
|
||||
|
||||
filename = spotdl.metadata.format_string(
|
||||
arguments["output_file"],
|
||||
metadata,
|
||||
output_extension=output_extension,
|
||||
sanitizer=lambda s: spotdl.util.sanitize(
|
||||
s, spaces_to_underscores=arguments["no_spaces"]
|
||||
)
|
||||
)
|
||||
print(filename)
|
||||
# log.info(filename)
|
||||
|
||||
to_skip = arguments["dry_run"]
|
||||
if not to_skip and os.path.isfile(filename):
|
||||
if arguments["overwrite"] == "force":
|
||||
to_skip = False
|
||||
elif arguments["overwrite"] == "prompt":
|
||||
to_skip = not input("overwrite? (y/N)").lower() == "y"
|
||||
else:
|
||||
to_skip = True
|
||||
|
||||
if to_skip:
|
||||
return
|
||||
|
||||
if Encoder is None:
|
||||
track.download(stream, filename)
|
||||
else:
|
||||
track.download_while_re_encoding(
|
||||
stream,
|
||||
filename,
|
||||
target_encoding=output_extension,
|
||||
encoder=Encoder()
|
||||
)
|
||||
|
||||
if not arguments["no_metadata"]:
|
||||
track.metadata["lyrics"] = track.metadata["lyrics"].join()
|
||||
try:
|
||||
track.apply_metadata(filename, encoding=output_extension)
|
||||
except TypeError:
|
||||
# log.warning("Cannot write metadata to given file")
|
||||
pass
|
||||
|
||||
|
||||
def download_tracks_from_file(path, arguments):
|
||||
# log.info(
|
||||
# "Checking and removing any duplicate tracks "
|
||||
# "in reading {}".format(path)
|
||||
# )
|
||||
with open(path, "r") as fin:
|
||||
# Read tracks into a list and remove any duplicates
|
||||
tracks = fin.read().splitlines()
|
||||
|
||||
# Remove duplicates and empty elements
|
||||
# Also strip whitespaces from elements (if any)
|
||||
spotdl.util.remove_duplicates(
|
||||
tracks,
|
||||
condition=lambda x: x,
|
||||
operation=str.strip
|
||||
)
|
||||
|
||||
# Overwrite file
|
||||
with open(path, "w") as fout:
|
||||
fout.writelines(tracks)
|
||||
|
||||
for number, track in enumerate(tracks, 1):
|
||||
try:
|
||||
metadata = search_metadata(next_track, arguments["search_format"])
|
||||
log_fmt=(str(number) + ". {artist} - {track-name}")
|
||||
# log.info(log_fmt)
|
||||
download_track_from_metadata(metadata, arguments)
|
||||
except (urllib.request.URLError, TypeError, IOError) as e:
|
||||
# log.exception(e.args[0])
|
||||
# log.warning("Failed. Will retry after other songs\n")
|
||||
tracks.append(track)
|
||||
else:
|
||||
if arguments["write_sucessful"]:
|
||||
with open(arguments["write_successful"], "a") as fout:
|
||||
fout.write(track)
|
||||
finally:
|
||||
with open(path, "w") as fout:
|
||||
fout.writelines(tracks[number-1:])
|
||||
|
||||
|
||||
def download_tracks_from_file_threaded(path, arguments):
|
||||
# FIXME: Can we make this function cleaner?
|
||||
|
||||
# log.info(
|
||||
# "Checking and removing any duplicate tracks "
|
||||
# "in reading {}".format(path)
|
||||
# )
|
||||
with open(path, "r") as fin:
|
||||
# Read tracks into a list and remove any duplicates
|
||||
tracks = fin.read().splitlines()
|
||||
|
||||
# Remove duplicates and empty elements
|
||||
# Also strip whitespaces from elements (if any)
|
||||
spotdl.util.remove_duplicates(
|
||||
tracks,
|
||||
condition=lambda x: x,
|
||||
operation=str.strip
|
||||
)
|
||||
|
||||
# Overwrite file
|
||||
with open(path, "w") as fout:
|
||||
fout.writelines(tracks)
|
||||
|
||||
tracks_count = len(tracks)
|
||||
current_iteration = 1
|
||||
|
||||
next_track = tracks.pop(0)
|
||||
metadata = {
|
||||
"current_track": None,
|
||||
"next_track": spotdl.util.ThreadWithReturnValue(
|
||||
target=search_metadata,
|
||||
args=(next_track, arguments["search_format"])
|
||||
)
|
||||
}
|
||||
metadata["next_track"].start()
|
||||
while tracks_count > 0:
|
||||
metadata["current_track"] = metadata["next_track"].join()
|
||||
metadata["next_track"] = None
|
||||
try:
|
||||
print(tracks_count)
|
||||
print(tracks)
|
||||
if tracks_count > 1:
|
||||
current_track = next_track
|
||||
next_track = tracks.pop(0)
|
||||
metadata["next_track"] = spotdl.util.ThreadWithReturnValue(
|
||||
target=search_metadata,
|
||||
args=(next_track, arguments["search_format"])
|
||||
)
|
||||
metadata["next_track"].start()
|
||||
|
||||
log_fmt=(str(current_iteration) + ". {artist} - {track-name}")
|
||||
# log.info(log_fmt)
|
||||
if metadata["current_track"] is None:
|
||||
# log.warning("Something went wrong. Will retry after downloading remaining tracks")
|
||||
pass
|
||||
print(metadata["current_track"]["name"])
|
||||
# download_track_from_metadata(
|
||||
# metadata["current_track"],
|
||||
# arguments
|
||||
# )
|
||||
except (urllib.request.URLError, TypeError, IOError) as e:
|
||||
# log.exception(e.args[0])
|
||||
# log.warning("Failed. Will retry after other songs\n")
|
||||
tracks.append(current_track)
|
||||
else:
|
||||
tracks_count -= 1
|
||||
if arguments["write_successful"]:
|
||||
with open(arguments["write_successful"], "a") as fout:
|
||||
fout.write(current_track)
|
||||
finally:
|
||||
current_iteration += 1
|
||||
with open(path, "w") as fout:
|
||||
fout.writelines(tracks)
|
||||
|
||||
400
spotdl/command_line/lib.py
Normal file
400
spotdl/command_line/lib.py
Normal file
@@ -0,0 +1,400 @@
|
||||
from spotdl.metadata.providers import ProviderSpotify
|
||||
from spotdl.metadata.providers import ProviderYouTube
|
||||
from spotdl.metadata.providers import YouTubeSearch
|
||||
from spotdl.metadata.embedders import EmbedderDefault
|
||||
from spotdl.metadata.exceptions import SpotifyMetadataNotFoundError
|
||||
import spotdl.metadata
|
||||
|
||||
from spotdl.encode.encoders import EncoderFFmpeg
|
||||
from spotdl.encode.encoders import EncoderAvconv
|
||||
|
||||
from spotdl.lyrics.providers import LyricWikia
|
||||
from spotdl.lyrics.providers import Genius
|
||||
from spotdl.lyrics.exceptions import LyricsNotFoundError
|
||||
|
||||
from spotdl.authorize.services import AuthorizeSpotify
|
||||
|
||||
from spotdl.track import Track
|
||||
import spotdl.util
|
||||
import spotdl.config
|
||||
|
||||
from spotdl.command_line.exceptions import NoYouTubeVideoError
|
||||
|
||||
import logzero
|
||||
import os
|
||||
import urllib.request
|
||||
|
||||
|
||||
class Spotdl:
|
||||
def __init__(self, arguments):
|
||||
if "config" in arguments:
|
||||
# Make sure we set the base configuration from the config file if
|
||||
# the config file has been passed.
|
||||
config = spotdl.util.merge(
|
||||
spotdl.config.DEFAULT_CONFIGURATION["spotify-downloader"],
|
||||
spotdl.config.get_config(arguments["config"])
|
||||
)
|
||||
else:
|
||||
# If config file has not been passed, set the base configuration
|
||||
# to the default confguration.
|
||||
config = spotdl.config.DEFAULT_CONFIGURATION["spotify-downloader"]
|
||||
|
||||
self.arguments = spotdl.util.merge(config, arguments)
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, type, value, traceback):
|
||||
del self
|
||||
|
||||
def authorize_spotify(self, client_id, client_secret):
|
||||
AuthorizeSpotify(
|
||||
client_id=client_id,
|
||||
client_secret=client_secret
|
||||
)
|
||||
|
||||
def match_arguments(self):
|
||||
self.authorize_spotify(
|
||||
self.arguments["spotify_client_id"],
|
||||
self.arguments["spotify_client_secret"]
|
||||
)
|
||||
logger = self.set_logger(self.arguments["log_level"])
|
||||
logger.debug(self.arguments)
|
||||
|
||||
# youtube_tools.set_api_key()
|
||||
if self.arguments["song"]:
|
||||
for track in self.arguments["song"]:
|
||||
if track == "-":
|
||||
for line in sys.stdin:
|
||||
self.download_track(
|
||||
line,
|
||||
self.arguments
|
||||
)
|
||||
else:
|
||||
self.download_track(track)
|
||||
elif self.arguments["list"]:
|
||||
if self.arguments["write_m3u"]:
|
||||
youtube_tools.generate_m3u(
|
||||
track_file=self.arguments["list"]
|
||||
)
|
||||
else:
|
||||
list_download = {
|
||||
"synchronous": self.download_tracks_from_file,
|
||||
"threaded" : self.download_tracks_from_file_threaded,
|
||||
}[self.arguments["processor"]]
|
||||
|
||||
list_download(
|
||||
self.arguments["list"],
|
||||
)
|
||||
elif self.arguments["playlist"]:
|
||||
spotify_tools.write_playlist(
|
||||
playlist_url=self.arguments["playlist"], text_file=self.arguments["write_to"]
|
||||
)
|
||||
elif self.arguments["album"]:
|
||||
spotify_tools.write_album(
|
||||
album_url=self.arguments["album"], text_file=self.arguments["write_to"]
|
||||
)
|
||||
elif self.arguments["all_albums"]:
|
||||
spotify_tools.write_all_albums_from_artist(
|
||||
artist_url=self.arguments["all_albums"], text_file=self.arguments["write_to"]
|
||||
)
|
||||
elif self.arguments["username"]:
|
||||
spotify_tools.write_user_playlist(
|
||||
username=self.arguments["username"], text_file=self.arguments["write_to"]
|
||||
)
|
||||
|
||||
def set_logger(self, level):
|
||||
fmt = "%(color)s%(levelname)s:%(end_color)s %(message)s"
|
||||
formatter = logzero.LogFormatter(fmt=fmt)
|
||||
logzero.formatter(formatter)
|
||||
logzero.loglevel(level)
|
||||
return logzero.logger
|
||||
|
||||
def search_lyrics(self, query):
|
||||
provider = Genius()
|
||||
try:
|
||||
lyrics = provider.from_query(query)
|
||||
except LyricsNotFoundError:
|
||||
lyrics = None
|
||||
return lyrics
|
||||
|
||||
def search_metadata_on_spotify(self, query):
|
||||
provider = ProviderSpotify()
|
||||
try:
|
||||
metadata = provider.from_query(query)
|
||||
except SpotifyMetadataNotFoundError:
|
||||
metadata = {}
|
||||
return metadata
|
||||
|
||||
def prompt_for_youtube_search_result(self, videos):
|
||||
urls = []
|
||||
print("0. Skip downloading this track")
|
||||
for index, video in enumerate(videos, 1):
|
||||
video_repr = "{index}. {title} [{duration}] ({url})".format(
|
||||
index=index,
|
||||
title=video["title"],
|
||||
duration=video["duration"],
|
||||
url=video["url"]
|
||||
)
|
||||
print(video_repr)
|
||||
urls.append(video["url"])
|
||||
return spotdl.util.prompt_user_for_selection(urls)
|
||||
|
||||
def search_metadata(self, track, search_format="{artist} - {track-name} lyrics", manual=False):
|
||||
# TODO: Clean this function
|
||||
youtube = ProviderYouTube()
|
||||
youtube_searcher = YouTubeSearch()
|
||||
|
||||
if spotdl.util.is_spotify(track):
|
||||
spotify = ProviderSpotify()
|
||||
spotify_metadata = spotify.from_url(track)
|
||||
lyric_query = spotdl.metadata.format_string(
|
||||
"{artist} - {track-name}",
|
||||
spotify_metadata,
|
||||
)
|
||||
search_query = spotdl.metadata.format_string(search_format, spotify_metadata)
|
||||
youtube_videos = youtube_searcher.search(search_query)
|
||||
if not youtube_videos:
|
||||
raise NoYouTubeVideoError(
|
||||
'No videos found for the search query: "{}"'.format(search_query)
|
||||
)
|
||||
if manual:
|
||||
youtube_video = prompt_for_youtube_search_result(youtube_videos)
|
||||
else:
|
||||
youtube_video = youtube_videos.bestmatch()["url"]
|
||||
youtube_metadata = youtube.from_url(youtube_video)
|
||||
metadata = spotdl.util.merge(
|
||||
youtube_metadata,
|
||||
spotify_metadata
|
||||
)
|
||||
|
||||
elif spotdl.util.is_youtube(track):
|
||||
metadata = youtube.from_url(track)
|
||||
lyric_query = spotdl.metadata.format_string(
|
||||
"{artist} - {track-name}",
|
||||
metadata,
|
||||
)
|
||||
|
||||
else:
|
||||
lyric_query = track
|
||||
spotify_metadata = spotdl.util.ThreadWithReturnValue(
|
||||
target=self.search_metadata_on_spotify,
|
||||
args=(track,)
|
||||
)
|
||||
spotify_metadata.start()
|
||||
youtube_videos = youtube_searcher.search(track)
|
||||
if not youtube_videos:
|
||||
raise NoYouTubeVideoError(
|
||||
'No videos found for the search query: "{}"'.format(track)
|
||||
)
|
||||
return
|
||||
if manual:
|
||||
youtube_video = prompt_for_youtube_search_result(youtube_videos)
|
||||
else:
|
||||
youtube_video = youtube_videos.bestmatch()["url"]
|
||||
youtube_metadata = youtube.from_url(youtube_video)
|
||||
metadata = spotdl.util.merge(
|
||||
youtube_metadata,
|
||||
spotify_metadata.join()
|
||||
)
|
||||
|
||||
metadata["lyrics"] = spotdl.util.ThreadWithReturnValue(
|
||||
target=self.search_lyrics,
|
||||
args=(lyric_query,)
|
||||
)
|
||||
|
||||
metadata["lyrics"].start()
|
||||
return metadata
|
||||
|
||||
def download_track(self, track):
|
||||
track_splits = track.split(":")
|
||||
if len(track_splits) == 2:
|
||||
youtube_track, spotify_track = track_splits
|
||||
metadata = self.search_metadata(
|
||||
track,
|
||||
search_format=self.arguments["search_format"],
|
||||
manual=self.arguments["manual"],
|
||||
)
|
||||
log_fmt = spotdl.metadata.format_string(
|
||||
self.arguments["output_file"],
|
||||
metadata,
|
||||
output_extension=self.arguments["output_ext"],
|
||||
)
|
||||
# log.info(log_fmt)
|
||||
self.download_track_from_metadata(metadata)
|
||||
|
||||
def download_track_from_metadata(self, metadata):
|
||||
# TODO: Add `-m` flag
|
||||
track = Track(metadata, cache_albumart=(not self.arguments["no_metadata"]))
|
||||
print(metadata["name"])
|
||||
|
||||
stream = metadata["streams"].get(
|
||||
quality=self.arguments["quality"],
|
||||
preftype=self.arguments["input_ext"],
|
||||
)
|
||||
# log.info(stream)
|
||||
|
||||
Encoder = {
|
||||
"ffmpeg": EncoderFFmpeg,
|
||||
"avconv": EncoderAvconv,
|
||||
}.get(self.arguments["encoder"])
|
||||
|
||||
if Encoder is None:
|
||||
output_extension = stream["encoding"]
|
||||
else:
|
||||
output_extension = self.arguments["output_ext"]
|
||||
|
||||
filename = spotdl.metadata.format_string(
|
||||
self.arguments["output_file"],
|
||||
metadata,
|
||||
output_extension=output_extension,
|
||||
sanitizer=lambda s: spotdl.util.sanitize(
|
||||
s, spaces_to_underscores=self.arguments["no_spaces"]
|
||||
)
|
||||
)
|
||||
print(filename)
|
||||
# log.info(filename)
|
||||
|
||||
to_skip = self.arguments["dry_run"]
|
||||
if not to_skip and os.path.isfile(filename):
|
||||
if self.arguments["overwrite"] == "force":
|
||||
to_skip = False
|
||||
elif self.arguments["overwrite"] == "prompt":
|
||||
to_skip = not input("overwrite? (y/N)").lower() == "y"
|
||||
else:
|
||||
to_skip = True
|
||||
|
||||
if to_skip:
|
||||
return
|
||||
|
||||
if Encoder is None:
|
||||
track.download(stream, filename)
|
||||
else:
|
||||
track.download_while_re_encoding(
|
||||
stream,
|
||||
filename,
|
||||
target_encoding=output_extension,
|
||||
encoder=Encoder()
|
||||
)
|
||||
|
||||
if not self.arguments["no_metadata"]:
|
||||
track.metadata["lyrics"] = track.metadata["lyrics"].join()
|
||||
try:
|
||||
track.apply_metadata(filename, encoding=output_extension)
|
||||
except TypeError:
|
||||
# log.warning("Cannot write metadata to given file")
|
||||
pass
|
||||
|
||||
def download_tracks_from_file(self, path):
|
||||
# log.info(
|
||||
# "Checking and removing any duplicate tracks "
|
||||
# "in reading {}".format(path)
|
||||
# )
|
||||
with open(path, "r") as fin:
|
||||
# Read tracks into a list and remove any duplicates
|
||||
tracks = fin.read().splitlines()
|
||||
|
||||
# Remove duplicates and empty elements
|
||||
# Also strip whitespaces from elements (if any)
|
||||
spotdl.util.remove_duplicates(
|
||||
tracks,
|
||||
condition=lambda x: x,
|
||||
operation=str.strip
|
||||
)
|
||||
|
||||
# Overwrite file
|
||||
with open(path, "w") as fout:
|
||||
fout.writelines(tracks)
|
||||
|
||||
for number, track in enumerate(tracks, 1):
|
||||
try:
|
||||
metadata = self.search_metadata(track, self.arguments["search_format"])
|
||||
log_fmt=(str(number) + ". {artist} - {track-name}")
|
||||
# log.info(log_fmt)
|
||||
self.download_track_from_metadata(metadata)
|
||||
except (urllib.request.URLError, TypeError, IOError) as e:
|
||||
# log.exception(e.args[0])
|
||||
# log.warning("Failed. Will retry after other songs\n")
|
||||
tracks.append(track)
|
||||
except NoYouTubeVideoError:
|
||||
# log.warning("Failed. No YouTube video found.\n")
|
||||
pass
|
||||
else:
|
||||
if self.arguments["write_successful"]:
|
||||
with open(self.arguments["write_successful"], "a") as fout:
|
||||
fout.write(track)
|
||||
finally:
|
||||
with open(path, "w") as fout:
|
||||
fout.writelines(tracks[number-1:])
|
||||
|
||||
def download_tracks_from_file_threaded(self, path):
|
||||
# FIXME: Can we make this function cleaner?
|
||||
|
||||
# log.info(
|
||||
# "Checking and removing any duplicate tracks "
|
||||
# "in reading {}".format(path)
|
||||
# )
|
||||
with open(path, "r") as fin:
|
||||
# Read tracks into a list and remove any duplicates
|
||||
tracks = fin.read().splitlines()
|
||||
|
||||
# Remove duplicates and empty elements
|
||||
# Also strip whitespaces from elements (if any)
|
||||
spotdl.util.remove_duplicates(
|
||||
tracks,
|
||||
condition=lambda x: x,
|
||||
operation=str.strip
|
||||
)
|
||||
|
||||
# Overwrite file
|
||||
with open(path, "w") as fout:
|
||||
fout.writelines(tracks)
|
||||
|
||||
tracks_count = len(tracks)
|
||||
current_iteration = 1
|
||||
|
||||
next_track = tracks.pop(0)
|
||||
metadata = {
|
||||
"current_track": None,
|
||||
"next_track": spotdl.util.ThreadWithReturnValue(
|
||||
target=self.search_metadata,
|
||||
args=(next_track, self.arguments["search_format"])
|
||||
)
|
||||
}
|
||||
metadata["next_track"].start()
|
||||
while tracks_count > 0:
|
||||
metadata["current_track"] = metadata["next_track"].join()
|
||||
metadata["next_track"] = None
|
||||
try:
|
||||
print(tracks_count)
|
||||
print(tracks)
|
||||
if tracks_count > 1:
|
||||
current_track = next_track
|
||||
next_track = tracks.pop(0)
|
||||
metadata["next_track"] = spotdl.util.ThreadWithReturnValue(
|
||||
target=self.search_metadata,
|
||||
args=(next_track, self.arguments["search_format"])
|
||||
)
|
||||
metadata["next_track"].start()
|
||||
|
||||
log_fmt=(str(current_iteration) + ". {artist} - {track-name}")
|
||||
# log.info(log_fmt)
|
||||
if metadata["current_track"] is None:
|
||||
# log.warning("Something went wrong. Will retry after downloading remaining tracks")
|
||||
pass
|
||||
print(metadata["current_track"]["name"])
|
||||
# self.download_track_from_metadata(metadata["current_track"])
|
||||
except (urllib.request.URLError, TypeError, IOError) as e:
|
||||
# log.exception(e.args[0])
|
||||
# log.warning("Failed. Will retry after other songs\n")
|
||||
tracks.append(current_track)
|
||||
else:
|
||||
tracks_count -= 1
|
||||
if self.arguments["write_successful"]:
|
||||
with open(self.arguments["write_successful"], "a") as fout:
|
||||
fout.write(current_track)
|
||||
finally:
|
||||
current_iteration += 1
|
||||
with open(path, "w") as fout:
|
||||
fout.writelines(tracks)
|
||||
|
||||
@@ -7,28 +7,28 @@ import spotdl.util
|
||||
DEFAULT_CONFIGURATION = {
|
||||
"spotify-downloader": {
|
||||
"manual": False,
|
||||
"no-metadata": False,
|
||||
"no-fallback-metadata": False,
|
||||
"no_metadata": False,
|
||||
"no_fallback_metadata": False,
|
||||
"encoder": "ffmpeg",
|
||||
"overwrite": "prompt",
|
||||
"quality": "best",
|
||||
"input-ext": "automatic",
|
||||
"output-ext": "mp3",
|
||||
"write-to": None,
|
||||
"trim-silence": False,
|
||||
"download-only-metadata": False,
|
||||
"dry-run": False,
|
||||
"music-videos-only": False,
|
||||
"no-spaces": False,
|
||||
"input_ext": "automatic",
|
||||
"output_ext": "mp3",
|
||||
"write_to": None,
|
||||
"trim_silence": False,
|
||||
"download_only_metadata": False,
|
||||
"dry_run": False,
|
||||
"music_videos_only": False,
|
||||
"no_spaces": False,
|
||||
"processor": "synchronous",
|
||||
"output-file": "{artist} - {track-name}.{output-ext}",
|
||||
"search-format": "{artist} - {track-name} lyrics",
|
||||
"youtube-api-key": None,
|
||||
"output_file": "{artist} - {track-name}.{output-ext}",
|
||||
"search_format": "{artist} - {track-name} lyrics",
|
||||
"youtube_api_key": None,
|
||||
"skip": None,
|
||||
"write-successful": None,
|
||||
"log-level": "INFO",
|
||||
"spotify-client-id": "4fe3fecfe5334023a1472516cc99d805",
|
||||
"spotify-client-secret": "0f02b7c483c04257984695007a4a8d5c",
|
||||
"write_successful": None,
|
||||
"log_level": "INFO",
|
||||
"spotify_client_id": "4fe3fecfe5334023a1472516cc99d805",
|
||||
"spotify_client_secret": "0f02b7c483c04257984695007a4a8d5c",
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user