Clone/zspotify
1
0
Fork 0
mirror of https://github.com/THIS-IS-NOT-A-BACKUP/zspotify.git synced 2024-11-29 19:24:34 +01:00
Code Issues Packages Projects Releases Wiki Activity

added m3u playlist file creation

Browse Source 
fixed sonar issues
This commit is contained in:
Raju komati 2021-10-24 18:35:40 +05:30
parent ad4c730578
commit beff9627a8
No known key found for this signature in database
GPG Key ID: D0390D7534A9E462
6 changed files with 193 additions and 176 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

110
src/app.py
View File

@ -6,10 +6,11 @@ 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
from playlist import get_playlist_songs, get_playlist_info, download_from_user_playlist, download_playlist
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 sanitize_data, splash, split_input, regex_input_for_urls
from utils import splash, split_input, regex_input_for_urls
from zspotify import ZSpotify
SEARCH_URL = 'https://api.spotify.com/v1/search'
@ -29,89 +30,68 @@ def client() -> None:
while True:
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)
process_args_input()
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)
process_url_input(search_text, call_search=True)
# wait()
def process_args_input():
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):
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:
if track[EXPLICIT]:
explicit = '[E]'
else:
explicit = ''
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', 'Artists'], tablefmt='pretty'))
print('\n')
else:
total_tracks = 0
albums = resp[ALBUMS][ITEMS]
if len(albums) > 0:
@ -123,8 +103,6 @@ def search(search_term):
total_albums = counter - total_tracks - 1
print(tabulate(album_data, headers=['S.NO', 'Album', 'Artists'], tablefmt='pretty'))
print('\n')
else:
total_albums = 0
artists = resp[ARTISTS][ITEMS]
if len(artists) > 0:
@ -136,8 +114,6 @@ def search(search_term):
total_artists = counter - total_tracks - total_albums - 1
print(tabulate(artist_data, headers=['S.NO', 'Name'], tablefmt='pretty'))
print('\n')
else:
total_artists = 0
playlists = resp[PLAYLISTS][ITEMS]
print('### PLAYLISTS ###')
@ -147,7 +123,11 @@ def search(search_term):
counter += 1
print(tabulate(playlist_data, headers=['S.NO', 'Name', 'Owner'], 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):
if len(tracks) + len(albums) + len(playlists) == 0:
print('NO RESULTS FOUND - EXITING...')
else:

14
src/const.py
View File

@ -93,3 +93,17 @@ 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'

15
src/playlist.py
View File

@ -47,17 +47,20 @@ def get_playlist_info(playlist_id):
return resp['name'].strip(), resp['owner']['display_name'].strip()
def download_playlist(playlists, playlist_number):
"""Downloads all the songs from a playlist"""
playlist_songs = [song for song in get_playlist_songs(playlists[int(playlist_number) - 1][ID]) if song[TRACK][ID]]
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(playlists[int(playlist_number) - 1][NAME].strip()) + '/',
disable_progressbar=True)
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()

22
src/podcast.py
View File

@ -1,6 +1,5 @@
from typing import Optional
from librespot.audio.decoders import VorbisOnlyAudioQuality
from librespot.metadata import EpisodeId
from tqdm import tqdm
@ -8,25 +7,24 @@ 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}')
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_str) -> list:
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_str}/episodes', limit=limit, offset=offset)
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])
@ -41,7 +39,7 @@ def download_episode(episode_id) -> None:
extra_paths = podcast_name + '/'
if podcast_name is None:
if not podcast_name:
print('### SKIPPING: (EPISODE NOT FOUND) ###')
else:
filename = podcast_name + ' - ' + episode_name
@ -54,11 +52,11 @@ def download_episode(episode_id) -> None:
total_size = stream.input_stream.size
with open(ZSpotify.get_config(ROOT_PODCAST_PATH) + extra_paths + filename + MusicFormat.OGG.value,
'wb') as file, tqdm(
desc=filename,
total=total_size,
unit='B',
unit_scale=True,
unit_divisor=1024
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(

152
src/track.py
View File

@ -1,3 +1,4 @@
import math
import os
import time
from typing import Any
@ -9,7 +10,7 @@ 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
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
@ -31,7 +32,7 @@ def get_saved_tracks() -> list:
return songs
def get_song_info(song_id) -> tuple[list[str], str, str, Any, Any, Any, Any, Any, Any]:
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')
@ -46,75 +47,118 @@ def get_song_info(song_id) -> tuple[list[str], str, str, Any, Any, Any, Any, Any
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
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='', prefix=False, prefix_value='', disable_progressbar=False) -> None:
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) = get_song_info(track_id)
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 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))
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:
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(ZSpotify.get_config(ROOT_PATH) + extra_paths)
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:
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) ###')
print(e)
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() + ' ###')

56
src/utils.py
View File

@ -3,18 +3,19 @@ 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
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)
@ -135,46 +136,23 @@ def regex_input_for_urls(search_input) -> tuple[str, str, str, str, str, str]:
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
return (
extract_info_from_regex_response(TRACK_ID, track_uri_search, track_url_search),
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
extract_info_from_regex_response(ALBUM_ID, album_uri_search, album_url_search),
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
extract_info_from_regex_response(PLAYLIST_ID, playlist_uri_search, playlist_url_search),
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
extract_info_from_regex_response(EPISODE_ID, episode_uri_search, episode_url_search),
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
extract_info_from_regex_response(SHOW_ID, show_uri_search, show_url_search),
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
extract_info_from_regex_response(ARTIST_ID, artist_uri_search, artist_url_search)
)
return track_id_str, album_id_str, playlist_id_str, episode_id_str, show_id_str, artist_id_str
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