mirror of
https://github.com/THIS-IS-NOT-A-BACKUP/zspotify.git
synced 2025-12-27 19:10:28 +00:00
Revert "Fix #105 renamed src to zspotify"
This reverts commit cf1b98cce3.
This commit is contained in:
el-gringo-alto
@@ -1,4 +0,0 @@
|
||||
from app import client
|
||||
|
||||
if __name__ == '__main__':
|
||||
client()
|
||||
@@ -1,60 +0,0 @@
|
||||
from tqdm import tqdm
|
||||
|
||||
from const import ITEMS, ARTISTS, NAME, ID
|
||||
from track import download_track
|
||||
from utils import sanitize_data
|
||||
from zspotify import ZSpotify
|
||||
|
||||
ALBUM_URL = 'https://api.spotify.com/v1/albums'
|
||||
ARTIST_URL = 'https://api.spotify.com/v1/artists'
|
||||
|
||||
|
||||
def get_album_tracks(album_id):
|
||||
""" Returns album tracklist """
|
||||
songs = []
|
||||
offset = 0
|
||||
limit = 50
|
||||
|
||||
while True:
|
||||
resp = ZSpotify.invoke_url_with_params(f'{ALBUM_URL}/{album_id}/tracks', limit=limit, offset=offset)
|
||||
offset += limit
|
||||
songs.extend(resp[ITEMS])
|
||||
if len(resp[ITEMS]) < limit:
|
||||
break
|
||||
|
||||
return songs
|
||||
|
||||
|
||||
def get_album_name(album_id):
|
||||
""" Returns album name """
|
||||
resp = ZSpotify.invoke_url(f'{ALBUM_URL}/{album_id}')
|
||||
return resp[ARTISTS][0][NAME], sanitize_data(resp[NAME])
|
||||
|
||||
|
||||
def get_artist_albums(artist_id):
|
||||
""" Returns artist's albums """
|
||||
resp = ZSpotify.invoke_url(f'{ARTIST_URL}/{artist_id}/albums')
|
||||
# Return a list each album's id
|
||||
album_ids = [resp[ITEMS][i][ID] for i in range(len(resp[ITEMS]))]
|
||||
# Recursive requests to get all albums including singles an EPs
|
||||
while resp['next'] is not None:
|
||||
resp = ZSpotify.invoke_url(resp['next'])
|
||||
album_ids.extend([resp[ITEMS][i][ID] for i in range(len(resp[ITEMS]))])
|
||||
|
||||
return album_ids
|
||||
|
||||
|
||||
def download_album(album):
|
||||
""" Downloads songs from an album """
|
||||
artist, album_name = get_album_name(album)
|
||||
tracks = get_album_tracks(album)
|
||||
for n, track in tqdm(enumerate(tracks, start=1), unit_scale=True, unit='Song', total=len(tracks)):
|
||||
download_track(track[ID], f'{artist}/{album_name}',
|
||||
prefix=True, prefix_value=str(n), disable_progressbar=True)
|
||||
|
||||
|
||||
def download_artist_albums(artist):
|
||||
""" Downloads albums of an artist """
|
||||
albums = get_artist_albums(artist)
|
||||
for album_id in albums:
|
||||
download_album(album_id)
|
||||
264
zspotify/app.py
264
zspotify/app.py
@@ -1,264 +0,0 @@
|
||||
import sys
|
||||
|
||||
from librespot.audio.decoders import AudioQuality
|
||||
from tabulate import tabulate
|
||||
|
||||
from album import download_album, download_artist_albums
|
||||
from const import TRACK, NAME, ID, ARTIST, ARTISTS, ITEMS, TRACKS, EXPLICIT, ALBUM, ALBUMS, \
|
||||
OWNER, PLAYLIST, PLAYLISTS, DISPLAY_NAME
|
||||
from playlist import get_playlist_songs, get_playlist_info, download_from_user_playlist, download_playlist
|
||||
from podcast import download_episode, get_show_episodes
|
||||
from track import download_track, get_saved_tracks
|
||||
from utils import sanitize_data, splash, split_input, regex_input_for_urls
|
||||
from zspotify import ZSpotify
|
||||
|
||||
SEARCH_URL = 'https://api.spotify.com/v1/search'
|
||||
|
||||
|
||||
def client() -> None:
|
||||
""" Connects to spotify to perform query's and get songs to download """
|
||||
ZSpotify()
|
||||
splash()
|
||||
|
||||
if ZSpotify.check_premium():
|
||||
print('[ DETECTED PREMIUM ACCOUNT - USING VERY_HIGH QUALITY ]\n\n')
|
||||
ZSpotify.DOWNLOAD_QUALITY = AudioQuality.VERY_HIGH
|
||||
else:
|
||||
print('[ DETECTED FREE ACCOUNT - USING HIGH QUALITY ]\n\n')
|
||||
ZSpotify.DOWNLOAD_QUALITY = AudioQuality.HIGH
|
||||
|
||||
if len(sys.argv) > 1:
|
||||
if sys.argv[1] == '-p' or sys.argv[1] == '--playlist':
|
||||
download_from_user_playlist()
|
||||
elif sys.argv[1] == '-ls' or sys.argv[1] == '--liked-songs':
|
||||
for song in get_saved_tracks():
|
||||
if not song[TRACK][NAME]:
|
||||
print('### SKIPPING: SONG DOES NOT EXISTS ON SPOTIFY ANYMORE ###')
|
||||
else:
|
||||
download_track(song[TRACK][ID], 'Liked Songs/')
|
||||
print('\n')
|
||||
else:
|
||||
track_id, album_id, playlist_id, episode_id, show_id, artist_id = regex_input_for_urls(sys.argv[1])
|
||||
|
||||
if track_id is not None:
|
||||
download_track(track_id)
|
||||
elif artist_id is not None:
|
||||
download_artist_albums(artist_id)
|
||||
elif album_id is not None:
|
||||
download_album(album_id)
|
||||
elif playlist_id is not None:
|
||||
playlist_songs = get_playlist_songs(playlist_id)
|
||||
name, _ = get_playlist_info(playlist_id)
|
||||
for song in playlist_songs:
|
||||
download_track(song[TRACK][ID],
|
||||
sanitize_data(name) + '/')
|
||||
print('\n')
|
||||
elif episode_id is not None:
|
||||
download_episode(episode_id)
|
||||
elif show_id is not None:
|
||||
for episode in get_show_episodes(show_id):
|
||||
download_episode(episode)
|
||||
|
||||
else:
|
||||
search_text = ''
|
||||
while len(search_text) == 0:
|
||||
search_text = input('Enter search or URL: ')
|
||||
|
||||
track_id, album_id, playlist_id, episode_id, show_id, artist_id = regex_input_for_urls(search_text)
|
||||
|
||||
if track_id is not None:
|
||||
download_track(track_id)
|
||||
elif artist_id is not None:
|
||||
download_artist_albums(artist_id)
|
||||
elif album_id is not None:
|
||||
download_album(album_id)
|
||||
elif playlist_id is not None:
|
||||
playlist_songs = get_playlist_songs(playlist_id)
|
||||
name, _ = get_playlist_info(playlist_id)
|
||||
for song in playlist_songs:
|
||||
download_track(song[TRACK][ID], sanitize_data(name) + '/')
|
||||
print('\n')
|
||||
elif episode_id is not None:
|
||||
download_episode(episode_id)
|
||||
elif show_id is not None:
|
||||
for episode in get_show_episodes(show_id):
|
||||
download_episode(episode)
|
||||
else:
|
||||
search(search_text)
|
||||
|
||||
|
||||
def search(search_term):
|
||||
""" Searches Spotify's API for relevant data """
|
||||
params = {'limit': '10',
|
||||
'offset': '0',
|
||||
'q': search_term,
|
||||
'type': 'track,album,artist,playlist'}
|
||||
|
||||
# Parse args
|
||||
splits = search_term.split()
|
||||
for split in splits:
|
||||
index = splits.index(split)
|
||||
|
||||
if split[0] == '-' and len(split) > 1:
|
||||
if len(splits)-1 == index:
|
||||
raise IndexError('No parameters passed after option: {}\n'.
|
||||
format(split))
|
||||
|
||||
if split == '-l' or split == '-limit':
|
||||
try:
|
||||
int(splits[index+1])
|
||||
except ValueError:
|
||||
raise ValueError('Paramater passed after {} option must be an integer.\n'.
|
||||
format(split))
|
||||
if int(splits[index+1]) > 50:
|
||||
raise ValueError('Invalid limit passed. Max is 50.\n')
|
||||
params['limit'] = splits[index+1]
|
||||
|
||||
|
||||
if split == '-t' or split == '-type':
|
||||
|
||||
allowed_types = ['track', 'playlist', 'album', 'artist']
|
||||
passed_types = []
|
||||
for i in range(index+1, len(splits)):
|
||||
if splits[i][0] == '-':
|
||||
break
|
||||
|
||||
if splits[i] not in allowed_types:
|
||||
raise ValueError('Parameters passed after {} option must be from this list:\n{}'.
|
||||
format(split, '\n'.join(allowed_types)))
|
||||
|
||||
passed_types.append(splits[i])
|
||||
params['type'] = ','.join(passed_types)
|
||||
|
||||
if len(params['type']) == 0:
|
||||
params['type'] = 'track,album,artist,playlist'
|
||||
|
||||
# Clean search term
|
||||
search_term_list = []
|
||||
for split in splits:
|
||||
if split[0] == "-":
|
||||
break
|
||||
search_term_list.append(split)
|
||||
if not search_term_list:
|
||||
raise ValueError("Invalid query.")
|
||||
params["q"] = ' '.join(search_term_list)
|
||||
|
||||
resp = ZSpotify.invoke_url_with_params(SEARCH_URL, **params)
|
||||
|
||||
counter = 1
|
||||
dics = []
|
||||
|
||||
if TRACK in params['type'].split(','):
|
||||
tracks = resp[TRACKS][ITEMS]
|
||||
if len(tracks) > 0:
|
||||
print('### TRACKS ###')
|
||||
track_data = []
|
||||
for track in tracks:
|
||||
if track[EXPLICIT]:
|
||||
explicit = '[E]'
|
||||
else:
|
||||
explicit = ''
|
||||
|
||||
track_data.append([counter, f'{track[NAME]} {explicit}',
|
||||
','.join([artist[NAME] for artist in track[ARTISTS]])])
|
||||
dics.append({
|
||||
ID : track[ID],
|
||||
NAME : track[NAME],
|
||||
'type' : TRACK,
|
||||
})
|
||||
|
||||
counter += 1
|
||||
total_tracks = counter - 1
|
||||
print(tabulate(track_data, headers=['S.NO', 'Name', 'Artists'], tablefmt='pretty'))
|
||||
print('\n')
|
||||
del tracks
|
||||
del track_data
|
||||
else:
|
||||
total_tracks = 0
|
||||
|
||||
if ALBUM in params['type'].split(','):
|
||||
albums = resp[ALBUMS][ITEMS]
|
||||
if len(albums) > 0:
|
||||
print('### ALBUMS ###')
|
||||
album_data = []
|
||||
for album in albums:
|
||||
album_data.append([counter, album[NAME],
|
||||
','.join([artist[NAME] for artist in album[ARTISTS]])])
|
||||
dics.append({
|
||||
ID : album[ID],
|
||||
NAME : album[NAME],
|
||||
'type' : ALBUM,
|
||||
})
|
||||
|
||||
counter += 1
|
||||
total_albums = counter - total_tracks - 1
|
||||
print(tabulate(album_data, headers=['S.NO', 'Album', 'Artists'], tablefmt='pretty'))
|
||||
print('\n')
|
||||
del albums
|
||||
del album_data
|
||||
else:
|
||||
total_albums = 0
|
||||
|
||||
if ARTIST in params['type'].split(','):
|
||||
artists = resp[ARTISTS][ITEMS]
|
||||
if len(artists) > 0:
|
||||
print('### ARTISTS ###')
|
||||
artist_data = []
|
||||
for artist in artists:
|
||||
artist_data.append([counter, artist[NAME]])
|
||||
dics.append({
|
||||
ID : artist[ID],
|
||||
NAME : artist[NAME],
|
||||
'type' : ARTIST,
|
||||
})
|
||||
counter += 1
|
||||
total_artists = counter - total_tracks - total_albums - 1
|
||||
print(tabulate(artist_data, headers=['S.NO', 'Name'], tablefmt='pretty'))
|
||||
print('\n')
|
||||
del artists
|
||||
del artist_data
|
||||
else:
|
||||
total_artists = 0
|
||||
|
||||
if PLAYLIST in params['type'].split(','):
|
||||
playlists = resp[PLAYLISTS][ITEMS]
|
||||
if len(playlists) > 0:
|
||||
print('### PLAYLISTS ###')
|
||||
playlist_data = []
|
||||
for playlist in playlists:
|
||||
playlist_data.append([counter, playlist[NAME], playlist[OWNER][DISPLAY_NAME]])
|
||||
dics.append({
|
||||
ID : playlist[ID],
|
||||
NAME : playlist[NAME],
|
||||
'type' : PLAYLIST,
|
||||
})
|
||||
counter += 1
|
||||
total_playlists = counter - total_artists - total_tracks - total_albums - 1
|
||||
print(tabulate(playlist_data, headers=['S.NO', 'Name', 'Owner'], tablefmt='pretty'))
|
||||
print('\n')
|
||||
del playlists
|
||||
del playlist_data
|
||||
else:
|
||||
total_playlists = 0
|
||||
|
||||
if total_tracks + total_albums + total_artists + total_playlists == 0:
|
||||
print('NO RESULTS FOUND - EXITING...')
|
||||
else:
|
||||
selection = ''
|
||||
while len(selection) == 0:
|
||||
selection = str(input('SELECT ITEM(S) BY S.NO: '))
|
||||
inputs = split_input(selection)
|
||||
for pos in inputs:
|
||||
position = int(pos)
|
||||
for dic in dics:
|
||||
print_pos = dics.index(dic) + 1
|
||||
if print_pos == position:
|
||||
if dic['type'] == TRACK:
|
||||
download_track(dic[ID])
|
||||
elif dic['type'] == ALBUM:
|
||||
download_album(dic[ID])
|
||||
elif dic['type'] == ARTIST:
|
||||
download_artist_albums(dic[ID])
|
||||
else:
|
||||
download_playlist(dic)
|
||||
|
||||
@@ -1,97 +0,0 @@
|
||||
SANITIZE = ('\\', '/', ':', '*', '?', '\'', '<', '>', '"')
|
||||
|
||||
SAVED_TRACKS_URL = 'https://api.spotify.com/v1/me/tracks'
|
||||
|
||||
TRACKS_URL = 'https://api.spotify.com/v1/tracks'
|
||||
|
||||
TRACKNUMBER = 'tracknumber'
|
||||
|
||||
DISCNUMBER = 'discnumber'
|
||||
|
||||
YEAR = 'year'
|
||||
|
||||
ALBUM = 'album'
|
||||
|
||||
TRACKTITLE = 'tracktitle'
|
||||
|
||||
ARTIST = 'artist'
|
||||
|
||||
ARTISTS = 'artists'
|
||||
|
||||
ARTWORK = 'artwork'
|
||||
|
||||
TRACKS = 'tracks'
|
||||
|
||||
TRACK = 'track'
|
||||
|
||||
ITEMS = 'items'
|
||||
|
||||
NAME = 'name'
|
||||
|
||||
ID = 'id'
|
||||
|
||||
URL = 'url'
|
||||
|
||||
RELEASE_DATE = 'release_date'
|
||||
|
||||
IMAGES = 'images'
|
||||
|
||||
LIMIT = 'limit'
|
||||
|
||||
OFFSET = 'offset'
|
||||
|
||||
AUTHORIZATION = 'Authorization'
|
||||
|
||||
IS_PLAYABLE = 'is_playable'
|
||||
|
||||
TRACK_NUMBER = 'track_number'
|
||||
|
||||
DISC_NUMBER = 'disc_number'
|
||||
|
||||
SHOW = 'show'
|
||||
|
||||
ERROR = 'error'
|
||||
|
||||
EXPLICIT = 'explicit'
|
||||
|
||||
PLAYLIST = 'playlist'
|
||||
|
||||
PLAYLISTS = 'playlists'
|
||||
|
||||
OWNER = 'owner'
|
||||
|
||||
DISPLAY_NAME = 'display_name'
|
||||
|
||||
ALBUMS = 'albums'
|
||||
|
||||
TYPE = 'type'
|
||||
|
||||
PREMIUM = 'premium'
|
||||
|
||||
USER_READ_EMAIL = 'user-read-email'
|
||||
|
||||
PLAYLIST_READ_PRIVATE = 'playlist-read-private'
|
||||
|
||||
WINDOWS_SYSTEM = 'Windows'
|
||||
|
||||
CREDENTIALS_JSON = 'credentials.json'
|
||||
|
||||
CONFIG_FILE_PATH = '../zs_config.json'
|
||||
|
||||
ROOT_PATH = 'ROOT_PATH'
|
||||
|
||||
ROOT_PODCAST_PATH = 'ROOT_PODCAST_PATH'
|
||||
|
||||
SKIP_EXISTING_FILES = 'SKIP_EXISTING_FILES'
|
||||
|
||||
DOWNLOAD_FORMAT = 'DOWNLOAD_FORMAT'
|
||||
|
||||
FORCE_PREMIUM = 'FORCE_PREMIUM'
|
||||
|
||||
ANTI_BAN_WAIT_TIME = 'ANTI_BAN_WAIT_TIME'
|
||||
|
||||
OVERRIDE_AUTO_WAIT = 'OVERRIDE_AUTO_WAIT'
|
||||
|
||||
CHUNK_SIZE = 'CHUNK_SIZE'
|
||||
|
||||
SPLIT_ALBUM_DISCS = 'SPLIT_ALBUM_DISCS'
|
||||
@@ -1,87 +0,0 @@
|
||||
from tqdm import tqdm
|
||||
|
||||
from const import ITEMS, ID, TRACK, NAME
|
||||
from track import download_track
|
||||
from utils import sanitize_data
|
||||
from zspotify import ZSpotify
|
||||
|
||||
MY_PLAYLISTS_URL = 'https://api.spotify.com/v1/me/playlists'
|
||||
PLAYLISTS_URL = 'https://api.spotify.com/v1/playlists'
|
||||
|
||||
|
||||
def get_all_playlists():
|
||||
""" Returns list of users playlists """
|
||||
playlists = []
|
||||
limit = 50
|
||||
offset = 0
|
||||
|
||||
while True:
|
||||
resp = ZSpotify.invoke_url_with_params(MY_PLAYLISTS_URL, limit=limit, offset=offset)
|
||||
offset += limit
|
||||
playlists.extend(resp[ITEMS])
|
||||
if len(resp[ITEMS]) < limit:
|
||||
break
|
||||
|
||||
return playlists
|
||||
|
||||
|
||||
def get_playlist_songs(playlist_id):
|
||||
""" returns list of songs in a playlist """
|
||||
songs = []
|
||||
offset = 0
|
||||
limit = 100
|
||||
|
||||
while True:
|
||||
resp = ZSpotify.invoke_url_with_params(f'{PLAYLISTS_URL}/{playlist_id}/tracks', limit=limit, offset=offset)
|
||||
offset += limit
|
||||
songs.extend(resp[ITEMS])
|
||||
if len(resp[ITEMS]) < limit:
|
||||
break
|
||||
|
||||
return songs
|
||||
|
||||
|
||||
def get_playlist_info(playlist_id):
|
||||
""" Returns information scraped from playlist """
|
||||
resp = ZSpotify.invoke_url(f'{PLAYLISTS_URL}/{playlist_id}?fields=name,owner(display_name)&market=from_token')
|
||||
return resp['name'].strip(), resp['owner']['display_name'].strip()
|
||||
|
||||
|
||||
def download_playlist(playlist):
|
||||
"""Downloads all the songs from a playlist"""
|
||||
|
||||
playlist_songs = [song for song in get_playlist_songs(playlist[ID]) if song[TRACK][ID]]
|
||||
p_bar = tqdm(playlist_songs, unit='song', total=len(playlist_songs), unit_scale=True)
|
||||
for song in p_bar:
|
||||
download_track(song[TRACK][ID], sanitize_data(playlist[NAME].strip()) + '/',
|
||||
disable_progressbar=True)
|
||||
p_bar.set_description(song[TRACK][NAME])
|
||||
|
||||
|
||||
def download_from_user_playlist():
|
||||
""" Select which playlist(s) to download """
|
||||
playlists = get_all_playlists()
|
||||
|
||||
count = 1
|
||||
for playlist in playlists:
|
||||
print(str(count) + ': ' + playlist[NAME].strip())
|
||||
count += 1
|
||||
|
||||
print('\n> SELECT A PLAYLIST BY ID')
|
||||
print('> SELECT A RANGE BY ADDING A DASH BETWEEN BOTH ID\'s')
|
||||
print('> For example, typing 10 to get one playlist or 10-20 to get\nevery playlist from 10-20 (inclusive)\n')
|
||||
|
||||
playlist_choices = input('ID(s): ').split('-')
|
||||
|
||||
if len(playlist_choices) == 1:
|
||||
download_playlist(playlists, playlist_choices[0])
|
||||
else:
|
||||
start = int(playlist_choices[0])
|
||||
end = int(playlist_choices[1]) + 1
|
||||
|
||||
print(f'Downloading from {start} to {end}...')
|
||||
|
||||
for playlist_number in range(start, end):
|
||||
download_playlist(playlists, playlist_number)
|
||||
|
||||
print('\n**All playlists have been downloaded**\n')
|
||||
@@ -1,70 +0,0 @@
|
||||
import os
|
||||
from typing import Optional, Tuple
|
||||
|
||||
from librespot.audio.decoders import VorbisOnlyAudioQuality
|
||||
from librespot.metadata import EpisodeId
|
||||
from tqdm import tqdm
|
||||
|
||||
from const import NAME, ERROR, SHOW, ITEMS, ID, ROOT_PODCAST_PATH, CHUNK_SIZE
|
||||
from utils import sanitize_data, create_download_directory, MusicFormat
|
||||
from zspotify import ZSpotify
|
||||
|
||||
|
||||
EPISODE_INFO_URL = 'https://api.spotify.com/v1/episodes'
|
||||
SHOWS_URL = 'https://api.spotify.com/v1/shows'
|
||||
|
||||
|
||||
def get_episode_info(episode_id_str) -> Tuple[Optional[str], Optional[str]]:
|
||||
info = ZSpotify.invoke_url(f'{EPISODE_INFO_URL}/{episode_id_str}')
|
||||
if ERROR in info:
|
||||
return None, None
|
||||
return sanitize_data(info[SHOW][NAME]), sanitize_data(info[NAME])
|
||||
|
||||
|
||||
def get_show_episodes(show_id_str) -> list:
|
||||
episodes = []
|
||||
offset = 0
|
||||
limit = 50
|
||||
|
||||
while True:
|
||||
resp = ZSpotify.invoke_url_with_params(f'{SHOWS_URL}/{show_id_str}/episodes', limit=limit, offset=offset)
|
||||
offset += limit
|
||||
for episode in resp[ITEMS]:
|
||||
episodes.append(episode[ID])
|
||||
if len(resp[ITEMS]) < limit:
|
||||
break
|
||||
|
||||
return episodes
|
||||
|
||||
|
||||
def download_episode(episode_id) -> None:
|
||||
podcast_name, episode_name = get_episode_info(episode_id)
|
||||
|
||||
extra_paths = podcast_name + '/'
|
||||
|
||||
if podcast_name is None:
|
||||
print('### SKIPPING: (EPISODE NOT FOUND) ###')
|
||||
else:
|
||||
filename = podcast_name + ' - ' + episode_name
|
||||
|
||||
episode_id = EpisodeId.from_base62(episode_id)
|
||||
stream = ZSpotify.get_content_stream(episode_id, ZSpotify.DOWNLOAD_QUALITY)
|
||||
|
||||
download_directory = os.path.dirname(__file__) + ZSpotify.get_config(ROOT_PODCAST_PATH) + extra_paths
|
||||
create_download_directory(download_directory)
|
||||
|
||||
total_size = stream.input_stream.size
|
||||
with open(download_directory + filename + MusicFormat.OGG.value,
|
||||
'wb') as file, tqdm(
|
||||
desc=filename,
|
||||
total=total_size,
|
||||
unit='B',
|
||||
unit_scale=True,
|
||||
unit_divisor=1024
|
||||
) as bar:
|
||||
for _ in range(int(total_size / ZSpotify.get_config(CHUNK_SIZE)) + 1):
|
||||
bar.update(file.write(
|
||||
stream.input_stream.stream().read(ZSpotify.get_config(CHUNK_SIZE))))
|
||||
|
||||
# convert_audio_format(ROOT_PODCAST_PATH +
|
||||
# extra_paths + filename + '.ogg')
|
||||
@@ -1,128 +0,0 @@
|
||||
import os
|
||||
import time
|
||||
from typing import Any, Tuple, List
|
||||
|
||||
from librespot.audio.decoders import AudioQuality
|
||||
from librespot.metadata import TrackId
|
||||
from pydub import AudioSegment
|
||||
from tqdm import tqdm
|
||||
|
||||
from const import TRACKS, ALBUM, NAME, ITEMS, DISC_NUMBER, TRACK_NUMBER, IS_PLAYABLE, ARTISTS, IMAGES, URL, \
|
||||
RELEASE_DATE, ID, TRACKS_URL, SAVED_TRACKS_URL, SPLIT_ALBUM_DISCS, ROOT_PATH, DOWNLOAD_FORMAT, CHUNK_SIZE, \
|
||||
SKIP_EXISTING_FILES, ANTI_BAN_WAIT_TIME, OVERRIDE_AUTO_WAIT
|
||||
from utils import sanitize_data, set_audio_tags, set_music_thumbnail, create_download_directory, \
|
||||
MusicFormat
|
||||
from zspotify import ZSpotify
|
||||
|
||||
|
||||
def get_saved_tracks() -> list:
|
||||
""" Returns user's saved tracks """
|
||||
songs = []
|
||||
offset = 0
|
||||
limit = 50
|
||||
|
||||
while True:
|
||||
resp = ZSpotify.invoke_url_with_params(SAVED_TRACKS_URL, limit=limit, offset=offset)
|
||||
offset += limit
|
||||
songs.extend(resp[ITEMS])
|
||||
if len(resp[ITEMS]) < limit:
|
||||
break
|
||||
|
||||
return songs
|
||||
|
||||
|
||||
def get_song_info(song_id) -> Tuple[List[str], str, str, Any, Any, Any, Any, Any, Any]:
|
||||
""" Retrieves metadata for downloaded songs """
|
||||
info = ZSpotify.invoke_url(f'{TRACKS_URL}?ids={song_id}&market=from_token')
|
||||
|
||||
artists = []
|
||||
for data in info[TRACKS][0][ARTISTS]:
|
||||
artists.append(sanitize_data(data[NAME]))
|
||||
album_name = sanitize_data(info[TRACKS][0][ALBUM][NAME])
|
||||
name = sanitize_data(info[TRACKS][0][NAME])
|
||||
image_url = info[TRACKS][0][ALBUM][IMAGES][0][URL]
|
||||
release_year = info[TRACKS][0][ALBUM][RELEASE_DATE].split('-')[0]
|
||||
disc_number = info[TRACKS][0][DISC_NUMBER]
|
||||
track_number = info[TRACKS][0][TRACK_NUMBER]
|
||||
scraped_song_id = info[TRACKS][0][ID]
|
||||
is_playable = info[TRACKS][0][IS_PLAYABLE]
|
||||
|
||||
return artists, album_name, name, image_url, release_year, disc_number, track_number, scraped_song_id, is_playable
|
||||
|
||||
|
||||
# noinspection PyBroadException
|
||||
def download_track(track_id: str, extra_paths='', prefix=False, prefix_value='', disable_progressbar=False) -> None:
|
||||
""" Downloads raw song audio from Spotify """
|
||||
download_directory = os.path.join(os.path.dirname(__file__), ZSpotify.get_config(ROOT_PATH), extra_paths)
|
||||
try:
|
||||
(artists, album_name, name, image_url, release_year, disc_number,
|
||||
track_number, scraped_song_id, is_playable) = get_song_info(track_id)
|
||||
|
||||
song_name = artists[0] + ' - ' + name
|
||||
if prefix:
|
||||
song_name = f'{prefix_value.zfill(2)} - {song_name}' if prefix_value.isdigit(
|
||||
) else f'{prefix_value} - {song_name}'
|
||||
|
||||
if ZSpotify.get_config(SPLIT_ALBUM_DISCS):
|
||||
filename = os.path.join(download_directory, f'Disc {disc_number}',
|
||||
f'{song_name}.{ZSpotify.get_config(DOWNLOAD_FORMAT)}')
|
||||
else:
|
||||
filename = os.path.join(download_directory,
|
||||
f'{song_name}.{ZSpotify.get_config(DOWNLOAD_FORMAT)}')
|
||||
except Exception:
|
||||
print('### SKIPPING SONG - FAILED TO QUERY METADATA ###')
|
||||
else:
|
||||
try:
|
||||
if not is_playable:
|
||||
print('\n### SKIPPING:', song_name,
|
||||
'(SONG IS UNAVAILABLE) ###')
|
||||
else:
|
||||
if os.path.isfile(filename) and os.path.getsize(filename) and ZSpotify.get_config(SKIP_EXISTING_FILES):
|
||||
print('\n### SKIPPING:', song_name,
|
||||
'(SONG ALREADY EXISTS) ###')
|
||||
else:
|
||||
if track_id != scraped_song_id:
|
||||
track_id = scraped_song_id
|
||||
track_id = TrackId.from_base62(track_id)
|
||||
stream = ZSpotify.get_content_stream(track_id, ZSpotify.DOWNLOAD_QUALITY)
|
||||
create_download_directory(download_directory)
|
||||
total_size = stream.input_stream.size
|
||||
|
||||
with open(filename, 'wb') as file, tqdm(
|
||||
desc=song_name,
|
||||
total=total_size,
|
||||
unit='B',
|
||||
unit_scale=True,
|
||||
unit_divisor=1024,
|
||||
disable=disable_progressbar
|
||||
) as p_bar:
|
||||
for _ in range(int(total_size / ZSpotify.get_config(CHUNK_SIZE)) + 1):
|
||||
p_bar.update(file.write(
|
||||
stream.input_stream.stream().read(ZSpotify.get_config(CHUNK_SIZE))))
|
||||
|
||||
if ZSpotify.get_config(DOWNLOAD_FORMAT) == 'mp3':
|
||||
convert_audio_format(filename)
|
||||
set_audio_tags(filename, artists, name, album_name,
|
||||
release_year, disc_number, track_number)
|
||||
set_music_thumbnail(filename, image_url)
|
||||
|
||||
if not ZSpotify.get_config(OVERRIDE_AUTO_WAIT):
|
||||
time.sleep(ZSpotify.get_config(ANTI_BAN_WAIT_TIME))
|
||||
except Exception as e:
|
||||
print('### SKIPPING:', song_name,
|
||||
'(GENERAL DOWNLOAD ERROR) ###')
|
||||
print(e)
|
||||
if os.path.exists(filename):
|
||||
os.remove(filename)
|
||||
|
||||
|
||||
def convert_audio_format(filename) -> None:
|
||||
""" Converts raw audio into playable mp3 """
|
||||
# print('### CONVERTING TO ' + MUSIC_FORMAT.upper() + ' ###')
|
||||
raw_audio = AudioSegment.from_file(filename, format=MusicFormat.OGG.value,
|
||||
frame_rate=44100, channels=2, sample_width=2)
|
||||
if ZSpotify.DOWNLOAD_QUALITY == AudioQuality.VERY_HIGH:
|
||||
bitrate = '320k'
|
||||
else:
|
||||
bitrate = '160k'
|
||||
raw_audio.export(filename, format=ZSpotify.get_config(DOWNLOAD_FORMAT), bitrate=bitrate)
|
||||
@@ -1,181 +0,0 @@
|
||||
import os
|
||||
import platform
|
||||
import re
|
||||
import time
|
||||
from enum import Enum
|
||||
from typing import List, Tuple
|
||||
|
||||
import music_tag
|
||||
import requests
|
||||
|
||||
from const import SANITIZE, ARTIST, TRACKTITLE, ALBUM, YEAR, DISCNUMBER, TRACKNUMBER, ARTWORK, \
|
||||
WINDOWS_SYSTEM
|
||||
|
||||
|
||||
class MusicFormat(str, Enum):
|
||||
MP3 = 'mp3',
|
||||
OGG = 'ogg',
|
||||
|
||||
|
||||
def create_download_directory(download_path: str) -> None:
|
||||
os.makedirs(download_path, exist_ok=True)
|
||||
|
||||
|
||||
def wait(seconds: int = 3) -> None:
|
||||
""" Pause for a set number of seconds """
|
||||
for second in range(seconds)[::-1]:
|
||||
print(f'\rWait for {second + 1} second(s)...', end='')
|
||||
time.sleep(1)
|
||||
|
||||
|
||||
def split_input(selection) -> List[str]:
|
||||
""" Returns a list of inputted strings """
|
||||
inputs = []
|
||||
if '-' in selection:
|
||||
for number in range(int(selection.split('-')[0]), int(selection.split('-')[1]) + 1):
|
||||
inputs.append(number)
|
||||
else:
|
||||
selections = selection.split(',')
|
||||
for i in selections:
|
||||
inputs.append(i.strip())
|
||||
return inputs
|
||||
|
||||
|
||||
def splash() -> None:
|
||||
""" Displays splash screen """
|
||||
print("""
|
||||
███████ ███████ ██████ ██████ ████████ ██ ███████ ██ ██
|
||||
███ ██ ██ ██ ██ ██ ██ ██ ██ ██ ██
|
||||
███ ███████ ██████ ██ ██ ██ ██ █████ ████
|
||||
███ ██ ██ ██ ██ ██ ██ ██ ██
|
||||
███████ ███████ ██ ██████ ██ ██ ██ ██
|
||||
""")
|
||||
|
||||
|
||||
def clear() -> None:
|
||||
""" Clear the console window """
|
||||
if platform.system() == WINDOWS_SYSTEM:
|
||||
os.system('cls')
|
||||
else:
|
||||
os.system('clear')
|
||||
|
||||
|
||||
def sanitize_data(value) -> str:
|
||||
""" Returns given string with problematic removed """
|
||||
for pattern in SANITIZE:
|
||||
value = value.replace(pattern, '')
|
||||
return value.replace('|', '-')
|
||||
|
||||
|
||||
def set_audio_tags(filename, artists, name, album_name, release_year, disc_number, track_number) -> None:
|
||||
""" sets music_tag metadata """
|
||||
tags = music_tag.load_file(filename)
|
||||
tags[ARTIST] = conv_artist_format(artists)
|
||||
tags[TRACKTITLE] = name
|
||||
tags[ALBUM] = album_name
|
||||
tags[YEAR] = release_year
|
||||
tags[DISCNUMBER] = disc_number
|
||||
tags[TRACKNUMBER] = track_number
|
||||
tags.save()
|
||||
|
||||
|
||||
def conv_artist_format(artists) -> str:
|
||||
""" Returns converted artist format """
|
||||
return ', '.join(artists)
|
||||
|
||||
|
||||
def set_music_thumbnail(filename, image_url) -> None:
|
||||
""" Downloads cover artwork """
|
||||
img = requests.get(image_url).content
|
||||
tags = music_tag.load_file(filename)
|
||||
tags[ARTWORK] = img
|
||||
tags.save()
|
||||
|
||||
|
||||
def regex_input_for_urls(search_input) -> Tuple[str, str, str, str, str, str]:
|
||||
""" Since many kinds of search may be passed at the command line, process them all here. """
|
||||
track_uri_search = re.search(
|
||||
r'^spotify:track:(?P<TrackID>[0-9a-zA-Z]{22})$', search_input)
|
||||
track_url_search = re.search(
|
||||
r'^(https?://)?open\.spotify\.com/track/(?P<TrackID>[0-9a-zA-Z]{22})(\?si=.+?)?$',
|
||||
search_input,
|
||||
)
|
||||
|
||||
album_uri_search = re.search(
|
||||
r'^spotify:album:(?P<AlbumID>[0-9a-zA-Z]{22})$', search_input)
|
||||
album_url_search = re.search(
|
||||
r'^(https?://)?open\.spotify\.com/album/(?P<AlbumID>[0-9a-zA-Z]{22})(\?si=.+?)?$',
|
||||
search_input,
|
||||
)
|
||||
|
||||
playlist_uri_search = re.search(
|
||||
r'^spotify:playlist:(?P<PlaylistID>[0-9a-zA-Z]{22})$', search_input)
|
||||
playlist_url_search = re.search(
|
||||
r'^(https?://)?open\.spotify\.com/playlist/(?P<PlaylistID>[0-9a-zA-Z]{22})(\?si=.+?)?$',
|
||||
search_input,
|
||||
)
|
||||
|
||||
episode_uri_search = re.search(
|
||||
r'^spotify:episode:(?P<EpisodeID>[0-9a-zA-Z]{22})$', search_input)
|
||||
episode_url_search = re.search(
|
||||
r'^(https?://)?open\.spotify\.com/episode/(?P<EpisodeID>[0-9a-zA-Z]{22})(\?si=.+?)?$',
|
||||
search_input,
|
||||
)
|
||||
|
||||
show_uri_search = re.search(
|
||||
r'^spotify:show:(?P<ShowID>[0-9a-zA-Z]{22})$', search_input)
|
||||
show_url_search = re.search(
|
||||
r'^(https?://)?open\.spotify\.com/show/(?P<ShowID>[0-9a-zA-Z]{22})(\?si=.+?)?$',
|
||||
search_input,
|
||||
)
|
||||
|
||||
artist_uri_search = re.search(
|
||||
r'^spotify:artist:(?P<ArtistID>[0-9a-zA-Z]{22})$', search_input)
|
||||
artist_url_search = re.search(
|
||||
r'^(https?://)?open\.spotify\.com/artist/(?P<ArtistID>[0-9a-zA-Z]{22})(\?si=.+?)?$',
|
||||
search_input,
|
||||
)
|
||||
|
||||
if track_uri_search is not None or track_url_search is not None:
|
||||
track_id_str = (track_uri_search
|
||||
if track_uri_search is not None else
|
||||
track_url_search).group('TrackID')
|
||||
else:
|
||||
track_id_str = None
|
||||
|
||||
if album_uri_search is not None or album_url_search is not None:
|
||||
album_id_str = (album_uri_search
|
||||
if album_uri_search is not None else
|
||||
album_url_search).group('AlbumID')
|
||||
else:
|
||||
album_id_str = None
|
||||
|
||||
if playlist_uri_search is not None or playlist_url_search is not None:
|
||||
playlist_id_str = (playlist_uri_search
|
||||
if playlist_uri_search is not None else
|
||||
playlist_url_search).group('PlaylistID')
|
||||
else:
|
||||
playlist_id_str = None
|
||||
|
||||
if episode_uri_search is not None or episode_url_search is not None:
|
||||
episode_id_str = (episode_uri_search
|
||||
if episode_uri_search is not None else
|
||||
episode_url_search).group('EpisodeID')
|
||||
else:
|
||||
episode_id_str = None
|
||||
|
||||
if show_uri_search is not None or show_url_search is not None:
|
||||
show_id_str = (show_uri_search
|
||||
if show_uri_search is not None else
|
||||
show_url_search).group('ShowID')
|
||||
else:
|
||||
show_id_str = None
|
||||
|
||||
if artist_uri_search is not None or artist_url_search is not None:
|
||||
artist_id_str = (artist_uri_search
|
||||
if artist_uri_search is not None else
|
||||
artist_url_search).group('ArtistID')
|
||||
else:
|
||||
artist_id_str = None
|
||||
|
||||
return track_id_str, album_id_str, playlist_id_str, episode_id_str, show_id_str, artist_id_str
|
||||
@@ -1,96 +0,0 @@
|
||||
#! /usr/bin/env python3
|
||||
|
||||
"""
|
||||
ZSpotify
|
||||
It's like youtube-dl, but for Spotify.
|
||||
|
||||
(Made by Deathmonger/Footsiefat - @doomslayer117:matrix.org)
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import os.path
|
||||
from getpass import getpass
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
from librespot.audio.decoders import VorbisOnlyAudioQuality
|
||||
from librespot.core import Session
|
||||
|
||||
from const import CREDENTIALS_JSON, TYPE, \
|
||||
PREMIUM, USER_READ_EMAIL, AUTHORIZATION, OFFSET, LIMIT, CONFIG_FILE_PATH, FORCE_PREMIUM, \
|
||||
PLAYLIST_READ_PRIVATE
|
||||
from utils import MusicFormat
|
||||
|
||||
|
||||
class ZSpotify:
|
||||
SESSION: Session = None
|
||||
DOWNLOAD_QUALITY = None
|
||||
CONFIG = {}
|
||||
|
||||
def __init__(self):
|
||||
ZSpotify.load_config()
|
||||
ZSpotify.login()
|
||||
|
||||
@classmethod
|
||||
def login(cls):
|
||||
""" Authenticates with Spotify and saves credentials to a file """
|
||||
|
||||
if os.path.isfile(CREDENTIALS_JSON):
|
||||
try:
|
||||
cls.SESSION = Session.Builder().stored_file().create()
|
||||
return
|
||||
except RuntimeError:
|
||||
pass
|
||||
while True:
|
||||
user_name = ''
|
||||
while len(user_name) == 0:
|
||||
user_name = input('Username: ')
|
||||
password = getpass()
|
||||
try:
|
||||
cls.SESSION = Session.Builder().user_pass(user_name, password).create()
|
||||
return
|
||||
except RuntimeError:
|
||||
pass
|
||||
|
||||
@classmethod
|
||||
def load_config(cls) -> None:
|
||||
app_dir = os.path.dirname(__file__)
|
||||
with open(os.path.join(app_dir, CONFIG_FILE_PATH), encoding='utf-8') as config_file:
|
||||
cls.CONFIG = json.load(config_file)
|
||||
|
||||
@classmethod
|
||||
def get_config(cls, key) -> Any:
|
||||
return cls.CONFIG.get(key)
|
||||
|
||||
@classmethod
|
||||
def get_content_stream(cls, content_id, quality):
|
||||
return cls.SESSION.content_feeder().load(content_id, VorbisOnlyAudioQuality(quality), False, None)
|
||||
|
||||
@classmethod
|
||||
def __get_auth_token(cls):
|
||||
return cls.SESSION.tokens().get_token(USER_READ_EMAIL, PLAYLIST_READ_PRIVATE).access_token
|
||||
|
||||
@classmethod
|
||||
def get_auth_header(cls):
|
||||
return {
|
||||
AUTHORIZATION: f'Bearer {cls.__get_auth_token()}'}
|
||||
|
||||
@classmethod
|
||||
def get_auth_header_and_params(cls, limit, offset):
|
||||
return {AUTHORIZATION: f'Bearer {cls.__get_auth_token()}'}, {LIMIT: limit, OFFSET: offset}
|
||||
|
||||
@classmethod
|
||||
def invoke_url_with_params(cls, url, limit, offset, **kwargs):
|
||||
headers, params = cls.get_auth_header_and_params(limit=limit, offset=offset)
|
||||
params.update(kwargs)
|
||||
return requests.get(url, headers=headers, params=params).json()
|
||||
|
||||
@classmethod
|
||||
def invoke_url(cls, url):
|
||||
headers = cls.get_auth_header()
|
||||
return requests.get(url, headers=headers).json()
|
||||
|
||||
@classmethod
|
||||
def check_premium(cls) -> bool:
|
||||
""" If user has spotify premium return true """
|
||||
return (cls.SESSION.get_user_attribute(TYPE) == PREMIUM) or cls.get_config(FORCE_PREMIUM)
|
||||
Reference in New Issue
Block a user