fixed funny bug
This commit is contained in:
Hellow
parent
5387301ed2
commit
267bf52847
@ -38,18 +38,18 @@ logging.getLogger("musicbrainzngs").setLevel(logging.WARNING)
|
||||
musicbrainzngs.set_useragent("metadata receiver", "0.1", "https://github.com/HeIIow2/music-downloader")
|
||||
|
||||
|
||||
def get_options_from_query(query: str) -> List[objects.MusicObject]:
|
||||
def get_options_from_query(query: str) -> List[objects.DatabaseObject]:
|
||||
options = []
|
||||
for MetadataPage in pages.MetadataPages:
|
||||
options.extend(MetadataPage.search_by_query(query=query))
|
||||
return options
|
||||
|
||||
def get_options_from_option(option: objects.MusicObject) -> List[objects.MusicObject]:
|
||||
def get_options_from_option(option: objects.DatabaseObject) -> List[objects.DatabaseObject]:
|
||||
for MetadataPage in pages.MetadataPages:
|
||||
option = MetadataPage.fetch_details(option, flat=False)
|
||||
return option.get_options()
|
||||
|
||||
def print_options(options: List[objects.MusicObject]):
|
||||
def print_options(options: List[objects.DatabaseObject]):
|
||||
print("\n".join([f"{str(j).zfill(2)}: {i.get_option_string()}" for j, i in enumerate(options)]))
|
||||
|
||||
def cli():
|
||||
|
||||
@ -1,19 +0,0 @@
|
||||
from . import database
|
||||
from .. import objects
|
||||
|
||||
MusicObject = objects.MusicObject
|
||||
|
||||
ID3Timestamp = objects.ID3Timestamp
|
||||
SourceTypes = objects.SourceTypes
|
||||
SourcePages = objects.SourcePages
|
||||
Song = objects.Song
|
||||
Source = objects.Source
|
||||
Target = objects.Target
|
||||
Lyrics = objects.Lyrics
|
||||
Album = objects.Album
|
||||
Artist = objects.Artist
|
||||
|
||||
FormattedText = objects.FormattedText
|
||||
|
||||
Database = database.Database
|
||||
# cache = temp_database.TempDatabase()
|
||||
@ -130,7 +130,7 @@ class Database:
|
||||
|
||||
print(model._meta.fields)
|
||||
|
||||
def push(self, database_object: objects.MusicObject):
|
||||
def push(self, database_object: objects.DatabaseObject):
|
||||
"""
|
||||
Adds a new music object to the database using the corresponding method from the `write` session.
|
||||
When possible, rather use the `push_many` function.
|
||||
@ -153,7 +153,7 @@ class Database:
|
||||
if isinstance(database_object, objects.Artist):
|
||||
return writing_session.add_artist(database_object)
|
||||
|
||||
def push_many(self, database_objects: List[objects.MusicObject]) -> None:
|
||||
def push_many(self, database_objects: List[objects.DatabaseObject]) -> None:
|
||||
"""
|
||||
Adds a list of MusicObject instances to the database.
|
||||
This function sends only needs one querry for each type of table added.
|
||||
|
||||
@ -2,7 +2,7 @@ from collections import defaultdict
|
||||
from typing import Dict, List, Optional
|
||||
import weakref
|
||||
|
||||
from src.music_kraken.objects import MusicObject
|
||||
from src.music_kraken.objects import DatabaseObject
|
||||
|
||||
"""
|
||||
This is a cache for the objects, that et pulled out of the database.
|
||||
@ -32,14 +32,14 @@ class ObjectCache:
|
||||
:method extent: Add a list of MusicObjects to the cache.
|
||||
:method remove: Remove a MusicObject from the cache by its id.
|
||||
:method get: Retrieve a MusicObject from the cache by its id. """
|
||||
object_to_id: Dict[str, MusicObject]
|
||||
object_to_id: Dict[str, DatabaseObject]
|
||||
weakref_map: Dict[weakref.ref, str]
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.object_to_id = dict()
|
||||
self.weakref_map = defaultdict()
|
||||
|
||||
def exists(self, music_object: MusicObject) -> bool:
|
||||
def exists(self, music_object: DatabaseObject) -> bool:
|
||||
"""
|
||||
Check if a MusicObject with the same id already exists in the cache.
|
||||
|
||||
@ -60,7 +60,7 @@ class ObjectCache:
|
||||
data_id = self.weakref_map.pop(weakref_)
|
||||
self.object_to_id.pop(data_id)
|
||||
|
||||
def append(self, music_object: MusicObject) -> bool:
|
||||
def append(self, music_object: DatabaseObject) -> bool:
|
||||
"""
|
||||
Add a MusicObject to the cache.
|
||||
|
||||
@ -75,7 +75,7 @@ class ObjectCache:
|
||||
|
||||
return False
|
||||
|
||||
def extent(self, music_object_list: List[MusicObject]):
|
||||
def extent(self, music_object_list: List[DatabaseObject]):
|
||||
"""
|
||||
adjacent to the extent method of list, this appends n Object
|
||||
"""
|
||||
@ -93,7 +93,7 @@ class ObjectCache:
|
||||
self.weakref_map.pop(weakref.ref(data))
|
||||
self.object_to_id.pop(_id)
|
||||
|
||||
def __getitem__(self, item) -> Optional[MusicObject]:
|
||||
def __getitem__(self, item) -> Optional[DatabaseObject]:
|
||||
"""
|
||||
this returns the data obj
|
||||
:param item: the id of the music object
|
||||
@ -102,5 +102,5 @@ class ObjectCache:
|
||||
|
||||
return self.object_to_id.get(item)
|
||||
|
||||
def get(self, _id: str) -> Optional[MusicObject]:
|
||||
def get(self, _id: str) -> Optional[DatabaseObject]:
|
||||
return self.__getitem__(_id)
|
||||
|
||||
@ -9,7 +9,7 @@ from . import (
|
||||
collection
|
||||
)
|
||||
|
||||
MusicObject = parents.DatabaseObject
|
||||
DatabaseObject = parents.DatabaseObject
|
||||
|
||||
ID3Mapping = metadata.Mapping
|
||||
ID3Timestamp = metadata.ID3Timestamp
|
||||
|
||||
@ -1,9 +1,16 @@
|
||||
from typing import List, Iterable, Dict
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass
|
||||
|
||||
from .parents import DatabaseObject
|
||||
|
||||
|
||||
@dataclass
|
||||
class AppendResult:
|
||||
was_in_collection: bool
|
||||
current_element: DatabaseObject
|
||||
|
||||
|
||||
class Collection:
|
||||
"""
|
||||
This a class for the iterables
|
||||
@ -62,7 +69,8 @@ class Collection:
|
||||
except KeyError:
|
||||
pass
|
||||
|
||||
def append(self, element: DatabaseObject, merge_on_conflict: bool = True, merge_into_existing: bool = True) -> DatabaseObject:
|
||||
def append(self, element: DatabaseObject, merge_on_conflict: bool = True,
|
||||
merge_into_existing: bool = True) -> AppendResult:
|
||||
"""
|
||||
:param element:
|
||||
:param merge_on_conflict:
|
||||
@ -79,7 +87,7 @@ class Collection:
|
||||
existing_object = self._attribute_to_object_map[name][value]
|
||||
|
||||
if not merge_on_conflict:
|
||||
return existing_object
|
||||
return AppendResult(True, existing_object)
|
||||
|
||||
# if the object does already exist
|
||||
# thus merging and don't add it afterwards
|
||||
@ -87,7 +95,7 @@ class Collection:
|
||||
existing_object.merge(element)
|
||||
# in case any relevant data has been added (e.g. it remaps the old object)
|
||||
self.map_element(existing_object)
|
||||
return existing_object
|
||||
return AppendResult(True, existing_object)
|
||||
|
||||
element.merge(existing_object)
|
||||
|
||||
@ -96,22 +104,20 @@ class Collection:
|
||||
|
||||
self.unmap_element(existing_object)
|
||||
self.map_element(element)
|
||||
return element
|
||||
return AppendResult(True, existing_object)
|
||||
|
||||
self._data.append(element)
|
||||
self.map_element(element)
|
||||
|
||||
return element
|
||||
return AppendResult(False, element)
|
||||
|
||||
def append_is_already_in_collection(self, element: DatabaseObject, merge_on_conflict: bool = True, merge_into_existing: bool = True) -> bool:
|
||||
object_representing_the_data = self.append(element, merge_on_conflict=merge_on_conflict, merge_into_existing=merge_into_existing)
|
||||
|
||||
def extend(self, element_list: Iterable[DatabaseObject], merge_on_conflict: bool = True):
|
||||
def extend(self, element_list: Iterable[DatabaseObject], merge_on_conflict: bool = True,
|
||||
merge_into_existing: bool = True):
|
||||
for element in element_list:
|
||||
self.append(element, merge_on_conflict=merge_on_conflict)
|
||||
self.append(element, merge_on_conflict=merge_on_conflict, merge_into_existing=merge_into_existing)
|
||||
|
||||
def __iter__(self):
|
||||
for element in self._data:
|
||||
for element in self.shallow_list:
|
||||
yield element
|
||||
|
||||
def __str__(self) -> str:
|
||||
@ -120,11 +126,22 @@ class Collection:
|
||||
def __len__(self) -> int:
|
||||
return len(self._data)
|
||||
|
||||
def __getitem__(self, item):
|
||||
if type(item) != int:
|
||||
def __getitem__(self, key):
|
||||
if type(key) != int:
|
||||
return ValueError("key needs to be an integer")
|
||||
|
||||
return self._data[item]
|
||||
return self._data[key]
|
||||
|
||||
def __setitem__(self, key, value: DatabaseObject):
|
||||
print(key, value)
|
||||
if type(key) != int:
|
||||
return ValueError("key needs to be an integer")
|
||||
|
||||
old_item = self._data[key]
|
||||
self.unmap_element(old_item)
|
||||
self.map_element(value)
|
||||
|
||||
self._data[key] = value
|
||||
|
||||
@property
|
||||
def shallow_list(self) -> List[DatabaseObject]:
|
||||
|
||||
@ -99,6 +99,7 @@ class DatabaseObject:
|
||||
pass
|
||||
|
||||
|
||||
|
||||
class MainObject(DatabaseObject):
|
||||
"""
|
||||
This is the parent class for all "main" data objects:
|
||||
|
||||
@ -1,11 +1,8 @@
|
||||
from typing import Optional, Union, Type
|
||||
from typing import Optional, Union, Type, Dict
|
||||
import requests
|
||||
import logging
|
||||
|
||||
LOGGER = logging.getLogger("this shouldn't be used")
|
||||
|
||||
from ..utils import shared
|
||||
|
||||
from ..objects import (
|
||||
Song,
|
||||
Source,
|
||||
@ -13,13 +10,15 @@ from ..objects import (
|
||||
Artist,
|
||||
Lyrics,
|
||||
Target,
|
||||
MusicObject,
|
||||
DatabaseObject,
|
||||
Options,
|
||||
SourcePages,
|
||||
Collection,
|
||||
Label
|
||||
)
|
||||
|
||||
LOGGER = logging.getLogger("this shouldn't be used")
|
||||
|
||||
|
||||
class Page:
|
||||
"""
|
||||
@ -139,7 +138,7 @@ class Page:
|
||||
return Options()
|
||||
|
||||
@classmethod
|
||||
def fetch_details(cls, music_object: Union[Song, Album, Artist, Label], stop_at_level: int = 1) -> MusicObject:
|
||||
def fetch_details(cls, music_object: Union[Song, Album, Artist, Label], stop_at_level: int = 1) -> DatabaseObject:
|
||||
"""
|
||||
when a music object with laccing data is passed in, it returns
|
||||
the SAME object **(no copy)** with more detailed data.
|
||||
@ -156,21 +155,28 @@ class Page:
|
||||
:return detailed_music_object: IT MODIFIES THE INPUT OBJ
|
||||
"""
|
||||
|
||||
new_music_object: MusicObject = type(music_object).__init__()
|
||||
new_music_object: DatabaseObject = type(music_object)()
|
||||
|
||||
source: Source
|
||||
for source in music_object.source_collection:
|
||||
new_music_object.merge(cls.fetch_object_from_source(source=source, obj_type=type(music_object), stop_at_level=stop_at_level))
|
||||
new_music_object.merge(cls._fetch_object_from_source(source=source, obj_type=type(music_object), stop_at_level=stop_at_level))
|
||||
|
||||
collections = {
|
||||
Label: Collection(element_type=Label),
|
||||
Artist: Collection(element_type=Artist),
|
||||
Album: Collection(element_type=Album),
|
||||
Song: Collection(element_type=Song)
|
||||
}
|
||||
|
||||
cls._clean_music_object(new_music_object, collections)
|
||||
|
||||
music_object.merge(new_music_object)
|
||||
music_object.compile()
|
||||
# music_object.compile()
|
||||
|
||||
return music_object
|
||||
|
||||
@classmethod
|
||||
def fetch_object_from_source(cls, source: Source, obj_type: Union[Type[Song], Type[Album], Type[Artist], Type[Label]], stop_at_level: int = 1):
|
||||
def _fetch_object_from_source(cls, source: Source, obj_type: Union[Type[Song], Type[Album], Type[Artist], Type[Label]], stop_at_level: int = 1):
|
||||
if obj_type == Artist:
|
||||
return cls.fetch_artist_from_source(source=source, stop_at_level=stop_at_level)
|
||||
|
||||
@ -183,6 +189,54 @@ class Page:
|
||||
if obj_type == Label:
|
||||
return cls.fetch_label_from_source(source=source, stop_at_level=stop_at_level)
|
||||
|
||||
@classmethod
|
||||
def _clean_music_object(cls, music_object: Union[Label, Album, Artist, Song], collections: Dict[Union[Type[Song], Type[Album], Type[Artist], Type[Label]], Collection]):
|
||||
if type(music_object) == Label:
|
||||
return cls._clean_label(label=music_object, collections=collections)
|
||||
if type(music_object) == Artist:
|
||||
return cls._clean_artist(artist=music_object, collections=collections)
|
||||
if type(music_object) == Album:
|
||||
return cls._clean_album(album=music_object, collections=collections)
|
||||
if type(music_object) == Song:
|
||||
return cls._clean_song(song=music_object, collections=collections)
|
||||
|
||||
@classmethod
|
||||
def _clean_collection(cls, collection: Collection, collection_dict: Dict[Union[Type[Song], Type[Album], Type[Artist], Type[Label]], Collection]):
|
||||
if collection.element_type not in collection_dict:
|
||||
return
|
||||
|
||||
for i, element in enumerate(collection):
|
||||
r = collection_dict[collection.element_type].append(element)
|
||||
if not r.was_in_collection:
|
||||
cls._clean_music_object(r.current_element, collection_dict)
|
||||
continue
|
||||
|
||||
collection[i] = r.current_element
|
||||
cls._clean_music_object(r.current_element, collection_dict)
|
||||
|
||||
@classmethod
|
||||
def _clean_label(cls, label: Label, collections: Dict[Union[Type[Song], Type[Album], Type[Artist], Type[Label]], Collection]):
|
||||
cls._clean_collection(label.current_artist_collection, collections)
|
||||
cls._clean_collection(label.album_collection, collections)
|
||||
|
||||
@classmethod
|
||||
def _clean_artist(cls, artist: Artist, collections: Dict[Union[Type[Song], Type[Album], Type[Artist], Type[Label]], Collection]):
|
||||
cls._clean_collection(artist.main_album_collection, collections)
|
||||
cls._clean_collection(artist.feature_song_collection, collections)
|
||||
cls._clean_collection(artist.label_collection, collections)
|
||||
|
||||
@classmethod
|
||||
def _clean_album(cls, album: Album, collections: Dict[Union[Type[Song], Type[Album], Type[Artist], Type[Label]], Collection]):
|
||||
cls._clean_collection(album.label_collection, collections)
|
||||
cls._clean_collection(album.song_collection, collections)
|
||||
cls._clean_collection(album.artist_collection, collections)
|
||||
|
||||
@classmethod
|
||||
def _clean_song(cls, song: Song, collections: Dict[Union[Type[Song], Type[Album], Type[Artist], Type[Label]], Collection]):
|
||||
cls._clean_collection(song.album_collection, collections)
|
||||
cls._clean_collection(song.feature_artist_collection, collections)
|
||||
cls._clean_collection(song.main_artist_collection, collections)
|
||||
|
||||
@classmethod
|
||||
def fetch_song_from_source(cls, source: Source, stop_at_level: int = 1) -> Song:
|
||||
return Song()
|
||||
@ -196,5 +250,6 @@ class Page:
|
||||
def fetch_artist_from_source(cls, source: Source, stop_at_level: int = 1) -> Artist:
|
||||
return Artist()
|
||||
|
||||
def fetch_label_from_source(source: Source, stop_at_level: int = 1) -> Label:
|
||||
@classmethod
|
||||
def fetch_label_from_source(cls, source: Source, stop_at_level: int = 1) -> Label:
|
||||
return Label()
|
||||
|
||||
@ -9,7 +9,7 @@ from ..utils.shared import (
|
||||
|
||||
from .abstract import Page
|
||||
from ..objects import (
|
||||
MusicObject,
|
||||
DatabaseObject,
|
||||
Artist,
|
||||
Source,
|
||||
SourcePages,
|
||||
|
||||
@ -14,7 +14,7 @@ from ..utils.shared import (
|
||||
|
||||
from .abstract import Page
|
||||
from ..objects import (
|
||||
MusicObject,
|
||||
DatabaseObject,
|
||||
Artist,
|
||||
Source,
|
||||
SourcePages,
|
||||
@ -545,7 +545,7 @@ class Musify(Page):
|
||||
))
|
||||
|
||||
@classmethod
|
||||
def get_discography(cls, url: MusifyUrl, artist_name: str = None, flat=False) -> List[Album]:
|
||||
def get_discography(cls, url: MusifyUrl, artist_name: str = None, stop_at_level: int = 1) -> List[Album]:
|
||||
"""
|
||||
POST https://musify.club/artist/filteralbums
|
||||
ArtistID: 280348
|
||||
@ -570,9 +570,9 @@ class Musify(Page):
|
||||
for card_soup in soup.find_all("div", {"class": "card"}):
|
||||
new_album: Album = cls.parse_album_card(card_soup, artist_name)
|
||||
album_source: Source
|
||||
if not flat:
|
||||
if stop_at_level > 1:
|
||||
for album_source in new_album.source_collection.get_sources_from_page(cls.SOURCE_TYPE):
|
||||
new_album.merge(cls.fetch_album_from_source(album_source))
|
||||
new_album.merge(cls.fetch_album_from_source(album_source, stop_at_level=stop_at_level-1))
|
||||
|
||||
discography.append(new_album)
|
||||
|
||||
@ -709,7 +709,7 @@ class Musify(Page):
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def fetch_artist_from_source(cls, source: Source, flat: bool = False) -> Artist:
|
||||
def fetch_artist_from_source(cls, source: Source, stop_at_level: int = 1) -> Artist:
|
||||
"""
|
||||
fetches artist from source
|
||||
|
||||
@ -719,7 +719,7 @@ class Musify(Page):
|
||||
|
||||
Args:
|
||||
source (Source): the source to fetch
|
||||
flat (bool, optional): if it is false, every album from discograohy will be fetched. Defaults to False.
|
||||
stop_at_level: int = 1: if it is false, every album from discograohy will be fetched. Defaults to False.
|
||||
|
||||
Returns:
|
||||
Artist: the artist fetched
|
||||
@ -851,7 +851,7 @@ class Musify(Page):
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def fetch_album_from_source(cls, source: Source, flat: bool = False) -> Album:
|
||||
def fetch_album_from_source(cls, source: Source, stop_at_level: int = 1) -> Album:
|
||||
"""
|
||||
fetches album from source:
|
||||
eg. 'https://musify.club/release/linkin-park-hybrid-theory-2000-188'
|
||||
@ -861,8 +861,8 @@ class Musify(Page):
|
||||
[] attributes
|
||||
[] ratings
|
||||
|
||||
:param stop_at_level:
|
||||
:param source:
|
||||
:param flat:
|
||||
:return:
|
||||
"""
|
||||
album = Album(title="Hi :)")
|
||||
|
||||
@ -22,9 +22,15 @@ def fetch_album():
|
||||
"https://musify.club/release/linkin-park-hybrid-theory-2000-188")]
|
||||
)
|
||||
|
||||
album = Musify.fetch_details(album)
|
||||
album: objects.Album = Musify.fetch_details(album)
|
||||
print(album.options)
|
||||
|
||||
song: objects.Song
|
||||
for artist in album.artist_collection:
|
||||
print(artist.id, artist.name)
|
||||
for song in album.song_collection:
|
||||
for artist in song.main_artist_collection:
|
||||
print(artist.id, artist.name)
|
||||
|
||||
if __name__ == "__main__":
|
||||
fetch_album()
|
||||
|
||||
Loading…
Reference in New Issue
Block a user