mirror of
https://zotify.xyz/zotify/zotify.git
synced 2025-06-23 03:06:43 +00:00
various changes
This commit is contained in:
parent
360e342bc2
commit
b361976504
17 changed files with 573 additions and 353 deletions
|
@ -97,7 +97,7 @@ class Session(LibrespotSession):
|
|||
self.__language = language
|
||||
|
||||
@staticmethod
|
||||
def from_file(cred_file: Path, language: str = "en") -> Session:
|
||||
def from_file(cred_file: Path | str, language: str = "en") -> Session:
|
||||
"""
|
||||
Creates session using saved credentials file
|
||||
Args:
|
||||
|
@ -106,6 +106,8 @@ class Session(LibrespotSession):
|
|||
Returns:
|
||||
Zotify session
|
||||
"""
|
||||
if not isinstance(cred_file, Path):
|
||||
cred_file = Path(cred_file).expanduser()
|
||||
conf = (
|
||||
LibrespotSession.Configuration.Builder()
|
||||
.set_store_credentials(False)
|
||||
|
@ -118,7 +120,7 @@ class Session(LibrespotSession):
|
|||
def from_userpass(
|
||||
username: str,
|
||||
password: str,
|
||||
save_file: Path | None = None,
|
||||
save_file: Path | str | None = None,
|
||||
language: str = "en",
|
||||
) -> Session:
|
||||
"""
|
||||
|
@ -133,6 +135,8 @@ class Session(LibrespotSession):
|
|||
"""
|
||||
builder = LibrespotSession.Configuration.Builder()
|
||||
if save_file:
|
||||
if not isinstance(save_file, Path):
|
||||
save_file = Path(save_file).expanduser()
|
||||
save_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
builder.set_stored_credential_file(str(save_file))
|
||||
else:
|
||||
|
@ -144,7 +148,9 @@ class Session(LibrespotSession):
|
|||
return Session(session, language)
|
||||
|
||||
@staticmethod
|
||||
def from_prompt(save_file: Path | None = None, language: str = "en") -> Session:
|
||||
def from_prompt(
|
||||
save_file: Path | str | None = None, language: str = "en"
|
||||
) -> Session:
|
||||
"""
|
||||
Creates a session with username + password supplied from CLI prompt
|
||||
Args:
|
||||
|
|
|
@ -5,16 +5,15 @@ from pathlib import Path
|
|||
|
||||
from zotify.app import App
|
||||
from zotify.config import CONFIG_PATHS, CONFIG_VALUES
|
||||
from zotify.utils import OptionalOrFalse, SimpleHelpFormatter
|
||||
from zotify.utils import OptionalOrFalse
|
||||
|
||||
VERSION = "0.9.4"
|
||||
VERSION = "0.9.5"
|
||||
|
||||
|
||||
def main():
|
||||
parser = ArgumentParser(
|
||||
prog="zotify",
|
||||
description="A fast and customizable music and podcast downloader",
|
||||
formatter_class=SimpleHelpFormatter,
|
||||
)
|
||||
parser.add_argument(
|
||||
"-v",
|
||||
|
@ -53,7 +52,7 @@ def main():
|
|||
)
|
||||
parser.add_argument("--username", type=str, default="", help="Account username")
|
||||
parser.add_argument("--password", type=str, default="", help="Account password")
|
||||
group = parser.add_mutually_exclusive_group(required=False)
|
||||
group = parser.add_mutually_exclusive_group(required=True)
|
||||
group.add_argument(
|
||||
"urls",
|
||||
type=str,
|
||||
|
|
134
zotify/app.py
134
zotify/app.py
|
@ -8,11 +8,7 @@ from zotify.config import Config
|
|||
from zotify.file import TranscodingError
|
||||
from zotify.loader import Loader
|
||||
from zotify.logger import LogChannel, Logger
|
||||
from zotify.utils import (
|
||||
AudioFormat,
|
||||
CollectionType,
|
||||
PlayableType,
|
||||
)
|
||||
from zotify.utils import AudioFormat, PlayableType
|
||||
|
||||
|
||||
class ParseError(ValueError): ...
|
||||
|
@ -32,7 +28,7 @@ class Selection:
|
|||
def search(
|
||||
self,
|
||||
search_text: str,
|
||||
category: list = [
|
||||
category: list[str] = [
|
||||
"track",
|
||||
"album",
|
||||
"artist",
|
||||
|
@ -56,12 +52,13 @@ class Selection:
|
|||
offset=0,
|
||||
)
|
||||
|
||||
print(f'Search results for "{search_text}"')
|
||||
count = 0
|
||||
for cat in categories.split(","):
|
||||
label = cat + "s"
|
||||
items = resp[label]["items"]
|
||||
if len(items) > 0:
|
||||
print(f"\n### {label.capitalize()} ###")
|
||||
print(f"\n{label.capitalize()}:")
|
||||
try:
|
||||
self.__print(count, items, *self.__print_labels[cat])
|
||||
except KeyError:
|
||||
|
@ -109,7 +106,7 @@ class Selection:
|
|||
|
||||
def __print(self, count: int, items: list[dict[str, Any]], *args: str) -> None:
|
||||
arg_range = range(len(args))
|
||||
category_str = " " + " ".join("{:<38}" for _ in arg_range)
|
||||
category_str = " # " + " ".join("{:<38}" for _ in arg_range)
|
||||
print(category_str.format(*[s.upper() for s in list(args)]))
|
||||
for item in items:
|
||||
count += 1
|
||||
|
@ -149,30 +146,21 @@ class App:
|
|||
self.__config = Config(args)
|
||||
Logger(self.__config)
|
||||
|
||||
# Check options
|
||||
if self.__config.audio_format == AudioFormat.VORBIS and (
|
||||
self.__config.ffmpeg_args != "" or self.__config.ffmpeg_path != ""
|
||||
):
|
||||
Logger.log(
|
||||
LogChannel.WARNINGS,
|
||||
"FFmpeg options will be ignored since no transcoding is required",
|
||||
)
|
||||
|
||||
# Create session
|
||||
if args.username != "" and args.password != "":
|
||||
self.__session = Session.from_userpass(
|
||||
args.username,
|
||||
args.password,
|
||||
self.__config.credentials,
|
||||
self.__config.credentials_path,
|
||||
self.__config.language,
|
||||
)
|
||||
elif self.__config.credentials.is_file():
|
||||
elif self.__config.credentials_path.is_file():
|
||||
self.__session = Session.from_file(
|
||||
self.__config.credentials, self.__config.language
|
||||
self.__config.credentials_path, self.__config.language
|
||||
)
|
||||
else:
|
||||
self.__session = Session.from_prompt(
|
||||
self.__config.credentials, self.__config.language
|
||||
self.__config.credentials_path, self.__config.language
|
||||
)
|
||||
|
||||
# Get items to download
|
||||
|
@ -182,6 +170,7 @@ class App:
|
|||
collections = self.parse(ids)
|
||||
except ParseError as e:
|
||||
Logger.log(LogChannel.ERRORS, str(e))
|
||||
exit(1)
|
||||
if len(collections) > 0:
|
||||
self.download_all(collections)
|
||||
else:
|
||||
|
@ -208,11 +197,12 @@ class App:
|
|||
return ids
|
||||
elif args.urls:
|
||||
return args.urls
|
||||
except (FileNotFoundError, ValueError):
|
||||
Logger.log(LogChannel.WARNINGS, "there is nothing to do")
|
||||
except KeyboardInterrupt:
|
||||
Logger.log(LogChannel.WARNINGS, "\nthere is nothing to do")
|
||||
exit(130)
|
||||
except (FileNotFoundError, ValueError):
|
||||
pass
|
||||
Logger.log(LogChannel.WARNINGS, "there is nothing to do")
|
||||
exit(0)
|
||||
|
||||
def parse(self, links: list[str]) -> list[Collection]:
|
||||
|
@ -226,28 +216,28 @@ class App:
|
|||
except IndexError:
|
||||
raise ParseError(f'Could not parse "{link}"')
|
||||
|
||||
match id_type:
|
||||
case "album":
|
||||
collections.append(Album(self.__session, _id))
|
||||
case "artist":
|
||||
collections.append(Artist(self.__session, _id))
|
||||
case "show":
|
||||
collections.append(Show(self.__session, _id))
|
||||
case "track":
|
||||
collections.append(Track(self.__session, _id))
|
||||
case "episode":
|
||||
collections.append(Episode(self.__session, _id))
|
||||
case "playlist":
|
||||
collections.append(Playlist(self.__session, _id))
|
||||
case _:
|
||||
raise ParseError(f'Unsupported content type "{id_type}"')
|
||||
collection_types = {
|
||||
"album": Album,
|
||||
"artist": Artist,
|
||||
"show": Show,
|
||||
"track": Track,
|
||||
"episode": Episode,
|
||||
"playlist": Playlist,
|
||||
}
|
||||
try:
|
||||
collections.append(
|
||||
collection_types[id_type](_id, self.__session.api(), self.__config)
|
||||
)
|
||||
except ValueError:
|
||||
raise ParseError(f'Unsupported content type "{id_type}"')
|
||||
return collections
|
||||
|
||||
def download_all(self, collections: list[Collection]) -> None:
|
||||
"""Downloads playable to local file"""
|
||||
count = 0
|
||||
total = sum(len(c.playables) for c in collections)
|
||||
for collection in collections:
|
||||
for i in range(len(collection.playables)):
|
||||
playable = collection.playables[i]
|
||||
for playable in collection.playables:
|
||||
count += 1
|
||||
|
||||
# Get track data
|
||||
if playable.type == PlayableType.TRACK:
|
||||
|
@ -263,43 +253,51 @@ class App:
|
|||
LogChannel.SKIPS,
|
||||
f'Download Error: Unknown playable content "{playable.type}"',
|
||||
)
|
||||
return
|
||||
continue
|
||||
|
||||
# Create download location and generate file name
|
||||
match collection.type():
|
||||
case CollectionType.PLAYLIST:
|
||||
# TODO: add playlist name to track metadata
|
||||
library = self.__config.playlist_library
|
||||
template = (
|
||||
self.__config.output_playlist_track
|
||||
if playable.type == PlayableType.TRACK
|
||||
else self.__config.output_playlist_episode
|
||||
)
|
||||
case CollectionType.SHOW | CollectionType.EPISODE:
|
||||
library = self.__config.podcast_library
|
||||
template = self.__config.output_podcast
|
||||
case _:
|
||||
library = self.__config.music_library
|
||||
template = self.__config.output_album
|
||||
output = track.create_output(
|
||||
library, template, self.__config.replace_existing
|
||||
)
|
||||
track.metadata.extend(playable.metadata)
|
||||
try:
|
||||
output = track.create_output(
|
||||
playable.library,
|
||||
playable.output_template,
|
||||
self.__config.replace_existing,
|
||||
)
|
||||
except FileExistsError:
|
||||
Logger.log(
|
||||
LogChannel.SKIPS,
|
||||
f'Skipping "{track.name}": Already exists at specified output',
|
||||
)
|
||||
|
||||
file = track.write_audio_stream(output)
|
||||
# Download track
|
||||
with Logger.progress(
|
||||
desc=f"({count}/{total}) {track.name}",
|
||||
total=track.input_stream.size,
|
||||
) as p_bar:
|
||||
file = track.write_audio_stream(output, p_bar)
|
||||
|
||||
# Download lyrics
|
||||
if playable.type == PlayableType.TRACK and self.__config.lyrics_file:
|
||||
with Loader("Fetching lyrics..."):
|
||||
try:
|
||||
track.get_lyrics().save(output)
|
||||
except FileNotFoundError as e:
|
||||
Logger.log(LogChannel.SKIPS, str(e))
|
||||
if not self.__session.is_premium():
|
||||
Logger.log(
|
||||
LogChannel.SKIPS,
|
||||
f'Failed to save lyrics for "{track.name}": Lyrics are only available to premium users',
|
||||
)
|
||||
else:
|
||||
with Loader("Fetching lyrics..."):
|
||||
try:
|
||||
track.lyrics().save(output)
|
||||
except FileNotFoundError as e:
|
||||
Logger.log(LogChannel.SKIPS, str(e))
|
||||
Logger.log(LogChannel.DOWNLOADS, f"\nDownloaded {track.name}")
|
||||
|
||||
# Transcode audio
|
||||
if self.__config.audio_format != AudioFormat.VORBIS:
|
||||
if (
|
||||
self.__config.audio_format != AudioFormat.VORBIS
|
||||
or self.__config.ffmpeg_args != ""
|
||||
):
|
||||
try:
|
||||
with Loader(LogChannel.PROGRESS, "Converting audio..."):
|
||||
with Loader("Converting audio..."):
|
||||
file.transcode(
|
||||
self.__config.audio_format,
|
||||
self.__config.transcode_bitrate,
|
||||
|
|
|
@ -5,80 +5,105 @@ from librespot.metadata import (
|
|||
ShowId,
|
||||
)
|
||||
|
||||
from zotify import Session
|
||||
from zotify.utils import CollectionType, PlayableData, PlayableType, bytes_to_base62
|
||||
from zotify import Api
|
||||
from zotify.config import Config
|
||||
from zotify.utils import MetadataEntry, PlayableData, PlayableType, bytes_to_base62
|
||||
|
||||
|
||||
class Collection:
|
||||
playables: list[PlayableData] = []
|
||||
|
||||
def type(self) -> CollectionType:
|
||||
return CollectionType(self.__class__.__name__.lower())
|
||||
def __init__(self, b62_id: str, api: Api, config: Config = Config()):
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class Album(Collection):
|
||||
def __init__(self, session: Session, b62_id: str):
|
||||
album = session.api().get_metadata_4_album(AlbumId.from_base62(b62_id))
|
||||
def __init__(self, b62_id: str, api: Api, config: Config = Config()):
|
||||
album = api.get_metadata_4_album(AlbumId.from_base62(b62_id))
|
||||
for disc in album.disc:
|
||||
for track in disc.track:
|
||||
self.playables.append(
|
||||
PlayableData(
|
||||
PlayableType.TRACK,
|
||||
bytes_to_base62(track.gid),
|
||||
config.album_library,
|
||||
config.output_album,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class Artist(Collection):
|
||||
def __init__(self, session: Session, b62_id: str):
|
||||
artist = session.api().get_metadata_4_artist(ArtistId.from_base62(b62_id))
|
||||
def __init__(self, b62_id: str, api: Api, config: Config = Config()):
|
||||
artist = api.get_metadata_4_artist(ArtistId.from_base62(b62_id))
|
||||
for album_group in (
|
||||
artist.album_group
|
||||
and artist.single_group
|
||||
and artist.compilation_group
|
||||
and artist.appears_on_group
|
||||
):
|
||||
album = session.api().get_metadata_4_album(
|
||||
AlbumId.from_hex(album_group.album[0].gid)
|
||||
)
|
||||
album = api.get_metadata_4_album(AlbumId.from_hex(album_group.album[0].gid))
|
||||
for disc in album.disc:
|
||||
for track in disc.track:
|
||||
self.playables.append(
|
||||
PlayableData(
|
||||
PlayableType.TRACK,
|
||||
bytes_to_base62(track.gid),
|
||||
config.album_library,
|
||||
config.output_album,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class Show(Collection):
|
||||
def __init__(self, session: Session, b62_id: str):
|
||||
show = session.api().get_metadata_4_show(ShowId.from_base62(b62_id))
|
||||
def __init__(self, b62_id: str, api: Api, config: Config = Config()):
|
||||
show = api.get_metadata_4_show(ShowId.from_base62(b62_id))
|
||||
for episode in show.episode:
|
||||
self.playables.append(
|
||||
PlayableData(PlayableType.EPISODE, bytes_to_base62(episode.gid))
|
||||
PlayableData(
|
||||
PlayableType.EPISODE,
|
||||
bytes_to_base62(episode.gid),
|
||||
config.podcast_library,
|
||||
config.output_podcast,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class Playlist(Collection):
|
||||
def __init__(self, session: Session, b62_id: str):
|
||||
playlist = session.api().get_playlist(PlaylistId(b62_id))
|
||||
# self.name = playlist.title
|
||||
for item in playlist.contents.items:
|
||||
def __init__(self, b62_id: str, api: Api, config: Config = Config()):
|
||||
playlist = api.get_playlist(PlaylistId(b62_id))
|
||||
for i in range(len(playlist.contents.items)):
|
||||
item = playlist.contents.items[i]
|
||||
split = item.uri.split(":")
|
||||
playable_type = split[1]
|
||||
playable_id = split[2]
|
||||
metadata = [
|
||||
MetadataEntry("playlist", playlist.attributes.name),
|
||||
MetadataEntry("playlist_length", playlist.length),
|
||||
MetadataEntry("playlist_owner", playlist.owner_username),
|
||||
MetadataEntry(
|
||||
"playlist_number",
|
||||
i + 1,
|
||||
str(i + 1).zfill(len(str(playlist.length + 1))),
|
||||
),
|
||||
]
|
||||
if playable_type == "track":
|
||||
self.playables.append(
|
||||
PlayableData(
|
||||
PlayableType.TRACK,
|
||||
split[2],
|
||||
playable_id,
|
||||
config.playlist_library,
|
||||
config.output_playlist_track,
|
||||
metadata,
|
||||
)
|
||||
)
|
||||
elif playable_type == "episode":
|
||||
self.playables.append(
|
||||
PlayableData(
|
||||
PlayableType.EPISODE,
|
||||
split[2],
|
||||
playable_id,
|
||||
config.playlist_library,
|
||||
config.output_playlist_episode,
|
||||
metadata,
|
||||
)
|
||||
)
|
||||
else:
|
||||
|
@ -86,10 +111,21 @@ class Playlist(Collection):
|
|||
|
||||
|
||||
class Track(Collection):
|
||||
def __init__(self, session: Session, b62_id: str):
|
||||
self.playables.append(PlayableData(PlayableType.TRACK, b62_id))
|
||||
def __init__(self, b62_id: str, api: Api, config: Config = Config()):
|
||||
self.playables.append(
|
||||
PlayableData(
|
||||
PlayableType.TRACK, b62_id, config.album_library, config.output_album
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class Episode(Collection):
|
||||
def __init__(self, session: Session, b62_id: str):
|
||||
self.playables.append(PlayableData(PlayableType.EPISODE, b62_id))
|
||||
def __init__(self, b62_id: str, api: Api, config: Config = Config()):
|
||||
self.playables.append(
|
||||
PlayableData(
|
||||
PlayableType.EPISODE,
|
||||
b62_id,
|
||||
config.podcast_library,
|
||||
config.output_podcast,
|
||||
)
|
||||
)
|
||||
|
|
|
@ -7,18 +7,18 @@ from typing import Any
|
|||
|
||||
from zotify.utils import AudioFormat, ImageSize, Quality
|
||||
|
||||
ALBUM_LIBRARY = "album_library"
|
||||
ALL_ARTISTS = "all_artists"
|
||||
ARTWORK_SIZE = "artwork_size"
|
||||
AUDIO_FORMAT = "audio_format"
|
||||
CREATE_PLAYLIST_FILE = "create_playlist_file"
|
||||
CREDENTIALS = "credentials"
|
||||
CREDENTIALS_PATH = "credentials_path"
|
||||
DOWNLOAD_QUALITY = "download_quality"
|
||||
FFMPEG_ARGS = "ffmpeg_args"
|
||||
FFMPEG_PATH = "ffmpeg_path"
|
||||
LANGUAGE = "language"
|
||||
LYRICS_FILE = "lyrics_file"
|
||||
LYRICS_ONLY = "lyrics_only"
|
||||
MUSIC_LIBRARY = "music_library"
|
||||
OUTPUT = "output"
|
||||
OUTPUT_ALBUM = "output_album"
|
||||
OUTPUT_PLAYLIST_TRACK = "output_playlist_track"
|
||||
|
@ -49,7 +49,7 @@ SYSTEM_PATHS = {
|
|||
}
|
||||
|
||||
LIBRARY_PATHS = {
|
||||
"music": Path.home().joinpath("Music/Zotify Music"),
|
||||
"album": Path.home().joinpath("Music/Zotify Albums"),
|
||||
"podcast": Path.home().joinpath("Music/Zotify Podcasts"),
|
||||
"playlist": Path.home().joinpath("Music/Zotify Playlists"),
|
||||
}
|
||||
|
@ -68,7 +68,7 @@ OUTPUT_PATHS = {
|
|||
}
|
||||
|
||||
CONFIG_VALUES = {
|
||||
CREDENTIALS: {
|
||||
CREDENTIALS_PATH: {
|
||||
"default": CONFIG_PATHS["creds"],
|
||||
"type": Path,
|
||||
"args": ["--credentials"],
|
||||
|
@ -80,11 +80,11 @@ CONFIG_VALUES = {
|
|||
"args": ["--archive"],
|
||||
"help": "Path to track archive file",
|
||||
},
|
||||
MUSIC_LIBRARY: {
|
||||
"default": LIBRARY_PATHS["music"],
|
||||
ALBUM_LIBRARY: {
|
||||
"default": LIBRARY_PATHS["album"],
|
||||
"type": Path,
|
||||
"args": ["--music-library"],
|
||||
"help": "Path to root of music library",
|
||||
"args": ["--album-library"],
|
||||
"help": "Path to root of album library",
|
||||
},
|
||||
PODCAST_LIBRARY: {
|
||||
"default": LIBRARY_PATHS["podcast"],
|
||||
|
@ -138,8 +138,8 @@ CONFIG_VALUES = {
|
|||
},
|
||||
AUDIO_FORMAT: {
|
||||
"default": "vorbis",
|
||||
"type": AudioFormat,
|
||||
"choices": [n.value.name for n in AudioFormat],
|
||||
"type": AudioFormat.from_string,
|
||||
"choices": list(AudioFormat),
|
||||
"args": ["--audio-format"],
|
||||
"help": "Audio format of final track output",
|
||||
},
|
||||
|
@ -256,13 +256,13 @@ CONFIG_VALUES = {
|
|||
|
||||
class Config:
|
||||
__config_file: Path | None
|
||||
album_library: Path
|
||||
artwork_size: ImageSize
|
||||
audio_format: AudioFormat
|
||||
credentials: Path
|
||||
credentials_path: Path
|
||||
download_quality: Quality
|
||||
ffmpeg_args: str
|
||||
ffmpeg_path: str
|
||||
music_library: Path
|
||||
language: str
|
||||
lyrics_file: bool
|
||||
output_album: str
|
||||
|
@ -276,9 +276,9 @@ class Config:
|
|||
save_metadata: bool
|
||||
transcode_bitrate: int
|
||||
|
||||
def __init__(self, args: Namespace = Namespace()):
|
||||
def __init__(self, args: Namespace | None = None):
|
||||
jsonvalues = {}
|
||||
if args.config:
|
||||
if args is not None and args.config:
|
||||
self.__config_file = Path(args.config)
|
||||
# Valid config file found
|
||||
if self.__config_file.exists():
|
||||
|
@ -300,7 +300,7 @@ class Config:
|
|||
|
||||
for key in CONFIG_VALUES:
|
||||
# Override config with commandline arguments
|
||||
if key in vars(args) and vars(args)[key] is not None:
|
||||
if args is not None and key in vars(args) and vars(args)[key] is not None:
|
||||
setattr(self, key, self.__parse_arg_value(key, vars(args)[key]))
|
||||
# If no command option specified use config
|
||||
elif key in jsonvalues:
|
||||
|
@ -314,14 +314,13 @@ class Config:
|
|||
)
|
||||
|
||||
# "library" arg overrides all *_library options
|
||||
if args.library:
|
||||
print("args.library")
|
||||
self.music_library = Path(args.library).expanduser().resolve()
|
||||
if args is not None and args.library:
|
||||
self.album_library = Path(args.library).expanduser().resolve()
|
||||
self.playlist_library = Path(args.library).expanduser().resolve()
|
||||
self.podcast_library = Path(args.library).expanduser().resolve()
|
||||
|
||||
# "output" arg overrides all output_* options
|
||||
if args.output:
|
||||
if args is not None and args.output:
|
||||
self.output_album = args.output
|
||||
self.output_podcast = args.output
|
||||
self.output_playlist_track = args.output
|
||||
|
@ -334,8 +333,8 @@ class Config:
|
|||
return value
|
||||
elif config_type == Path:
|
||||
return Path(value).expanduser().resolve()
|
||||
elif config_type == AudioFormat:
|
||||
return AudioFormat[value.upper()]
|
||||
elif config_type == AudioFormat.from_string:
|
||||
return AudioFormat.from_string(value)
|
||||
elif config_type == ImageSize.from_string:
|
||||
return ImageSize.from_string(value)
|
||||
elif config_type == Quality.from_string:
|
||||
|
|
|
@ -4,7 +4,7 @@ from __future__ import annotations
|
|||
|
||||
from itertools import cycle
|
||||
from shutil import get_terminal_size
|
||||
from sys import platform
|
||||
from sys import platform as PLATFORM
|
||||
from threading import Thread
|
||||
from time import sleep
|
||||
|
||||
|
@ -22,7 +22,7 @@ class Loader:
|
|||
pass
|
||||
"""
|
||||
|
||||
def __init__(self, desc="Loading...", end="", timeout=0.1, mode="std3") -> None:
|
||||
def __init__(self, desc: str = "Loading...", end: str = "", timeout: float = 0.1):
|
||||
"""
|
||||
A loader-like context manager
|
||||
Args:
|
||||
|
@ -35,7 +35,8 @@ class Loader:
|
|||
self.timeout = timeout
|
||||
|
||||
self.__thread = Thread(target=self.__animate, daemon=True)
|
||||
if platform == "win32":
|
||||
# Cool loader looks awful in cmd
|
||||
if PLATFORM == "win32":
|
||||
self.steps = ["/", "-", "\\", "|"]
|
||||
else:
|
||||
self.steps = ["⣾", "⣽", "⣻", "⢿", "⡿", "⣟", "⣯", "⣷"]
|
||||
|
|
|
@ -22,7 +22,7 @@ class LogChannel(Enum):
|
|||
|
||||
|
||||
class Logger:
|
||||
__config: Config
|
||||
__config: Config = Config()
|
||||
|
||||
@classmethod
|
||||
def __init__(cls, config: Config):
|
||||
|
@ -50,9 +50,9 @@ class Logger:
|
|||
total=None,
|
||||
leave=False,
|
||||
position=0,
|
||||
unit="it",
|
||||
unit_scale=False,
|
||||
unit_divisor=1000,
|
||||
unit="B",
|
||||
unit_scale=True,
|
||||
unit_divisor=1024,
|
||||
) -> tqdm:
|
||||
"""
|
||||
Prints progress bar
|
||||
|
|
|
@ -7,14 +7,13 @@ from librespot.metadata import AlbumId
|
|||
from librespot.structure import GeneralAudioStream
|
||||
from librespot.util import bytes_to_hex
|
||||
from requests import get
|
||||
from tqdm import tqdm
|
||||
|
||||
from zotify.file import LocalFile
|
||||
from zotify.logger import Logger
|
||||
from zotify.utils import (
|
||||
AudioFormat,
|
||||
ImageSize,
|
||||
MetadataEntry,
|
||||
PlayableType,
|
||||
bytes_to_base62,
|
||||
fix_filename,
|
||||
)
|
||||
|
@ -40,13 +39,15 @@ class Lyrics:
|
|||
f"[{ts_minutes}:{ts_seconds}.{ts_millis}]{line.words}\n"
|
||||
)
|
||||
|
||||
def save(self, path: Path, prefer_synced: bool = True) -> None:
|
||||
def save(self, path: Path | str, prefer_synced: bool = True) -> None:
|
||||
"""
|
||||
Saves lyrics to file
|
||||
Args:
|
||||
location: path to target lyrics file
|
||||
prefer_synced: Use line synced lyrics if available
|
||||
"""
|
||||
if not isinstance(path, Path):
|
||||
path = Path(path).expanduser()
|
||||
if self.__sync_type == "line_synced" and prefer_synced:
|
||||
with open(f"{path}.lrc", "w+", encoding="utf-8") as f:
|
||||
f.writelines(self.__lines_synced)
|
||||
|
@ -60,10 +61,12 @@ class Playable:
|
|||
input_stream: GeneralAudioStream
|
||||
metadata: list[MetadataEntry]
|
||||
name: str
|
||||
type: PlayableType
|
||||
|
||||
def create_output(
|
||||
self, library: Path = Path("./"), output: str = "{title}", replace: bool = False
|
||||
self,
|
||||
library: Path | str = Path("./"),
|
||||
output: str = "{title}",
|
||||
replace: bool = False,
|
||||
) -> Path:
|
||||
"""
|
||||
Creates save directory for the output file
|
||||
|
@ -74,6 +77,8 @@ class Playable:
|
|||
Returns:
|
||||
File path for the track
|
||||
"""
|
||||
if not isinstance(library, Path):
|
||||
library = Path(library)
|
||||
for meta in self.metadata:
|
||||
if meta.string is not None:
|
||||
output = output.replace(
|
||||
|
@ -87,26 +92,20 @@ class Playable:
|
|||
return file_path
|
||||
|
||||
def write_audio_stream(
|
||||
self,
|
||||
output: Path,
|
||||
self, output: Path | str, p_bar: tqdm = tqdm(disable=True)
|
||||
) -> LocalFile:
|
||||
"""
|
||||
Writes audio stream to file
|
||||
Args:
|
||||
output: File path of saved audio stream
|
||||
p_bar: tqdm progress bar
|
||||
Returns:
|
||||
LocalFile object
|
||||
"""
|
||||
if not isinstance(output, Path):
|
||||
output = Path(output).expanduser()
|
||||
file = f"{output}.ogg"
|
||||
with open(file, "wb") as f, Logger.progress(
|
||||
desc=self.name,
|
||||
total=self.input_stream.size,
|
||||
unit="B",
|
||||
unit_scale=True,
|
||||
unit_divisor=1024,
|
||||
position=0,
|
||||
leave=False,
|
||||
) as p_bar:
|
||||
with open(file, "wb") as f, p_bar as p_bar:
|
||||
chunk = None
|
||||
while chunk != b"":
|
||||
chunk = self.input_stream.stream().read(1024)
|
||||
|
@ -127,6 +126,8 @@ class Playable:
|
|||
|
||||
|
||||
class Track(PlayableContentFeeder.LoadedStream, Playable):
|
||||
__lyrics: Lyrics
|
||||
|
||||
def __init__(self, track: PlayableContentFeeder.LoadedStream, api):
|
||||
super(Track, self).__init__(
|
||||
track.track,
|
||||
|
@ -135,10 +136,8 @@ class Track(PlayableContentFeeder.LoadedStream, Playable):
|
|||
track.metrics,
|
||||
)
|
||||
self.__api = api
|
||||
self.__lyrics: Lyrics
|
||||
self.cover_images = self.album.cover_group.image
|
||||
self.metadata = self.__default_metadata()
|
||||
self.type = PlayableType.TRACK
|
||||
|
||||
def __getattr__(self, name):
|
||||
try:
|
||||
|
@ -154,7 +153,8 @@ class Track(PlayableContentFeeder.LoadedStream, Playable):
|
|||
)
|
||||
return [
|
||||
MetadataEntry("album", self.album.name),
|
||||
MetadataEntry("album_artist", [a.name for a in self.album.artist]),
|
||||
MetadataEntry("album_artist", self.album.artist[0].name),
|
||||
MetadataEntry("album_artists", [a.name for a in self.album.artist]),
|
||||
MetadataEntry("artist", self.artist[0].name),
|
||||
MetadataEntry("artists", [a.name for a in self.artist]),
|
||||
MetadataEntry("date", f"{date.year}-{date.month}-{date.day}"),
|
||||
|
@ -180,7 +180,7 @@ class Track(PlayableContentFeeder.LoadedStream, Playable):
|
|||
),
|
||||
]
|
||||
|
||||
def lyrics(self) -> Lyrics:
|
||||
def get_lyrics(self) -> Lyrics:
|
||||
"""Returns track lyrics if available"""
|
||||
if not self.track.has_lyrics:
|
||||
raise FileNotFoundError(
|
||||
|
@ -208,7 +208,6 @@ class Episode(PlayableContentFeeder.LoadedStream, Playable):
|
|||
self.__api = api
|
||||
self.cover_images = self.episode.cover_image.image
|
||||
self.metadata = self.__default_metadata()
|
||||
self.type = PlayableType.EPISODE
|
||||
|
||||
def __getattr__(self, name):
|
||||
try:
|
||||
|
@ -228,29 +227,26 @@ class Episode(PlayableContentFeeder.LoadedStream, Playable):
|
|||
MetadataEntry("title", self.name),
|
||||
]
|
||||
|
||||
def write_audio_stream(self, output: Path) -> LocalFile:
|
||||
def write_audio_stream(
|
||||
self, output: Path | str, p_bar: tqdm = tqdm(disable=True)
|
||||
) -> LocalFile:
|
||||
"""
|
||||
Writes audio stream to file.
|
||||
Uses external source if available for faster download.
|
||||
Args:
|
||||
output: File path of saved audio stream
|
||||
p_bar: tqdm progress bar
|
||||
Returns:
|
||||
LocalFile object
|
||||
"""
|
||||
if not isinstance(output, Path):
|
||||
output = Path(output).expanduser()
|
||||
if not bool(self.external_url):
|
||||
return super().write_audio_stream(output)
|
||||
file = f"{output}.{self.external_url.rsplit('.', 1)[-1]}"
|
||||
with get(self.external_url, stream=True) as r, open(
|
||||
file, "wb"
|
||||
) as f, Logger.progress(
|
||||
desc=self.name,
|
||||
total=self.input_stream.size,
|
||||
unit="B",
|
||||
unit_scale=True,
|
||||
unit_divisor=1024,
|
||||
position=0,
|
||||
leave=False,
|
||||
) as p_bar:
|
||||
) as f, p_bar as p_bar:
|
||||
for chunk in r.iter_content(chunk_size=1024):
|
||||
p_bar.update(f.write(chunk))
|
||||
return LocalFile(Path(file))
|
||||
|
|
|
@ -1,9 +1,7 @@
|
|||
from argparse import Action, ArgumentError, HelpFormatter
|
||||
from argparse import Action, ArgumentError
|
||||
from enum import Enum, IntEnum
|
||||
from pathlib import Path
|
||||
from re import IGNORECASE, sub
|
||||
from sys import exit
|
||||
from sys import platform as PLATFORM
|
||||
from sys import stderr
|
||||
from typing import Any, NamedTuple
|
||||
|
||||
from librespot.audio.decoders import AudioQuality
|
||||
|
@ -25,7 +23,20 @@ class AudioFormat(Enum):
|
|||
OPUS = AudioCodec("opus", "ogg")
|
||||
VORBIS = AudioCodec("vorbis", "ogg")
|
||||
WAV = AudioCodec("wav", "wav")
|
||||
WV = AudioCodec("wavpack", "wv")
|
||||
WAVPACK = AudioCodec("wavpack", "wv")
|
||||
|
||||
def __str__(self):
|
||||
return self.name.lower()
|
||||
|
||||
def __repr__(self):
|
||||
return str(self)
|
||||
|
||||
@staticmethod
|
||||
def from_string(s):
|
||||
try:
|
||||
return AudioFormat[s.upper()]
|
||||
except Exception:
|
||||
return s
|
||||
|
||||
|
||||
class Quality(Enum):
|
||||
|
@ -94,15 +105,6 @@ class MetadataEntry:
|
|||
self.string = str(string_value)
|
||||
|
||||
|
||||
class CollectionType(Enum):
|
||||
ALBUM = "album"
|
||||
ARTIST = "artist"
|
||||
SHOW = "show"
|
||||
PLAYLIST = "playlist"
|
||||
TRACK = "track"
|
||||
EPISODE = "episode"
|
||||
|
||||
|
||||
class PlayableType(Enum):
|
||||
TRACK = "track"
|
||||
EPISODE = "episode"
|
||||
|
@ -111,14 +113,9 @@ class PlayableType(Enum):
|
|||
class PlayableData(NamedTuple):
|
||||
type: PlayableType
|
||||
id: str
|
||||
|
||||
|
||||
class SimpleHelpFormatter(HelpFormatter):
|
||||
def _format_usage(self, usage, actions, groups, prefix):
|
||||
if usage is not None:
|
||||
super()._format_usage(usage, actions, groups, prefix)
|
||||
stderr.write('zotify: error: unrecognized arguments - try "zotify -h"\n')
|
||||
exit(2)
|
||||
library: Path
|
||||
output_template: str
|
||||
metadata: list[MetadataEntry] = []
|
||||
|
||||
|
||||
class OptionalOrFalse(Action):
|
||||
|
@ -171,24 +168,22 @@ class OptionalOrFalse(Action):
|
|||
)
|
||||
|
||||
|
||||
def fix_filename(filename: str, substitute: str = "_", platform: str = PLATFORM) -> str:
|
||||
def fix_filename(
|
||||
filename: str,
|
||||
substitute: str = "_",
|
||||
) -> str:
|
||||
"""
|
||||
Replace invalid characters on Linux/Windows/MacOS with underscores.
|
||||
Replace invalid characters. Trailing spaces & periods are ignored.
|
||||
Original list from https://stackoverflow.com/a/31976060/819417
|
||||
Trailing spaces & periods are ignored on Windows.
|
||||
Args:
|
||||
filename: The name of the file to repair
|
||||
platform: Host operating system
|
||||
substitute: Replacement character for disallowed characters
|
||||
Returns:
|
||||
Filename with replaced characters
|
||||
"""
|
||||
if platform == "linux":
|
||||
regex = r"[/\0]|^(?![^.])|[\s]$"
|
||||
elif platform == "darwin":
|
||||
regex = r"[/\0:]|^(?![^.])|[\s]$"
|
||||
else:
|
||||
regex = r"[/\\:|<>\"?*\0-\x1f]|^(AUX|COM[1-9]|CON|LPT[1-9]|NUL|PRN)(?![^.])|^\s|[\s.]$"
|
||||
regex = (
|
||||
r"[/\\:|<>\"?*\0-\x1f]|^(AUX|COM[1-9]|CON|LPT[1-9]|NUL|PRN)(?![^.])|^\s|[\s.]$"
|
||||
)
|
||||
return sub(regex, substitute, str(filename), flags=IGNORECASE)
|
||||
|
||||
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue