mirror of
https://github.com/glomatico/votify.git
synced 2026-10-04 15:02:17 +02:00
Add CLI entrypoint and config management
This commit is contained in:
Rafael Moraes
@@ -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)")
|
||||
@@ -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,
|
||||
),
|
||||
]
|
||||
@@ -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
|
||||
@@ -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 "{}"'
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user