mirror of
https://github.com/migatu/conjurer.git
synced 2026-07-14 13:34:40 +00:00
Fixed version of spotify_dl
This commit is contained in:
@@ -0,0 +1,12 @@
|
||||
import signal
|
||||
import sys
|
||||
|
||||
|
||||
def signal_handler(sig, frame):
|
||||
# Signal handler to handle SIGINT, usually when Ctrl+C is pressed.. or if SIGINT is sent.
|
||||
# Thanks to https://stackoverflow.com/a/1112350/92837
|
||||
print("\nCaught interrupt(did you press Ctrl+C?), stopping spotify_dl")
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
signal.signal(signal.SIGINT, signal_handler)
|
||||
@@ -0,0 +1,13 @@
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
__all__ = ["VERSION"]
|
||||
|
||||
VERSION = "8.8.1"
|
||||
|
||||
if os.getenv("XDG_CACHE_HOME") is not None:
|
||||
SAVE_PATH = os.getenv("XDG_CACHE_HOME") + "/spotifydl"
|
||||
else:
|
||||
SAVE_PATH = str(Path.home()) + "/.cache/spotifydl"
|
||||
|
||||
DOWNLOAD_LIST = "download_list.log"
|
||||
@@ -0,0 +1,63 @@
|
||||
import logging
|
||||
from os import getenv
|
||||
import sentry_sdk
|
||||
from rich.logging import RichHandler
|
||||
from rich.console import Console
|
||||
|
||||
__all__ = ["log", "setLogLevel", "get_tokens", "console"]
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(message)s",
|
||||
datefmt="[%X]",
|
||||
handlers=[RichHandler(show_level=False, show_time=False)],
|
||||
)
|
||||
console = Console()
|
||||
log = logging.getLogger("sdl")
|
||||
sentry_sdk.init(
|
||||
"https://fc66a23d79634b9bba1690ea13e289f0@o321064.ingest.sentry.io/2383261"
|
||||
)
|
||||
|
||||
|
||||
def setLogLevel(level):
|
||||
"""
|
||||
Sets the log level of the global logger to the passed level
|
||||
:param level: the log level
|
||||
"""
|
||||
logging.getLogger().setLevel(level)
|
||||
|
||||
|
||||
def get_tokens():
|
||||
"""
|
||||
Checks if the required API keys for Spotify has been set.
|
||||
:param name: Name to be cleaned up
|
||||
:return string containing the cleaned name
|
||||
"""
|
||||
log.debug("Checking for tokens")
|
||||
CLIENT_ID = getenv("SPOTIPY_CLIENT_ID")
|
||||
CLIENT_SECRET = getenv("SPOTIPY_CLIENT_SECRET")
|
||||
log.debug("Tokens fetched : %s %s ", CLIENT_ID, CLIENT_SECRET)
|
||||
|
||||
if CLIENT_ID is None or CLIENT_SECRET is None:
|
||||
print(
|
||||
"""
|
||||
You need to set your Spotify API credentials. You can do this by
|
||||
setting environment variables like so:
|
||||
Linux:
|
||||
export SPOTIPY_CLIENT_ID='your-spotify-client-id'
|
||||
export SPOTIPY_CLIENT_SECRET='your-spotify-client-secret'
|
||||
|
||||
Windows Powershell:
|
||||
$env:SPOTIPY_CLIENT_ID='your-spotify-client-id'
|
||||
$env:SPOTIPY_CLIENT_SECRET='your-spotify-client-secret'
|
||||
|
||||
Windows CMD:
|
||||
set SPOTIPY_CLIENT_ID=your-spotify-client-id
|
||||
set SPOTIPY_CLIENT_SECRET=your-spotify-client-secret
|
||||
|
||||
Get your credentials at
|
||||
https://developer.spotify.com/dashboard
|
||||
"""
|
||||
)
|
||||
return None
|
||||
return CLIENT_ID, CLIENT_SECRET
|
||||
@@ -0,0 +1,261 @@
|
||||
import sys
|
||||
from spotify_dl.scaffold import log
|
||||
from spotify_dl.utils import sanitize
|
||||
from rich.progress import Progress
|
||||
|
||||
|
||||
def fetch_tracks(sp, item_type, url):
|
||||
"""
|
||||
Fetches tracks from the provided URL.
|
||||
:param sp: Spotify client
|
||||
:param item_type: Type of item being requested for: album/playlist/track
|
||||
:param url: URL of the item
|
||||
:return Dictionary of song and artist
|
||||
"""
|
||||
songs_list = []
|
||||
offset = 0
|
||||
songs_fetched = 0
|
||||
|
||||
if item_type == "playlist":
|
||||
with Progress() as progress:
|
||||
songs_task = progress.add_task(description="Fetching songs from playlist..")
|
||||
while True:
|
||||
items = sp.playlist_items(
|
||||
playlist_id=url,
|
||||
fields="items.track.name,items.track.artists(name, uri),"
|
||||
"items.track.album(name, release_date, total_tracks, images),"
|
||||
"items.track.track_number,total, next,offset,"
|
||||
"items.track.id",
|
||||
additional_types=["track"],
|
||||
offset=offset,
|
||||
)
|
||||
total_songs = items.get("total")
|
||||
track_info_task = progress.add_task(
|
||||
description="Fetching track info", total=len(items["items"])
|
||||
)
|
||||
for item in items["items"]:
|
||||
track_info = item.get("track")
|
||||
# If the user has a podcast in their playlist, there will be no track
|
||||
# Without this conditional, the program will fail later on when the metadata is fetched
|
||||
if track_info is None:
|
||||
offset += 1
|
||||
continue
|
||||
track_album_info = track_info.get("album")
|
||||
track_num = track_info.get("track_number")
|
||||
spotify_id = track_info.get("id")
|
||||
track_name = track_info.get("name")
|
||||
track_artist = ", ".join(
|
||||
[artist["name"] for artist in track_info.get("artists")]
|
||||
)
|
||||
if track_album_info:
|
||||
track_album = track_album_info.get("name")
|
||||
track_year = (
|
||||
track_album_info.get("release_date")[:4]
|
||||
if track_album_info.get("release_date")
|
||||
else ""
|
||||
)
|
||||
album_total = track_album_info.get("total_tracks")
|
||||
if len(item["track"]["album"]["images"]) > 0:
|
||||
cover = item["track"]["album"]["images"][0]["url"]
|
||||
else:
|
||||
cover = None
|
||||
|
||||
artists = track_info.get("artists")
|
||||
main_artist_id = (
|
||||
artists[0].get("uri", None) if len(artists) > 0 else None
|
||||
)
|
||||
genres = (
|
||||
sp.artist(artist_id=main_artist_id).get("genres", [])
|
||||
if main_artist_id
|
||||
else []
|
||||
)
|
||||
if len(genres) > 0:
|
||||
genre = genres[0]
|
||||
else:
|
||||
genre = ""
|
||||
songs_list.append(
|
||||
{
|
||||
"name": track_name,
|
||||
"artist": track_artist,
|
||||
"album": track_album,
|
||||
"year": track_year,
|
||||
"num_tracks": album_total,
|
||||
"num": track_num,
|
||||
"playlist_num": offset + 1,
|
||||
"cover": cover,
|
||||
"genre": genre,
|
||||
"spotify_id": spotify_id,
|
||||
"track_url": None,
|
||||
}
|
||||
)
|
||||
offset += 1
|
||||
songs_fetched += 1
|
||||
progress.update(
|
||||
task_id=track_info_task,
|
||||
description=f"Fetching track info for \n{track_name}",
|
||||
advance=1,
|
||||
)
|
||||
|
||||
progress.update(
|
||||
task_id=songs_task,
|
||||
description=f"Fetched {songs_fetched} of {total_songs} songs from the playlist",
|
||||
advance=100,
|
||||
total=total_songs,
|
||||
)
|
||||
if total_songs == offset:
|
||||
break
|
||||
|
||||
elif item_type == "album":
|
||||
with Progress() as progress:
|
||||
album_songs_task = progress.add_task(
|
||||
description="Fetching songs from the album.."
|
||||
)
|
||||
while True:
|
||||
album_info = sp.album(album_id=url)
|
||||
items = sp.album_tracks(album_id=url, offset=offset)
|
||||
total_songs = items.get("total")
|
||||
track_album = album_info.get("name")
|
||||
track_year = (
|
||||
album_info.get("release_date")[:4]
|
||||
if album_info.get("release_date")
|
||||
else ""
|
||||
)
|
||||
album_total = album_info.get("total_tracks")
|
||||
if len(album_info["images"]) > 0:
|
||||
cover = album_info["images"][0]["url"]
|
||||
else:
|
||||
cover = None
|
||||
if (
|
||||
len(sp.artist(artist_id=album_info["artists"][0]["uri"])["genres"])
|
||||
> 0
|
||||
):
|
||||
genre = sp.artist(artist_id=album_info["artists"][0]["uri"])[
|
||||
"genres"
|
||||
][0]
|
||||
else:
|
||||
genre = ""
|
||||
for item in items["items"]:
|
||||
track_name = item.get("name")
|
||||
track_artist = ", ".join(
|
||||
[artist["name"] for artist in item["artists"]]
|
||||
)
|
||||
track_num = item["track_number"]
|
||||
spotify_id = item.get("id")
|
||||
songs_list.append(
|
||||
{
|
||||
"name": track_name,
|
||||
"artist": track_artist,
|
||||
"album": track_album,
|
||||
"year": track_year,
|
||||
"num_tracks": album_total,
|
||||
"num": track_num,
|
||||
"track_url": None,
|
||||
"playlist_num": offset + 1,
|
||||
"cover": cover,
|
||||
"genre": genre,
|
||||
"spotify_id": spotify_id,
|
||||
}
|
||||
)
|
||||
offset += 1
|
||||
|
||||
progress.update(
|
||||
task_id=album_songs_task,
|
||||
description=f"Fetched {offset} of {album_total} songs from the album {track_album}",
|
||||
advance=offset,
|
||||
total=album_total,
|
||||
)
|
||||
if album_total == offset:
|
||||
break
|
||||
|
||||
elif item_type == "track":
|
||||
items = sp.track(track_id=url)
|
||||
track_name = items.get("name")
|
||||
album_info = items.get("album")
|
||||
track_artist = ", ".join([artist["name"] for artist in items["artists"]])
|
||||
if album_info:
|
||||
track_album = album_info.get("name")
|
||||
track_year = (
|
||||
album_info.get("release_date")[:4]
|
||||
if album_info.get("release_date")
|
||||
else ""
|
||||
)
|
||||
album_total = album_info.get("total_tracks")
|
||||
track_num = items["track_number"]
|
||||
spotify_id = items["id"]
|
||||
if len(items["album"]["images"]) > 0:
|
||||
cover = items["album"]["images"][0]["url"]
|
||||
else:
|
||||
cover = None
|
||||
if len(sp.artist(artist_id=items["artists"][0]["uri"])["genres"]) > 0:
|
||||
genre = sp.artist(artist_id=items["artists"][0]["uri"])["genres"][0]
|
||||
else:
|
||||
genre = ""
|
||||
songs_list.append(
|
||||
{
|
||||
"name": track_name,
|
||||
"artist": track_artist,
|
||||
"album": track_album,
|
||||
"year": track_year,
|
||||
"num_tracks": album_total,
|
||||
"num": track_num,
|
||||
"playlist_num": offset + 1,
|
||||
"cover": cover,
|
||||
"genre": genre,
|
||||
"track_url": None,
|
||||
"spotify_id": spotify_id,
|
||||
}
|
||||
)
|
||||
|
||||
return songs_list
|
||||
|
||||
|
||||
def parse_spotify_url(url):
|
||||
"""
|
||||
Parse the provided Spotify playlist URL and determine if it is a playlist, track or album.
|
||||
:param url: URL to be parsed
|
||||
|
||||
:return tuple indicating the type and id of the item
|
||||
"""
|
||||
if url.startswith("spotify:"):
|
||||
log.error("Spotify URI was provided instead of a playlist/album/track URL.")
|
||||
sys.exit(1)
|
||||
parsed_url = url.replace("https://open.spotify.com/", "").split("?")[0]
|
||||
item_type = parsed_url.split("/")[0]
|
||||
item_id = parsed_url.split("/")[1]
|
||||
return item_type, item_id
|
||||
|
||||
|
||||
def get_item_name(sp, item_type, item_id):
|
||||
"""
|
||||
Fetch the name of the item.
|
||||
:param sp: Spotify Client
|
||||
:param item_type: Type of the item
|
||||
:param item_id: id of the item
|
||||
:return String indicating the name of the item
|
||||
"""
|
||||
if item_type == "playlist":
|
||||
name = sp.playlist(playlist_id=item_id, fields="name").get("name")
|
||||
elif item_type == "album":
|
||||
name = sp.album(album_id=item_id).get("name")
|
||||
elif item_type == "track":
|
||||
name = sp.track(track_id=item_id).get("name")
|
||||
return sanitize(name)
|
||||
|
||||
|
||||
def validate_spotify_urls(urls):
|
||||
"""
|
||||
Validate the URLs and determine if the item types they have are supported
|
||||
: return a list of valid urls
|
||||
"""
|
||||
valid_urls = []
|
||||
for url in urls:
|
||||
item_type, item_id = parse_spotify_url(url)
|
||||
log.debug("Got item_type %s and item_id %s", item_type, item_id)
|
||||
if item_type not in ["album", "track", "playlist"]:
|
||||
log.info("Only albums/tracks/playlists are supported and not %s", item_type)
|
||||
continue
|
||||
if item_id is None:
|
||||
log.info("Couln't get valid item_id for %s", url)
|
||||
else:
|
||||
valid_urls.append(url)
|
||||
return valid_urls
|
||||
@@ -0,0 +1,227 @@
|
||||
#!/usr/bin/env python3
|
||||
import argparse
|
||||
import time
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from logging import DEBUG, ERROR
|
||||
from pathlib import Path, PurePath
|
||||
import spotipy
|
||||
from spotipy.oauth2 import SpotifyClientCredentials
|
||||
|
||||
from spotify_dl.constants import VERSION
|
||||
from spotify_dl.scaffold import log, setLogLevel, console, get_tokens
|
||||
from spotify_dl.spotify import (
|
||||
fetch_tracks,
|
||||
parse_spotify_url,
|
||||
validate_spotify_urls,
|
||||
get_item_name,
|
||||
)
|
||||
|
||||
from spotify_dl.youtube import (
|
||||
download_songs,
|
||||
default_filename,
|
||||
playlist_num_filename,
|
||||
dump_json,
|
||||
)
|
||||
|
||||
|
||||
def spotify_dl():
|
||||
"""Main entry point of the script."""
|
||||
parser = argparse.ArgumentParser(prog="spotify_dl")
|
||||
parser.add_argument(
|
||||
"-l",
|
||||
"--url",
|
||||
action="store",
|
||||
help="Spotify Playlist link URL",
|
||||
type=str,
|
||||
nargs="+",
|
||||
required=False, # this has to be set to false to prevent useless prompt for url when all user wants is the script version
|
||||
)
|
||||
parser.add_argument(
|
||||
"-o",
|
||||
"--output",
|
||||
type=str,
|
||||
action="store",
|
||||
help="Specify download directory.",
|
||||
required=False,
|
||||
default=".",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-d",
|
||||
"--download",
|
||||
action="store_true",
|
||||
help="Download using youtube-dl",
|
||||
default=True,
|
||||
)
|
||||
parser.add_argument(
|
||||
"-j",
|
||||
"--dump-json",
|
||||
action="store_true",
|
||||
help="Dump info-json using youtube-dl",
|
||||
default=False,
|
||||
)
|
||||
parser.add_argument(
|
||||
"-f",
|
||||
"--format_str",
|
||||
type=str,
|
||||
action="store",
|
||||
help="Specify youtube-dl format string.",
|
||||
default="bestaudio/best",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-k",
|
||||
"--keep_playlist_order",
|
||||
default=False,
|
||||
action="store_true",
|
||||
help="Whether to keep original playlist ordering or not.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-m",
|
||||
"--skip_mp3",
|
||||
action="store_true",
|
||||
help="Don't convert downloaded songs to mp3",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-s",
|
||||
"--use_sponsorblock",
|
||||
default="no",
|
||||
action="store",
|
||||
help="Whether to skip non-music sections using SponsorBlock API. Pass y or yes to skip using SponsorBlock",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-w",
|
||||
"--no-overwrites",
|
||||
action="store_true",
|
||||
help="Whether we should avoid overwriting the target audio file if it already exists",
|
||||
default=False,
|
||||
)
|
||||
parser.add_argument(
|
||||
"-r",
|
||||
"--remove-trailing-tracks",
|
||||
default="no",
|
||||
action="store",
|
||||
help="Whether we should delete tracks that were previously downloaded but are not longer in the playlist",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-V",
|
||||
"--verbose",
|
||||
action="store_true",
|
||||
help="Show more information on what" "s happening.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-v",
|
||||
"--version",
|
||||
action="store_true",
|
||||
help="Shows current version of the program",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-mc",
|
||||
"--multi_core",
|
||||
action="store",
|
||||
type=str,
|
||||
default=0,
|
||||
help="Use multiprocessing [-m [int:numcores]",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-p",
|
||||
"--proxy",
|
||||
action="store",
|
||||
type=str,
|
||||
default="",
|
||||
help="Download through a proxy. Support HTTP & SOCKS5. Use 'http://username:password@hostname:port' or 'http://hostname:port'",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
num_cores = os.cpu_count()
|
||||
args.multi_core = int(args.multi_core)
|
||||
|
||||
if args.dump_json:
|
||||
setLogLevel(ERROR)
|
||||
if args.verbose:
|
||||
setLogLevel(DEBUG)
|
||||
|
||||
log.info("Starting spotify_dl v%s", VERSION)
|
||||
log.debug("Setting debug mode on spotify_dl")
|
||||
|
||||
if args.multi_core > (num_cores - 1):
|
||||
log.info(
|
||||
"Requested cores %d exceeds available %d, using %d cores.",
|
||||
args.multi_core,
|
||||
num_cores,
|
||||
num_cores - 1,
|
||||
)
|
||||
args.multi_core = num_cores - 1
|
||||
if args.version:
|
||||
console.print(f"spotify_dl [bold green]v{VERSION}[/bold green]")
|
||||
sys.exit(0)
|
||||
|
||||
if os.path.isfile(os.path.expanduser("~/.spotify_dl_settings")):
|
||||
with open(os.path.expanduser("~/.spotify_dl_settings")) as file:
|
||||
config = json.load(file)
|
||||
print(config)
|
||||
|
||||
for key, value in config.items():
|
||||
if (isinstance(value, bool) and value) or (
|
||||
isinstance(value, str) and value and value.lower() in ["true", "t"]
|
||||
):
|
||||
setattr(args, key, True)
|
||||
else:
|
||||
setattr(args, key, value)
|
||||
|
||||
if not args.url:
|
||||
raise (Exception("No playlist url provided:"))
|
||||
|
||||
tokens = get_tokens()
|
||||
if tokens is None:
|
||||
sys.exit(1)
|
||||
client_id, client_secret = tokens
|
||||
|
||||
sp = spotipy.Spotify(
|
||||
auth_manager=SpotifyClientCredentials(
|
||||
client_id=client_id, client_secret=client_secret
|
||||
)
|
||||
)
|
||||
log.debug("Arguments: %s ", args)
|
||||
log.info("Sponsorblock enabled?: %s", args.use_sponsorblock)
|
||||
valid_urls = validate_spotify_urls(args.url)
|
||||
if not valid_urls:
|
||||
sys.exit(1)
|
||||
|
||||
url_data = {"urls": []}
|
||||
start_time = time.time()
|
||||
for url in valid_urls:
|
||||
url_dict = {}
|
||||
item_type, item_id = parse_spotify_url(url)
|
||||
directory_name = get_item_name(sp, item_type, item_id)
|
||||
url_dict["save_path"] = Path(
|
||||
PurePath.joinpath(Path(args.output), Path(directory_name))
|
||||
)
|
||||
url_dict["save_path"].mkdir(parents=True, exist_ok=True)
|
||||
log.info("Saving songs to %s directory", directory_name)
|
||||
url_dict["songs"] = fetch_tracks(sp, item_type, url)
|
||||
url_data["urls"].append(url_dict.copy())
|
||||
if args.dump_json is True:
|
||||
dump_json(url_dict["songs"])
|
||||
elif args.download is True:
|
||||
file_name_f = default_filename
|
||||
if args.keep_playlist_order:
|
||||
file_name_f = playlist_num_filename
|
||||
|
||||
download_songs(
|
||||
songs=url_data,
|
||||
output_dir=args.output,
|
||||
format_str=args.format_str,
|
||||
skip_mp3=args.skip_mp3,
|
||||
keep_playlist_order=args.keep_playlist_order,
|
||||
no_overwrites=args.no_overwrites,
|
||||
remove_trailing_tracks=(args.remove_trailing_tracks[0].lower()),
|
||||
use_sponsorblock=args.use_sponsorblock,
|
||||
file_name_f=file_name_f,
|
||||
multi_core=args.multi_core,
|
||||
proxy=args.proxy,
|
||||
)
|
||||
log.info("Download completed in %.2f seconds.", time.time() - start_time)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
spotify_dl()
|
||||
@@ -0,0 +1,10 @@
|
||||
def sanitize(name, replace_with=""):
|
||||
"""
|
||||
Removes some of the reserved characters from the name so it can be saved
|
||||
:param name: Name to be cleaned up
|
||||
:return string containing the cleaned name
|
||||
"""
|
||||
clean_up_list = ["\\", "/", ":", "*", "?", '"', "<", ">", "|", "\0", "$", "\""]
|
||||
for x in clean_up_list:
|
||||
name = name.replace(x, replace_with)
|
||||
return name
|
||||
@@ -0,0 +1,327 @@
|
||||
import urllib.request
|
||||
from os import path
|
||||
import os
|
||||
import multiprocessing
|
||||
|
||||
import shutil
|
||||
import json
|
||||
import mutagen
|
||||
import csv
|
||||
import yt_dlp
|
||||
from mutagen.easyid3 import EasyID3
|
||||
from mutagen.id3 import APIC, ID3
|
||||
from mutagen.mp3 import MP3
|
||||
from spotify_dl.scaffold import log
|
||||
from spotify_dl.utils import sanitize
|
||||
from spotify_dl.constants import DOWNLOAD_LIST
|
||||
|
||||
|
||||
def default_filename(**kwargs):
|
||||
"""name without number"""
|
||||
return sanitize(
|
||||
f"{kwargs['artist']} - {kwargs['name']}", "#"
|
||||
) # youtube-dl automatically replaces with #
|
||||
|
||||
|
||||
def playlist_num_filename(**kwargs):
|
||||
"""name with track number"""
|
||||
return f"{kwargs['track_num']} - {default_filename(**kwargs)}"
|
||||
|
||||
|
||||
def dump_json(songs):
|
||||
"""
|
||||
Outputs the JSON response of ydl.extract_info to stdout
|
||||
:param songs: the songs for which the JSON should be output
|
||||
"""
|
||||
for song in songs:
|
||||
query = f"{song.get('artist')} - {song.get('name')} Lyrics".replace(
|
||||
":", ""
|
||||
).replace('"', "")
|
||||
|
||||
ydl_opts = {"quiet": True}
|
||||
|
||||
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
||||
try:
|
||||
ytJson = ydl.extract_info("ytsearch:" + query, False)
|
||||
print(json.dumps(ytJson.get("entries")))
|
||||
except Exception as e: # skipcq: PYL-W0703
|
||||
log.debug(e)
|
||||
print(
|
||||
f"Failed to download {song.get('name')}, make sure yt_dlp is up to date"
|
||||
)
|
||||
continue
|
||||
|
||||
|
||||
def write_tracks(tracks_file, song_dict):
|
||||
"""
|
||||
Writes the information of all tracks in the playlist[s] to a text file in csv kind of format
|
||||
This includins the name, artist, and spotify URL. Each is delimited by a comma.
|
||||
:param tracks_file: name of file towhich the songs are to be written
|
||||
:param song_dict: the songs to be written to tracks_file
|
||||
"""
|
||||
track_db = []
|
||||
|
||||
with open(tracks_file, "w+", encoding="utf-8", newline="") as file_out:
|
||||
i = 0
|
||||
writer = csv.writer(file_out, delimiter=";")
|
||||
for url_dict in song_dict["urls"]:
|
||||
# for track in url_dict['songs']:
|
||||
for track in url_dict["songs"]:
|
||||
track_url = track["track_url"] # here
|
||||
track_name = track["name"]
|
||||
track_artist = track["artist"]
|
||||
track_num = track["num"]
|
||||
track_album = track["album"]
|
||||
track["save_path"] = url_dict["save_path"]
|
||||
track_db.append(track)
|
||||
track_index = i
|
||||
i += 1
|
||||
csv_row = [
|
||||
track_name,
|
||||
track_artist,
|
||||
track_url,
|
||||
str(track_num),
|
||||
track_album,
|
||||
str(track_index),
|
||||
]
|
||||
try:
|
||||
writer.writerow(csv_row)
|
||||
except UnicodeEncodeError:
|
||||
print(
|
||||
"Track named {track_name} failed due to an encoding error. This is \
|
||||
most likely due to this song having a non-English name."
|
||||
)
|
||||
return track_db
|
||||
|
||||
|
||||
def set_tags(temp, filename, kwargs):
|
||||
"""
|
||||
sets song tags after they are downloaded
|
||||
:param temp: contains index used to obtain more info about song being editted
|
||||
:param filename: location of song whose tags are to be editted
|
||||
:param kwargs: a dictionary of extra arguments to be used in tag editing
|
||||
"""
|
||||
song = kwargs["track_db"][int(temp[-1])]
|
||||
try:
|
||||
song_file = MP3(filename, ID3=EasyID3)
|
||||
except mutagen.MutagenError as e:
|
||||
log.debug(e)
|
||||
print(
|
||||
f"Failed to download: {filename}, please ensure YouTubeDL is up-to-date. "
|
||||
)
|
||||
|
||||
return
|
||||
song_file["date"] = song.get("year")
|
||||
if kwargs["keep_playlist_order"]:
|
||||
song_file["tracknumber"] = str(song.get("playlist_num"))
|
||||
else:
|
||||
song_file["tracknumber"] = (
|
||||
str(song.get("num")) + "/" + str(song.get("num_tracks"))
|
||||
)
|
||||
|
||||
song_file["genre"] = song.get("genre")
|
||||
song_file.save()
|
||||
song_file = MP3(filename, ID3=ID3)
|
||||
cover = song.get("cover")
|
||||
if cover is not None:
|
||||
if cover.lower().startswith("http"):
|
||||
req = urllib.request.Request(cover)
|
||||
else:
|
||||
raise ValueError from None
|
||||
with urllib.request.urlopen(req) as resp: # nosec
|
||||
song_file.tags["APIC"] = APIC(
|
||||
encoding=3,
|
||||
mime="image/jpeg",
|
||||
type=3,
|
||||
desc="Cover",
|
||||
data=resp.read(),
|
||||
)
|
||||
song_file.save()
|
||||
|
||||
|
||||
def find_and_download_songs(kwargs):
|
||||
"""
|
||||
function handles actual download of the songs
|
||||
the youtube_search lib is used to search for songs and get best url
|
||||
:param kwargs: dictionary of key value arguments to be used in download
|
||||
"""
|
||||
sponsorblock_postprocessor = []
|
||||
reference_file = kwargs["reference_file"]
|
||||
files = {}
|
||||
with open(reference_file, "r", encoding="utf-8") as file:
|
||||
for line in file:
|
||||
temp = line.split(";")
|
||||
name, artist, album, i = (
|
||||
temp[0],
|
||||
temp[1],
|
||||
temp[4],
|
||||
int(temp[-1].replace("\n", "")),
|
||||
)
|
||||
|
||||
query = f"{artist} - {name} Lyrics".replace(":", "").replace('"', "")
|
||||
print(f"Initiating download for {query}.")
|
||||
|
||||
file_name = kwargs["file_name_f"](
|
||||
name=name, artist=artist, track_num=kwargs["track_db"][i].get("num")
|
||||
)
|
||||
|
||||
if kwargs["use_sponsorblock"][0].lower() == "y":
|
||||
sponsorblock_postprocessor = [
|
||||
{
|
||||
"key": "SponsorBlock",
|
||||
"categories": ["skip_non_music_sections"],
|
||||
},
|
||||
{
|
||||
"key": "ModifyChapters",
|
||||
"remove_sponsor_segments": ["music_offtopic"],
|
||||
"force_keyframes": True,
|
||||
},
|
||||
]
|
||||
save_path = kwargs["track_db"][i]["save_path"]
|
||||
file_path = path.join(save_path, file_name)
|
||||
|
||||
mp3file_path = f"{file_path}.mp3"
|
||||
|
||||
if save_path not in files:
|
||||
path_files = set()
|
||||
files[save_path] = path_files
|
||||
else:
|
||||
path_files = files[save_path]
|
||||
|
||||
path_files.add(f"{file_name}.mp3")
|
||||
|
||||
if (
|
||||
kwargs["no_overwrites"]
|
||||
and not kwargs["skip_mp3"]
|
||||
and path.exists(mp3file_path)
|
||||
):
|
||||
print(f"File {mp3file_path} already exists, we do not overwrite it ")
|
||||
continue
|
||||
|
||||
outtmpl = f"{file_path}.%(ext)s"
|
||||
ydl_opts = {
|
||||
"proxy": kwargs.get("proxy"),
|
||||
"default_search": "ytsearch",
|
||||
"format": "bestaudio/best",
|
||||
"outtmpl": outtmpl,
|
||||
"postprocessors": sponsorblock_postprocessor,
|
||||
"noplaylist": True,
|
||||
"no_color": False,
|
||||
"postprocessor_args": [
|
||||
"-metadata",
|
||||
"title=" + name,
|
||||
"-metadata",
|
||||
"artist=" + artist,
|
||||
"-metadata",
|
||||
"album=" + album,
|
||||
],
|
||||
}
|
||||
if not kwargs["skip_mp3"]:
|
||||
mp3_postprocess_opts = {
|
||||
"key": "FFmpegExtractAudio",
|
||||
"preferredcodec": "mp3",
|
||||
"preferredquality": "192",
|
||||
}
|
||||
ydl_opts["postprocessors"].append(mp3_postprocess_opts.copy())
|
||||
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
||||
try:
|
||||
ydl.download([query])
|
||||
except Exception as e: # skipcq: PYL-W0703
|
||||
log.debug(e)
|
||||
print(f"Failed to download {name}, make sure yt_dlp is up to date")
|
||||
if not kwargs["skip_mp3"]:
|
||||
set_tags(temp, mp3file_path, kwargs)
|
||||
if kwargs["remove_trailing_tracks"] == "y":
|
||||
for save_path in files:
|
||||
for f in os.listdir(save_path):
|
||||
if f not in files[save_path]:
|
||||
print(f"File {f} is not in the playlist anymore, we delete it")
|
||||
os.remove(path.join(save_path, f))
|
||||
|
||||
|
||||
def multicore_find_and_download_songs(kwargs):
|
||||
"""
|
||||
function handles divinding songs to be downloaded among the specified number of CPU's
|
||||
extra songs are shared among the CPU's
|
||||
each cpu then handles its own batch through the multihandler fn
|
||||
"""
|
||||
reference_file = kwargs["reference_file"]
|
||||
lines = []
|
||||
with open(reference_file, "r", encoding="utf-8") as file:
|
||||
for line in file:
|
||||
lines.append(line)
|
||||
cpu_count = kwargs["multi_core"]
|
||||
number_of_songs = len(lines)
|
||||
songs_per_cpu = number_of_songs // cpu_count
|
||||
extra_songs = number_of_songs % cpu_count
|
||||
|
||||
cpu_count_list = []
|
||||
for cpu in range(cpu_count):
|
||||
songs = songs_per_cpu
|
||||
if cpu < extra_songs:
|
||||
songs = songs + 1
|
||||
cpu_count_list.append(songs)
|
||||
|
||||
index = 0
|
||||
file_segments = []
|
||||
for cpu in cpu_count_list:
|
||||
right = cpu + index
|
||||
segment = lines[index:right]
|
||||
index = index + cpu
|
||||
file_segments.append(segment)
|
||||
|
||||
processes = []
|
||||
segment_index = 0
|
||||
for segment in file_segments:
|
||||
p = multiprocessing.Process(
|
||||
target=multicore_handler, args=(segment_index, segment, kwargs.copy())
|
||||
)
|
||||
processes.append(p)
|
||||
segment_index += 1
|
||||
|
||||
for p in processes:
|
||||
p.start()
|
||||
for p in processes:
|
||||
p.join()
|
||||
|
||||
|
||||
def multicore_handler(segment_index, segment, kwargs):
|
||||
"""
|
||||
function to handle each unique processor spawned download job
|
||||
:param segment_index: to be used for naming the reference file to be used for processor's download batch
|
||||
:param segment: list of songs to be downloaded using spawning processor
|
||||
"""
|
||||
reference_filename = f"{segment_index}.txt"
|
||||
with open(reference_filename, "w+", encoding="utf-8") as file_out:
|
||||
for line in segment:
|
||||
file_out.write(line)
|
||||
|
||||
kwargs["reference_file"] = reference_filename
|
||||
find_and_download_songs(kwargs)
|
||||
|
||||
if os.path.exists(reference_filename):
|
||||
os.remove(reference_filename)
|
||||
|
||||
|
||||
def download_songs(**kwargs):
|
||||
"""
|
||||
Downloads songs from the YouTube URL passed to either current directory or download_directory, as it is passed. [made small typo change]
|
||||
:param kwargs: keyword arguments to be passed on between functions when downloading
|
||||
"""
|
||||
for url in kwargs["songs"]["urls"]:
|
||||
log.debug("Downloading to %s", url["save_path"])
|
||||
reference_file = DOWNLOAD_LIST
|
||||
track_db = write_tracks(reference_file, kwargs["songs"])
|
||||
try:
|
||||
shutil.copy(reference_file, kwargs["output_dir"] + "/" + reference_file)
|
||||
os.remove(reference_file)
|
||||
except:
|
||||
os.rename(reference_file, kwargs["output_dir"] + "/" + reference_file)
|
||||
reference_file = str(kwargs["output_dir"]) + "/" + reference_file
|
||||
kwargs["reference_file"] = reference_file
|
||||
kwargs["track_db"] = track_db
|
||||
if kwargs["multi_core"] > 1:
|
||||
multicore_find_and_download_songs(kwargs)
|
||||
else:
|
||||
find_and_download_songs(kwargs)
|
||||
os.remove(reference_file)
|
||||
Reference in New Issue
Block a user