Clone/zspotify
1
0
Fork 0
mirror of https://github.com/THIS-IS-NOT-A-BACKUP/zspotify.git synced 2025-12-27 19:10:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

updated folder name

Browse Source
This commit is contained in:
Raju komati
2021-10-25 09:22:53 +05:30
parent 441da8eef0
commit dab823d09e
8 changed files with 37 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
zspotify/album.py Normal file
View File

@@ -0,0 +1,57 @@
"""It's provides functions for downloading the albums"""
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
return [resp[ITEMS][i][ID] for i in range(len(resp[ITEMS]))]
def download_album(album):
""" Downloads songs from an album """
artist, album_name = get_album_name(album)
tracks = get_album_tracks(album)
for album_number, 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(album_number), 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)

167
zspotify/app.py Normal file
View File

@@ -0,0 +1,167 @@
"""Entrypoint of ZSpotify app. It provides functions for searching"""
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, ARTISTS, ITEMS, TRACKS, EXPLICIT, ALBUMS, OWNER, \
PLAYLISTS, DISPLAY_NAME, LIMIT, OFFSET, TYPE, S_NO, ALBUM
from playlist import download_from_user_playlist, download_playlist, \
download_playlist_with_id
from podcast import download_episode, get_show_episodes
from track import download_track, get_saved_tracks
from utils import 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
while True:
if len(sys.argv) > 1:
process_args_input()
else:
search_text = ''
while len(search_text) == 0:
search_text = input('Enter search or URL: ')
process_url_input(search_text, call_search=True)
# wait()
def process_args_input():
"""
process the sys args
"""
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:
process_url_input(sys.argv[1])
def process_url_input(url, call_search=False):
"""
process the input and calls appropriate download method
@param url: input url
@param call_search: boolean variable to notify calling search method
"""
track_id, album_id, playlist_id, episode_id, show_id, artist_id = regex_input_for_urls(url)
if track_id:
download_track(track_id)
elif artist_id:
download_artist_albums(artist_id)
elif album_id:
download_album(album_id)
elif playlist_id:
download_playlist_with_id(playlist_id)
elif episode_id:
download_episode(episode_id)
elif show_id:
for episode in get_show_episodes(show_id):
download_episode(episode)
elif call_search:
search(url)
def search(search_term):
""" Searches Spotify's API for relevant data """
params = {LIMIT: '10', OFFSET: '0', 'q': search_term, TYPE: 'track,album,artist,playlist'}
resp = ZSpotify.invoke_url_with_params(SEARCH_URL, **params)
total_tracks = total_albums = total_artists = 0
counter = 1
tracks = resp[TRACKS][ITEMS]
if len(tracks) > 0:
print('### TRACKS ###')
track_data = []
for track in tracks:
explicit = '[E]' if track[EXPLICIT] else ''
track_data.append([counter, f'{track[NAME]} {explicit}',
','.join([artist[NAME] for artist in track[ARTISTS]])])
counter += 1
total_tracks = counter - 1
print(tabulate(track_data, headers=[S_NO, NAME.title(), ARTISTS.title()],
tablefmt='pretty'))
print('\n')
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]])])
counter += 1
total_albums = counter - total_tracks - 1
print(tabulate(album_data, headers=[S_NO, ALBUM.title(), ARTISTS.title()],
tablefmt='pretty'))
print('\n')
artists = resp[ARTISTS][ITEMS]
if len(artists) > 0:
print('### ARTISTS ###')
artist_data = []
for artist in artists:
artist_data.append([counter, artist[NAME]])
counter += 1
total_artists = counter - total_tracks - total_albums - 1
print(tabulate(artist_data, headers=[S_NO, NAME.title()], tablefmt='pretty'))
print('\n')
playlists = resp[PLAYLISTS][ITEMS]
print('### PLAYLISTS ###')
playlist_data = []
for playlist in playlists:
playlist_data.append([counter, playlist[NAME], playlist[OWNER][DISPLAY_NAME]])
counter += 1
print(tabulate(playlist_data, headers=[S_NO, NAME.title(), OWNER.title()], tablefmt='pretty'))
print('\n')
perform_action(tracks, albums, playlists, artists, total_tracks, total_albums, total_artists)
def perform_action(tracks: list, albums: list, playlists: list, artists: list,
total_tracks: int, total_albums: int, total_artists: int):
"""
process and downloads the user selection
"""
if len(tracks) + len(albums) + len(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)
if position <= total_tracks:
track_id = tracks[position - 1][ID]
download_track(track_id)
elif position <= total_albums + total_tracks:
download_album(albums[position - total_tracks - 1][ID])
elif position <= total_artists + total_tracks + total_albums:
download_artist_albums(artists[position - total_tracks - total_albums - 1][ID])
else:
download_playlist(playlists, position - total_tracks - total_albums - total_artists)
if __name__ == '__main__':
client()

112
zspotify/const.py Normal file
View File

@@ -0,0 +1,112 @@
""" provides commonly used string across different modules"""
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'
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'
DURATION_MS = 'duration_ms'
ARTIST_ID = 'ArtistID'
SHOW_ID = 'ShowID'
EPISODE_ID = 'EpisodeID'
PLAYLIST_ID = 'PlaylistID'
ALBUM_ID = 'AlbumID'
TRACK_ID = 'TrackID'
S_NO = 'S.NO'

92
zspotify/playlist.py Normal file
View File

@@ -0,0 +1,92 @@
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_with_id(playlist_id):
name, _ = get_playlist_info(playlist_id)
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(name.strip()) + '/', disable_progressbar=True,
create_m3u_file=True)
p_bar.set_description(song[TRACK][NAME])
def download_playlist(playlists, playlist_number):
"""Downloads all the songs from a playlist"""
download_playlist_with_id(playlists[int(playlist_number) - 1][ID])
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')

68
zspotify/podcast.py Normal file
View File

@@ -0,0 +1,68 @@
import os
from typing import Optional
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) -> tuple[Optional[str], Optional[str]]:
info = ZSpotify.invoke_url(f'{EPISODE_INFO_URL}/{episode_id}')
if ERROR in info:
return None, None
return sanitize_data(info[SHOW][NAME]), sanitize_data(info[NAME])
def get_show_episodes(show_id) -> list:
episodes = []
offset = 0
limit = 50
while True:
resp = ZSpotify.invoke_url_with_params(f'{SHOWS_URL}/{show_id}/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 not podcast_name:
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')

171
zspotify/track.py Normal file
View File

@@ -0,0 +1,171 @@
import math
import os
import time
from typing import Any
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, DURATION_MS
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, 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(data[NAME])
album_name = info[TRACKS][0][ALBUM][NAME]
name = 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]
duration = math.ceil(info[TRACKS][0][DURATION_MS] / 1000)
return (artists, album_name, name, image_url, release_year, disc_number, track_number, scraped_song_id, is_playable,
duration)
# noinspection PyBroadException
def download_track(track_id: str, extra_paths: str = '', prefix: bool = False, prefix_value='',
disable_progressbar: bool = False,
create_m3u_file: bool = False) -> None:
""" Downloads raw song audio from Spotify """
try:
(artists, album_name, name, image_url, release_year, disc_number,
track_number, scraped_song_id, is_playable, duration) = get_song_info(track_id)
song_name, filename, m3u_filename = process_track_metadata(artists, name, disc_number, extra_paths, prefix,
prefix_value, create_m3u_file)
except Exception:
print('### SKIPPING SONG - FAILED TO QUERY METADATA ###')
else:
try:
track_info = (
artists, album_name, name, image_url, release_year, disc_number, track_number, scraped_song_id,
is_playable, duration)
playlist_info = (create_m3u_file, m3u_filename)
creat_track(track_id, extra_paths, song_name, filename, disable_progressbar, track_info, playlist_info)
except Exception:
print('### SKIPPING:', song_name,
'(GENERAL DOWNLOAD ERROR) ###')
if os.path.exists(filename):
os.remove(filename)
def process_track_metadata(artists: list, name: str, disc_number: Any, extra_paths: str, prefix: bool,
prefix_value: str,
create_m3u_file: bool):
m3u_filename = None
song_name = sanitize_data(artists[0]) + ' - ' + sanitize_data(name)
if prefix:
song_name = f'{prefix_value.zfill(2)} - {song_name}' if prefix_value.isdigit(
) else f'{prefix_value} - {song_name}'
if create_m3u_file:
m3u_filename = f'{os.path.join(ZSpotify.get_config(ROOT_PATH), extra_paths, extra_paths)[:-1]}.m3u'
if ZSpotify.get_config(SPLIT_ALBUM_DISCS):
filename = os.path.join(ZSpotify.get_config(ROOT_PATH), extra_paths, 'Disc ' + str(
disc_number) + '/' + song_name + '.' + ZSpotify.get_config(DOWNLOAD_FORMAT))
else:
filename = os.path.join(ZSpotify.get_config(ROOT_PATH), extra_paths,
song_name + '.' + ZSpotify.get_config(DOWNLOAD_FORMAT))
return song_name, filename, m3u_filename
def creat_track(track_id, extra_paths: str, song_name: str, filename: str, disable_progressbar: bool, track_info: tuple,
playlist_info: tuple):
(artists, album_name, name, image_url, release_year, disc_number,
track_number, scraped_song_id, is_playable, duration) = track_info
create_m3u, m3u_filename = playlist_info
if not is_playable:
print(f'\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):
playlist_data = f'#EXTINF:{duration}, {artists[0]} - {name}\n{os.path.abspath(filename)}'
create_playlist_file(create_m3u, playlist_data, m3u_filename)
print(f'\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(ZSpotify.get_config(ROOT_PATH) + extra_paths)
total_size = stream.input_stream.size
write_stream_to_file(stream, filename, song_name, total_size, disable_progressbar, track_info,
playlist_info)
def write_stream_to_file(stream, filename: str, song_name: str, total_size: Any, disable_progressbar: bool,
track_info: tuple, playlist_info: tuple):
(artists, album_name, name, image_url, release_year, disc_number,
track_number, scraped_song_id, is_playable, duration) = track_info
create_m3u, m3u_filename = playlist_info
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))))
playlist_data = f'#EXTINF:{duration}, {artists[0]} - {name}\n{os.path.abspath(filename)}'
if ZSpotify.get_config(DOWNLOAD_FORMAT) == MusicFormat.MP3.value:
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))
create_playlist_file(create_m3u, playlist_data, m3u_filename)
def create_playlist_file(create_m3u: bool, playlist_data: str, m3u_filename: str):
if create_m3u and m3u_filename and playlist_data:
with open(m3u_filename, 'a+') as pfile:
if os.path.getsize(m3u_filename) == 0:
pfile.write('#EXTM3U\n')
pfile.write(playlist_data + '\n')
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)

158
zspotify/utils.py Normal file
View File

@@ -0,0 +1,158 @@
import os
import platform
import re
import time
from enum import Enum
from typing import Match
import music_tag
import requests
from const import SANITIZE, ARTIST, TRACKTITLE, ALBUM, YEAR, DISCNUMBER, TRACKNUMBER, ARTWORK, \
WINDOWS_SYSTEM, TRACK_ID, ALBUM_ID, PLAYLIST_ID, EPISODE_ID, SHOW_ID, ARTIST_ID
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,
)
return (
extract_info_from_regex_response(TRACK_ID, track_uri_search, track_url_search),
extract_info_from_regex_response(ALBUM_ID, album_uri_search, album_url_search),
extract_info_from_regex_response(PLAYLIST_ID, playlist_uri_search, playlist_url_search),
extract_info_from_regex_response(EPISODE_ID, episode_uri_search, episode_url_search),
extract_info_from_regex_response(SHOW_ID, show_uri_search, show_url_search),
extract_info_from_regex_response(ARTIST_ID, artist_uri_search, artist_url_search)
)
def extract_info_from_regex_response(key, uri_data: Match[str], url_data: Match[str]):
if uri_data or url_data:
return (uri_data if uri_data else url_data).group(key)
else:
return None

96
zspotify/zspotify.py Normal file
View File

@@ -0,0 +1,96 @@
#! /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)