refactor
This commit is contained in:
Hellow2
@@ -3,8 +3,3 @@ from .musify import Musify
|
||||
|
||||
EncyclopaediaMetallum = EncyclopaediaMetallum
|
||||
Musify = Musify
|
||||
|
||||
from . import download_center
|
||||
|
||||
Search = download_center.Search
|
||||
|
||||
|
||||
@@ -1,5 +0,0 @@
|
||||
from . import search
|
||||
from . import download
|
||||
|
||||
Download = download.Download
|
||||
Search = search.Search
|
||||
@@ -1,45 +0,0 @@
|
||||
from typing import Optional, Tuple, Type, Set, Union, List
|
||||
|
||||
from . import page_attributes
|
||||
from ..abstract import Page
|
||||
from ...objects import Song, Album, Artist, Label, Source
|
||||
|
||||
MusicObject = Union[Song, Album, Artist, Label]
|
||||
|
||||
|
||||
class Download:
|
||||
def __init__(
|
||||
self,
|
||||
pages: Tuple[Type[Page]] = page_attributes.ALL_PAGES,
|
||||
exclude_pages: Set[Type[Page]] = set(),
|
||||
exclude_shady: bool = False,
|
||||
) -> None:
|
||||
_page_list: List[Type[Page]] = []
|
||||
_audio_page_list: List[Type[Page]] = []
|
||||
|
||||
for page in pages:
|
||||
if exclude_shady and page in page_attributes.SHADY_PAGES:
|
||||
continue
|
||||
if page in exclude_pages:
|
||||
continue
|
||||
|
||||
_page_list.append(page)
|
||||
|
||||
if page in page_attributes.AUDIO_PAGES:
|
||||
_audio_page_list.append(page)
|
||||
|
||||
self.pages: Tuple[Type[Page]] = tuple(_page_list)
|
||||
self.audio_pages: Tuple[Type[Page]] = tuple(_audio_page_list)
|
||||
|
||||
def fetch_details(self, music_object: MusicObject) -> MusicObject:
|
||||
for page in self.pages:
|
||||
page.fetch_details(music_object=music_object)
|
||||
return music_object
|
||||
|
||||
def fetch_source(self, source: Source) -> Optional[MusicObject]:
|
||||
source_page = page_attributes.SOURCE_PAGE_MAP[source.page_enum]
|
||||
|
||||
if source_page not in self.pages:
|
||||
return
|
||||
|
||||
return source_page.fetch_object_from_source(source)
|
||||
@@ -1,100 +0,0 @@
|
||||
from collections import defaultdict
|
||||
from typing import Tuple, List, Dict, Type
|
||||
|
||||
from . import page_attributes
|
||||
from ..abstract import Page
|
||||
from ...objects import Options, DatabaseObject, Source
|
||||
|
||||
|
||||
class MultiPageOptions:
|
||||
def __init__(
|
||||
self,
|
||||
max_displayed_options: int = 10,
|
||||
option_digits: int = 3,
|
||||
derived_from: DatabaseObject = None
|
||||
) -> None:
|
||||
self.max_displayed_options = max_displayed_options
|
||||
self.option_digits: int = option_digits
|
||||
|
||||
self._length = 0
|
||||
self._current_option_dict: Dict[Type[Page], Options] = defaultdict(lambda: Options())
|
||||
|
||||
self._derive_from = derived_from
|
||||
|
||||
def __getitem__(self, key: Type[Page]):
|
||||
return self._current_option_dict[key]
|
||||
|
||||
def __setitem__(self, key: Type[Page], value: Options):
|
||||
self._current_option_dict[key] = value
|
||||
|
||||
self._length = 0
|
||||
for key in self._current_option_dict:
|
||||
self._length += 1
|
||||
|
||||
def __len__(self) -> int:
|
||||
return self._length
|
||||
|
||||
def get_page_str(self, page: Type[Page]) -> str:
|
||||
page_name_fill = "-"
|
||||
max_page_len = 21
|
||||
|
||||
return f"({page_attributes.PAGE_NAME_MAP[page]}) ------------------------{page.__name__:{page_name_fill}<{max_page_len}}------------"
|
||||
|
||||
def string_from_all_pages(self) -> str:
|
||||
if self._length == 1:
|
||||
for key in self._current_option_dict:
|
||||
return self.string_from_single_page(key)
|
||||
|
||||
lines: List[str] = []
|
||||
|
||||
j = 0
|
||||
for page, options in self._current_option_dict.items():
|
||||
lines.append(self.get_page_str(page))
|
||||
|
||||
i = -1
|
||||
|
||||
option_obj: DatabaseObject
|
||||
for i, option_obj in enumerate(options):
|
||||
if i >= self.max_displayed_options:
|
||||
lines.append("...")
|
||||
break
|
||||
|
||||
lines.append(f"{j + i:0{self.option_digits}} {option_obj.option_string}")
|
||||
|
||||
j += i + 1
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
def choose_from_all_pages(self, index: int) -> Tuple[DatabaseObject, Type[Page]]:
|
||||
if self._length == 1:
|
||||
for key in self._current_option_dict:
|
||||
return self.choose_from_single_page(key, index), key
|
||||
|
||||
sum_of_length = 0
|
||||
for page, options in self._current_option_dict.items():
|
||||
option_len = min((len(options), self.max_displayed_options))
|
||||
|
||||
index_of_list = index - sum_of_length
|
||||
|
||||
if index_of_list < option_len:
|
||||
return options[index_of_list], page
|
||||
|
||||
sum_of_length += option_len
|
||||
|
||||
raise IndexError("index is out of range")
|
||||
|
||||
def string_from_single_page(self, page: Type[Page]) -> str:
|
||||
lines: List[str] = [self.get_page_str(page)]
|
||||
|
||||
option_obj: DatabaseObject
|
||||
for i, option_obj in enumerate(self._current_option_dict[page]):
|
||||
lines.append(f"{i:0{self.option_digits}} {option_obj.option_string}")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
def choose_from_single_page(self, page: Type[Page], index: int) -> DatabaseObject:
|
||||
return self._current_option_dict[page][index]
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return self.string_from_all_pages()
|
||||
|
||||
@@ -1,34 +0,0 @@
|
||||
from typing import Tuple, Type, Dict
|
||||
|
||||
from ...utils.enums.source import SourcePages
|
||||
from ..abstract import Page
|
||||
from ..encyclopaedia_metallum import EncyclopaediaMetallum
|
||||
from ..musify import Musify
|
||||
|
||||
NAME_PAGE_MAP: Dict[str, Type[Page]] = dict()
|
||||
PAGE_NAME_MAP: Dict[Type[Page], str] = dict()
|
||||
SOURCE_PAGE_MAP: Dict[SourcePages, Type[Page]] = dict()
|
||||
|
||||
ALL_PAGES: Tuple[Type[Page]] = (
|
||||
EncyclopaediaMetallum,
|
||||
Musify
|
||||
)
|
||||
|
||||
AUDIO_PAGES: Tuple[Type[Page]] = (
|
||||
Musify,
|
||||
)
|
||||
|
||||
SHADY_PAGES: Tuple[Type[Page]] = (
|
||||
Musify,
|
||||
)
|
||||
|
||||
# this needs to be case insensitive
|
||||
SHORTHANDS = ('a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z')
|
||||
for i, page in enumerate(ALL_PAGES):
|
||||
NAME_PAGE_MAP[page.__name__.lower()] = page
|
||||
NAME_PAGE_MAP[SHORTHANDS[i].lower()] = page
|
||||
|
||||
PAGE_NAME_MAP[page] = SHORTHANDS[i]
|
||||
|
||||
SOURCE_PAGE_MAP[page.SOURCE_TYPE] = page
|
||||
|
||||
@@ -1,146 +0,0 @@
|
||||
from typing import Tuple, List, Set, Type, Optional
|
||||
|
||||
from . import page_attributes
|
||||
from .download import Download
|
||||
from .multiple_options import MultiPageOptions
|
||||
from ..abstract import Page
|
||||
from ..support_classes.download_result import DownloadResult
|
||||
from ...objects import DatabaseObject, Source
|
||||
from ...utils.enums.source import SourcePages
|
||||
|
||||
|
||||
class Search(Download):
|
||||
def __init__(
|
||||
self,
|
||||
pages: Tuple[Type[Page]] = page_attributes.ALL_PAGES,
|
||||
exclude_pages: Set[Type[Page]] = set(),
|
||||
exclude_shady: bool = False,
|
||||
max_displayed_options: int = 10,
|
||||
option_digits: int = 3,
|
||||
) -> None:
|
||||
super().__init__(
|
||||
pages=pages,
|
||||
exclude_pages=exclude_pages,
|
||||
exclude_shady=exclude_shady
|
||||
)
|
||||
|
||||
self.max_displayed_options = max_displayed_options
|
||||
self.option_digits: int = option_digits
|
||||
|
||||
self._option_history: List[MultiPageOptions] = []
|
||||
|
||||
self._current_option: MultiPageOptions = self.next_options()
|
||||
|
||||
|
||||
def __repr__(self):
|
||||
return self._current_option.__repr__()
|
||||
|
||||
def next_options(self, derive_from: DatabaseObject = None) -> MultiPageOptions:
|
||||
mpo = MultiPageOptions(
|
||||
max_displayed_options=self.max_displayed_options,
|
||||
option_digits=self.option_digits,
|
||||
derived_from=derive_from
|
||||
)
|
||||
|
||||
self._option_history.append(mpo)
|
||||
self._current_option = mpo
|
||||
|
||||
return mpo
|
||||
|
||||
def _previous_options(self) -> MultiPageOptions:
|
||||
self._option_history.pop()
|
||||
self._current_option = self._option_history[-1]
|
||||
|
||||
return self._option_history[-1]
|
||||
|
||||
def search(self, query: str):
|
||||
"""
|
||||
# The Query
|
||||
|
||||
You can define a new parameter with "#",
|
||||
the letter behind it defines the *type* of parameter,
|
||||
followed by a space "#a Psychonaut 4 #r Tired, Numb and #t Drop by Drop"
|
||||
if no # is in the query it gets treated as "unspecified query"
|
||||
|
||||
doesn't set derived_from thus,
|
||||
can't download right after
|
||||
"""
|
||||
|
||||
for page in self.pages:
|
||||
self._current_option[page] = page._raw_search(query=query)
|
||||
|
||||
def choose_page(self, page: Type[Page]):
|
||||
"""
|
||||
doesn't set derived_from thus,
|
||||
can't download right after
|
||||
"""
|
||||
|
||||
if page not in page_attributes.ALL_PAGES:
|
||||
raise ValueError(f"Page \"{page.__name__}\" does not exist in page_attributes.ALL_PAGES")
|
||||
|
||||
prev_mpo = self._current_option
|
||||
mpo = self.next_options()
|
||||
|
||||
mpo[page] = prev_mpo[page]
|
||||
|
||||
def get_page_from_query(self, query: str) -> Optional[Type[Page]]:
|
||||
"""
|
||||
query can be for example:
|
||||
"a" or "EncyclopaediaMetallum" to choose a page
|
||||
"""
|
||||
|
||||
page = page_attributes.NAME_PAGE_MAP.get(query.lower().strip())
|
||||
|
||||
if page in self.pages:
|
||||
return page
|
||||
|
||||
def _get_page_from_source(self, source: Source) -> Optional[Type[Page]]:
|
||||
return page_attributes.SOURCE_PAGE_MAP.get(source.page_enum)
|
||||
|
||||
def choose_index(self, index: int):
|
||||
db_object, page = self._current_option.choose_from_all_pages(index=index)
|
||||
|
||||
music_object = self.fetch_details(db_object)
|
||||
mpo = self.next_options(derive_from=music_object)
|
||||
|
||||
mpo[page] = music_object.options
|
||||
|
||||
def goto_previous(self):
|
||||
try:
|
||||
self._previous_options()
|
||||
except IndexError:
|
||||
pass
|
||||
|
||||
def search_url(self, url: str) -> bool:
|
||||
"""
|
||||
sets derived_from, thus
|
||||
can download directly after
|
||||
"""
|
||||
|
||||
source = Source.match_url(url=url, referer_page=SourcePages.MANUAL)
|
||||
if source is None:
|
||||
return False
|
||||
|
||||
new_object = self.fetch_source(source)
|
||||
if new_object is None:
|
||||
return False
|
||||
|
||||
page = page_attributes.SOURCE_PAGE_MAP[source.page_enum]
|
||||
mpo = self.next_options(derive_from=new_object)
|
||||
mpo[page] = new_object.options
|
||||
|
||||
return True
|
||||
|
||||
def download_chosen(self, genre: str = None, download_all: bool = False, **kwargs) -> DownloadResult:
|
||||
if self._current_option._derive_from is None:
|
||||
return DownloadResult(error_message="No option has been chosen yet.")
|
||||
|
||||
source: Source
|
||||
for source in self._current_option._derive_from.source_collection:
|
||||
page = self._get_page_from_source(source=source)
|
||||
|
||||
if page in self.audio_pages:
|
||||
return page.download(music_object=self._current_option._derive_from, genre=genre, download_all=download_all, **kwargs)
|
||||
|
||||
return DownloadResult(error_message=f"Didn't find a source for {self._current_option._derive_from.option_string}.")
|
||||
|
||||
Reference in New Issue
Block a user