started to migrate to new config

This commit is contained in:
Hellow 2023-09-10 16:27:09 +02:00
parent 80c4117629
commit d2dd015817
30 changed files with 798 additions and 634 deletions

View File

@ -3,11 +3,22 @@ import logging
import gc import gc
import musicbrainzngs import musicbrainzngs
from .utils.config import read_config from .utils.config import logging_settings, main_settings, read_config
from .utils.shared import MODIFY_GC read_config()
from . import cli from . import cli
if MODIFY_GC:
# configure logger default
logging.basicConfig(
level=logging_settings['log_level'],
format=logging_settings['logging_format'],
handlers=[
logging.FileHandler(main_settings['log_file']),
logging.StreamHandler()
]
)
if main_settings['modify_gc']:
""" """
At the start I modify the garbage collector to run a bit fewer times. At the start I modify the garbage collector to run a bit fewer times.
This should increase speed: This should increase speed:
@ -21,6 +32,3 @@ if MODIFY_GC:
gen1 = gen1 * 2 gen1 = gen1 * 2
gen2 = gen2 * 2 gen2 = gen2 * 2
gc.set_threshold(allocs, gen1, gen2) gc.set_threshold(allocs, gen1, gen2)
logging.getLogger("musicbrainzngs").setLevel(logging.WARNING)
musicbrainzngs.set_useragent("metadata receiver", "0.1", "https://github.com/HeIIow2/music-downloader")

View File

@ -2,11 +2,14 @@ from typing import List, Tuple
from tqdm import tqdm from tqdm import tqdm
from ffmpeg_progress_yield import FfmpegProgress from ffmpeg_progress_yield import FfmpegProgress
from ..utils.shared import BITRATE, AUDIO_FORMAT, CODEX_LOGGER as LOGGER, FFMPEG_BINARY from ..utils.config import main_settings, logging_settings
from ..objects import Target from ..objects import Target
def correct_codec(target: Target, bitrate_kb: int = BITRATE, audio_format: str = AUDIO_FORMAT, interval_list: List[Tuple[float, float]] = None): LOGGER = logging_settings["codex_logger"]
def correct_codec(target: Target, bitrate_kb: int = main_settings["bitrate"], audio_format: str = main_settings["audio_format"], interval_list: List[Tuple[float, float]] = None):
if not target.exists: if not target.exists:
LOGGER.warning(f"Target doesn't exist: {target.file_path}") LOGGER.warning(f"Target doesn't exist: {target.file_path}")
return return
@ -35,7 +38,7 @@ def correct_codec(target: Target, bitrate_kb: int = BITRATE, audio_format: str =
# build the ffmpeg command # build the ffmpeg command
ffmpeg_command = [ ffmpeg_command = [
str(FFMPEG_BINARY), str(main_settings["ffmpeg_binary"]),
"-i", str(target.file_path), "-i", str(target.file_path),
"-af", select, "-af", select,
"-b", str(bitrate_b), "-b", str(bitrate_b),

View File

@ -4,12 +4,13 @@ from pathlib import Path
from typing import List from typing import List
import logging import logging
from ..utils.shared import ( from ..utils.config import logging_settings
TAGGING_LOGGER as LOGGER
)
from ..objects import Song, Target, Metadata from ..objects import Song, Target, Metadata
LOGGER = logging_settings["tagging_logger"]
class AudioMetadata: class AudioMetadata:
def __init__(self, file_location: str = None) -> None: def __init__(self, file_location: str = None) -> None:
self._file_location = None self._file_location = None

View File

@ -8,7 +8,7 @@ import requests
from tqdm import tqdm from tqdm import tqdm
from .rotating import RotatingProxy from .rotating import RotatingProxy
from ..utils.shared import PROXIES_LIST, CHUNK_SIZE from ..utils.config import main_settings
from ..utils.support_classes import DownloadResult from ..utils.support_classes import DownloadResult
from ..objects import Target from ..objects import Target
@ -18,7 +18,7 @@ class Connection:
self, self,
host: str, host: str,
proxies: List[dict] = None, proxies: List[dict] = None,
tries: int = (len(PROXIES_LIST) + 1) * 4, tries: int = (len(main_settings["proxies"]) + 1) * 4,
timeout: int = 7, timeout: int = 7,
logger: logging.Logger = logging.getLogger("connection"), logger: logging.Logger = logging.getLogger("connection"),
header_values: Dict[str, str] = None, header_values: Dict[str, str] = None,
@ -28,7 +28,7 @@ class Connection:
hearthbeat_interval = 0, hearthbeat_interval = 0,
): ):
if proxies is None: if proxies is None:
proxies = PROXIES_LIST proxies = main_settings["proxies"]
if header_values is None: if header_values is None:
header_values = dict() header_values = dict()
@ -266,7 +266,7 @@ class Connection:
timeout: float = None, timeout: float = None,
headers: dict = None, headers: dict = None,
raw_url: bool = False, raw_url: bool = False,
chunk_size: int = CHUNK_SIZE, chunk_size: int = main_settings["chunk_size"],
try_count: int = 0, try_count: int = 0,
progress: int = 0, progress: int = 0,
**kwargs **kwargs

View File

@ -4,7 +4,10 @@ from typing import Optional, Dict, Tuple, List
from .metadata import Metadata from .metadata import Metadata
from .option import Options from .option import Options
from ..utils.shared import ID_RANGE, OBJECT_LOGGER as LOGGER from ..utils.config import main_settings, logging_settings
LOGGER = logging_settings["object_logger"]
class DatabaseObject: class DatabaseObject:
@ -25,7 +28,7 @@ class DatabaseObject:
64 bit integer, but this is defined in shared.py in ID_BITS 64 bit integer, but this is defined in shared.py in ID_BITS
the range is defined in the Tuple ID_RANGE the range is defined in the Tuple ID_RANGE
""" """
_id = random.randint(*ID_RANGE) _id = random.randint(*main_settings['id_bits'])
self.automatic_id = True self.automatic_id = True
LOGGER.debug(f"Id for {type(self).__name__} isn't set. Setting to {_id}") LOGGER.debug(f"Id for {type(self).__name__} isn't set. Setting to {_id}")

View File

@ -19,7 +19,7 @@ from .source import Source, SourceCollection
from .target import Target from .target import Target
from ..utils.string_processing import unify from ..utils.string_processing import unify
from ..utils import settings from ..utils.config import main_settings
""" """
All Objects dependent All Objects dependent
@ -513,7 +513,7 @@ class Artist(MainObject):
AlbumType.STUDIO_ALBUM: 0, AlbumType.STUDIO_ALBUM: 0,
AlbumType.EP: 0, AlbumType.EP: 0,
AlbumType.SINGLE: 1 AlbumType.SINGLE: 1
}) if settings["sort_album_by_type"] else defaultdict(lambda: 0) }) if main_settings["sort_album_by_type"] else defaultdict(lambda: 0)
sections = defaultdict(list) sections = defaultdict(list)
@ -526,7 +526,7 @@ class Artist(MainObject):
# album is just a value used in loops # album is just a value used in loops
nonlocal album nonlocal album
if settings["sort_by_date"]: if main_settings["sort_by_date"]:
_section.sort(key=lambda _album: _album.date, reverse=True) _section.sort(key=lambda _album: _album.date, reverse=True)
new_last_albumsort = last_albumsort new_last_albumsort = last_albumsort

View File

@ -4,7 +4,8 @@ from typing import List, Dict, Set, Tuple, Optional
from urllib.parse import urlparse from urllib.parse import urlparse
from ..utils.enums.source import SourcePages, SourceTypes from ..utils.enums.source import SourcePages, SourceTypes
from ..utils.shared import ALL_YOUTUBE_URLS from ..utils.config import youtube_settings
from .metadata import Mapping, Metadata from .metadata import Mapping, Metadata
from .parents import DatabaseObject from .parents import DatabaseObject
from .collection import Collection from .collection import Collection
@ -54,7 +55,7 @@ class Source(DatabaseObject):
if "musify" in parsed.netloc: if "musify" in parsed.netloc:
return cls(SourcePages.MUSIFY, url, referer_page=referer_page) return cls(SourcePages.MUSIFY, url, referer_page=referer_page)
if parsed.netloc in [_url.netloc for _url in ALL_YOUTUBE_URLS]: if parsed.netloc in [_url.netloc for _url in youtube_settings['youtube_url']]:
return cls(SourcePages.YOUTUBE, url, referer_page=referer_page) return cls(SourcePages.YOUTUBE, url, referer_page=referer_page)
if url.startswith("https://www.deezer"): if url.startswith("https://www.deezer"):

View File

@ -23,9 +23,10 @@ from ..utils.enums.source import SourcePages
from ..utils.enums.album import AlbumType from ..utils.enums.album import AlbumType
from ..audio import write_metadata_to_target, correct_codec from ..audio import write_metadata_to_target, correct_codec
from ..utils import shared from ..utils import shared
from ..utils.shared import DOWNLOAD_PATH, DOWNLOAD_FILE, AUDIO_FORMAT from ..utils.config import main_settings
from ..utils.support_classes import Query, DownloadResult from ..utils.support_classes import Query, DownloadResult
INDEPENDENT_DB_OBJECTS = Union[Label, Album, Artist, Song] INDEPENDENT_DB_OBJECTS = Union[Label, Album, Artist, Song]
INDEPENDENT_DB_TYPES = Union[Type[Song], Type[Album], Type[Artist], Type[Label]] INDEPENDENT_DB_TYPES = Union[Type[Song], Type[Album], Type[Artist], Type[Label]]
@ -44,7 +45,7 @@ class NamingDict(dict):
self.object_mappings: Dict[str, DatabaseObject] = object_mappings or dict() self.object_mappings: Dict[str, DatabaseObject] = object_mappings or dict()
super().__init__(values) super().__init__(values)
self["audio_format"] = AUDIO_FORMAT self["audio_format"] = main_settings["audio_format"]
def add_object(self, music_object: DatabaseObject): def add_object(self, music_object: DatabaseObject):
self.object_mappings[type(music_object).__name__.lower()] = music_object self.object_mappings[type(music_object).__name__.lower()] = music_object
@ -380,12 +381,12 @@ class Page:
if song.genre is None: if song.genre is None:
song.genre = naming_dict["genre"] song.genre = naming_dict["genre"]
path_parts = Formatter().parse(DOWNLOAD_PATH) path_parts = Formatter().parse(main_settings["download_path"])
file_parts = Formatter().parse(DOWNLOAD_FILE) file_parts = Formatter().parse(main_settings["download_file"])
new_target = Target( new_target = Target(
relative_to_music_dir=True, relative_to_music_dir=True,
path=DOWNLOAD_PATH.format(**{part[1]: naming_dict[part[1]] for part in path_parts}), path=main_settings["download_path"].format(**{part[1]: naming_dict[part[1]] for part in path_parts}),
file=DOWNLOAD_FILE.format(**{part[1]: naming_dict[part[1]] for part in file_parts}) file=main_settings["download_file"].format(**{part[1]: naming_dict[part[1]] for part in file_parts})
) )

View File

@ -5,7 +5,7 @@ import pycountry
from urllib.parse import urlparse from urllib.parse import urlparse
from ..connection import Connection from ..connection import Connection
from ..utils.shared import ENCYCLOPAEDIA_METALLUM_LOGGER from ..utils.config import logging_settings
from .abstract import Page from .abstract import Page
from ..utils.enums.source import SourcePages from ..utils.enums.source import SourcePages
from ..utils.enums.album import AlbumType from ..utils.enums.album import AlbumType
@ -108,12 +108,12 @@ def _album_from_json(album_html=None, release_type=None, artist_html=None) -> Al
class EncyclopaediaMetallum(Page): class EncyclopaediaMetallum(Page):
SOURCE_TYPE = SourcePages.ENCYCLOPAEDIA_METALLUM SOURCE_TYPE = SourcePages.ENCYCLOPAEDIA_METALLUM
LOGGER = ENCYCLOPAEDIA_METALLUM_LOGGER LOGGER = logging_settings["metal_archives_logger"]
def __init__(self, **kwargs): def __init__(self, **kwargs):
self.connection: Connection = Connection( self.connection: Connection = Connection(
host="https://www.metal-archives.com/", host="https://www.metal-archives.com/",
logger=ENCYCLOPAEDIA_METALLUM_LOGGER logger=self.LOGGER
) )
super().__init__(**kwargs) super().__init__(**kwargs)

View File

@ -23,7 +23,7 @@ from ..objects import (
DatabaseObject, DatabaseObject,
Lyrics Lyrics
) )
from ..utils.shared import MUSIFY_LOGGER from ..utils.config import logging_settings
from ..utils import string_processing, shared from ..utils import string_processing, shared
from ..utils.support_classes import DownloadResult, Query from ..utils.support_classes import DownloadResult, Query
@ -95,7 +95,7 @@ def parse_url(url: str) -> MusifyUrl:
try: try:
type_enum = MusifyTypes(path[1]) type_enum = MusifyTypes(path[1])
except ValueError as e: except ValueError as e:
MUSIFY_LOGGER.warning(f"{path[1]} is not yet implemented, add it to MusifyTypes") logging_settings["musify_logger"].warning(f"{path[1]} is not yet implemented, add it to MusifyTypes")
raise e raise e
return MusifyUrl( return MusifyUrl(
@ -110,7 +110,7 @@ def parse_url(url: str) -> MusifyUrl:
class Musify(Page): class Musify(Page):
# CHANGE # CHANGE
SOURCE_TYPE = SourcePages.MUSIFY SOURCE_TYPE = SourcePages.MUSIFY
LOGGER = MUSIFY_LOGGER LOGGER = logging_settings["musify_logger"]
HOST = "https://musify.club" HOST = "https://musify.club"

View File

@ -20,7 +20,7 @@ from ..objects import (
) )
from ..connection import Connection from ..connection import Connection
from ..utils.support_classes import DownloadResult from ..utils.support_classes import DownloadResult
from ..utils.shared import YOUTUBE_LOGGER, INVIDIOUS_INSTANCE, BITRATE, ENABLE_SPONSOR_BLOCK, PIPED_INSTANCE, SLEEP_AFTER_YOUTUBE_403 from ..utils.config import youtube_settings, main_settings, logging_settings
from .youtube_music.super_youtube import SuperYouTube, YouTubeUrl, get_invidious_url, YouTubeUrlType from .youtube_music.super_youtube import SuperYouTube, YouTubeUrl, get_invidious_url, YouTubeUrlType
@ -34,13 +34,13 @@ from .youtube_music.super_youtube import SuperYouTube, YouTubeUrl, get_invidious
def get_piped_url(path: str = "", params: str = "", query: str = "", fragment: str = "") -> str: def get_piped_url(path: str = "", params: str = "", query: str = "", fragment: str = "") -> str:
return urlunparse((PIPED_INSTANCE.scheme, PIPED_INSTANCE.netloc, path, params, query, fragment)) return urlunparse((youtube_settings["piped_instance"].scheme, youtube_settings["piped_instance"].netloc, path, params, query, fragment))
class YouTube(SuperYouTube): class YouTube(SuperYouTube):
# CHANGE # CHANGE
SOURCE_TYPE = SourcePages.YOUTUBE SOURCE_TYPE = SourcePages.YOUTUBE
LOGGER = YOUTUBE_LOGGER LOGGER = logging_settings["youtube_logger"]
NO_ADDITIONAL_DATA_FROM_SONG = True NO_ADDITIONAL_DATA_FROM_SONG = True
@ -58,7 +58,7 @@ class YouTube(SuperYouTube):
self.download_connection: Connection = Connection( self.download_connection: Connection = Connection(
host="https://www.youtube.com/", host="https://www.youtube.com/",
logger=self.LOGGER, logger=self.LOGGER,
sleep_after_404=SLEEP_AFTER_YOUTUBE_403 sleep_after_404=youtube_settings["sleep_after_youtube_403"]
) )
# the stuff with the connection is, to ensure sponsorblock uses the proxies, my programm does # the stuff with the connection is, to ensure sponsorblock uses the proxies, my programm does
@ -307,7 +307,7 @@ class YouTube(SuperYouTube):
bitrate = int(possible_format.get("bitrate", 0)) bitrate = int(possible_format.get("bitrate", 0))
if bitrate >= BITRATE: if bitrate >= main_settings["bitrate"]:
best_bitrate = bitrate best_bitrate = bitrate
audio_format = possible_format audio_format = possible_format
break break
@ -325,7 +325,7 @@ class YouTube(SuperYouTube):
def get_skip_intervals(self, song: Song, source: Source) -> List[Tuple[float, float]]: def get_skip_intervals(self, song: Song, source: Source) -> List[Tuple[float, float]]:
if not ENABLE_SPONSOR_BLOCK: if not youtube_settings["use_sponsor_block"]:
return [] return []
parsed = YouTubeUrl(source.url) parsed = YouTubeUrl(source.url)

View File

@ -7,7 +7,8 @@ from dataclasses import dataclass
import re import re
from ...utils.exception.config import SettingValueError from ...utils.exception.config import SettingValueError
from ...utils.shared import PROXIES_LIST, YOUTUBE_MUSIC_LOGGER, DEBUG from ...utils.config import main_settings, youtube_settings, logging_settings
from ...utils.shared import DEBUG
from ...utils.config import CONNECTION_SECTION, write_config from ...utils.config import CONNECTION_SECTION, write_config
from ...utils.functions import get_current_millis from ...utils.functions import get_current_millis
if DEBUG: if DEBUG:
@ -94,7 +95,7 @@ class YouTubeMusicCredentials:
class YoutubeMusic(SuperYouTube): class YoutubeMusic(SuperYouTube):
# CHANGE # CHANGE
SOURCE_TYPE = SourcePages.YOUTUBE_MUSIC SOURCE_TYPE = SourcePages.YOUTUBE_MUSIC
LOGGER = YOUTUBE_MUSIC_LOGGER LOGGER = logging_settings["youtube_music_logger"]
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
self.connection: YoutubeMusicConnection = YoutubeMusicConnection(logger=self.LOGGER, accept_language="en-US,en;q=0.5") self.connection: YoutubeMusicConnection = YoutubeMusicConnection(logger=self.LOGGER, accept_language="en-US,en;q=0.5")

View File

@ -1,3 +1 @@
from .config import config, read_config, write_config from .config import config, read_config, write_config
from .config.settings import settings

View File

@ -1,13 +1,36 @@
from .sections.logging import LOGGING_SECTION from typing import Tuple
from .sections.audio import AUDIO_SECTION
from .sections.connection import CONNECTION_SECTION
from .sections.misc import MISC_SECTION
from .sections.paths import PATHS_SECTION
from .sections.paths import LOCATIONS
from .config import Config from .config import Config
from .config_files import (
main_config,
logging_config,
youtube_config,
)
from .settings import read_config, write_config, load, set_name_to_value _sections: Tuple[Config, ...] = (
main_config.config,
logging_config.config,
youtube_config.config
)
def read_config():
for section in _sections:
section.read()
load() # special cases
if main_settings['tor']:
main_settings['proxies'] = {
'http': f'socks5h://127.0.0.1:{main_settings["tor_port"]}',
'https': f'socks5h://127.0.0.1:{main_settings["tor_port"]}'
}
def write_config():
for section in _sections:
section.write()
def set_name_to_value():
pass
main_settings: main_config.SettingsStructure = main_config.config.loaded_settings
logging_settings: logging_config.SettingsStructure = logging_config.config.loaded_settings
youtube_settings: youtube_config.SettingsStructure = youtube_config.config.loaded_settings

View File

@ -0,0 +1,127 @@
from typing import Union, Tuple, Dict, Iterable, List
from datetime import datetime
import logging
import os
from ..exception.config import SettingNotFound, SettingValueError
from ..path_manager import LOCATIONS
from .base_classes import Description, Attribute, Section, EmptyLine, COMMENT_PREFIX
from .sections.audio import AUDIO_SECTION
from .sections.logging import LOGGING_SECTION
from .sections.connection import CONNECTION_SECTION
from .sections.misc import MISC_SECTION
from .sections.paths import PATHS_SECTION
LOGGER = logging.getLogger("config")
class Config:
def __init__(self):
self.config_elements: Tuple[Union[Description, Attribute, Section], ...] = (
Description("IMPORTANT: If you modify this file, the changes for the actual setting, will be kept as is.\n"
"The changes you make to the comments, will be discarded, next time you run music-kraken. "
"Have fun!"),
Description(f"Latest reset: {datetime.now()}"),
Description("Those are all Settings for the audio codec.\n"
"If you, for some reason wanna fill your drive real quickly, I mean enjoy HIFI music,\n"
"feel free to tinker with the Bitrate or smth. :)"),
AUDIO_SECTION,
Description("Modify how Music-Kraken connects to the internet:"),
CONNECTION_SECTION,
Description("Modify all your paths, except your config file..."),
PATHS_SECTION,
Description("For all your Logging needs.\n"
"If you found a bug, and wan't to report it, please set the Logging level to 0,\n"
"reproduce the bug, and attach the logfile in the bugreport. ^w^"),
LOGGING_SECTION,
Description("If there are stupid settings, they are here."),
MISC_SECTION,
Description("🏳️‍⚧️🏳️‍⚧️ Protect trans youth. 🏳️‍⚧️🏳️‍⚧️\n"),
)
self._length = 0
self._section_list: List[Section] = []
self.name_section_map: Dict[str, Section] = dict()
for element in self.config_elements:
if not isinstance(element, Section):
continue
self._section_list.append(element)
for name in element.name_attribute_map:
if name in self.name_section_map:
raise ValueError(f"Two sections have the same name: "
f"{name}: "
f"{element.__class__.__name__} {self.name_section_map[name].__class__.__name__}")
self.name_section_map[name] = element
self._length += 1
def set_name_to_value(self, name: str, value: str, silent: bool = True):
"""
:raises SettingValueError, SettingNotFound:
:param name:
:param value:
:return:
"""
if name not in self.name_section_map:
if silent:
LOGGER.warning(f"The setting \"{name}\" is either deprecated, or doesn't exist.")
return
raise SettingNotFound(setting_name=name)
LOGGER.debug(f"setting: {name} value: {value}")
self.name_section_map[name].modify_setting(setting_name=name, new_value=value)
def __len__(self):
return self._length
@property
def config_string(self) -> str:
return "\n\n".join(str(element) for element in self.config_elements)
def _parse_conf_line(self, line: str, index: int):
"""
:raises SettingValueError, SettingNotFound:
:param line:
:param index:
:return:
"""
line = line.strip()
if line.startswith(COMMENT_PREFIX):
return
if line == "":
return
if "=" not in line:
"""
TODO
No value error but custom conf error
"""
raise ValueError(f"Couldn't find the '=' in line {index}.")
line_segments = line.split("=")
name = line_segments[0]
value = "=".join(line_segments[1:])
self.set_name_to_value(name, value)
def read_from_config_file(self, path: os.PathLike):
with open(path, "r", encoding=LOCATIONS.FILE_ENCODING) as conf_file:
for section in self._section_list:
section.reset_list_attribute()
for i, line in enumerate(conf_file):
self._parse_conf_line(line, i+1)
def write_to_config_file(self, path: os.PathLike):
with open(path, "w", encoding=LOCATIONS.FILE_ENCODING) as conf_file:
conf_file.write(self.config_string)
def __iter__(self) -> Iterable[Attribute]:
for section in self._section_list:
for name, attribute in section.name_attribute_map.items():
yield attribute

View File

@ -1,60 +1,123 @@
import re import re
from typing import Optional, List, Union, Iterable, Callable
from dataclasses import dataclass
import logging
import toml
from copy import deepcopy
from ...exception.config import SettingValueError from ...exception.config import SettingValueError
from ..utils import comment from ..utils import comment
LOGGER = logging.getLogger("config")
COMMENT_PREFIX = "#"
def comment_string(uncommented: str) -> str:
unprocessed_lines = uncommented.split("\n")
processed_lines: List[str] = []
for line in unprocessed_lines:
line: str = line.strip()
if line.startswith(COMMENT_PREFIX) or line == "":
processed_lines.append(line)
continue
line = COMMENT_PREFIX + " " + line
processed_lines.append(line)
return "\n".join(processed_lines)
@dataclass
class Description: class Description:
def __init__(self, string: str) -> None: description: str
self.string = string
@property @property
def config_string(self) -> str: def toml_string(self):
return comment(self.string) return comment_string(self.description)
class EmptyLine(Description):
def __init__(self):
self.description = ""
class Attribute: class Attribute:
pattern: str = r'^.*a$' def __init__(
rule: str = "This is a default string, it has no rule." self,
string_value: str = "" name: str,
default_value: any,
def __init__(self, name: str, description: str, pattern: str = None, rule: str = None) -> None: description: Optional[str] = None,
if pattern is not None: ):
self.pattern = pattern
if rule is not None:
self.rule = rule
self.name = name self.name = name
self.description = description
def validate(self, input_string: str) -> bool: self.raw_data = {name: default_value}
return re.match(self.REGEX, input_string) is None self.value = default_value
def output_parse(self): self.description: Optional[str] = description
return self.string_value.strip()
def input_parse(self, input_string: str) -> str: def unparse_simple_value(self, value: any) -> any:
match_result = re.match(self.pattern, input_string) return value
if match_result is None: def parse_simple_value(self, value: any) -> any:
raise SettingValueError( return value
setting_name=self.name,
setting_value=input_string, def _recursive_parse_object(self, __object, callback: Callable):
rule=self.rule if isinstance(__object, dict):
) for key, value in __object.items():
__object[key] = self._recursive_parse_object(value, callback)
return __object
if isinstance(__object, Union[list, tuple]):
for i, item in enumerate(__object):
__object[i] = self._recursive_parse_object(item, callback)
return __object
return callback(__object)
def load_toml(self, loaded_toml: dict, loaded_settings: dict) -> bool:
"""
returns true if succesfull
"""
if self.name not in loaded_toml:
LOGGER.warning(f"No setting by the name {self.name} found in the settings file.")
loaded_settings[self.name] = self.value
return
self.raw_data = loaded_toml[self.name]
_object = deepcopy(loaded_toml[self.name])
try:
self._recursive_parse_object(_object, self.parse_simple_value)
except SettingValueError as settings_error:
logging.warning(settings_error)
return False
self.value = _object
loaded_settings[self.name] = self.value
return True
return match_result.string
@property @property
def value(self) -> str: def toml_string(self) -> str:
raise NotImplementedError() string = ""
@property if self.description is not None:
def config_string(self) -> str: string += comment(self.description) + "\n"
return NotImplementedError()
string += toml.dumps(self.raw_data)
return string
def __str__(self):
return f"{self.description}\n{self.name}={self.value}"
attr = Attribute(name="hello world", description="fuck you", value="defaulte")
attr.input_parse("fafda")
attr.input_parse("eeee")

View File

@ -1,34 +0,0 @@
from typing import List
from .attribute import Attribute
from ..utils import comment
class ListAttribute(Attribute):
def __init__(self, name: str, description: str, value: List[str], pattern: str = None, rule: str = None) -> None:
super().__init__(name, description, pattern, rule)
self.string_value_list = []
self.set_to_list(value)
def set_to_list(self, input_value_list: List[str]):
self.string_value_list = []
for input_value in input_value_list:
self.string_value_list.append(input_value)
def append(self, input_value: str):
self.string_value_list.append(self.input_parse(input_value))
@property
def value(self) -> str:
return [self.output_parse(element) for element in self.string_value_list]
@property
def config_string(self) -> str:
NEWLINE = "\n"
return f"[{self.name}.start]" \
f"{comment(self.description)}\n" \
f"{NEWLINE.join(self.name+'='+v for v in self.string_value_list)}\n" \
f"{comment('RULE: ' + self.rule)}\n" \
f"[{self.name}.end]"

View File

@ -1,18 +0,0 @@
from ..utils import comment
from .attribute import Attribute
class SingleAttribute(Attribute):
def __init__(self, name: str, description: str, value: str, pattern: str = None, rule: str = None) -> None:
super().__init__(name, description, pattern, rule)
self.string_value = self.input_parse(value)
@property
def value(self) -> str:
return self.output_parse(self.string_value)
@property
def config_string(self) -> str:
return f"{comment(self.description)}\n" \
f"{self.name}={self.value}\n" \
f"{comment('RULE: ' + self.rule)}" \

View File

@ -0,0 +1,149 @@
from pathlib import Path
from typing import Optional, Dict, Set
from urllib.parse import urlparse, urlunparse
import logging
from .attribute import Attribute
from ...exception.config import SettingValueError
class UrlAttribute(Attribute):
def parse_simple_value(self, value: any) -> any:
return urlparse(value)
def unparse_simple_value(self, value: any) -> any:
return urlunparse((value.scheme, value.netloc, value.path, value.params, value.query, value.fragment))
class PathAttribute(Attribute):
def parse_simple_value(self, value: any) -> any:
return Path(value)
def unparse_simple_value(self, value: any) -> any:
return value.resolve()
class SelectAttribute(Attribute):
def __init__(self, name: str, default_value: any, options: tuple, description: Optional[str] = None, ignore_options_for_description = False):
self.options: tuple = options
new_description = ""
if description is not None:
new_description += description
new_description += "\n"
if not ignore_options_for_description:
new_description += f"{{{', '.join(self.options)}}}"
super().__init__(name, default_value, description)
def parse_simple_value(self, value: any) -> any:
if value in self.options:
return value
raise SettingValueError(
setting_name=self.name,
setting_value=value,
rule=f"has to be in the options: {{{', '.join(self.options)}}}."
)
def unparse_simple_value(self, value: any) -> any:
return value
class IntegerSelect(Attribute):
def __init__(self, name: str, default_value: any, options: Dict[int, str], description: Optional[str] = None, ignore_options_for_description = False):
self.options: Dict[str, int] = options
self.option_values: Set[int] = set(self.options.values())
new_description = ""
if description is not None:
new_description += description
description_lines = []
if description is not None:
description_lines.append(description)
description_lines.append("The values can be either an integer or one of the following values:")
for number, option in self.options.items():
description_lines.append(f"{number}: {option}")
super().__init__(name, default_value, "\n".join(description_lines))
def parse_simple_value(self, value: any) -> any:
if isinstance(value, str):
if value not in self.options:
raise SettingValueError(
setting_name=self.name,
setting_value=value,
rule=f"has to be in the options: {{{', '.join(self.options.keys())}}}, if it is a string."
)
return self.options[value]
return value
def unparse_simple_value(self, value: int) -> any:
if value in self.option_values:
for option, v in self.options.items():
if v == value:
return value
return value
ID3_2_FILE_FORMATS = frozenset((
"mp3", "mp2", "mp1", # MPEG-1 ID3.2
"wav", "wave", "rmi", # RIFF (including WAV) ID3.2
"aiff", "aif", "aifc", # AIFF ID3.2
"aac", "aacp", # Raw AAC ID3.2
"tta", # True Audio ID3.2
))
_sorted_id3_2_formats = sorted(ID3_2_FILE_FORMATS)
ID3_1_FILE_FORMATS = frozenset((
"ape", # Monkey's Audio ID3.1
"mpc", "mpp", "mp+", # MusePack ID3.1
"wv", # WavPack ID3.1
"ofr", "ofs" # OptimFrog ID3.1
))
_sorted_id3_1_formats = sorted(ID3_1_FILE_FORMATS)
class AudioFormatAttribute(Attribute):
def __init__(self, name: str, default_value: any, description: Optional[str] = None, ignore_options_for_description = False):
new_description = ""
if description is not None:
new_description += description
new_description += "\n"
new_description += f"ID3.2: {{{', '.join(ID3_2_FILE_FORMATS)}}}\n"
new_description += f"ID3.1: {{{', '.join(ID3_1_FILE_FORMATS)}}}"
super().__init__(name, default_value, description)
def parse_simple_value(self, value: any) -> any:
value = value.strip().lower()
if value in ID3_2_FILE_FORMATS:
return value
if value in ID3_1_FILE_FORMATS:
logging.debug(f"setting audio format to a format that only supports ID3.1: {v}")
return value
raise SettingValueError(
setting_name=self.name,
setting_value=value,
rule="has to be a valid audio format, supporting id3 metadata"
)
def unparse_simple_value(self, value: any) -> any:
return value
class LoggerAttribute(Attribute):
def parse_simple_value(self, value: str) -> logging.Logger:
return logging.getLogger(self.value)
def unparse_simple_value(self, value: logging.Logger) -> any:
return value.name

View File

@ -1,234 +0,0 @@
import logging
from dataclasses import dataclass
from typing import Optional, List, Union, Dict
from ..exception.config import SettingNotFound, SettingValueError
LOGGER = logging.getLogger("config")
COMMENT_PREFIX = "#"
def comment_string(uncommented: str) -> str:
unprocessed_lines = uncommented.split("\n")
processed_lines: List[str] = []
for line in unprocessed_lines:
line: str = line.strip()
if line.startswith(COMMENT_PREFIX) or line == "":
processed_lines.append(line)
continue
line = COMMENT_PREFIX + " " + line
processed_lines.append(line)
return "\n".join(processed_lines)
@dataclass
class Attribute:
name: str
description: Optional[str]
value: Union[str, List[str]]
def validate(self, value: str):
"""
This function validates a new value without setting it.
:raise SettingValueError:
:param value:
:return:
"""
pass
def set_value(self, value: str):
"""
:raise SettingValueError: if the value is invalid for this setting
:param value:
:return:
"""
self.validate(value)
self.value = value
@property
def description_as_comment(self):
return comment_string(self.description)
@property
def object_from_value(self):
return self.value
def __str__(self):
return f"{self.description_as_comment}\n{self.name}={self.value}"
class SingleAttribute(Attribute):
value: str
class StringAttribute(SingleAttribute):
@property
def object_from_value(self) -> str:
return self.value.strip()
class IntAttribute(SingleAttribute):
def validate(self, value: str):
if not value.isdigit():
raise SettingValueError(
setting_name=self.name,
setting_value=value,
rule="has to be a digit (an int)"
)
@property
def object_from_value(self) -> int:
if self.value.isdigit():
return int(self.value)
class BoolAttribute(SingleAttribute):
def validate(self, value: str):
if value.lower().strip() not in {"true", "false"}:
raise SettingValueError(
setting_name=self.name,
setting_value=value,
rule="has to be a bool (true/false)"
)
@property
def object_from_value(self) -> bool:
return self.value.lower().strip() in {"yes", "y", "t", "true"}
class FloatAttribute(SingleAttribute):
def validate(self, value: str):
try:
float(value)
except ValueError:
raise SettingValueError(
setting_name=self.name,
setting_value=value,
rule="has to be numeric (an int or float)"
)
@property
def object_from_value(self) -> float:
return float(self.value)
class ListAttribute(Attribute):
value: List[str]
has_default_values: bool = True
def __len__(self):
return len(self.value)
def set_value(self, value: str):
"""
Due to lists being represented as multiple lines with the same key,
this appends, rather than setting anything.
:raise SettingValueError:
:param value:
:return:
"""
self.validate(value)
# resetting the list to an empty list, if this is the first config line to load
if self.has_default_values:
self.value = []
self.has_default_values = False
if value in self.value:
return
self.value.append(value)
def __str__(self):
return f"{self.description_as_comment}\n" + \
"\n".join(f"{self.name}={element}" for element in self.value)
def single_object_from_element(self, value: str):
return value
@property
def object_from_value(self) -> list:
"""
THIS IS NOT THE PROPERTY TO OVERRIDE WHEN INHERITING ListAttribute
single_object_from_element
:return:
"""
parsed = list()
for raw in self.value:
parsed.append(self.single_object_from_element(raw))
return parsed
@dataclass
class Description:
description: str
def __str__(self):
return comment_string(self.description)
class EmptyLine(Description):
def __init__(self):
self.description = ""
class Section:
"""
A placeholder class
"""
attribute_list: List[Union[
Attribute,
Description
]]
def __init__(self):
self.name_attribute_map: Dict[str, Attribute] = dict()
self.index_values()
def __str__(self):
return "\n".join(attribute.__str__() for attribute in self.attribute_list)
def index_values(self):
for element in self.attribute_list:
if not isinstance(element, Attribute):
continue
if element.name in self.name_attribute_map:
raise ValueError(f"Two different Attributes have the same name: "
f"{self.name_attribute_map[element.name]} {element}")
self.name_attribute_map[element.name] = element
def modify_setting(self, setting_name: str, new_value: str):
"""
:raise SettingValueError, SettingNotFound:
:param setting_name:
:param new_value:
:return:
"""
if setting_name not in self.name_attribute_map:
raise SettingNotFound(
setting_name=setting_name
)
self.name_attribute_map[setting_name].set_value(new_value)
def reset_list_attribute(self):
for attribute in self.attribute_list:
if not isinstance(attribute, ListAttribute):
continue
attribute.has_default_values = True

View File

@ -1,127 +1,37 @@
from typing import Union, Tuple, Dict, Iterable, List from typing import Tuple, Union
from datetime import datetime from pathlib import Path
import logging import logging
import os
from ..exception.config import SettingNotFound, SettingValueError import toml
from ..path_manager import LOCATIONS
from .base_classes import Description, Attribute, Section, EmptyLine, COMMENT_PREFIX
from .sections.audio import AUDIO_SECTION
from .sections.logging import LOGGING_SECTION
from .sections.connection import CONNECTION_SECTION
from .sections.misc import MISC_SECTION
from .sections.paths import PATHS_SECTION
from .attributes.attribute import Attribute, Description, EmptyLine
LOGGER = logging.getLogger("config")
class Config: class Config:
def __init__(self): def __init__(self, componet_list: Tuple[Union[Attribute, Description, EmptyLine]], config_file: Path) -> None:
self.config_elements: Tuple[Union[Description, Attribute, Section], ...] = ( self.config_file: Path = config_file
Description("IMPORTANT: If you modify this file, the changes for the actual setting, will be kept as is.\n"
"The changes you make to the comments, will be discarded, next time you run music-kraken. "
"Have fun!"),
Description(f"Latest reset: {datetime.now()}"),
Description("Those are all Settings for the audio codec.\n"
"If you, for some reason wanna fill your drive real quickly, I mean enjoy HIFI music,\n"
"feel free to tinker with the Bitrate or smth. :)"),
AUDIO_SECTION,
Description("Modify how Music-Kraken connects to the internet:"),
CONNECTION_SECTION,
Description("Modify all your paths, except your config file..."),
PATHS_SECTION,
Description("For all your Logging needs.\n"
"If you found a bug, and wan't to report it, please set the Logging level to 0,\n"
"reproduce the bug, and attach the logfile in the bugreport. ^w^"),
LOGGING_SECTION,
Description("If there are stupid settings, they are here."),
MISC_SECTION,
Description("🏳️‍⚧️🏳️‍⚧️ Protect trans youth. 🏳️‍⚧️🏳️‍⚧️\n"),
)
self._length = 0 self.component_list: Tuple[Union[Attribute, Description, EmptyLine]] = componet_list
self._section_list: List[Section] = [] self.loaded_settings: dict = {}
self.name_section_map: Dict[str, Section] = dict()
for element in self.config_elements:
if not isinstance(element, Section):
continue
self._section_list.append(element)
for name in element.name_attribute_map:
if name in self.name_section_map:
raise ValueError(f"Two sections have the same name: "
f"{name}: "
f"{element.__class__.__name__} {self.name_section_map[name].__class__.__name__}")
self.name_section_map[name] = element
self._length += 1
def set_name_to_value(self, name: str, value: str, silent: bool = True):
"""
:raises SettingValueError, SettingNotFound:
:param name:
:param value:
:return:
"""
if name not in self.name_section_map:
if silent:
LOGGER.warning(f"The setting \"{name}\" is either deprecated, or doesn't exist.")
return
raise SettingNotFound(setting_name=name)
LOGGER.debug(f"setting: {name} value: {value}")
self.name_section_map[name].modify_setting(setting_name=name, new_value=value)
def __len__(self):
return self._length
@property @property
def config_string(self) -> str: def toml_string(self):
return "\n\n".join(str(element) for element in self.config_elements) "\n\n".join(component.toml_string for component in self.component_list)
def _parse_conf_line(self, line: str, index: int): def write(self):
""" with self.config_file.open("w") as conf_file:
:raises SettingValueError, SettingNotFound: conf_file.write(self.toml_string)
:param line:
:param index: def read(self):
:return: if not self.config_file.is_file():
""" logging.info(f"Config file at '{self.config_file}' doesn't exist => generating")
line = line.strip() self.write()
if line.startswith(COMMENT_PREFIX):
return return
if line == "": toml_data = {}
return with self.config_file.open("r") as conf_file:
toml_data = toml.load(conf_file)
if "=" not in line: for component in self.component_list:
""" if isinstance(component, Attribute):
TODO component.load_toml(toml_data, self.loaded_settings)
No value error but custom conf error
"""
raise ValueError(f"Couldn't find the '=' in line {index}.")
line_segments = line.split("=")
name = line_segments[0]
value = "=".join(line_segments[1:])
self.set_name_to_value(name, value)
def read_from_config_file(self, path: os.PathLike):
with open(path, "r", encoding=LOCATIONS.FILE_ENCODING) as conf_file:
for section in self._section_list:
section.reset_list_attribute()
for i, line in enumerate(conf_file):
self._parse_conf_line(line, i+1)
def write_to_config_file(self, path: os.PathLike):
with open(path, "w", encoding=LOCATIONS.FILE_ENCODING) as conf_file:
conf_file.write(self.config_string)
def __iter__(self) -> Iterable[Attribute]:
for section in self._section_list:
for name, attribute in section.name_attribute_map.items():
yield attribute

View File

@ -0,0 +1,99 @@
from typing import TypedDict, List
from urllib.parse import ParseResult
from logging import Logger
from pathlib import Path
import logging
from ...path_manager import LOCATIONS
from ..config import Config
from ..attributes.attribute import Attribute, EmptyLine
from ..attributes.special_attributes import (
IntegerSelect,
LoggerAttribute
)
config = Config([
Attribute(name="logging_format", default_value="%(levelname)s:%(name)s:%(message)s", description="""Logging settings for the actual logging:
Reference for the logging formats: https://docs.python.org/3/library/logging.html#logrecord-attributes"""),
IntegerSelect(
name="log_level",
default_value=str(logging.INFO),
options={
"CRITICAL": 50,
"ERROR": 40,
"WARNING": 30,
"INFO": 20,
"DEBUG": 10,
"NOTSET": 0
}
),
LoggerAttribute(
name="download_logger",
description="The logger for downloading.",
default_value="download"
),
LoggerAttribute(
name="tagging_logger",
description="The logger for tagging id3 containers.",
default_value="tagging"
),
LoggerAttribute(
name="codex_logger",
description="The logger for streaming the audio into an uniform codex.",
default_value="codex"
),
LoggerAttribute(
name="object_logger",
description="The logger for creating Data-Objects.",
default_value="object"
),
LoggerAttribute(
name="database_logger",
description="The logger for Database operations.",
default_value="database"
),
LoggerAttribute(
name="musify_logger",
description="The logger for the musify scraper.",
default_value="musify"
),
LoggerAttribute(
name="youtube_logger",
description="The logger for the youtube scraper.",
default_value="youtube"
),
LoggerAttribute(
name="youtube_music_logger",
description="The logger for the youtube music scraper.\n(The scraper is seperate to the youtube scraper)",
default_value="youtube_music"
),
LoggerAttribute(
name="metal_archives_logger",
description="The logger for the metal archives scraper.",
default_value="metal_archives"
),
LoggerAttribute(
name="genius_logger",
description="The logger for the genius scraper",
default_value="genius"
),
], LOCATIONS.get_config_file("logging"))
class SettingsStructure(TypedDict):
# logging
logging_format: str
log_level: int
download_logger: Logger
tagging_logger: Logger
codex_logger: Logger
object_logger: Logger
database_logger: Logger
musify_logger: Logger
youtube_logger: Logger
youtube_music_logger: Logger
metal_archives_logger: Logger
genius_logger: Logger

View File

@ -0,0 +1,142 @@
from typing import TypedDict, List
from datetime import datetime
from urllib.parse import ParseResult
from logging import Logger
from pathlib import Path
from ...path_manager import LOCATIONS
from ..config import Config
from ..attributes.attribute import Attribute, EmptyLine, Description
from ..attributes.special_attributes import (
SelectAttribute,
PathAttribute,
AudioFormatAttribute,
)
config = Config([
Description(f"""IMPORTANT: If you modify this file, the changes for the actual setting, will be kept as is.
The changes you make to the comments, will be discarded, next time you run music-kraken. Have fun!
Latest reset: {datetime.now()}
_________ __
\\_ ___ \\ __ __ _/ |_ ____
/ \\ \\/ | | \\\\ __\\_/ __ \\
\\ \\____| | / | | \\ ___/
\\______ /|____/ |__| \\___ >
\\/ \\/
"""),
Attribute(name="hasnt_yet_started", default_value=False, description="This will be set automatically, to look if it needs to run the scripts that run on start."),
Attribute(name="bitrate", default_value=125, description="Streams the audio with given bitrate [kB/s]. Can't stream with a higher Bitrate, than the audio source provides."),
AudioFormatAttribute(name="audio_format", default_value="mp3", description="""Music Kraken will stream the audio into this format.
You can use Audio formats which support ID3.2 and ID3.1,
but you will have cleaner Metadata using ID3.2."""),
Attribute(name="result_history", default_value=False, description="""If enabled, you can go back to the previous results.
The consequence is a higher meory consumption, because every result is saved."""),
Attribute(name="history_length", default_value=8, description="""You can choose how far back you can go in the result history.
The further you choose to be able to go back, the higher the memory usage.
'-1' removes the Limit entirely."""),
EmptyLine(),
Attribute(name="sort_by_date", default_value=True, description="If this is set to true, it will set the albumsort attribute such that,\nthe albums are sorted by date"),
Attribute(name="sort_album_by_type", default_value=True, description="""If this is set to true, it will set the albumsort attribute such that,
the albums are put into categories before being sorted.
This means for example, the Studio Albums and EP's are always in front of Singles, and Compilations are in the back."""),
Attribute(name="download_path", default_value="{genre}/{artist}/{album}", description="""There are multiple fields, you can use for the path and file name:
- genre
- label
- artist
- album
- song
- album_type
The folder music kraken should put the songs into."""),
Attribute(name="download_file", default_value="{song}.{audio_format}", description="The filename of the audio file."),
SelectAttribute(name="album_type_blacklist", default_value=[
"Compilation Album",
"Live Album",
"Mixtape"
], options=("Studio Album", "EP (Extended Play)", "Single", "Live Album", "Compilation Album", "Mixtape", "Demo", "Other"), description="""Music Kraken ignores all albums of those types.
Following album types exist in the programm:"""),
EmptyLine(),
Attribute(name="proxies", default_value=[], description="This is a dictionary."),
Attribute(name="tor", default_value=False, description="""Route ALL traffic through Tor.
If you use Tor, make sure the Tor browser is installed, and running.I can't guarantee maximum security though!"""),
Attribute(name="tor_port", default_value=9150, description="The port, tor is listening. If tor is already working, don't change it."),
Attribute(name="chunk_size", default_value=1024, description="Size of the chunks that are streamed.\nHere could be some room for improvement."),
Attribute(name="show_download_errors_threshold", default_value=0.3, description="""If the percentage of failed downloads goes over this threshold,
all the error messages are shown."""),
EmptyLine(),
PathAttribute(name="music_directory", default_value=LOCATIONS.MUSIC_DIRECTORY, description="The directory, all the music will be downloaded to."),
PathAttribute(name="temp_directory", default_value=LOCATIONS.TEMP_DIRECTORY, description="All temporary stuff is gonna be dumped in this directory."),
PathAttribute(name="log_file", default_value=LOCATIONS.get_log_file("download_logs.log")),
PathAttribute(name="ffmpeg_binary", default_value=LOCATIONS.FFMPEG_BIN, description="Set the path to the ffmpeg binary."),
Attribute(
name="not_a_genre_regex",
description="These regular expressions tell music-kraken, which sub-folders of the music-directory\n"
"it should ignore, and not count to genres",
default_value=[
r'^\.' # is hidden/starts with a "."
]
),
EmptyLine(),
Attribute(name="happy_messages", default_value=[
"Support the artist.",
"Star Me: https://github.com/HeIIow2/music-downloader",
"🏳️‍⚧️🏳️‍⚧️ Trans rights are human rights. 🏳️‍⚧️🏳️‍⚧️",
"🏳️‍⚧️🏳️‍⚧️ Trans women are women, trans men are men, and enbies are enbies. 🏳️‍⚧️🏳️‍⚧️",
"🏴‍☠️🏴‍☠️ Unite under one flag, fck borders. 🏴‍☠️🏴‍☠️",
"Join my Matrix Space: https://matrix.to/#/#music-kraken:matrix.org",
"BPJM does cencorship.",
"🏳️‍⚧️🏳️‍⚧️ Protect trans youth. 🏳️‍⚧️🏳️‍⚧️",
"Klassenkampf",
"Rise Proletarians!!"
], description="""Just some nice and wholesome messages.
If your mindset has traits of a [file corruption], you might not agree.
But anyways... Freedom of thought, so go ahead and change the messages."""),
Attribute(name="id_bits", default_value=64, description="I really dunno why I even made this a setting.. Modifying this is a REALLY dumb idea."),
Description("🏳️‍⚧️🏳️‍⚧️ Protect trans youth. 🏳️‍⚧️🏳️‍⚧️\n"),
], LOCATIONS.get_config_file("main"))
class SettingsStructure(TypedDict):
hasnt_yet_started: bool
result_history: bool
history_length: int
happy_messages: List[str]
modify_gc: bool
id_bits: int
# audio
bitrate: int
audio_format: str
sort_by_date: bool
sort_album_by_type: bool
download_path: str
download_file: str
album_type_blacklist: List[str]
# connection
proxies: List[dict[str, str]]
tor: bool
tor_port: int
chunk_size: int
show_download_errors_threshold: float
# paths
music_directory: Path
temp_directory: Path
log_file: Path
not_a_genre_regex: List[str]
ffmpeg_binary: Path

View File

@ -0,0 +1,45 @@
from typing import TypedDict, List
from urllib.parse import ParseResult
from logging import Logger
from pathlib import Path
from ...path_manager import LOCATIONS
from ..config import Config
from ..attributes.attribute import Attribute
from ..attributes.special_attributes import SelectAttribute, PathAttribute, UrlAttribute
config = Config([
UrlAttribute(name="invidious_instance", default_value="https://yt.artemislena.eu", description="""This is an attribute, where you can define the invidious instances,
the youtube downloader should use.
Here is a list of active ones: https://docs.invidious.io/instances/
Instances that use cloudflare or have source code changes could cause issues.
Hidden instances (.onion) will only work, when setting 'tor=true'."""),
UrlAttribute(name="piped_instance", default_value="https://piped-api.privacy.com.de", description="""This is an attribute, where you can define the pioed instances,
the youtube downloader should use.
Here is a list of active ones: https://github.com/TeamPiped/Piped/wiki/Instances
Instances that use cloudflare or have source code changes could cause issues.
Hidden instances (.onion) will only work, when setting 'tor=true"""),
Attribute(name="sleep_after_youtube_403", default_value=30, description="The time to wait, after youtube returned 403 (in seconds)"),
Attribute(name="youtube_music_api_key", default_value="AIzaSyC9XL3ZjWddXya6X74dJoCTL-WEYFDNX30", description="""This is the API key used by YouTube-Music internally.
Dw. if it is empty, Rachel will fetch it automatically for you <333
(she will also update outdated api keys/those that don't work)"""),
Attribute(name="youtube_music_clean_data", default_value=True, description="If set to true, it exclusively fetches artists/albums/songs, not things like user channels etc."),
UrlAttribute(name="youtube_url", default_value=[
"https://www.youtube.com/",
"https://www.youtu.be/"
], description="""This is used to detect, if an url is from youtube, or any alternativ frontend.
If any instance seems to be missing, run music kraken with the -f flag."""),
Attribute(name="use_sponsor_block", default_value=True, description="Use sponsor block to remove adds or simmilar from the youtube videos.")
], LOCATIONS.get_config_file("youtube"))
class SettingsStructure(TypedDict):
# youtube
invidious_instance: ParseResult
piped_instance: ParseResult
sleep_after_youtube_403: float
youtube_music_api_key: str
youtube_music_clean_data: bool
youtube_url: List[ParseResult]
use_sponsor_block: bool

View File

@ -4,9 +4,6 @@ from urllib.parse import ParseResult
from logging import Logger from logging import Logger
from pathlib import Path from pathlib import Path
from .sections.paths import LOCATIONS
from .config import Config
from .base_classes import Section, Attribute
class SettingsStructure(TypedDict): class SettingsStructure(TypedDict):
@ -62,34 +59,3 @@ class SettingsStructure(TypedDict):
log_file: Path log_file: Path
not_a_genre_regex: List[str] not_a_genre_regex: List[str]
ffmpeg_binary: Path ffmpeg_binary: Path
settings: SettingsStructure = {}
config = Config()
set_name_to_value = config.set_name_to_value
def read_config():
if not LOCATIONS.CONFIG_FILE.is_file():
write_config()
config.read_from_config_file(LOCATIONS.CONFIG_FILE)
def write_config():
config.write_to_config_file(LOCATIONS.CONFIG_FILE)
def load():
read_config()
for section in config.config_elements:
if not isinstance(section, Section):
continue
for attribute in section.attribute_list:
if not isinstance(attribute, Attribute):
continue
settings[attribute.name] = attribute.object_from_value

View File

@ -24,5 +24,8 @@ class Locations:
self.FFMPEG_BIN = Path(FFmpeg(enable_log=False).get_ffmpeg_bin()) self.FFMPEG_BIN = Path(FFmpeg(enable_log=False).get_ffmpeg_bin())
def get_config_file(self, config_name: str) -> Path:
return Path(self.CONFIG_DIRECTORY, f"{config_name}.toml")
def get_log_file(self, file_name: os.PathLike) -> Path: def get_log_file(self, file_name: os.PathLike) -> Path:
return Path(self.TEMP_DIRECTORY, file_name) return Path(self.TEMP_DIRECTORY, file_name)

View File

@ -5,108 +5,17 @@ from typing import List, Tuple, Set, Dict
from urllib.parse import ParseResult from urllib.parse import ParseResult
from .path_manager import LOCATIONS from .path_manager import LOCATIONS
from .config import LOGGING_SECTION, AUDIO_SECTION, CONNECTION_SECTION, MISC_SECTION, PATHS_SECTION from .config import main_settings, logging_settings, youtube_settings
from .enums.album import AlbumType from .enums.album import AlbumType
CONFIG_FILE = LOCATIONS.CONFIG_FILE DEBUG = True
if DEBUG:
# modifies the garbage collector to speed up the program print("DEBUG ACTIVE")
# https://mkennedy.codes/posts/python-gc-settings-change-this-and-make-your-app-go-20pc-faster/
# https://web.archive.org/web/20221124122222/https://mkennedy.codes/posts/python-gc-settings-change-this-and-make-your-app-go-20pc-faster/
MODIFY_GC: bool = MISC_SECTION.MODIFY_GC.object_from_value
ID_BITS: int = MISC_SECTION.ID_BITS.object_from_value
ID_RANGE: Tuple[int, int] = (0, int(2 ** ID_BITS))
"""
I will now and then use those messages in the programm.
But I won't overuse them dw.
I will keep those messages, if you disagree with me on the messages,
feel free to fork the programm and edit them, or just edit them in the config
file once I implemented it. (I did it is in ~/.config/music-kraken/music-kraken.conf)
"""
HAPPY_MESSAGES: List[str] = MISC_SECTION.HAPPY_MESSAGES.object_from_value
def get_random_message() -> str: def get_random_message() -> str:
return random.choice(HAPPY_MESSAGES) return random.choice(main_settings['happy_messages'])
TEMP_DIR = PATHS_SECTION.TEMP_DIRECTORY.object_from_value
LOG_PATH = PATHS_SECTION.LOG_PATH.object_from_value
MUSIC_DIR: Path = PATHS_SECTION.MUSIC_DIRECTORY.object_from_value
NOT_A_GENRE_REGEX: Tuple[str] = PATHS_SECTION.NOT_A_GENRE_REGEX.object_from_value
# configure logger default
logging.basicConfig(
level=LOGGING_SECTION.LOG_LEVEL.object_from_value,
format=LOGGING_SECTION.FORMAT.object_from_value,
handlers=[
logging.FileHandler(LOG_PATH),
logging.StreamHandler()
]
)
OBJECT_LOGGER = LOGGING_SECTION.OBJECT_LOGGER.object_from_value
DATABASE_LOGGER = LOGGING_SECTION.DATABASE_LOGGER.object_from_value
YOUTUBE_LOGGER = LOGGING_SECTION.YOUTUBE_LOGGER.object_from_value
YOUTUBE_MUSIC_LOGGER = LOGGING_SECTION.YOUTUBE_MUSIC_LOGGER.object_from_value
MUSIFY_LOGGER = LOGGING_SECTION.MUSIFY_LOGGER.object_from_value
GENIUS_LOGGER = LOGGING_SECTION.GENIUS_LOGGER
ENCYCLOPAEDIA_METALLUM_LOGGER = LOGGING_SECTION.ENCYCLOPAEDIA_METALLUM_LOGGER.object_from_value
DOWNLOAD_LOGGER = LOGGING_SECTION.DOWNLOAD_LOGGER.object_from_value
TAGGING_LOGGER = LOGGING_SECTION.TAGGING_LOGGER.object_from_value
CODEX_LOGGER = LOGGING_SECTION.CODEX_LOGGER.object_from_value
# kB per second
BITRATE = AUDIO_SECTION.BITRATE.object_from_value
AUDIO_FORMAT = AUDIO_SECTION.AUDIO_FORMAT.object_from_value
DOWNLOAD_PATH = AUDIO_SECTION.DOWNLOAD_PATH.object_from_value
DOWNLOAD_FILE = AUDIO_SECTION.DOWNLOAD_FILE.object_from_value
TOR: bool = CONNECTION_SECTION.USE_TOR.object_from_value
PROXIES_LIST: List[Dict[str, str]] = CONNECTION_SECTION.PROXIES.object_from_value
proxies = {}
if len(CONNECTION_SECTION.PROXIES) > 0:
"""
TODO
rotating proxies
"""
proxies = CONNECTION_SECTION.PROXIES.object_from_value[0]
if TOR:
proxies = {
'http': f'socks5h://127.0.0.1:{CONNECTION_SECTION.TOR_PORT.object_from_value}',
'https': f'socks5h://127.0.0.1:{CONNECTION_SECTION.TOR_PORT.object_from_value}'
}
INVIDIOUS_INSTANCE: ParseResult = CONNECTION_SECTION.INVIDIOUS_INSTANCE.object_from_value
PIPED_INSTANCE: ParseResult = CONNECTION_SECTION.PIPED_INSTANCE.object_from_value
ALL_YOUTUBE_URLS: List[ParseResult] = CONNECTION_SECTION.ALL_YOUTUBE_URLS.object_from_value
ENABLE_SPONSOR_BLOCK: bool = CONNECTION_SECTION.SPONSOR_BLOCK.object_from_value
# size of the chunks that are streamed
CHUNK_SIZE = CONNECTION_SECTION.CHUNK_SIZE.object_from_value
# this is a percentage describing the percentage of failed downloads,
# relative to the total downloads.
# If the percentage goes over this threshold DownloadResult returns the download errors
# in the __str__ method
SHOW_DOWNLOAD_ERRORS_THRESHOLD = CONNECTION_SECTION.SHOW_DOWNLOAD_ERRORS_THRESHOLD.object_from_value
SORT_BY_DATE = AUDIO_SECTION.SORT_BY_DATE.object_from_value
SORT_BY_ALBUM_TYPE = AUDIO_SECTION.SORT_BY_ALBUM_TYPE.object_from_value
ALBUM_TYPE_BLACKLIST: Set[AlbumType] = set(AUDIO_SECTION.ALBUM_TYPE_BLACKLIST.object_from_value)
THREADED = False
ENABLE_RESULT_HISTORY: bool = MISC_SECTION.ENABLE_RESULT_HISTORY.object_from_value
HISTORY_LENGTH: int = MISC_SECTION.HISTORY_LENGTH.object_from_value
HELP_MESSAGE = """ HELP_MESSAGE = """
to search: to search:
> s: {query or url} > s: {query or url}
@ -121,14 +30,3 @@ to download:
have fun :3 have fun :3
""".strip() """.strip()
FFMPEG_BINARY: Path = PATHS_SECTION.FFMPEG_BINARY.object_from_value
HASNT_YET_STARTED: bool = MISC_SECTION.HASNT_YET_STARTED.object_from_value
SLEEP_AFTER_YOUTUBE_403: float = CONNECTION_SECTION.SLEEP_AFTER_YOUTUBE_403.object_from_value
DEBUG = True
if DEBUG:
print("DEBUG ACTIVE")
YOUTUBE_MUSIC_CLEAN_DATA: bool = CONNECTION_SECTION.YOUTUBE_MUSIC_CLEAN_DATA.object_from_value

View File

@ -1,13 +1,16 @@
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import List, Tuple from typing import List, Tuple
from ...utils.shared import SHOW_DOWNLOAD_ERRORS_THRESHOLD, DOWNLOAD_LOGGER as LOGGER from ...utils.config import main_settings, logging_settings
from ...objects import Target from ...objects import Target
UNIT_PREFIXES: List[str] = ["", "k", "m", "g", "t"] UNIT_PREFIXES: List[str] = ["", "k", "m", "g", "t"]
UNIT_DIVISOR = 1024 UNIT_DIVISOR = 1024
LOGGER = logging_settings["download_logger"]
@dataclass @dataclass
class DownloadResult: class DownloadResult:
total: int = 0 total: int = 0
@ -44,7 +47,7 @@ class DownloadResult:
if self.is_fatal_error: if self.is_fatal_error:
return True return True
return self.failure_percentage > SHOW_DOWNLOAD_ERRORS_THRESHOLD return self.failure_percentage > main_settings["show_download_errors_threshold"]
def _size_val_unit_pref_ind(self, val: float, ind: int) -> Tuple[float, int]: def _size_val_unit_pref_ind(self, val: float, ind: int) -> Tuple[float, int]:
if val < UNIT_DIVISOR: if val < UNIT_DIVISOR:

6
src/settings.py Normal file
View File

@ -0,0 +1,6 @@
from pathlib import Path
import tomllib
data = tomllib.load(Path("/home/lars/music-kraken.conf").open("r"))
print(data)