From df1743c695a5786c04483fe770c7439d1c45fe9d Mon Sep 17 00:00:00 2001 From: Lars Noack Date: Mon, 3 Jun 2024 15:04:47 +0200 Subject: [PATCH] feat: implemented better components --- music_kraken/__init__.py | 6 +- music_kraken/cli/genre.py | 45 +++ music_kraken/cli/main_downloader.py | 24 +- music_kraken/download/components.py | 303 ------------------- music_kraken/download/components/__init__.py | 158 ++++++++++ music_kraken/download/components/generic.py | 1 + music_kraken/objects/song.py | 41 ++- 7 files changed, 227 insertions(+), 351 deletions(-) create mode 100644 music_kraken/cli/genre.py delete mode 100644 music_kraken/download/components.py create mode 100644 music_kraken/download/components/__init__.py create mode 100644 music_kraken/download/components/generic.py diff --git a/music_kraken/__init__.py b/music_kraken/__init__.py index 5176b38..e9dbc41 100644 --- a/music_kraken/__init__.py +++ b/music_kraken/__init__.py @@ -1,13 +1,13 @@ -import logging import gc +import logging import sys from pathlib import Path -from rich.logging import RichHandler from rich.console import Console +from rich.logging import RichHandler -from .utils.shared import DEBUG, DEBUG_LOGGING from .utils.config import logging_settings, main_settings, read_config +from .utils.shared import DEBUG, DEBUG_LOGGING read_config() diff --git a/music_kraken/cli/genre.py b/music_kraken/cli/genre.py new file mode 100644 index 0000000..78b67ef --- /dev/null +++ b/music_kraken/cli/genre.py @@ -0,0 +1,45 @@ +class GenreIO(components.HumanIO): + @staticmethod + def ask_to_create(option: components.Option) -> bool: + output() + return ask_for_bool(f"create the genre {BColors.OKBLUE.value}{option.value}{BColors.ENDC.value}") + + @staticmethod + def not_found(key: str) -> None: + output(f"\ngenre {BColors.BOLD.value}{key}{BColors.ENDC.value} not found\n", color=BColors.FAIL) + + +def _genre_generator() -> Generator[str, None, None]: + def is_valid_genre(genre: Path) -> bool: + """ + gets the name of all subdirectories of shared.MUSIC_DIR, + but filters out all directories, where the name matches with any Patern + from shared.NOT_A_GENRE_REGEX. + """ + if not genre.is_dir(): + return False + + if any(re.match(regex_pattern, genre.name) for regex_pattern in main_settings["not_a_genre_regex"]): + return False + + return True + + for genre in filter(is_valid_genre, main_settings["music_directory"].iterdir()): + yield genre.name + +def get_genre() -> str: + select_genre = components.Select( + human_io=GenreIO, + can_create_options=True, + data=_genre_generator(), + ) + genre: Optional[components.Option[str]] = None + + while genre is None: + print(select_genre.pprint()) + print() + + genre = select_genre.choose(input("> ")) + + return genre.value + \ No newline at end of file diff --git a/music_kraken/cli/main_downloader.py b/music_kraken/cli/main_downloader.py index 12d076e..eb01c1a 100644 --- a/music_kraken/cli/main_downloader.py +++ b/music_kraken/cli/main_downloader.py @@ -18,6 +18,7 @@ from ..utils.shared import HELP_MESSAGE, URL_PATTERN from ..utils.string_processing import fit_to_file_system from ..utils.support_classes.download_result import DownloadResult from ..utils.support_classes.query import Query +from .genre import get_genre from .options.first_config import initial_config from .utils import ask_for_bool, cli_function @@ -27,30 +28,7 @@ PAGE_NAME_FILL = "-" MAX_PAGE_LEN = 21 -class GenreIO(components.HumanIO): - @staticmethod - def ask_to_create(option: components.Option) -> bool: - output() - return ask_for_bool(f"create the genre {BColors.OKBLUE.value}{option.value}{BColors.ENDC.value}") - @staticmethod - def not_found(key: str) -> None: - output(f"\ngenre {BColors.BOLD.value}{key}{BColors.ENDC.value} not found\n", color=BColors.FAIL) - - -def get_genre(): - select_genre = components.GenreSelect() - select_genre.human_io = GenreIO - - genre: Optional[components.Option] = None - - while genre is None: - print(select_genre.pprint()) - print() - - genre = select_genre.choose(input("> ")) - - return genre.value def help_message(): diff --git a/music_kraken/download/components.py b/music_kraken/download/components.py deleted file mode 100644 index 413f794..0000000 --- a/music_kraken/download/components.py +++ /dev/null @@ -1,303 +0,0 @@ -from __future__ import annotations - -import re -from collections import defaultdict -from pathlib import Path -from typing import Any, Callable, Dict, Generator, List, Optional - -from ..objects import OuterProxy as DataObject -from ..utils import BColors -from ..utils.config import main_settings -from ..utils.enums import SourceType -from ..utils.exception import MKComposeException -from ..utils.shared import ALPHABET -from ..utils.string_processing import unify - - -class HumanIO: - @staticmethod - def ask_to_create(option: Option) -> bool: - return True - - @staticmethod - def not_found(key: Any) -> None: - return None - - -class Option: - """ - This could represent a data object, a string or a page. - """ - - def __init__( - self, - value: Any, - text: Optional[str] = None, - keys: List[Any] = None, - hidden: bool = False, - parse_key: Callable[[Any], Any] = lambda x: x, - index: int = None, - ): - self._parse_key: Callable[[Any], Any] = parse_key - - self._index = index - self.value = value - self._text = text or str(value) - self.hidden = hidden - - self._raw_keys = set(keys or []) - self._raw_keys.add(self.text) - try: - self._raw_keys.add(self.value) - except TypeError: - pass - self._raw_keys.add(str(self.value)) - self._raw_keys.add(self._index) - self.keys = set(self.parse_key(key) for key in self._raw_keys) - - @property - def text(self) -> str: - return self._text.replace("{index}", str(self.index)) - - @text.setter - def text(self, value: str): - self._text = value - - @property - def index(self) -> int: - return self._index - - @index.setter - def index(self, value: int): - p = self._parse_key(self._index) - if p in self.keys: - self.keys.remove(p) - self._index = value - self.keys.add(p) - - def register_key(self, key: Any): - self._raw_keys.add(key) - self.keys.add(self._parse_key(key)) - - @property - def parse_key(self) -> Callable[[Any], Any]: - return self._parse_key - - @parse_key.setter - def parse_key(self, value: Callable[[Any], Any]): - self._parse_key = value - self.keys = set(self._parse_key(key) for key in self._raw_keys) - - def __str__(self): - return self.text - - -class Select: - def __init__( - self, - options: Generator[Option, None, None] = None, - option_factory: Callable[[Any], Option] = None, - raw_options: List[Any] = None, - parse_option_key: Callable[[Any], Any] = lambda x: x, - human_io: HumanIO = HumanIO, - sort: bool = False, - **kwargs - ): - self._parse_option_key: Callable[[Any], Any] = parse_option_key - self.human_io: HumanIO = human_io - - self._key_to_option: Dict[Any, Option] = dict() - self._options: List[Option] = [] - - options = options or [] - self.option_factory: Optional[Callable[[Any], Option]] = option_factory - if self.can_create_options: - _raw_options = raw_options or [] - if sort: - _raw_options = sorted(_raw_options) - - for raw_option in _raw_options: - self.append(self.option_factory(raw_option)) - elif raw_options is not None: - raise MKComposeException("Cannot create options without a factory.") - - self.extend(options) - - @property - def can_create_options(self) -> bool: - return self.option_factory is not None - - def append(self, option: Option): - option.parse_key = self._parse_option_key - self._options.append(option) - for key in option.keys: - self._key_to_option[key] = option - - def _remap(self): - self._key_to_option = dict() - for option in self._options: - for key in option.keys: - self._key_to_option[key] = option - - def extend(self, options: List[Option]): - for option in options: - self.append(option) - - def __iter__(self) -> Generator[Option, None, None]: - for option in self._options: - if option.hidden: - continue - - yield option - - def __contains__(self, key: Any) -> bool: - return self._parse_option_key(key) in self._key_to_option - - def __getitem__(self, key: Any) -> Option: - r = self._key_to_option[self._parse_option_key(key)] - if callable(r): - r = r() - if callable(r.value): - r.value = r.value() - return r - - def create_option(self, key: Any, **kwargs) -> Option: - if not self.can_create_options: - raise MKComposeException("Cannot create options without a factory.") - - option = self.option_factory(key, **kwargs) - self.append(option) - return option - - def choose(self, key: Any) -> Optional[Option]: - if key not in self: - if self.can_create_options: - c = self.create_option(key) - if self.human_io.ask_to_create(c): - return c - - self.human_io.not_found(key) - return None - - return self[key] - - def pprint(self) -> str: - return "\n".join(str(option) for option in self) - - - -class StringSelect(Select): - def __init__(self, **kwargs): - self._current_index = 0 - kwargs["option_factory"] = self.next_option - kwargs["parse_option_key"] = lambda x: unify(str(x)) - - super().__init__(**kwargs) - - def next_option(self, value: Any) -> Optional[Option]: - o = Option(value=value, keys=[self._current_index], text=f"{BColors.BOLD.value}{self._current_index: >2}{BColors.ENDC.value}: {value}") - self._current_index += 1 - return o - - -class GenreSelect(StringSelect): - @staticmethod - def is_valid_genre(genre: Path) -> bool: - """ - gets the name of all subdirectories of shared.MUSIC_DIR, - but filters out all directories, where the name matches with any Patern - from shared.NOT_A_GENRE_REGEX. - """ - if not genre.is_dir(): - return False - - if any(re.match(regex_pattern, genre.name) for regex_pattern in main_settings["not_a_genre_regex"]): - return False - - return True - - def __init__(self): - super().__init__(sort=True, raw_options=(genre.name for genre in filter(self.is_valid_genre, main_settings["music_directory"].iterdir()))) - - - -class SourceTypeToOption(dict): - def __init__(self, callback): - super().__init__() - - self.callback = callback - - def __missing__(self, key): - self[key] = self.callback(key) - return self[key] - - -class DataObjectSelect(Select): - def __init__(self, data_objects: Generator[DataObject]): - self._source_type_to_data_objects: Dict[SourceType, List[Option]] = defaultdict(list) - self._source_type_to_option: Dict[SourceType, Option] = SourceTypeToOption(self.option_from_source_type) - - self._data_object_index: int = 0 - self._source_type_index: int = 0 - - super().__init__( - parse_option_key=lambda x: unify(str(x)), - ) - - self.extend(data_objects) - - def option_from_data_object(self, data_object: DataObject) -> Option: - index = self._data_object_index - self._data_object_index += 1 - - return Option( - value=data_object, - keys=[index, data_object.option_string, data_object.title_string], - text=f"{BColors.BOLD.value}{{index}}{BColors.ENDC.value}: {data_object.option_string}", - index=index, - ) - - def option_from_source_type(self, source_type: SourceType) -> Option: - index = ALPHABET[self._source_type_index % len(ALPHABET)] - self._source_type_index += 1 - - o = Option( - value=lambda: DataObjectSelect(self._source_type_to_data_objects[source_type]), - keys=[index, source_type], - text=f"{BColors.HEADER.value}({index}) --------------------------------{source_type.name:{'-'}<{21}}--------------------{BColors.ENDC.value}", - ) - - super().append(o) - - return o - - def append(self, option: Union[Option, DataObject]): - if isinstance(option, DataObject): - data_object = option - option = self.option_from_data_object(data_object) - else: - data_object = option.value - - for source_type in data_object.source_collection.source_types(only_with_page=True): - self._source_type_to_data_objects[source_type].append(option) - - super().append(option) - - def __iter__(self): - source_types = list(sorted(self._source_type_to_data_objects.keys(), key=lambda x: x.name)) - single_source = len(source_types) > 1 - - j = 0 - for st in source_types: - if single_source: - yield self._source_type_to_option[st] - - limit = min(15, len(self._source_type_to_data_objects[st])) if single_source else len(self._source_type_to_data_objects[st]) - - for i in range(limit): - o = self._source_type_to_data_objects[st][i] - o.index = j - yield o - j += 1 - - self._remap() \ No newline at end of file diff --git a/music_kraken/download/components/__init__.py b/music_kraken/download/components/__init__.py new file mode 100644 index 0000000..19d5cda --- /dev/null +++ b/music_kraken/download/components/__init__.py @@ -0,0 +1,158 @@ +from __future__ import annotations + +import re +from collections import defaultdict +from pathlib import Path +from typing import (Any, Callable, Dict, Generator, Generic, Hashable, List, + Optional, Tuple, TypeVar, Union) + +from ...objects import OuterProxy as DataObject +from ...utils import BColors +from ...utils.config import main_settings +from ...utils.enums import SourceType +from ...utils.exception import MKComposeException +from ...utils.shared import ALPHABET +from ...utils.string_processing import unify + +P = TypeVar('P') + + +class HumanIO: + @staticmethod + def ask_to_create(option: Option) -> bool: + return True + + @staticmethod + def not_found(key: Any) -> None: + return None + + +class Option(Generic[P]): + """ + This could represent a data object, a string or a page. + """ + TEXT_TEMPLATE: str = f"{BColors.BOLD.value}{{index}}{BColors.ENDC.value}: {{value}}" + ATTRIBUTES_FORMATTING: Tuple[str, ...] = ("index", "value") + ATTRIBUTES_KEY: Tuple[str, ...] = ("index", ) + + def __init__( + self, + value: P, + hidden: bool = False, + additional_keys: List[Hashable] = None, + **kwargs + ): + self.value = value + self.hidden = hidden + self._additional_keys = set(self._to_hash(key) for key in additional_keys or []) + + for key, value in kwargs.items(): + setattr(self, key, value) + + def _to_option_string(self, value: Any) -> str: + if hasattr(value, "option_string"): + return value.option_string + + return str(value) + + @property + def text(self) -> str: + text = self.TEXT_TEMPLATE + + for attribute_key in self.ATTRIBUTES_FORMATTING: + text = text.replace(f"{{{attribute_key}}}", self._to_option_string(getattr(self, attribute_key))) + + return text + + def _to_hash(self, key: Any) -> int: + try: + key = int(key) + except ValueError: + pass + + if isinstance(key, str): + return hash(unify(key)) + + return hash(key) + + @property + def keys(self) -> set: + keys = self._additional_keys.copy() + + for key in self.ATTRIBUTES_KEY: + keys.add(self._to_hash(getattr(self, key))) + + def __contains__(self, key: Any) -> bool: + return self._to_hash(key) in self.keys + + def __str__(self): + return self.text + + +class Select(Generic[P]): + OPTION: Type[Option[P]] = Option + HUMAN_IO: Type[HumanIO] = HumanIO + CAN_CREATE_OPTIONS: bool = False + + def __init__( + self, + data: Generator[P, None, None], + **kwargs + ): + self.option: Type[Option[P]] = kwargs.get("option", self.OPTION) + self.human_io: Type[HumanIO] = kwargs.get("human_io", self.HUMAN_IO) + self.can_create_options: bool = kwargs.get("can_create_options", self.CAN_CREATE_OPTIONS) + + self._options: List[Option[P]] = [] + self.extend(data) + + def append(self, value: P) -> Option[P]: + option = self.option(value) + self._options.append(option) + return option + + def extend(self, values: Generator[P, None, None]): + for value in values: + self.append(value) + + @property + def _options_to_show(self) -> Generator[Option[P], None, None]: + for option in self._options: + if option.hidden: + continue + + yield option + + def __iter__(self) -> Generator[Option, None, None]: + _index = 0 + + for i, option in enumerate(self._options_to_show): + option.index = _index + yield option + _index += 1 + + def __contains__(self, key: Any) -> bool: + for option in self._options: + if key in option: + return True + + return False + + def __getitem__(self, key: Any) -> Option[P]: + for option in self._options: + if key in option: + return option + + raise KeyError(key) + + def choose(self, key: Any) -> Optional[Option[P]]: + try: + return self[key] + except KeyError: + if self.can_create_options: + return self.append(key) + + self.human_io.not_found(key) + + def pprint(self) -> str: + return "\n".join(str(option) for option in self) diff --git a/music_kraken/download/components/generic.py b/music_kraken/download/components/generic.py new file mode 100644 index 0000000..be88017 --- /dev/null +++ b/music_kraken/download/components/generic.py @@ -0,0 +1 @@ +from . import Option, Select diff --git a/music_kraken/objects/song.py b/music_kraken/objects/song.py index 980bc08..a25252e 100644 --- a/music_kraken/objects/song.py +++ b/music_kraken/objects/song.py @@ -1,35 +1,32 @@ from __future__ import annotations +import copy import random from collections import defaultdict -from typing import List, Optional, Dict, Tuple, Type, Union -import copy +from typing import Dict, List, Optional, Tuple, Type, Union import pycountry -from ..utils.enums.album import AlbumType, AlbumStatus -from .collection import Collection -from .formatted_text import FormattedText -from .lyrics import Lyrics -from .contact import Contact -from .artwork import Artwork -from .metadata import ( - Mapping as id3Mapping, - ID3Timestamp, - Metadata -) -from .option import Options -from .parents import OuterProxy, P -from .source import Source, SourceCollection -from .target import Target -from .country import Language, Country +from ..utils.config import main_settings +from ..utils.enums.album import AlbumStatus, AlbumType +from ..utils.enums.colors import BColors from ..utils.shared import DEBUG_PRINT_ID from ..utils.string_processing import unify - +from .artwork import Artwork +from .collection import Collection +from .contact import Contact +from .country import Country, Language +from .formatted_text import FormattedText +from .lyrics import Lyrics +from .metadata import ID3Timestamp +from .metadata import Mapping as id3Mapping +from .metadata import Metadata +from .option import Options +from .parents import OuterProxy from .parents import OuterProxy as Base - -from ..utils.config import main_settings -from ..utils.enums.colors import BColors +from .parents import P +from .source import Source, SourceCollection +from .target import Target """ All Objects dependent