From 08ab495baab5f35753dd0885ac6a81908a1d7517 Mon Sep 17 00:00:00 2001 From: Rafael Moraes <50295204+glomatico@users.noreply.github.com> Date: Thu, 26 Feb 2026 14:03:03 -0300 Subject: [PATCH] Add CLI entrypoint and config management --- votify/cli/__init__.py | 0 votify/cli/cli.py | 203 ++++++++++++++++++ votify/cli/cli_config.py | 422 ++++++++++++++++++++++++++++++++++++++ votify/cli/config_file.py | 167 +++++++++++++++ votify/cli/constants.py | 9 + votify/cli/utils.py | 96 +++++++++ 6 files changed, 897 insertions(+) create mode 100644 votify/cli/__init__.py create mode 100644 votify/cli/cli.py create mode 100644 votify/cli/cli_config.py create mode 100644 votify/cli/config_file.py create mode 100644 votify/cli/constants.py create mode 100644 votify/cli/utils.py diff --git a/votify/cli/__init__.py b/votify/cli/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/votify/cli/cli.py b/votify/cli/cli.py new file mode 100644 index 0000000..c41bddf --- /dev/null +++ b/votify/cli/cli.py @@ -0,0 +1,203 @@ +import asyncio +import logging +from functools import wraps +from pathlib import Path + +import click +import colorama +from dataclass_click import dataclass_click + +from .. import __version__ +from ..api.api import SpotifyApi +from ..downloader.audio import SpotifyAudioDownloader +from ..downloader.base import SpotifyBaseDownloader +from ..downloader.downloader import SpotifyDownloader +from ..downloader.video import SpotifyVideoDownloader +from ..interface.audio import SpotifyAudioInterface +from ..interface.base import SpotifyBaseInterface +from ..interface.episode import SpotifyEpisodeInterface +from ..interface.exceptions import VotifyUrlParseException, VotifyMediaException +from ..interface.episode_video import SpotifyEpisodeVideoInterface +from ..interface.interface import SpotifyInterface +from ..downloader.exceptions import VotifyDownloaderException +from ..interface.music_video import SpotifyMusicVideoInterface +from ..interface.song import SpotifySongInterface +from ..interface.video import SpotifyVideoInterface +from .cli_config import CliConfig +from .config_file import ConfigFile +from .constants import X_NOT_IN_PATH +from .utils import Csv, CustomLoggerFormatter, prompt_path + +logger = logging.getLogger(__name__) + + +def make_sync(func): + @wraps(func) + def wrapper(*args, **kwargs): + return asyncio.run(func(*args, **kwargs)) + + return wrapper + + +@click.command() +@click.help_option("-h", "--help") +@click.version_option(__version__, "-v", "--version") +@dataclass_click(CliConfig) +@ConfigFile.loader +@make_sync +async def main(config: CliConfig): + colorama.just_fix_windows_console() + + root_logger = logging.getLogger(__name__.split(".")[0]) + root_logger.setLevel(config.log_level) + root_logger.propagate = False + + stream_handler = logging.StreamHandler() + stream_handler.setFormatter(CustomLoggerFormatter()) + root_logger.addHandler(stream_handler) + + if config.log_file: + file_handler = logging.FileHandler(config.log_file, encoding="utf-8") + file_handler.setFormatter(CustomLoggerFormatter(use_colors=False)) + root_logger.addHandler(file_handler) + + logger.info(f"Starting Votify {__version__}") + + cookies_path = prompt_path(config.cookies_path) + + wvd_path = None + if not config.no_drm: + wvd_path = prompt_path(config.wvd_path) + + api = await SpotifyApi.create_from_netscape_cookies(cookies_path) + + base_interface = SpotifyBaseInterface( + api=api, + cover_size=config.cover_size, + prefer_video=config.prefer_video, + no_drm=config.no_drm, + wvd_path=wvd_path, + ) + video_interface = SpotifyVideoInterface( + base=base_interface, + video_format=config.video_format, + resolution=config.video_resolution, + ) + audio_interface = SpotifyAudioInterface( + base=base_interface, + audio_quality=config.audio_quality, + ) + song_interface = SpotifySongInterface(audio_interface) + episode_interface = SpotifyEpisodeInterface(audio_interface) + music_video_interface = SpotifyMusicVideoInterface(video_interface) + episode_video_interface = SpotifyEpisodeVideoInterface(video_interface) + + interface = SpotifyInterface( + base=audio_interface, + song=song_interface, + episode=episode_interface, + music_video=music_video_interface, + episode_video=episode_video_interface, + ) + + base_downloader = SpotifyBaseDownloader( + interface=interface, + output_path=config.output, + temp_path=config.temp, + aria2c_path=config.aria2c_path, + curl_path=config.curl_path, + ffmpeg_path=config.ffmpeg_path, + mp4box_path=config.mp4box_path, + mp4decrypt_path=config.mp4decrypt_path, + shaka_packager_path=config.shaka_packager_path, + album_folder_template=config.album_folder_template, + compilation_folder_template=config.compilation_folder_template, + no_album_folder_template=config.no_album_folder_template, + single_disc_file_template=config.single_disc_file_template, + multi_disc_file_template=config.multi_disc_file_template, + no_album_file_template=config.no_album_file_template, + playlist_file_template=config.playlist_file_template, + date_tag_template=config.date_tag_template, + exclude_tags=config.exclude_tags, + truncate=config.truncate, + ) + audio_downloader = SpotifyAudioDownloader( + base=base_downloader, + download_mode=config.audio_download_mode, + remux_mode=config.audio_remux_mode, + ) + video_downloader = SpotifyVideoDownloader( + base=base_downloader, + remux_mode=config.video_remux_mode, + ) + downloader = SpotifyDownloader( + base=base_downloader, + audio=audio_downloader, + video=video_downloader, + ) + + if config.read_urls_as_txt: + urls_from_file = [] + for url in config.urls: + if Path(url).is_file() and Path(url).exists(): + urls_from_file.extend( + [ + line.strip() + for line in Path(url).read_text(encoding="utf-8").splitlines() + if line.strip() + ] + ) + urls = urls_from_file + else: + urls = config.urls + + error_count = 0 + for url_index, url in enumerate(urls, 1): + url_progress = click.style(f"[URL {url_index}/{len(urls)}]", dim=True) + logger.info(url_progress + f' Processing "{url}"') + download_queue = downloader.get_download_item(url) + download_index = 0 + while True: + item = None + try: + item = await download_queue.__anext__() + download_queue_progress = click.style( + f"[Track {download_index}]", + dim=True, + ) + media_title = item.media.media_metadata["name"] + logger.info(download_queue_progress + f' Downloading "{media_title}"') + + await downloader.download(item) + + download_index += 1 + except StopAsyncIteration: + break + except VotifyUrlParseException as e: + logger.error(url_progress + f" {str(e)}") + break + except VotifyMediaException as e: + media_title = ( + e.media_metadata["name"] if e.media_metadata else "Unknown Title" + ) + logger.warning( + download_queue_progress + f' Skipping "{media_title}": {str(e)}' + ) + except VotifyDownloaderException as e: + media_title = ( + item.media.media_metadata["name"] if item else "Unknown Title" + ) + logger.warning( + download_queue_progress + f' Skipping "{media_title}": {str(e)}' + ) + except Exception: + error_count += 1 + media_title = ( + item.media.media_metadata["name"] if item else "Unknown Title" + ) + logger.error( + download_queue_progress + f' Error downloading "{media_title}"', + exc_info=not config.no_exceptions, + ) + + logger.info(f"Finished with {error_count} error(s)") diff --git a/votify/cli/cli_config.py b/votify/cli/cli_config.py new file mode 100644 index 0000000..cc67f28 --- /dev/null +++ b/votify/cli/cli_config.py @@ -0,0 +1,422 @@ +import inspect +from dataclasses import dataclass +from pathlib import Path +from typing import Annotated + +import click +from dataclass_click import argument, option + +from ..api.api import SpotifyApi +from ..downloader.audio import SpotifyAudioDownloader +from ..downloader.base import SpotifyBaseDownloader +from ..downloader.downloader import SpotifyDownloader +from ..downloader.enums import AudioDownloadMode, AudioRemuxMode, VideoRemuxMode +from ..downloader.video import SpotifyVideoDownloader +from ..interface.audio import SpotifyAudioInterface +from ..interface.base import SpotifyBaseInterface +from ..interface.enums import AudioQuality, CoverSize, VideoFormat, VideoResolution +from ..interface.video import SpotifyVideoInterface +from .utils import Csv + +api_from_cookies_sig = inspect.signature(SpotifyApi.create_from_netscape_cookies) + +base_interface_sig = inspect.signature(SpotifyBaseInterface.__init__) +audio_interface_sig = inspect.signature(SpotifyAudioInterface.__init__) +video_interface_sig = inspect.signature(SpotifyVideoInterface.__init__) + +base_downloader_sig = inspect.signature(SpotifyBaseDownloader.__init__) +video_downloader_sig = inspect.signature(SpotifyVideoDownloader.__init__) +audio_downloader_sig = inspect.signature(SpotifyAudioDownloader.__init__) +downloader_sig = inspect.signature(SpotifyDownloader.__init__) + + +@dataclass +class CliConfig: + # CLI specific options + urls: Annotated[ + list[str], + argument( + nargs=-1, + type=str, + required=True, + ), + ] + read_urls_as_txt: Annotated[ + bool, + option( + "--read-urls-as-txt", + "-r", + help="Read URLs from text files", + is_flag=True, + ), + ] + config_path: Annotated[ + str, + option( + "--config-path", + help="Config file path", + default=str(Path.home() / ".votify" / "config.ini"), + type=click.Path( + file_okay=True, + dir_okay=False, + writable=True, + resolve_path=True, + ), + ), + ] + log_level: Annotated[ + str, + option( + "--log-level", + help="Logging level", + default="INFO", + type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"]), + ), + ] + log_file: Annotated[ + str, + option( + "--log-file", + help="Log file path", + default=None, + type=click.Path( + file_okay=True, + dir_okay=False, + writable=True, + resolve_path=True, + ), + ), + ] + no_exceptions: Annotated[ + bool, + option( + "--no-exceptions", + help="Don't print exceptions", + is_flag=True, + ), + ] + # API specific options + cookies_path: Annotated[ + str, + option( + "--cookies-path", + "-c", + help="Cookies file path", + default=api_from_cookies_sig.parameters["cookies_path"].default, + type=click.Path( + file_okay=True, + dir_okay=False, + readable=True, + resolve_path=True, + ), + ), + ] + # Base Interface specific options + cover_size: Annotated[ + CoverSize, + option( + "--cover-size", + help="Cover size to use", + default=base_interface_sig.parameters["cover_size"].default, + type=CoverSize, + ), + ] + prefer_video: Annotated[ + bool, + option( + "--prefer-video", + help="Prefer video streams when available", + is_flag=True, + ), + ] + no_drm: Annotated[ + bool, + option( + "--no-drm", + help="Don't allow DRM-protected media", + is_flag=True, + ), + ] + wvd_path: Annotated[ + str, + option( + "--wvd-path", + help=".wvd file path", + default=base_interface_sig.parameters["wvd_path"].default, + type=click.Path( + file_okay=True, + dir_okay=False, + readable=True, + resolve_path=True, + ), + ), + ] + # Audio Interface specific options + audio_quality: Annotated[ + AudioQuality, + option( + "--audio-quality", + help="Audio quality to use", + default=audio_interface_sig.parameters["audio_quality"].default, + type=AudioQuality, + ), + ] + # Video Interface specific options + video_format: Annotated[ + VideoFormat, + option( + "--video-format", + help="Video format to use", + default=video_interface_sig.parameters["video_format"].default, + type=VideoFormat, + ), + ] + video_resolution: Annotated[ + VideoResolution, + option( + "--video-resolution", + help="Video resolution to use", + default=video_interface_sig.parameters["resolution"].default, + type=VideoResolution, + ), + ] + # Base Downloader specific options + output: Annotated[ + str, + option( + "--output", + "-o", + help="Output directory path", + default=base_downloader_sig.parameters["output_path"].default, + type=click.Path( + file_okay=True, + dir_okay=True, + writable=True, + resolve_path=True, + ), + ), + ] + temp: Annotated[ + str, + option( + "--temp", + help="Temporary directory path", + default=base_downloader_sig.parameters["temp_path"].default, + type=click.Path( + file_okay=True, + dir_okay=True, + writable=True, + resolve_path=True, + ), + ), + ] + aria2c_path: Annotated[ + str, + option( + "--aria2c-path", + help="aria2c executable path", + default=base_downloader_sig.parameters["aria2c_path"].default, + ), + ] + curl_path: Annotated[ + str, + option( + "--curl-path", + help="curl executable path", + default=base_downloader_sig.parameters["curl_path"].default, + ), + ] + ffmpeg_path: Annotated[ + str, + option( + "--ffmpeg-path", + help="ffmpeg executable path", + default=base_downloader_sig.parameters["ffmpeg_path"].default, + ), + ] + mp4box_path: Annotated[ + str, + option( + "--mp4box-path", + help="MP4Box executable path", + default=base_downloader_sig.parameters["mp4box_path"].default, + ), + ] + mp4decrypt_path: Annotated[ + str, + option( + "--mp4decrypt-path", + help="mp4decrypt executable path", + default=base_downloader_sig.parameters["mp4decrypt_path"].default, + ), + ] + shaka_packager_path: Annotated[ + str, + option( + "--shaka-packager-path", + help="Shaka Packager executable path", + default=base_downloader_sig.parameters["shaka_packager_path"].default, + ), + ] + album_folder_template: Annotated[ + str, + option( + "--album-folder-template", + help="Album folder template", + default=base_downloader_sig.parameters["album_folder_template"].default, + ), + ] + compilation_folder_template: Annotated[ + str, + option( + "--compilation-folder-template", + help="Compilation folder template", + default=base_downloader_sig.parameters[ + "compilation_folder_template" + ].default, + ), + ] + no_album_folder_template: Annotated[ + str, + option( + "--no-album-folder-template", + help="No album folder template", + default=base_downloader_sig.parameters["no_album_folder_template"].default, + ), + ] + single_disc_file_template: Annotated[ + str, + option( + "--single-disc-file-template", + help="Single disc file template", + default=base_downloader_sig.parameters["single_disc_file_template"].default, + ), + ] + multi_disc_file_template: Annotated[ + str, + option( + "--multi-disc-file-template", + help="Multi disc file template", + default=base_downloader_sig.parameters["multi_disc_file_template"].default, + ), + ] + no_album_file_template: Annotated[ + str, + option( + "--no-album-file-template", + help="No album file template", + default=base_downloader_sig.parameters["no_album_file_template"].default, + ), + ] + playlist_file_template: Annotated[ + str, + option( + "--playlist-file-template", + help="Playlist file template", + default=base_downloader_sig.parameters["playlist_file_template"].default, + ), + ] + date_tag_template: Annotated[ + str, + option( + "--date-tag-template", + help="Date tag template", + default=base_downloader_sig.parameters["date_tag_template"].default, + ), + ] + exclude_tags: Annotated[ + list[str], + option( + "--exclude-tags", + help="Comma-separated tags to exclude", + default=base_downloader_sig.parameters["exclude_tags"].default, + type=Csv(str), + ), + ] + truncate: Annotated[ + int | None, + option( + "--truncate", + help="Truncate file and folder names to the specified length", + default=base_downloader_sig.parameters["truncate"].default, + ), + ] + # Video Downloader specific options + video_remux_mode: Annotated[ + VideoRemuxMode, + option( + "--video-remux-mode", + help="Video remux mode to use", + default=video_downloader_sig.parameters["remux_mode"].default, + type=VideoRemuxMode, + ), + ] + # Audio Downloader specific options + audio_download_mode: Annotated[ + AudioDownloadMode, + option( + "--audio-download-mode", + help="Audio download mode to use", + default=audio_downloader_sig.parameters["download_mode"].default, + type=AudioDownloadMode, + ), + ] + audio_remux_mode: Annotated[ + AudioRemuxMode, + option( + "--audio-remux-mode", + help="Audio remux mode to use", + default=audio_downloader_sig.parameters["remux_mode"].default, + type=AudioRemuxMode, + ), + ] + # Downloader specific options + no_synced_lyrics_file: Annotated[ + bool, + option( + "--no-synced-lyrics-file", + help="Don't create synced lyrics file", + is_flag=True, + ), + ] + save_playlist_file: Annotated[ + bool, + option( + "--save-playlist-file", + help="Save playlist file", + is_flag=True, + ), + ] + save_cover_file: Annotated[ + bool, + option( + "--save-cover-file", + help="Save cover file", + is_flag=True, + ), + ] + overwrite: Annotated[ + bool, + option( + "--overwrite", + help="Overwrite existing media files", + is_flag=True, + ), + ] + synced_lyrics_only: Annotated[ + bool, + option( + "--synced-lyrics-only", + help="Only download synced lyrics file", + is_flag=True, + ), + ] + no_config_file: Annotated[ + bool, + option( + "--no-config-file", + "-n", + help="Don't use a config file", + is_flag=True, + ), + ] diff --git a/votify/cli/config_file.py b/votify/cli/config_file.py new file mode 100644 index 0000000..c5e0c62 --- /dev/null +++ b/votify/cli/config_file.py @@ -0,0 +1,167 @@ +import configparser +import typing +from functools import wraps +from pathlib import Path + +import click +import click.types as click_types + +from .cli_config import CliConfig +from .constants import EXCLUDED_CONFIG_FILE_PARAMS +from .utils import Csv + + +class ConfigFile: + def __init__( + self, + config_path: str, + section_name: str = "votify", + ) -> None: + self.config_path = config_path + self.section_name = section_name + + self.click_context = click.get_current_context() + self._read_config_file() + + def _read_config_file(self) -> None: + self.config = configparser.ConfigParser(interpolation=None) + + if Path(self.config_path).exists(): + self.config.read(self.config_path, encoding="utf-8") + else: + Path(self.config_path).parent.mkdir(parents=True, exist_ok=True) + + if not self.config.has_section(self.section_name): + self.config.add_section(self.section_name) + + def _write_config_file(self) -> None: + with open(self.config_path, "w", encoding="utf-8") as config_file: + self.config.write(config_file) + + def _serialize_param_default(self, param: click.Parameter) -> str: + if param.default is None: + return "null" + + if isinstance(param.type, Csv): + return ",".join( + item.value if hasattr(item, "value") else str(item) + for item in param.default + ) + + if isinstance(param.type, click_types.FuncParamType): + return param.default.value + + if isinstance(param.type, click_types.BoolParamType): + return "true" if param.default else "false" + + if isinstance( + param.type, + click_types.Choice + | click_types.Path + | click_types.StringParamType + | click_types.IntParamType, + ): + return str(param.default) + + raise NotImplementedError( + f"Serialization for parameter '{param.name}' of type " + f"'{type(param.type)}' is not implemented." + ) + + def _add_param_default_to_config( + self, + param: click.Parameter, + ) -> bool: + if self.config.has_option(self.section_name, param.name): + return False + + value = self._serialize_param_default(param) + self.config.set(self.section_name, param.name, value) + + return True + + def _parse_param_from_config( + self, + param: click.Parameter, + ) -> typing.Any: + value = self.config[self.section_name].get(param.name) + if value is None: + return param.default + + if value == "null": + return None + + if not isinstance(param.type, click_types.ParamType): + raise NotImplementedError( + f"Parsing for parameter '{param.name}' of type " + f"'{type(param.type)}' is not implemented." + ) + + return param.type.convert(value, None, None) + + def add_params_default_to_config(self) -> None: + has_changes = False + + for param in self.click_context.command.params: + if param.name in EXCLUDED_CONFIG_FILE_PARAMS: + continue + + has_changes = self._add_param_default_to_config(param) or has_changes + + if has_changes: + self._write_config_file() + + def cleanup_unknown_params(self) -> None: + param_names = {info.name for info in self.click_context.command.params} + has_changes = False + + for key in list(self.config[self.section_name].keys()): + if key not in param_names: + self.config.remove_option(self.section_name, key) + has_changes = True + + if has_changes: + self._write_config_file() + + def update_params_from_config(self) -> None: + for param in self.click_context.command.params: + if ( + self.click_context.get_parameter_source(param.name) + == click.core.ParameterSource.COMMANDLINE + ): + continue + + if self.config.has_option(self.section_name, param.name): + self.click_context.params[param.name] = self._parse_param_from_config( + param + ) + + def get_cli_config(self) -> CliConfig: + config_dict = {} + for param in self.click_context.command.params: + if param.name in {"help", "version"}: + continue + + config_dict[param.name] = self.click_context.params.get( + param.name, param.default + ) + return CliConfig(**config_dict) + + def load(self) -> CliConfig: + self.cleanup_unknown_params() + self.add_params_default_to_config() + self.update_params_from_config() + return self.get_cli_config() + + @staticmethod + def loader(func): + @wraps(func) + def wrapper(cli_config: CliConfig): + ctx = click.get_current_context() + config_path = ctx.params.get("config_path") + no_config_file = ctx.params.get("no_config_file") + if config_path and not no_config_file: + cli_config = ConfigFile(config_path).load() + return func(cli_config) + + return wrapper diff --git a/votify/cli/constants.py b/votify/cli/constants.py new file mode 100644 index 0000000..a07301d --- /dev/null +++ b/votify/cli/constants.py @@ -0,0 +1,9 @@ +EXCLUDED_CONFIG_FILE_PARAMS = { + "urls", + "config_path", + "read_urls_as_txt", + "no_config_file", + "version", + "help", +} +X_NOT_IN_PATH = '{} was not found in PATH at "{}"' diff --git a/votify/cli/utils.py b/votify/cli/utils.py new file mode 100644 index 0000000..89131cd --- /dev/null +++ b/votify/cli/utils.py @@ -0,0 +1,96 @@ +import logging +from enum import Enum +from pathlib import Path + +import click + + +class Csv(click.ParamType): + name = "csv" + + def __init__( + self, + subtype: Enum, + ) -> None: + self.subtype = subtype + + def convert( + self, + value: str, + param: click.Parameter, + ctx: click.Context, + ) -> list[Enum]: + if not isinstance(value, str): + return value + + items = [v.strip() for v in value.split(",") if v.strip()] + result = [] + + for item in items: + try: + result.append(self.subtype(item)) + except ValueError as e: + self.fail( + f"'{item}' is not a valid value for {self.subtype.__name__}", + param, + ctx, + ) + return result + + +class CustomLoggerFormatter(logging.Formatter): + base_format = "[%(levelname)-8s %(asctime)s]" + format_colors = { + logging.DEBUG: dict(dim=True), + logging.INFO: dict(fg="green"), + logging.WARNING: dict(fg="yellow"), + logging.ERROR: dict(fg="red"), + logging.CRITICAL: dict(fg="red", bold=True), + } + date_format = "%H:%M:%S" + + def __init__(self, use_colors: bool = True) -> None: + super().__init__() + self.use_colors = use_colors + + def format(self, record: logging.LogRecord) -> str: + return logging.Formatter( + ( + click.style(self.base_format, **self.format_colors.get(record.levelno)) + if self.use_colors + else self.base_format + ) + + " %(message)s", + datefmt=self.date_format, + ).format(record) + + +def prompt_path( + input_path: str, + is_dir: bool = False, +) -> str: + path_validator = click.Path( + exists=True, + file_okay=not is_dir, + dir_okay=is_dir, + ) + path_type = "directory" if is_dir else "file" + + while True: + try: + result_path = path_validator.convert(input_path, None, None) + break + except click.BadParameter as e: + input_path = click.prompt( + ( + f'{path_type.capitalize()} "{Path(input_path).absolute()}" does not exist. ' + f"Create the {path_type} at the specified path, " + f"type a new path or drag and drop the {path_type} here. " + "Then, press enter to continue" + ), + default=input_path, + show_default=False, + ) + input_path = input_path.strip('"') + + return result_path