Move config to config.py

This commit is contained in:
Mike Schwörer 2021-11-18 22:16:07 +01:00
parent 132e740201
commit 8cd868f317
No known key found for this signature in database
GPG Key ID: D3C7172E0A70F8CF
7 changed files with 146 additions and 92 deletions

View File

@ -10,6 +10,9 @@ if __name__ == '__main__':
parser.add_argument('-ns', '--no-splash',
action='store_true',
help='Suppress the splash screen when loading.')
parser.add_argument('--config-location',
type=str,
help='Specify the zs_config.json location')
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument('urls',
type=str,

View File

@ -16,7 +16,7 @@ SEARCH_URL = 'https://api.spotify.com/v1/search'
def client(args) -> None:
""" Connects to spotify to perform query's and get songs to download """
ZSpotify()
ZSpotify(args)
if not args.no_splash:
splash()

110
zspotify/config.py Normal file
View File

@ -0,0 +1,110 @@
import json
import os
import sys
from typing import Any
CONFIG_FILE_PATH = '../zs_config.json'
ROOT_PATH = 'ROOT_PATH'
ROOT_PODCAST_PATH = 'ROOT_PODCAST_PATH'
SKIP_EXISTING_FILES = 'SKIP_EXISTING_FILES'
SKIP_PREVIOUSLY_DOWNLOADED = 'SKIP_PREVIOUSLY_DOWNLOADED'
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'
DOWNLOAD_REAL_TIME = 'DOWNLOAD_REAL_TIME'
LANGUAGE = 'LANGUAGE'
BITRATE = 'BITRATE'
CONFIG_DEFAULT_SETTINGS = {
'ROOT_PATH': '../ZSpotify Music/',
'ROOT_PODCAST_PATH': '../ZSpotify Podcasts/',
'SKIP_EXISTING_FILES': True,
'SKIP_PREVIOUSLY_DOWNLOADED': False,
'DOWNLOAD_FORMAT': 'ogg',
'FORCE_PREMIUM': False,
'ANTI_BAN_WAIT_TIME': 1,
'OVERRIDE_AUTO_WAIT': False,
'CHUNK_SIZE': 50000,
'SPLIT_ALBUM_DISCS': False,
'DOWNLOAD_REAL_TIME': False,
'LANGUAGE': 'en',
'BITRATE': '',
}
class Config:
Values = {}
@classmethod
def load(cls, args) -> None:
app_dir = os.path.dirname(__file__)
true_config_file_path = os.path.join(app_dir, CONFIG_FILE_PATH)
if not os.path.exists(true_config_file_path):
with open(true_config_file_path, 'w', encoding='utf-8') as config_file:
json.dump(CONFIG_DEFAULT_SETTINGS, config_file, indent=4)
cls.Values = CONFIG_DEFAULT_SETTINGS
else:
with open(true_config_file_path, encoding='utf-8') as config_file:
cls.Values = json.load(config_file)
@classmethod
def get(cls, key) -> Any:
return cls.Values.get(key)
@classmethod
def getRootPath(cls) -> str:
return cls.get(ROOT_PATH)
@classmethod
def getRootPodcastPath(cls) -> str:
return cls.get(ROOT_PODCAST_PATH)
@classmethod
def getSkipExistingFiles(cls) -> bool:
return cls.get(SKIP_EXISTING_FILES)
@classmethod
def getSkipPreviouslyDownloaded(cls) -> bool:
return cls.get(SKIP_PREVIOUSLY_DOWNLOADED)
@classmethod
def getSplitAlbumDiscs(cls) -> bool:
return cls.get(SPLIT_ALBUM_DISCS)
@classmethod
def getChunkSize(cls) -> int():
return cls.get(CHUNK_SIZE)
@classmethod
def getOverrideAutoWait(cls) -> bool:
return cls.get(OVERRIDE_AUTO_WAIT)
@classmethod
def getForcePremium(cls) -> bool:
return cls.get(FORCE_PREMIUM)
@classmethod
def getDownloadFormat(cls) -> str:
return cls.get(DOWNLOAD_FORMAT)
@classmethod
def getAntiBanWaitTime(cls) -> int:
return cls.get(ANTI_BAN_WAIT_TIME)
@classmethod
def getLanguage(cls) -> str:
return cls.get(LANGUAGE)
@classmethod
def getDownloadRealTime(cls) -> bool:
return cls.get(DOWNLOAD_REAL_TIME)
@classmethod
def getBitrate(cls) -> str:
return cls.get(BITRATE)

View File

@ -82,32 +82,6 @@ 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'
SKIP_PREVIOUSLY_DOWNLOADED = 'SKIP_PREVIOUSLY_DOWNLOADED'
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'
DOWNLOAD_REAL_TIME = 'DOWNLOAD_REAL_TIME'
BITRATE = 'BITRATE'
CODEC_MAP = {
'aac': 'aac',
'fdk_aac': 'libfdk_aac',
@ -127,18 +101,3 @@ EXT_MAP = {
'opus': 'ogg',
'vorbis': 'ogg',
}
CONFIG_DEFAULT_SETTINGS = {
'ROOT_PATH': '../ZSpotify Music/',
'ROOT_PODCAST_PATH': '../ZSpotify Podcasts/',
'SKIP_EXISTING_FILES': True,
'SKIP_PREVIOUSLY_DOWNLOADED': False,
'DOWNLOAD_FORMAT': 'ogg',
'FORCE_PREMIUM': False,
'ANTI_BAN_WAIT_TIME': 1,
'OVERRIDE_AUTO_WAIT': False,
'CHUNK_SIZE': 50000,
'SPLIT_ALBUM_DISCS': False,
'DOWNLOAD_REAL_TIME': False,
'LANGUAGE': 'en'
}

View File

@ -5,8 +5,7 @@ from librespot.audio.decoders import VorbisOnlyAudioQuality
from librespot.metadata import EpisodeId
from tqdm import tqdm
from const import (CHUNK_SIZE, ERROR, ID, ITEMS, NAME, ROOT_PODCAST_PATH, SHOW,
SKIP_EXISTING_FILES)
from const import (ERROR, ID, ITEMS, NAME, SHOW)
from utils import create_download_directory, fix_filename
from zspotify import ZSpotify
@ -80,7 +79,7 @@ def download_episode(episode_id) -> None:
download_directory = os.path.join(
os.path.dirname(__file__),
ZSpotify.get_config(ROOT_PODCAST_PATH),
ZSpotify.CONFIG.getRootPodcastPath(),
extra_paths,
)
download_directory = os.path.realpath(download_directory)
@ -97,7 +96,7 @@ def download_episode(episode_id) -> None:
if (
os.path.isfile(filepath)
and os.path.getsize(filepath) == total_size
and ZSpotify.get_config(SKIP_EXISTING_FILES)
and ZSpotify.CONFIG.getSkipExistingFiles()
):
print(
"\n### SKIPPING:",
@ -115,9 +114,9 @@ def download_episode(episode_id) -> None:
unit_scale=True,
unit_divisor=1024
) as bar:
for _ in range(int(total_size / ZSpotify.get_config(CHUNK_SIZE)) + 1):
for _ in range(int(total_size / ZSpotify.CONFIG.getChunkSize()) + 1):
bar.update(file.write(
stream.input_stream.stream().read(ZSpotify.get_config(CHUNK_SIZE))))
stream.input_stream.stream().read(ZSpotify.CONFIG.getChunkSize())))
else:
filepath = os.path.join(download_directory, f"{filename}.mp3")
download_podcast_directly(direct_download_url, filepath)

View File

@ -10,9 +10,7 @@ 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, TRACK_STATS_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, DOWNLOAD_REAL_TIME, \
SKIP_PREVIOUSLY_DOWNLOADED, DURATION_MS
RELEASE_DATE, ID, TRACKS_URL, SAVED_TRACKS_URL, TRACK_STATS_URL, CODEC_MAP, EXT_MAP, DURATION_MS
from utils import fix_filename, set_audio_tags, set_music_thumbnail, create_download_directory, \
get_directory_song_ids, add_to_directory_song_ids, get_previously_downloaded, add_to_archive
from zspotify import ZSpotify
@ -78,12 +76,12 @@ def download_track(track_id: str, extra_paths='', prefix=False, prefix_value='',
(artists, album_name, name, image_url, release_year, disc_number,
track_number, scraped_song_id, is_playable, duration_ms) = get_song_info(track_id)
if ZSpotify.get_config(SPLIT_ALBUM_DISCS):
if ZSpotify.CONFIG.getSplitAlbumDiscs():
download_directory = os.path.join(os.path.dirname(
__file__), ZSpotify.get_config(ROOT_PATH), extra_paths, f'Disc {disc_number}')
__file__), ZSpotify.CONFIG.getRootPath(), extra_paths, f'Disc {disc_number}')
else:
download_directory = os.path.join(os.path.dirname(
__file__), ZSpotify.get_config(ROOT_PATH), extra_paths)
__file__), ZSpotify.CONFIG.getRootPath(), extra_paths)
song_name = fix_filename(artists[0]) + ' - ' + fix_filename(name)
if prefix:
@ -91,9 +89,9 @@ 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).lower())}')
download_directory, f'{song_name}.{EXT_MAP.get(ZSpotify.CONFIG.getDownloadFormat().lower())}')
archive_directory = os.path.join(os.path.dirname(__file__), ZSpotify.get_config(ROOT_PATH))
archive_directory = os.path.join(os.path.dirname(__file__), ZSpotify.CONFIG.getRootPath())
check_name = os.path.isfile(filename) and os.path.getsize(filename)
check_id = scraped_song_id in get_directory_song_ids(download_directory)
check_all_time = scraped_song_id in get_previously_downloaded(scraped_song_id, archive_directory)
@ -104,7 +102,7 @@ def download_track(track_id: str, extra_paths='', prefix=False, prefix_value='',
if re.search(f'^{song_name}_', file)]) + 1
filename = os.path.join(
download_directory, f'{song_name}_{c}.{EXT_MAP.get(ZSpotify.get_config(DOWNLOAD_FORMAT))}')
download_directory, f'{song_name}_{c}.{EXT_MAP.get(ZSpotify.CONFIG.getDownloadFormat())}')
except Exception as e:
@ -116,11 +114,11 @@ def download_track(track_id: str, extra_paths='', prefix=False, prefix_value='',
print('\n### SKIPPING:', song_name,
'(SONG IS UNAVAILABLE) ###')
else:
if check_id and check_name and ZSpotify.get_config(SKIP_EXISTING_FILES):
if check_id and check_name and ZSpotify.CONFIG.getSkipExistingFiles():
print('\n### SKIPPING:', song_name,
'(SONG ALREADY EXISTS) ###')
elif check_all_time and ZSpotify.get_config(SKIP_PREVIOUSLY_DOWNLOADED):
elif check_all_time and ZSpotify.CONFIG.getSkipPreviouslyDownloaded():
print('\n### SKIPPING:', song_name,
'(SONG ALREADY DOWNLOADED ONCE) ###')
@ -141,11 +139,11 @@ def download_track(track_id: str, extra_paths='', prefix=False, prefix_value='',
unit_divisor=1024,
disable=disable_progressbar
) as p_bar:
pause = duration_ms / 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))
pause = duration_ms / ZSpotify.CONFIG.getChunkSize()
for chunk in range(int(total_size / ZSpotify.CONFIG.getChunkSize()) + 1):
data = stream.input_stream.stream().read(ZSpotify.CONFIG.getChunkSize())
p_bar.update(file.write(data))
if ZSpotify.get_config(DOWNLOAD_REAL_TIME):
if ZSpotify.CONFIG.getDownloadRealTime():
time.sleep(pause)
convert_audio_format(filename)
@ -154,14 +152,14 @@ def download_track(track_id: str, extra_paths='', prefix=False, prefix_value='',
set_music_thumbnail(filename, image_url)
# add song id to archive file
if ZSpotify.get_config(SKIP_PREVIOUSLY_DOWNLOADED):
if ZSpotify.CONFIG.getSkipPreviouslyDownloaded():
add_to_archive(scraped_song_id, archive_directory)
# add song id to download directory's .song_ids file
if not check_id:
add_to_directory_song_ids(download_directory, scraped_song_id)
if not ZSpotify.get_config(OVERRIDE_AUTO_WAIT):
time.sleep(ZSpotify.get_config(ANTI_BAN_WAIT_TIME))
if not ZSpotify.CONFIG.getAntiBanWaitTime():
time.sleep(ZSpotify.CONFIG.getAntiBanWaitTime())
except Exception as e:
print('### SKIPPING:', song_name,
'(GENERAL DOWNLOAD ERROR) ###')
@ -175,10 +173,10 @@ def convert_audio_format(filename) -> None:
temp_filename = f'{os.path.splitext(filename)[0]}.tmp'
os.replace(filename, temp_filename)
download_format = ZSpotify.get_config(DOWNLOAD_FORMAT).lower()
download_format = ZSpotify.CONFIG.getDownloadFormat().lower()
file_codec = CODEC_MAP.get(download_format, 'copy')
if file_codec != 'copy':
bitrate = ZSpotify.get_config(BITRATE)
bitrate = ZSpotify.CONFIG.getBitrate()
if not bitrate:
if ZSpotify.DOWNLOAD_QUALITY == AudioQuality.VERY_HIGH:
bitrate = '320k'

View File

@ -17,18 +17,19 @@ 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, USER_LIBRARY_READ, CONFIG_DEFAULT_SETTINGS
PREMIUM, USER_READ_EMAIL, AUTHORIZATION, OFFSET, LIMIT, \
PLAYLIST_READ_PRIVATE, USER_LIBRARY_READ
from utils import MusicFormat
from config import Config
class ZSpotify:
SESSION: Session = None
DOWNLOAD_QUALITY = None
CONFIG = {}
CONFIG = Config()
def __init__(self):
ZSpotify.load_config()
def __init__(self, args):
ZSpotify.CONFIG.load(args)
ZSpotify.login()
@classmethod
@ -52,22 +53,6 @@ class ZSpotify:
except RuntimeError:
pass
@classmethod
def load_config(cls) -> None:
app_dir = os.path.dirname(__file__)
true_config_file_path = os.path.join(app_dir, CONFIG_FILE_PATH)
if not os.path.exists(true_config_file_path):
with open(true_config_file_path, 'w', encoding='utf-8') as config_file:
json.dump(CONFIG_DEFAULT_SETTINGS, config_file, indent=4)
cls.CONFIG = CONFIG_DEFAULT_SETTINGS
else:
with open(true_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)
@ -80,14 +65,14 @@ class ZSpotify:
def get_auth_header(cls):
return {
'Authorization': f'Bearer {cls.__get_auth_token()}',
'Accept-Language': f'{cls.CONFIG.get("LANGUAGE")}'
'Accept-Language': f'{cls.CONFIG.getLanguage()}'
}
@classmethod
def get_auth_header_and_params(cls, limit, offset):
return {
'Authorization': f'Bearer {cls.__get_auth_token()}',
'Accept-Language': f'{cls.CONFIG.get("LANGUAGE")}'
'Accept-Language': f'{cls.CONFIG.getLanguage()}'
}, {LIMIT: limit, OFFSET: offset}
@classmethod
@ -104,4 +89,4 @@ class ZSpotify:
@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)
return (cls.SESSION.get_user_attribute(TYPE) == PREMIUM) or cls.CONFIG.getForcePremium()