music-kraken-core/music_kraken/pages/_abstract.py

161 lines
5.1 KiB
Python
Raw Normal View History

import logging
import random
2024-01-22 17:36:16 +00:00
import re
2023-04-03 08:38:12 +00:00
from copy import copy
2024-01-15 09:56:59 +00:00
from pathlib import Path
2024-05-10 15:06:40 +00:00
from typing import Optional, Union, Type, Dict, Set, List, Tuple, TypedDict
2023-06-15 09:28:35 +00:00
from string import Formatter
2024-05-10 15:06:40 +00:00
from dataclasses import dataclass, field
import requests
from bs4 import BeautifulSoup
2023-04-20 17:45:29 +00:00
from ..connection import Connection
2023-03-10 09:13:35 +00:00
from ..objects import (
2023-01-23 13:53:35 +00:00
Song,
Source,
Album,
Artist,
Target,
2023-03-24 14:58:21 +00:00
DatabaseObject,
2023-03-20 13:40:32 +00:00
Options,
2023-03-24 13:28:19 +00:00
Collection,
2023-04-04 08:20:54 +00:00
Label,
2023-01-23 13:53:35 +00:00
)
2024-05-14 13:18:17 +00:00
from ..utils.enums import SourceType
2023-04-18 10:14:34 +00:00
from ..utils.enums.album import AlbumType
from ..audio import write_metadata_to_target, correct_codec
2023-09-10 14:27:09 +00:00
from ..utils.config import main_settings
2023-10-23 14:21:44 +00:00
from ..utils.support_classes.query import Query
from ..utils.support_classes.download_result import DownloadResult
2023-09-13 18:02:36 +00:00
from ..utils.string_processing import fit_to_file_system
2024-05-10 15:33:07 +00:00
from ..utils import trace, output, BColors
2023-04-04 17:17:58 +00:00
2023-05-23 16:09:53 +00:00
INDEPENDENT_DB_OBJECTS = Union[Label, Album, Artist, Song]
INDEPENDENT_DB_TYPES = Union[Type[Song], Type[Album], Type[Artist], Type[Label]]
2024-05-10 15:06:40 +00:00
@dataclass
class FetchOptions:
download_all: bool = False
album_type_blacklist: Set[AlbumType] = field(default_factory=lambda: set(AlbumType(a) for a in main_settings["album_type_blacklist"]))
@dataclass
class DownloadOptions:
download_all: bool = False
album_type_blacklist: Set[AlbumType] = field(default_factory=lambda: set(AlbumType(a) for a in main_settings["album_type_blacklist"]))
process_audio_if_found: bool = False
process_metadata_if_found: bool = True
2023-05-24 06:50:56 +00:00
2023-06-12 17:46:46 +00:00
class Page:
REGISTER = True
2024-05-14 13:18:17 +00:00
SOURCE_TYPE: SourceType
2024-05-15 11:16:11 +00:00
LOGGER: logging.Logger
2024-05-14 13:18:17 +00:00
def __new__(cls, *args, **kwargs):
cls.LOGGER = logging.getLogger(cls.__name__)
return super().__new__(cls)
2023-12-29 14:43:33 +00:00
2024-05-10 15:06:40 +00:00
def __init__(self, download_options: DownloadOptions = None, fetch_options: FetchOptions = None):
2024-05-15 11:16:11 +00:00
self.SOURCE_TYPE.register_page(self)
2024-05-10 15:06:40 +00:00
self.download_options: DownloadOptions = download_options or DownloadOptions()
self.fetch_options: FetchOptions = fetch_options or FetchOptions()
def __del__(self):
self.SOURCE_TYPE.deregister_page(self)
2024-01-22 17:36:16 +00:00
def _search_regex(self, pattern, string, default=None, fatal=True, flags=0, group=None):
"""
Perform a regex search on the given string, using a single or a list of
patterns returning the first matching group.
In case of failure return a default value or raise a WARNING or a
RegexNotFoundError, depending on fatal, specifying the field name.
"""
if isinstance(pattern, str):
mobj = re.search(pattern, string, flags)
else:
for p in pattern:
mobj = re.search(p, string, flags)
if mobj:
break
if mobj:
if group is None:
# return the first matching group
return next(g for g in mobj.groups() if g is not None)
elif isinstance(group, (list, tuple)):
return tuple(mobj.group(g) for g in group)
else:
return mobj.group(group)
return default
2023-05-24 08:12:03 +00:00
def get_source_type(self, source: Source) -> Optional[Type[DatabaseObject]]:
2023-05-24 06:50:56 +00:00
return None
2024-01-15 11:48:36 +00:00
2023-05-24 06:50:56 +00:00
def get_soup_from_response(self, r: requests.Response) -> BeautifulSoup:
return BeautifulSoup(r.content, "html.parser")
2023-05-23 14:21:12 +00:00
# to search stuff
def search(self, query: Query) -> List[DatabaseObject]:
music_object = query.music_object
2024-01-15 11:48:36 +00:00
2023-05-23 14:21:12 +00:00
search_functions = {
Song: self.song_search,
Album: self.album_search,
Artist: self.artist_search,
Label: self.label_search
}
2024-01-15 11:48:36 +00:00
2023-05-23 14:21:12 +00:00
if type(music_object) in search_functions:
r = search_functions[type(music_object)](music_object)
if r is not None and len(r) > 0:
2023-05-23 14:21:12 +00:00
return r
2024-01-15 11:48:36 +00:00
2023-05-23 14:21:12 +00:00
r = []
2023-05-23 08:49:52 +00:00
for default_query in query.default_search:
for single_option in self.general_search(default_query):
r.append(single_option)
2024-01-15 11:48:36 +00:00
2023-05-23 14:50:54 +00:00
return r
2024-01-15 11:48:36 +00:00
2023-05-23 14:21:12 +00:00
def general_search(self, search_query: str) -> List[DatabaseObject]:
return []
2024-01-15 11:48:36 +00:00
2023-05-23 14:21:12 +00:00
def label_search(self, label: Label) -> List[Label]:
return []
2024-01-15 11:48:36 +00:00
2023-05-23 14:21:12 +00:00
def artist_search(self, artist: Artist) -> List[Artist]:
return []
2024-01-15 11:48:36 +00:00
2023-05-23 14:21:12 +00:00
def album_search(self, album: Album) -> List[Album]:
2023-05-23 08:49:52 +00:00
return []
2024-01-15 11:48:36 +00:00
2023-05-23 14:21:12 +00:00
def song_search(self, song: Song) -> List[Song]:
return []
2023-01-23 13:53:35 +00:00
# to fetch stuff
2023-05-24 06:50:56 +00:00
def fetch_song(self, source: Source, stop_at_level: int = 1) -> Song:
return Song()
2023-03-20 13:40:32 +00:00
2023-05-24 06:50:56 +00:00
def fetch_album(self, source: Source, stop_at_level: int = 1) -> Album:
return Album()
2023-04-03 17:59:31 +00:00
2023-05-24 06:50:56 +00:00
def fetch_artist(self, source: Source, stop_at_level: int = 1) -> Artist:
return Artist()
2023-01-23 14:52:50 +00:00
2023-05-24 06:50:56 +00:00
def fetch_label(self, source: Source, stop_at_level: int = 1) -> Label:
return Label()
2023-03-24 14:58:21 +00:00
2024-05-15 11:16:11 +00:00
# to download stuff
2023-06-15 07:58:48 +00:00
def get_skip_intervals(self, song: Song, source: Source) -> List[Tuple[float, float]]:
return []
2024-01-15 11:48:36 +00:00
def post_process_hook(self, song: Song, temp_target: Target, **kwargs):
pass
2024-01-15 11:48:36 +00:00
2023-05-25 09:21:39 +00:00
def download_song_to_target(self, source: Source, target: Target, desc: str = None) -> DownloadResult:
2023-04-04 18:58:22 +00:00
return DownloadResult()