Compare commits
9 Commits
130f5edcfe
...
feature/cl
| Author | SHA1 | Date | |
|---|---|---|---|
| e53e50b5d2 | |||
| 240bd105f0 | |||
| ed2eeabd6a | |||
| b236291378 | |||
| 097211b3cd | |||
| f839cdf906 | |||
| c306da7934 | |||
| 684c90a7b4 | |||
| b30824baf9 |
3
music_kraken/__meta__.py
Normal file
3
music_kraken/__meta__.py
Normal file
@@ -0,0 +1,3 @@
|
||||
PROGRAMM: str = "music-kraken"
|
||||
DESCRIPTION: str = """This program will first get the metadata of various songs from metadata providers like musicbrainz, and then search for download links on pages like bandcamp.
|
||||
Then it will download the song and edit the metadata accordingly."""
|
||||
@@ -1,5 +0,0 @@
|
||||
from .informations import print_paths
|
||||
from .main_downloader import download
|
||||
from .options.settings import settings
|
||||
from .options.frontend import set_frontend
|
||||
|
||||
@@ -1,53 +0,0 @@
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Dict, Generator, List, Set, Type, Union
|
||||
|
||||
from ..download import Downloader, Page, components
|
||||
from ..utils.config import main_settings
|
||||
from .utils import ask_for_bool, cli_function
|
||||
|
||||
|
||||
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
|
||||
@@ -1,47 +0,0 @@
|
||||
from ..utils import BColors
|
||||
from ..utils.shared import get_random_message
|
||||
|
||||
|
||||
def cli_function(function):
|
||||
def wrapper(*args, **kwargs):
|
||||
silent = kwargs.get("no_cli", False)
|
||||
if "no_cli" in kwargs:
|
||||
del kwargs["no_cli"]
|
||||
|
||||
if silent:
|
||||
return function(*args, **kwargs)
|
||||
return
|
||||
|
||||
code = 0
|
||||
|
||||
print_cute_message()
|
||||
print()
|
||||
try:
|
||||
code = function(*args, **kwargs)
|
||||
except KeyboardInterrupt:
|
||||
print("\n\nRaise an issue if I fucked up:\nhttps://github.com/HeIIow2/music-downloader/issues")
|
||||
|
||||
finally:
|
||||
print()
|
||||
print_cute_message()
|
||||
print("See you soon! :3")
|
||||
|
||||
exit()
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
def print_cute_message():
|
||||
message = get_random_message()
|
||||
try:
|
||||
print(message)
|
||||
except UnicodeEncodeError:
|
||||
message = str(c for c in message if 0 < ord(c) < 127)
|
||||
print(message)
|
||||
|
||||
|
||||
AGREE_INPUTS = {"y", "yes", "ok"}
|
||||
def ask_for_bool(msg: str) -> bool:
|
||||
i = input(f"{msg} ({BColors.OKGREEN.value}Y{BColors.ENDC.value}/{BColors.FAIL.value}N{BColors.ENDC.value})? ").lower()
|
||||
return i in AGREE_INPUTS
|
||||
|
||||
82
music_kraken/development_cli/__init__.py
Normal file
82
music_kraken/development_cli/__init__.py
Normal file
@@ -0,0 +1,82 @@
|
||||
import argparse
|
||||
from functools import cached_property
|
||||
|
||||
from ..__meta__ import DESCRIPTION, PROGRAMM
|
||||
from ..download import Downloader
|
||||
from ..utils import BColors
|
||||
from ..utils.string_processing import unify
|
||||
from .utils import HELP_MESSAGE, ask_for_bool, ask_for_create
|
||||
|
||||
|
||||
class DevelopmentCli:
|
||||
def __init__(self, args: argparse.Namespace):
|
||||
self.args = args
|
||||
|
||||
if args.genre:
|
||||
self.genre = args.genre
|
||||
|
||||
self.downloader: Downloader = Downloader()
|
||||
|
||||
@cached_property
|
||||
def genre(self) -> str:
|
||||
"""This is a cached property, which means if it isn't set in the constructor or before it is accessed,
|
||||
the program will be thrown in a shell
|
||||
|
||||
Returns:
|
||||
str: the genre that should be used
|
||||
"""
|
||||
option_string = f"{BColors.HEADER}Genres{BColors.ENDC}"
|
||||
genre_map = {}
|
||||
|
||||
_string_list = []
|
||||
for i, genre in enumerate(self.downloader.get_existing_genres()):
|
||||
option_string += f"\n{BColors.BOLD}{i}{BColors.ENDC}: {genre}"
|
||||
|
||||
genre_map[str(i)] = genre
|
||||
genre_map[unify(genre)] = genre
|
||||
|
||||
genre = None
|
||||
while genre is None:
|
||||
print(option_string)
|
||||
print()
|
||||
|
||||
i = input("> ")
|
||||
u = unify(i)
|
||||
if u in genre_map:
|
||||
genre = genre_map[u]
|
||||
break
|
||||
|
||||
if ask_for_create("genre", i):
|
||||
genre = i
|
||||
break
|
||||
|
||||
return genre
|
||||
|
||||
|
||||
def help_screen(self) -> None:
|
||||
print(HELP_MESSAGE)
|
||||
|
||||
def shell(self) -> None:
|
||||
print(f"Welcome to the {PROGRAMM} shell!")
|
||||
print(f"Type '{BColors.OKBLUE}help{BColors.ENDC}' for a list of commands.")
|
||||
print("")
|
||||
|
||||
while True:
|
||||
i = input("> ")
|
||||
if i == "help":
|
||||
self.help_screen()
|
||||
elif i == "genre":
|
||||
self.genre
|
||||
elif i == "exit":
|
||||
break
|
||||
else:
|
||||
print("Unknown command. Type 'help' for a list of commands.")
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
prog=PROGRAMM,
|
||||
description=DESCRIPTION,
|
||||
epilog='This is just a development cli. The real frontend is coming soon.'
|
||||
)
|
||||
parser.add_argument('--genre', '-g', action='store_const', required=False, help="choose a genre to download from")
|
||||
args = parser.parse_args()
|
||||
85
music_kraken/development_cli/utils.py
Normal file
85
music_kraken/development_cli/utils.py
Normal file
@@ -0,0 +1,85 @@
|
||||
from ..utils import BColors
|
||||
from ..utils.shared import get_random_message
|
||||
|
||||
|
||||
def cli_function(function):
|
||||
def wrapper(*args, **kwargs):
|
||||
silent = kwargs.get("no_cli", False)
|
||||
if "no_cli" in kwargs:
|
||||
del kwargs["no_cli"]
|
||||
|
||||
if silent:
|
||||
return function(*args, **kwargs)
|
||||
return
|
||||
|
||||
code = 0
|
||||
|
||||
print_cute_message()
|
||||
print()
|
||||
try:
|
||||
code = function(*args, **kwargs)
|
||||
except KeyboardInterrupt:
|
||||
print("\n\nRaise an issue if I fucked up:\nhttps://github.com/HeIIow2/music-downloader/issues")
|
||||
|
||||
finally:
|
||||
print()
|
||||
print_cute_message()
|
||||
print("See you soon! :3")
|
||||
|
||||
exit()
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
def print_cute_message():
|
||||
message = get_random_message()
|
||||
try:
|
||||
print(message)
|
||||
except UnicodeEncodeError:
|
||||
message = str(c for c in message if 0 < ord(c) < 127)
|
||||
print(message)
|
||||
|
||||
|
||||
def highlight_placeholder(text: str) -> str:
|
||||
return text.replace("<", f"{BColors.BOLD}<").replace(">", f">{BColors.ENDC}")
|
||||
|
||||
|
||||
HELP_MESSAGE = highlight_placeholder(f"""{BColors.HEADER}To search:{BColors.ENDC}
|
||||
> s: <query/url>
|
||||
> s: https://musify.club/release/some-random-release-183028492
|
||||
> s: #a <artist> #r <release> #t <track>
|
||||
|
||||
If you found the same object twice from different sources you can merge those objects.
|
||||
Then it will use those sources. To do so, use the {BColors.BOLD}m{BColors.ENDC} command.
|
||||
|
||||
{BColors.HEADER}To download:{BColors.ENDC}
|
||||
> d: <id/url>
|
||||
> dm: 0, 3, 4 # merge all objects into one and download this object
|
||||
> d: 1
|
||||
> d: https://musify.club/release/some-random-release-183028492
|
||||
|
||||
{BColors.HEADER}To inspect an object:{BColors.ENDC}
|
||||
If you inspect an object, you see its adjacent object. This means for example the releases of an artist, or the tracks of a release.
|
||||
You can also merge objects with the {BColors.BOLD}m{BColors.ENDC} command here.
|
||||
|
||||
> g: <id/url>
|
||||
> gm: 0, 3, 4 # merge all objects into one and inspect this object
|
||||
> g: 1
|
||||
> g: https://musify.club/release/some-random-release-183028492""")
|
||||
|
||||
|
||||
class COMMANDS:
|
||||
AGREE = {"y", "yes", "ok"}
|
||||
DISAGREE = {"n", "no"}
|
||||
EXIT = {"exit"}
|
||||
HELP = {"help", "h"}
|
||||
|
||||
|
||||
def ask_for_bool(msg: str) -> bool:
|
||||
i = input(f"{msg} ({BColors.OKGREEN.value}Y{BColors.ENDC.value}/{BColors.FAIL.value}N{BColors.ENDC.value})? ").lower()
|
||||
return i in COMMANDS.AGREE
|
||||
|
||||
|
||||
def ask_for_create(name: str, value: str) -> bool:
|
||||
return ask_for_bool(f"Do you want to create the {name} {BColors.OKBLUE}{value}{BColors.ENDC}?")
|
||||
|
||||
@@ -19,7 +19,7 @@ from ..connection import Connection
|
||||
from ..objects import Album, Artist, Collection
|
||||
from ..objects import DatabaseObject as DataObject
|
||||
from ..objects import Label, Options, Song, Source, Target
|
||||
from ..utils import BColors, output, trace
|
||||
from ..utils import BColors, limit_generator, output, trace
|
||||
from ..utils.config import main_settings, youtube_settings
|
||||
from ..utils.enums import ALL_SOURCE_TYPES, SourceType
|
||||
from ..utils.enums.album import AlbumType
|
||||
@@ -74,6 +74,8 @@ class Downloader:
|
||||
if auto_register_pages:
|
||||
self.scan_for_pages(**kwargs)
|
||||
|
||||
# manage which pages to use
|
||||
|
||||
def register_page(self, page_type: Type[Page], **kwargs):
|
||||
if page_type in self._registered_pages:
|
||||
return
|
||||
@@ -122,22 +124,76 @@ class Downloader:
|
||||
for page_type in page_types:
|
||||
yield from self._registered_pages[page_type]
|
||||
|
||||
# fetching/downloading data
|
||||
|
||||
def search(self, query: Query) -> Generator[DataObject, None, None]:
|
||||
"""Yields all data objects that were found by the query.
|
||||
Other than `Downloader.search_yield_pages`, this function just yields all data objects.
|
||||
This looses the data, where th objects were searched originally, so this might not be the best choice.
|
||||
|
||||
Args:
|
||||
query (Query): The query to search for.
|
||||
|
||||
Yields:
|
||||
Generator[DataObject, None, None]: A generator that yields all found data objects.
|
||||
"""
|
||||
|
||||
for page in self.get_pages():
|
||||
yield from page.search(query=query)
|
||||
|
||||
def fetch_details(self, data_object: DataObject, stop_at_level: int = 1, **kwargs) -> DataObject:
|
||||
|
||||
def search_yield_pages(self, query: Query, results_per_page: Optional[int] = None) -> Generator[Tuple[Page, Generator[DataObject, None, None]], None, None]:
|
||||
"""Yields all data objects that were found by the query, grouped by the page they were found on.
|
||||
every yield is a tuple of the page and a generator that yields the data objects.
|
||||
So this could be how it is used:
|
||||
|
||||
```python
|
||||
for page, data_objects in downloader.search_yield_pages(query):
|
||||
print(f"Found on {page}:")
|
||||
for data_object in data_objects:
|
||||
print(data_object)
|
||||
```
|
||||
|
||||
Args:
|
||||
query (Query): The query to search for.
|
||||
results_per_page (Optional[int], optional): If this is set, the generators only yield this amount of data objects per page.
|
||||
|
||||
Yields:
|
||||
Generator[Tuple[Page, Generator[DataObject, None, None]], None, None]: yields the page and a generator that yields the data objects.
|
||||
"""
|
||||
|
||||
for page in self.get_pages():
|
||||
yield page, limit_generator(page.search(query=query), limit=results_per_page)
|
||||
|
||||
def fetch_details(self, data_object: DataObject, **kwargs) -> DataObject:
|
||||
"""Fetches more details for the given data object.
|
||||
This uses every source contained in data_object.source_collection that has a page.
|
||||
|
||||
Args:
|
||||
data_object (DataObject): The data object to fetch details for.
|
||||
|
||||
Returns:
|
||||
DataObject: The same data object, but with more details.
|
||||
"""
|
||||
|
||||
source: Source
|
||||
for source in data_object.source_collection.get_sources(source_type_sorting={
|
||||
"only_with_page": True,
|
||||
}):
|
||||
new_data_object = self.fetch_from_source(source=source, stop_at_level=stop_at_level)
|
||||
new_data_object = self.fetch_from_source(source=source, **kwargs)
|
||||
if new_data_object is not None:
|
||||
data_object.merge(new_data_object)
|
||||
|
||||
return data_object
|
||||
|
||||
def fetch_from_source(self, source: Source, **kwargs) -> Optional[DataObject]:
|
||||
"""Gets a data object from the given source.
|
||||
|
||||
Args:
|
||||
source (Source): The source to get the data object from.
|
||||
|
||||
Returns:
|
||||
Optional[DataObject]: If a data object can be retrieved, it is returned. Otherwise, None is returned.
|
||||
"""
|
||||
if not source.has_page:
|
||||
return None
|
||||
|
||||
@@ -154,6 +210,15 @@ class Downloader:
|
||||
return data_object
|
||||
|
||||
def fetch_from_url(self, url: str) -> Optional[DataObject]:
|
||||
"""This function tries to detect the source of the given url and fetches the data object from it.
|
||||
|
||||
Args:
|
||||
url (str): The url to fetch the data object from.
|
||||
|
||||
Returns:
|
||||
Optional[DataObject]: The fetched data object, or None if no source could be detected or if no object could be retrieved.
|
||||
"""
|
||||
|
||||
source = Source.match_url(url, ALL_SOURCE_TYPES.MANUAL)
|
||||
if source is None:
|
||||
return None
|
||||
@@ -161,6 +226,14 @@ class Downloader:
|
||||
return self.fetch_from_source(source=source)
|
||||
|
||||
def _skip_object(self, data_object: DataObject) -> bool:
|
||||
"""Determine if the given data object should be downloaded or not.
|
||||
|
||||
Args:
|
||||
data_object (DataObject): The data object in question.
|
||||
|
||||
Returns:
|
||||
bool: Returns True if the given data object should be skipped.
|
||||
"""
|
||||
if isinstance(data_object, Album):
|
||||
if not self.download_options.download_all and data_object.album_type in self.download_options.album_type_blacklist:
|
||||
return True
|
||||
@@ -168,6 +241,18 @@ class Downloader:
|
||||
return False
|
||||
|
||||
def download(self, data_object: DataObject, genre: str, **kwargs) -> DownloadResult:
|
||||
"""Downloads the given data object.
|
||||
It will recursively fetch all available details for this and every related object.
|
||||
Then it will create the folder structure and download the audio file.
|
||||
In the end the metadata will be written to the file.
|
||||
|
||||
Args:
|
||||
data_object (DataObject): The data object to download. If it is a artist, it will download the whole discography, if it is an Album it will download the whole Tracklist.
|
||||
|
||||
Returns:
|
||||
DownloadResult: Relevant information about the download success.
|
||||
"""
|
||||
|
||||
# fetch the given object
|
||||
self.fetch_details(data_object)
|
||||
output(f"\nDownloading {data_object.option_string}...", color=BColors.BOLD)
|
||||
@@ -330,6 +415,31 @@ class Downloader:
|
||||
|
||||
return source.page.fetch_object_from_source(source=source, **kwargs)
|
||||
|
||||
# misc function
|
||||
|
||||
def get_existing_genres(self) -> Generator[str, None, None]:
|
||||
"""Yields every existing genre, for the user to select from.
|
||||
|
||||
Yields:
|
||||
Generator[str, None, None]: a generator that yields every existing genre.
|
||||
"""
|
||||
|
||||
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
|
||||
|
||||
class Page:
|
||||
REGISTER = True
|
||||
|
||||
@@ -177,7 +177,7 @@ class Node(Generator[P]):
|
||||
):
|
||||
self.value = value
|
||||
self.depth = 0
|
||||
self.same_level_index: int = 0
|
||||
self.index: int = 0
|
||||
|
||||
|
||||
self.children: List[Node[P]] = kwargs.get("children", [])
|
||||
@@ -185,17 +185,6 @@ class Node(Generator[P]):
|
||||
|
||||
super(Node, self).__init__(**kwargs)
|
||||
|
||||
def hash_key(self, key: Any) -> int:
|
||||
try:
|
||||
key = int(key)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
if isinstance(key, str):
|
||||
return hash(unify(key))
|
||||
|
||||
return hash(key)
|
||||
|
||||
@property
|
||||
def is_root(self) -> bool:
|
||||
return self.parent is None
|
||||
@@ -207,7 +196,7 @@ class Node(Generator[P]):
|
||||
def __iter__(self, **kwargs) -> Generator[Node[P], None, None]:
|
||||
_level_index_map: Dict[int, int] = kwargs.get("level_index_map", defaultdict(lambda: 0))
|
||||
|
||||
self.same_level_index = _level_index_map[self.depth]
|
||||
self.index = _level_index_map[self.depth]
|
||||
yield self
|
||||
_level_index_map[self.depth] += 1
|
||||
|
||||
|
||||
@@ -2,6 +2,7 @@ import inspect
|
||||
import json
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from itertools import takewhile
|
||||
from pathlib import Path
|
||||
from typing import List, Union
|
||||
|
||||
@@ -127,3 +128,8 @@ def get_current_millis() -> int:
|
||||
|
||||
def get_unix_time() -> int:
|
||||
return int(datetime.now().timestamp())
|
||||
|
||||
|
||||
def limit_generator(generator, limit: Optional[int] = None):
|
||||
return takewhile(lambda x: x < limit, generator) if limit is not None else generator
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class BColors(Enum):
|
||||
class BColors:
|
||||
# https://stackoverflow.com/a/287944
|
||||
HEADER = "\033[95m"
|
||||
OKBLUE = "\033[94m"
|
||||
|
||||
Reference in New Issue
Block a user