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

Merge branch 'main' into Playlist-id-fix

Browse Source
This commit is contained in:
Logykk
2021-10-30 17:20:13 +13:00
committed by GitHub
parent 01c8dd49fd 265834746f
commit 9eabd55104
10 changed files with 140 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
zspotify/__main__.py
View File

@@ -1,4 +1,35 @@
import argparse
from app import client
if __name__ == '__main__':
client()
parser = argparse.ArgumentParser(prog='zspotify',
description='A Spotify downloader needing only a python interpreter and ffmpeg.')
parser.add_argument('-ns', '--no-splash',
action='store_true',
help='Suppress the splash screen when loading.')
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument('urls',
type=str,
# action='extend',
default='',
nargs='*',
help='Downloads the track, album, playlist, podcast episode, or all albums by an artist from a url. Can take multiple urls.')
group.add_argument('-ls', '--liked-songs',
dest='liked_songs',
action='store_true',
help='Downloads all the liked songs from your account.')
group.add_argument('-p', '--playlist',
action='store_true',
help='Downloads a saved playlist from your account.')
group.add_argument('-s', '--search',
dest='search_spotify',
action='store_true',
help='Loads search prompt to find then download a specific track, album or playlist')
parser.set_defaults(func=client)
args = parser.parse_args()
args.func(args)

44
zspotify/app.py
View File

@@ -1,5 +1,3 @@
import sys
from librespot.audio.decoders import AudioQuality
from tabulate import tabulate
@@ -15,32 +13,26 @@ from zspotify import ZSpotify
SEARCH_URL = 'https://api.spotify.com/v1/search'
def client() -> None:
def client(args) -> None:
""" Connects to spotify to perform query's and get songs to download """
ZSpotify()
splash()
if not args.no_splash:
splash()
if ZSpotify.check_premium():
print('[ DETECTED PREMIUM ACCOUNT - USING VERY_HIGH QUALITY ]\n\n')
if not args.no_splash:
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')
if not args.no_splash:
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 EXIST ON SPOTIFY ANYMORE ###')
else:
download_track(song[TRACK][ID], 'Liked Songs/')
print('\n')
else:
if args.urls:
for spotify_url in args.urls:
track_id, album_id, playlist_id, episode_id, show_id, artist_id = regex_input_for_urls(
sys.argv[1])
spotify_url)
if track_id is not None:
download_track(track_id)
@@ -61,7 +53,19 @@ def client() -> None:
for episode in get_show_episodes(show_id):
download_episode(episode)
else:
if args.playlist:
download_from_user_playlist()
if args.liked_songs:
for song in get_saved_tracks():
if not song[TRACK][NAME]:
print(
'### SKIPPING: SONG DOES NOT EXIST ON SPOTIFY ANYMORE ###')
else:
download_track(song[TRACK][ID], 'Liked Songs/')
print('\n')
if args.search_spotify:
search_text = ''
while len(search_text) == 0:
search_text = input('Enter search or URL: ')

7
zspotify/const.py
View File

@@ -72,6 +72,8 @@ USER_READ_EMAIL = 'user-read-email'
PLAYLIST_READ_PRIVATE = 'playlist-read-private'
USER_LIBRARY_READ = 'user-library-read'
WINDOWS_SYSTEM = 'Windows'
CREDENTIALS_JSON = 'credentials.json'
@@ -96,6 +98,8 @@ CHUNK_SIZE = 'CHUNK_SIZE'
SPLIT_ALBUM_DISCS = 'SPLIT_ALBUM_DISCS'
DOWNLOAD_REAL_TIME = 'DOWNLOAD_REAL_TIME'
BITRATE = 'BITRATE'
CODEC_MAP = {
@@ -127,5 +131,6 @@ CONFIG_DEFAULT_SETTINGS = {
'ANTI_BAN_WAIT_TIME': 1,
'OVERRIDE_AUTO_WAIT': False,
'CHUNK_SIZE': 50000,
'SPLIT_ALBUM_DISCS': False
'SPLIT_ALBUM_DISCS': False,
'DOWNLOAD_REAL_TIME': False
}

4
zspotify/playlist.py
View File

@@ -52,10 +52,12 @@ def download_playlist(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)
enum = 1
for song in p_bar:
download_track(song[TRACK][ID], sanitize_data(playlist[NAME].strip()) + '/',
disable_progressbar=True)
prefix=True, prefix_value=str(enum) ,disable_progressbar=True)
p_bar.set_description(song[TRACK][NAME])
enum += 1
def download_from_user_playlist():

36
zspotify/podcast.py
View File

@@ -5,11 +5,11 @@ 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 const import (CHUNK_SIZE, ERROR, ID, ITEMS, NAME, ROOT_PODCAST_PATH, SHOW,
SKIP_EXISTING_FILES)
from utils import create_download_directory, sanitize_data
from zspotify import ZSpotify
EPISODE_INFO_URL = 'https://api.spotify.com/v1/episodes'
SHOWS_URL = 'https://api.spotify.com/v1/shows'
@@ -55,16 +55,32 @@ def download_episode(episode_id) -> None:
ZSpotify.get_config(ROOT_PODCAST_PATH),
extra_paths,
)
download_directory = os.path.realpath(download_directory)
create_download_directory(download_directory)
total_size = stream.input_stream.size
with open(os.path.join(download_directory, f"{filename}.ogg"),
'wb') as file, tqdm(
desc=filename,
total=total_size,
unit='B',
unit_scale=True,
unit_divisor=1024
filepath = os.path.join(download_directory, f"{filename}.ogg")
if (
os.path.isfile(filepath)
and os.path.getsize(filepath) == total_size
and ZSpotify.get_config(SKIP_EXISTING_FILES)
):
print(
"\n### SKIPPING:",
podcast_name,
"-",
episode_name,
"(EPISODE ALREADY EXISTS) ###",
)
return
with open(filepath, '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(

38
zspotify/track.py
View File

@@ -5,11 +5,12 @@ from typing import Any, Tuple, List
from librespot.audio.decoders import AudioQuality
from librespot.metadata import TrackId
from ffmpy import FFmpeg
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, BITRATE, CODEC_MAP, EXT_MAP
SKIP_EXISTING_FILES, ANTI_BAN_WAIT_TIME, OVERRIDE_AUTO_WAIT, BITRATE, CODEC_MAP, EXT_MAP, DOWNLOAD_REAL_TIME
from utils import sanitize_data, set_audio_tags, set_music_thumbnail, create_download_directory
from zspotify import ZSpotify
@@ -71,7 +72,7 @@ def download_track(track_id: str, extra_paths='', prefix=False, prefix_value='',
) else f'{prefix_value} - {song_name}'
filename = os.path.join(
download_directory, f'{song_name}.{EXT_MAP.get(ZSpotify.get_config(DOWNLOAD_FORMAT))}')
download_directory, f'{song_name}.{EXT_MAP.get(ZSpotify.get_config(DOWNLOAD_FORMAT).lower())}')
except Exception as e:
print('### SKIPPING SONG - FAILED TO QUERY METADATA ###')
@@ -80,11 +81,11 @@ def download_track(track_id: str, extra_paths='', prefix=False, prefix_value='',
try:
if not is_playable:
print('\n### SKIPPING:', song_name,
'(SONG IS UNAVAILABLE) ###')
'(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) ###')
'(SONG ALREADY EXISTS) ###')
else:
if track_id != scraped_song_id:
track_id = scraped_song_id
@@ -102,13 +103,20 @@ def download_track(track_id: str, extra_paths='', prefix=False, prefix_value='',
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))))
for chunk in range(int(total_size / ZSpotify.get_config(CHUNK_SIZE)) + 1):
data = stream.input_stream.stream().read(ZSpotify.get_config(CHUNK_SIZE))
if data == b'':
break
p_bar.update(file.write(data))
if ZSpotify.get_config(DOWNLOAD_REAL_TIME):
if chunk == 0:
pause = get_segment_duration(p_bar)
if pause:
time.sleep(pause)
convert_audio_format(filename)
set_audio_tags(filename, artists, name, album_name,
release_year, disc_number, track_number)
release_year, disc_number, track_number)
set_music_thumbnail(filename, image_url)
if not ZSpotify.get_config(OVERRIDE_AUTO_WAIT):
@@ -121,12 +129,24 @@ def download_track(track_id: str, extra_paths='', prefix=False, prefix_value='',
os.remove(filename)
def get_segment_duration(segment):
""" Returns playback duration of given audio segment """
sound = AudioSegment(
data = segment,
sample_width = 2,
frame_rate = 44100,
channels = 2
)
duration = len(sound) / 5000
return duration
def convert_audio_format(filename) -> None:
""" Converts raw audio into playable file """
temp_filename = f'{os.path.splitext(filename)[0]}.tmp'
os.replace(filename, temp_filename)
download_format = ZSpotify.get_config(DOWNLOAD_FORMAT)
download_format = ZSpotify.get_config(DOWNLOAD_FORMAT).lower()
file_codec = CODEC_MAP.get(download_format, "copy")
if file_codec != 'copy':
bitrate = ZSpotify.get_config(BITRATE)

4
zspotify/zspotify.py
View File

@@ -18,7 +18,7 @@ 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, CONFIG_DEFAULT_SETTINGS
PLAYLIST_READ_PRIVATE, USER_LIBRARY_READ, CONFIG_DEFAULT_SETTINGS
from utils import MusicFormat
@@ -74,7 +74,7 @@ class ZSpotify:
@classmethod
def __get_auth_token(cls):
return cls.SESSION.tokens().get_token(USER_READ_EMAIL, PLAYLIST_READ_PRIVATE).access_token
return cls.SESSION.tokens().get_token(USER_READ_EMAIL, PLAYLIST_READ_PRIVATE, USER_LIBRARY_READ).access_token
@classmethod
def get_auth_header(cls):