mirror of
https://github.com/Dvorinka/swingmusic-extended.git
synced 2026-06-03 20:13:02 +00:00
rewrite album and artist stores using in-mem hashmap stores
This commit is contained in:
+36
-27
@@ -4,26 +4,24 @@ Contains all the album routes.
|
||||
|
||||
import random
|
||||
|
||||
from pydantic import Field
|
||||
from pydantic import BaseModel, Field
|
||||
from flask_openapi3 import Tag
|
||||
from flask_openapi3 import APIBlueprint
|
||||
from app.api.apischemas import AlbumHashSchema, AlbumLimitSchema, ArtistHashSchema
|
||||
|
||||
from app.config import UserConfig
|
||||
from app.db.libdata import ArtistTable
|
||||
from app.db.libdata import AlbumTable as AlbumDb, TrackTable as TrackDb
|
||||
from app.db.userdata import SimilarArtistTable
|
||||
from app.models.album import Album
|
||||
from app.settings import Defaults
|
||||
from app.utils import flatten
|
||||
from app.store.albums import AlbumStore
|
||||
from app.store.artists import ArtistStore
|
||||
from app.store.tracks import TrackStore
|
||||
from app.utils.hashing import create_hash
|
||||
from app.lib.albumslib import sort_by_track_no
|
||||
from app.serializers.album import serialize_for_card_many
|
||||
from app.serializers.track import serialize_tracks
|
||||
from app.db.sqlite.albumcolors import SQLiteAlbumMethods as adb
|
||||
from app.db.sqlite.favorite import SQLiteFavoriteMethods as favdb
|
||||
from app.db.sqlite.lastfm.similar_artists import SQLiteLastFMSimilarArtists as lastfmdb
|
||||
|
||||
get_albums_by_albumartist = adb.get_albums_by_albumartist
|
||||
check_is_fav = favdb.check_is_favorite
|
||||
|
||||
bp_tag = Tag(name="Album", description="Single album")
|
||||
@@ -39,12 +37,14 @@ def get_album_tracks_and_info(body: AlbumHashSchema):
|
||||
Returns album info and tracks for the given albumhash.
|
||||
"""
|
||||
albumhash = body.albumhash
|
||||
album = AlbumDb.get_album_by_albumhash(albumhash)
|
||||
# album = AlbumDb.get_album_by_albumhash(albumhash)
|
||||
albumentry = AlbumStore.albummap.get(albumhash)
|
||||
|
||||
if album is None:
|
||||
if albumentry is None:
|
||||
return {"error": "Album not found"}, 404
|
||||
|
||||
tracks = TrackDb.get_tracks_by_albumhash(albumhash)
|
||||
album = albumentry.album
|
||||
tracks = TrackStore.get_tracks_by_trackhashes(albumentry.trackhashes)
|
||||
album.trackcount = len(tracks)
|
||||
album.duration = sum(t.duration for t in tracks)
|
||||
album.check_type(
|
||||
@@ -52,6 +52,7 @@ def get_album_tracks_and_info(body: AlbumHashSchema):
|
||||
)
|
||||
|
||||
track_total = sum({int(t.extra.get("track_total", 1) or 1) for t in tracks})
|
||||
avg_bitrate = sum(t.bitrate for t in tracks) // (len(tracks) or 1)
|
||||
|
||||
return {
|
||||
"info": album,
|
||||
@@ -61,7 +62,7 @@ def get_album_tracks_and_info(body: AlbumHashSchema):
|
||||
# 1. All the tracks have the correct track totals
|
||||
# 2. Tracks with the same track total are from the same disc
|
||||
"track_total": track_total,
|
||||
"avg_bitrate": sum(t.bitrate for t in tracks) // len(tracks),
|
||||
"avg_bitrate": avg_bitrate,
|
||||
},
|
||||
"copyright": tracks[0].copyright,
|
||||
"tracks": serialize_tracks(tracks, remove_disc=False),
|
||||
@@ -76,7 +77,7 @@ def get_album_tracks(path: AlbumHashSchema):
|
||||
Returns all the tracks in the given album, sorted by disc and track number.
|
||||
NOTE: No album info is returned.
|
||||
"""
|
||||
tracks = TrackDb.get_tracks_by_albumhash(path.albumhash)
|
||||
tracks = AlbumStore.get_album_tracks(path.albumhash)
|
||||
tracks = sort_by_track_no(tracks)
|
||||
|
||||
return serialize_tracks(tracks)
|
||||
@@ -105,7 +106,11 @@ def get_more_from_artist(body: GetMoreFromArtistsBody):
|
||||
limit = body.limit
|
||||
base_title = body.base_title
|
||||
|
||||
all_albums = AlbumDb.get_albums_by_artisthashes(albumartists)
|
||||
all_albums: dict[str, list[Album]] = {}
|
||||
|
||||
for artisthash in albumartists:
|
||||
all_albums[artisthash] = AlbumStore.get_albums_by_artisthash(artisthash)
|
||||
|
||||
seen_hashes = set()
|
||||
|
||||
for artisthash, albums in all_albums.items():
|
||||
@@ -128,14 +133,15 @@ def get_more_from_artist(body: GetMoreFromArtistsBody):
|
||||
return all_albums
|
||||
|
||||
|
||||
class GetAlbumVersionsBody(ArtistHashSchema):
|
||||
class GetAlbumVersionsBody(BaseModel):
|
||||
og_album_title: str = Field(
|
||||
description="The original album title (album.og_title)",
|
||||
example=Defaults.API_ALBUMNAME,
|
||||
)
|
||||
base_title: str = Field(
|
||||
description="The base title of the album to exclude from the results.",
|
||||
example=Defaults.API_ALBUMNAME,
|
||||
|
||||
albumhash: str = Field(
|
||||
description="The album hash of the album to exclude from the results.",
|
||||
example=Defaults.API_ALBUMHASH,
|
||||
)
|
||||
|
||||
|
||||
@@ -146,18 +152,23 @@ def get_album_versions(body: GetAlbumVersionsBody):
|
||||
|
||||
Returns other versions of the given album.
|
||||
"""
|
||||
og_album_title = body.og_album_title
|
||||
base_title = body.base_title
|
||||
artisthash = body.artisthash
|
||||
albumhash = body.albumhash
|
||||
|
||||
albums = AlbumDb.get_albums_by_base_title(base_title)
|
||||
album = AlbumStore.albummap.get(albumhash)
|
||||
if not album:
|
||||
return []
|
||||
artisthash = album.album.artisthashes[0]
|
||||
albums = AlbumStore.get_albums_by_artisthash(artisthash)
|
||||
|
||||
basetitle = album.basetitle
|
||||
albums = [
|
||||
a
|
||||
for a in albums
|
||||
if a.og_title != og_album_title
|
||||
if a.og_title != album.album.og_title
|
||||
if a.base_title == basetitle
|
||||
and artisthash in {a["artisthash"] for a in a.albumartists}
|
||||
]
|
||||
print(albums)
|
||||
|
||||
return serialize_for_card_many(albums)
|
||||
|
||||
|
||||
@@ -181,10 +192,8 @@ def get_similar_albums(query: GetSimilarAlbumsQuery):
|
||||
return []
|
||||
|
||||
artisthashes = similar_artists.get_artist_hash_set()
|
||||
artists = ArtistTable.get_artists_by_artisthashes(artisthashes)
|
||||
|
||||
albums = AlbumDb.get_albums_by_artisthashes([a.artisthash for a in artists])
|
||||
albums = flatten(albums.values())
|
||||
artists = ArtistStore.get_artists_by_hashes(artisthashes)
|
||||
albums = AlbumStore.get_albums_by_artisthashes([a.artisthash for a in artists])
|
||||
sample = random.sample(albums, min(len(albums), limit))
|
||||
|
||||
return serialize_for_card_many(sample[:limit])
|
||||
|
||||
+17
-14
@@ -17,14 +17,16 @@ from app.api.apischemas import (
|
||||
)
|
||||
|
||||
from app.config import UserConfig
|
||||
from app.db.libdata import ArtistTable
|
||||
from app.db.libdata import AlbumTable, TrackTable
|
||||
from app.db.userdata import SimilarArtistTable
|
||||
|
||||
from app.serializers.album import serialize_for_card_many
|
||||
from app.serializers.artist import serialize_for_cards
|
||||
from app.serializers.track import serialize_tracks
|
||||
|
||||
from app.store.albums import AlbumStore
|
||||
from app.store.artists import ArtistStore
|
||||
from app.store.tracks import TrackStore
|
||||
|
||||
bp_tag = Tag(name="Artist", description="Single artist")
|
||||
api = APIBlueprint("artist", __name__, url_prefix="/artist", abp_tags=[bp_tag])
|
||||
|
||||
@@ -39,13 +41,15 @@ def get_artist(path: ArtistHashSchema, query: TrackLimitSchema):
|
||||
artisthash = path.artisthash
|
||||
limit = query.limit
|
||||
|
||||
artist = ArtistTable.get_artist_by_hash(artisthash)
|
||||
if artist is None:
|
||||
entry = ArtistStore.artistmap.get(artisthash)
|
||||
|
||||
if entry is None:
|
||||
return {"error": "Artist not found"}, 404
|
||||
|
||||
tracks = TrackTable.get_tracks_by_artisthash(artisthash)
|
||||
tracks = TrackStore.get_tracks_by_trackhashes(entry.trackhashes)
|
||||
tcount = len(tracks)
|
||||
|
||||
artist = entry.artist
|
||||
if artist.albumcount == 0 and tcount < 10:
|
||||
limit = tcount
|
||||
|
||||
@@ -85,19 +89,19 @@ def get_artist_albums(path: ArtistHashSchema, query: GetArtistAlbumsQuery):
|
||||
|
||||
limit = query.limit
|
||||
|
||||
artist = ArtistTable.get_artist_by_hash(artisthash)
|
||||
entry = ArtistStore.artistmap.get(artisthash)
|
||||
|
||||
if artist is None:
|
||||
if entry is None:
|
||||
return {"error": "Artist not found"}, 404
|
||||
|
||||
albums = AlbumTable.get_albums_by_artisthash(artisthash)
|
||||
tracks = TrackTable.get_tracks_by_artisthash(artisthash)
|
||||
albums = AlbumStore.get_albums_by_hashes(entry.albumhashes)
|
||||
tracks = TrackStore.get_tracks_by_trackhashes(entry.trackhashes)
|
||||
|
||||
missing_albumhashes = {
|
||||
t.albumhash for t in tracks if t.albumhash not in {a.albumhash for a in albums}
|
||||
}
|
||||
|
||||
albums.extend(AlbumTable.get_albums_by_albumhashes(missing_albumhashes))
|
||||
albums.extend(AlbumStore.get_albums_by_hashes(missing_albumhashes))
|
||||
albumdict = {a.albumhash: a for a in albums}
|
||||
|
||||
config = UserConfig()
|
||||
@@ -135,7 +139,7 @@ def get_artist_albums(path: ArtistHashSchema, query: GetArtistAlbumsQuery):
|
||||
for key, value in res.items():
|
||||
res[key] = serialize_for_card_many(value[:limit])
|
||||
|
||||
res["artistname"] = artist.name
|
||||
res["artistname"] = entry.artist.name
|
||||
return res
|
||||
|
||||
|
||||
@@ -146,8 +150,7 @@ def get_all_artist_tracks(path: ArtistHashSchema):
|
||||
|
||||
Returns all artists by a given artist.
|
||||
"""
|
||||
# tracks = TrackStore.get_tracks_by_artisthash(path.artisthash)
|
||||
tracks = TrackTable.get_tracks_by_artisthash(path.artisthash)
|
||||
tracks = ArtistStore.get_artist_tracks(path.artisthash)
|
||||
return serialize_tracks(tracks)
|
||||
|
||||
|
||||
@@ -162,7 +165,7 @@ def get_similar_artists(path: ArtistHashSchema, query: ArtistLimitSchema):
|
||||
if result is None:
|
||||
return []
|
||||
|
||||
similar = ArtistTable.get_artists_by_artisthashes(result.get_artist_hash_set())
|
||||
similar = ArtistStore.get_artists_by_hashes(result.get_artist_hash_set())
|
||||
|
||||
if len(similar) > limit:
|
||||
similar = random.sample(similar, min(limit, len(similar)))
|
||||
|
||||
@@ -317,7 +317,6 @@ class AlbumTable(Base):
|
||||
|
||||
for artist in artisthashes:
|
||||
result = conn.execute(
|
||||
# NOTE: The artist dict keys need to in the same order they appear in the db for this to work!
|
||||
select(AlbumTable).where(AlbumTable.artisthashes.contains(artist))
|
||||
)
|
||||
albums[artist] = albums_to_dataclasses(result.fetchall())
|
||||
|
||||
+2
-3
@@ -12,8 +12,7 @@ from app.db.sqlite.albumcolors import SQLiteAlbumMethods as aldb
|
||||
from app.db.sqlite.artistcolors import SQLiteArtistMethods as adb
|
||||
from app.db.sqlite.utils import SQLiteManager
|
||||
|
||||
from app.store.artists import ArtistStore
|
||||
from app.store.albums import AlbumStore
|
||||
# from app.store.artists import ArtistStore
|
||||
from app.logger import log
|
||||
from app.lib.errors import PopulateCancelledError
|
||||
from app.utils.progressbar import tqdm
|
||||
@@ -101,7 +100,7 @@ class ProcessArtistColors:
|
||||
"""
|
||||
|
||||
def __init__(self, instance_key: str) -> None:
|
||||
all_artists = [a for a in ArtistStore.artists if len(a.colors) == 0]
|
||||
# all_artists = [a for a in ArtistStore.artists if len(a.colors) == 0]
|
||||
|
||||
global PROCESS_ARTIST_COLORS_KEY
|
||||
PROCESS_ARTIST_COLORS_KEY = instance_key
|
||||
|
||||
+121
-122
@@ -2,13 +2,14 @@ import gc
|
||||
import os
|
||||
from pprint import pprint
|
||||
from time import time
|
||||
from typing import Generator
|
||||
from app import settings
|
||||
from app.config import UserConfig
|
||||
from app.db.libdata import ArtistTable
|
||||
from app.db.libdata import AlbumTable, TrackTable
|
||||
from app.db.libdata import TrackTable
|
||||
from app.lib.populate import CordinateMedia
|
||||
from app.lib.taglib import extract_thumb, get_tags
|
||||
from app.models.album import Album
|
||||
from app.models.artist import Artist
|
||||
from app.models.track import Track
|
||||
from app.store.folder import FolderStore
|
||||
from app.utils.filesystem import run_fast_scandir
|
||||
@@ -141,154 +142,152 @@ class IndexTracks:
|
||||
print("Done")
|
||||
|
||||
|
||||
class IndexAlbums:
|
||||
def __init__(self) -> None:
|
||||
albums = dict()
|
||||
all_tracks: list[Track] = TrackTable.get_all()
|
||||
# class IndexAlbums:
|
||||
def create_albums():
|
||||
albums = dict()
|
||||
all_tracks: list[Track] = TrackTable.get_all()
|
||||
|
||||
if len(all_tracks) == 0:
|
||||
return
|
||||
for track in all_tracks:
|
||||
if track.albumhash not in albums:
|
||||
albums[track.albumhash] = {
|
||||
"albumartists": track.albumartists,
|
||||
"artisthashes": [a["artisthash"] for a in track.albumartists],
|
||||
"albumhash": track.albumhash,
|
||||
"base_title": None,
|
||||
"color": None,
|
||||
"created_date": track.last_mod,
|
||||
"date": track.date,
|
||||
"duration": track.duration,
|
||||
"genres": [*track.genres] if track.genres else [],
|
||||
"og_title": track.og_album,
|
||||
"lastplayed": track.lastplayed,
|
||||
"playcount": track.playcount,
|
||||
"playduration": track.playduration,
|
||||
"title": track.album,
|
||||
"trackcount": 1,
|
||||
"extra": {}
|
||||
}
|
||||
else:
|
||||
album = albums[track.albumhash]
|
||||
album["trackcount"] += 1
|
||||
album["playcount"] += track.playcount
|
||||
album["playduration"] += track.playduration
|
||||
album["lastplayed"] = max(album["lastplayed"], track.lastplayed)
|
||||
album["duration"] += track.duration
|
||||
album["date"] = min(album["date"], track.date)
|
||||
album["created_date"] = min(album["created_date"], track.last_mod)
|
||||
|
||||
for track in all_tracks:
|
||||
if track.albumhash not in albums:
|
||||
albums[track.albumhash] = {
|
||||
"albumartists": track.albumartists,
|
||||
"artisthashes": [a["artisthash"] for a in track.albumartists],
|
||||
"albumhash": track.albumhash,
|
||||
"base_title": None,
|
||||
"color": None,
|
||||
if track.genres:
|
||||
album["genres"].extend(track.genres)
|
||||
|
||||
for album in albums.values():
|
||||
genres = []
|
||||
for genre in album["genres"]:
|
||||
if genre not in genres:
|
||||
genres.append(genre)
|
||||
|
||||
album["genres"] = genres
|
||||
album["genrehashes"] = " ".join([g['genrehash'] for g in genres])
|
||||
album["base_title"], _ = get_base_album_title(album["og_title"])
|
||||
|
||||
del genres
|
||||
|
||||
# AlbumTable.remove_all()
|
||||
# AlbumTable.insert_many(list(albums.values()))
|
||||
return [Album(**album) for album in albums.values()]
|
||||
|
||||
|
||||
# class IndexArtists:
|
||||
def create_artists():
|
||||
all_tracks: list[Track] = TrackTable.get_all()
|
||||
artists = dict()
|
||||
|
||||
for track in all_tracks:
|
||||
this_artists = track.artists
|
||||
|
||||
for a in track.albumartists:
|
||||
if a not in this_artists:
|
||||
a["in_track"] = False
|
||||
this_artists.append(a)
|
||||
|
||||
for thisartist in this_artists:
|
||||
if thisartist["artisthash"] not in artists:
|
||||
artists[thisartist["artisthash"]] = {
|
||||
"albumcount": None,
|
||||
"albums": {track.albumhash},
|
||||
"artisthash": thisartist["artisthash"],
|
||||
"created_date": track.last_mod,
|
||||
"date": track.date,
|
||||
"duration": track.duration,
|
||||
"genres": [*track.genres] if track.genres else [],
|
||||
"og_title": track.og_album,
|
||||
"genres": track.genres if track.genres else [],
|
||||
"name": None,
|
||||
"names": {thisartist["name"]},
|
||||
"lastplayed": track.lastplayed,
|
||||
"playcount": track.playcount,
|
||||
"playduration": track.playduration,
|
||||
"title": track.album,
|
||||
"trackcount": 1,
|
||||
"trackcount": None,
|
||||
"tracks": (
|
||||
{track.trackhash}
|
||||
if thisartist.get("in_track", True)
|
||||
else set()
|
||||
),
|
||||
"extra": {},
|
||||
}
|
||||
else:
|
||||
album = albums[track.albumhash]
|
||||
album["trackcount"] += 1
|
||||
album["playcount"] += track.playcount
|
||||
album["playduration"] += track.playduration
|
||||
album["lastplayed"] = max(album["lastplayed"], track.lastplayed)
|
||||
album["duration"] += track.duration
|
||||
album["date"] = min(album["date"], track.date)
|
||||
album["created_date"] = min(album["created_date"], track.last_mod)
|
||||
artist = artists[thisartist["artisthash"]]
|
||||
artist["duration"] += track.duration
|
||||
artist["playcount"] += track.playcount
|
||||
artist["playduration"] += track.playduration
|
||||
artist["albums"].add(track.albumhash)
|
||||
artist["date"] = min(artist["date"], track.date)
|
||||
artist["lastplayed"] = max(artist["lastplayed"], track.lastplayed)
|
||||
artist["created_date"] = min(artist["created_date"], track.last_mod)
|
||||
artist["names"].add(thisartist["name"])
|
||||
|
||||
if thisartist.get("in_track", True):
|
||||
artist["tracks"].add(track.trackhash)
|
||||
|
||||
if track.genres:
|
||||
album["genres"].extend(track.genres)
|
||||
artist["genres"].extend(track.genres)
|
||||
|
||||
for album in albums.values():
|
||||
genres = []
|
||||
for genre in album["genres"]:
|
||||
if genre not in genres:
|
||||
genres.append(genre)
|
||||
for artist in artists.values():
|
||||
artist["albumcount"] = len(artist["albums"])
|
||||
artist["trackcount"] = len(artist["tracks"])
|
||||
|
||||
album["genres"] = genres
|
||||
album["base_title"], _ = get_base_album_title(album["og_title"])
|
||||
genres = []
|
||||
|
||||
del genres
|
||||
for genre in artist["genres"]:
|
||||
if genre not in genres:
|
||||
genres.append(genre)
|
||||
|
||||
AlbumTable.remove_all()
|
||||
AlbumTable.insert_many(list(albums.values()))
|
||||
del albums
|
||||
artist["genres"] = genres
|
||||
artist["genrehashes"] = " ".join([g['genrehash'] for g in genres])
|
||||
artist["name"] = sorted(artist["names"])[0]
|
||||
|
||||
# INFO: Delete temporary keys
|
||||
del artist["names"]
|
||||
del artist["tracks"]
|
||||
del artist["albums"]
|
||||
|
||||
class IndexArtists:
|
||||
def __init__(self) -> None:
|
||||
all_tracks: list[Track] = TrackTable.get_all()
|
||||
artists = dict()
|
||||
# INFO: Delete local variables
|
||||
del genres
|
||||
|
||||
if len(all_tracks) == 0:
|
||||
return
|
||||
|
||||
for track in all_tracks:
|
||||
this_artists = track.artists
|
||||
|
||||
for a in track.albumartists:
|
||||
if a not in this_artists:
|
||||
a["in_track"] = False
|
||||
this_artists.append(a)
|
||||
|
||||
for thisartist in this_artists:
|
||||
if thisartist["artisthash"] not in artists:
|
||||
artists[thisartist["artisthash"]] = {
|
||||
"albumcount": None,
|
||||
"albums": {track.albumhash},
|
||||
"artisthash": thisartist["artisthash"],
|
||||
"created_date": track.last_mod,
|
||||
"date": track.date,
|
||||
"duration": track.duration,
|
||||
"genres": track.genres if track.genres else [],
|
||||
"name": None,
|
||||
"names": {thisartist["name"]},
|
||||
"lastplayed": track.lastplayed,
|
||||
"playcount": track.playcount,
|
||||
"playduration": track.playduration,
|
||||
"trackcount": None,
|
||||
"tracks": (
|
||||
{track.trackhash}
|
||||
if thisartist.get("in_track", True)
|
||||
else set()
|
||||
),
|
||||
}
|
||||
else:
|
||||
artist = artists[thisartist["artisthash"]]
|
||||
artist["duration"] += track.duration
|
||||
artist["playcount"] += track.playcount
|
||||
artist["playduration"] += track.playduration
|
||||
artist["albums"].add(track.albumhash)
|
||||
artist["date"] = min(artist["date"], track.date)
|
||||
artist["lastplayed"] = max(artist["lastplayed"], track.lastplayed)
|
||||
artist["created_date"] = min(artist["created_date"], track.last_mod)
|
||||
artist["names"].add(thisartist["name"])
|
||||
|
||||
|
||||
if thisartist.get("in_track", True):
|
||||
artist["tracks"].add(track.trackhash)
|
||||
|
||||
if track.genres:
|
||||
artist["genres"].extend(track.genres)
|
||||
|
||||
for artist in artists.values():
|
||||
artist["albumcount"] = len(artist["albums"])
|
||||
artist["trackcount"] = len(artist["tracks"])
|
||||
|
||||
genres = []
|
||||
|
||||
for genre in artist["genres"]:
|
||||
if genre not in genres:
|
||||
genres.append(genre)
|
||||
|
||||
artist["genres"] = genres
|
||||
artist["name"] = sorted(artist["names"])[0]
|
||||
|
||||
# INFO: Delete temporary keys
|
||||
del artist["names"]
|
||||
del artist["tracks"]
|
||||
del artist["albums"]
|
||||
|
||||
# INFO: Delete local variables
|
||||
del genres
|
||||
|
||||
ArtistTable.remove_all()
|
||||
ArtistTable.insert_many(list(artists.values()))
|
||||
del artists
|
||||
# ArtistTable.remove_all()
|
||||
# ArtistTable.insert_many(list(artists.values()))
|
||||
# del artists
|
||||
return [Artist(**artist) for artist in artists.values()]
|
||||
|
||||
|
||||
class IndexEverything:
|
||||
def __init__(self) -> None:
|
||||
IndexTracks(instance_key=time())
|
||||
IndexAlbums()
|
||||
IndexArtists()
|
||||
# IndexAlbums()
|
||||
# IndexArtists()
|
||||
FolderStore.load_filepaths()
|
||||
|
||||
# pass
|
||||
|
||||
CordinateMedia(instance_key=str(time()))
|
||||
# CordinateMedia(instance_key=str(time()))
|
||||
gc.collect()
|
||||
|
||||
|
||||
|
||||
+1
-1
@@ -14,7 +14,6 @@ class Album:
|
||||
Creates an album object
|
||||
"""
|
||||
|
||||
id: int
|
||||
albumartists: list[dict[str, str]]
|
||||
albumhash: str
|
||||
artisthashes: list[str]
|
||||
@@ -34,6 +33,7 @@ class Album:
|
||||
playduration: int
|
||||
extra: dict
|
||||
|
||||
id: int = -1
|
||||
type: str = "album"
|
||||
image: str = ""
|
||||
versions: list[str] = dataclasses.field(default_factory=list)
|
||||
|
||||
@@ -36,7 +36,6 @@ class Artist:
|
||||
Artist class
|
||||
"""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
albumcount: int
|
||||
artisthash: str
|
||||
@@ -53,6 +52,7 @@ class Artist:
|
||||
playduration: int
|
||||
extra: dict
|
||||
|
||||
id: int = -1
|
||||
image: str = ""
|
||||
|
||||
def __post_init__(self):
|
||||
|
||||
@@ -47,4 +47,6 @@ def load_into_mem():
|
||||
# AlbumStore.load_albums(instance_key)
|
||||
# ArtistStore.load_artists(instance_key)
|
||||
TrackStore.load_all_tracks(get_random_str())
|
||||
AlbumStore.load_albums('a')
|
||||
ArtistStore.load_artists('a')
|
||||
FolderStore.load_filepaths()
|
||||
+66
-48
@@ -1,9 +1,13 @@
|
||||
from itertools import groupby
|
||||
import json
|
||||
from pprint import pprint
|
||||
import random
|
||||
from typing import Iterable
|
||||
|
||||
from app.db.sqlite.albumcolors import SQLiteAlbumMethods as aldb
|
||||
from app.lib.tagger import create_albums
|
||||
from app.models import Album, Track
|
||||
from app.store.artists import ArtistStore
|
||||
from app.utils.customlist import CustomList
|
||||
from app.utils.remove_duplicates import remove_duplicates
|
||||
|
||||
@@ -14,8 +18,19 @@ from app.utils.progressbar import tqdm
|
||||
ALBUM_LOAD_KEY = ""
|
||||
|
||||
|
||||
class AlbumMapEntry:
|
||||
def __init__(self, album: Album) -> None:
|
||||
self.album = album
|
||||
self.trackhashes: set[str] = set()
|
||||
|
||||
@property
|
||||
def basetitle(self):
|
||||
return self.album.base_title
|
||||
|
||||
|
||||
class AlbumStore:
|
||||
albums: list[Album] = CustomList()
|
||||
albummap: dict[str, AlbumMapEntry] = {}
|
||||
|
||||
@staticmethod
|
||||
def create_album(track: Track):
|
||||
@@ -36,43 +51,28 @@ class AlbumStore:
|
||||
global ALBUM_LOAD_KEY
|
||||
ALBUM_LOAD_KEY = instance_key
|
||||
|
||||
cls.albums = CustomList()
|
||||
|
||||
print("Loading albums... ", end="")
|
||||
tracks = remove_duplicates(TrackStore.tracks)
|
||||
|
||||
cls.albummap = {
|
||||
album.albumhash: AlbumMapEntry(album=album) for album in create_albums()
|
||||
}
|
||||
tracks = remove_duplicates(TrackStore.get_flat_list())
|
||||
tracks = sorted(tracks, key=lambda t: t.albumhash)
|
||||
grouped = groupby(tracks, lambda t: t.albumhash)
|
||||
|
||||
for albumhash, tracks in grouped:
|
||||
tracks = list(tracks)
|
||||
sample = tracks[0]
|
||||
cls.albummap[albumhash].trackhashes = {t.trackhash for t in tracks}
|
||||
|
||||
if sample is None:
|
||||
continue
|
||||
# db_albums: list[tuple] = aldb.get_all_albums()
|
||||
|
||||
count = len(list(tracks))
|
||||
duration = sum(t.duration for t in tracks)
|
||||
created_date = min(t.created_date for t in tracks)
|
||||
# for album in db_albums:
|
||||
# albumhash = album[1]
|
||||
# colors = json.loads(album[2])
|
||||
|
||||
album = AlbumStore.create_album(sample)
|
||||
|
||||
album.get_date_from_tracks(tracks)
|
||||
album.set_count(count)
|
||||
album.set_duration(duration)
|
||||
album.set_created_date(created_date)
|
||||
|
||||
cls.albums.append(album)
|
||||
|
||||
db_albums: list[tuple] = aldb.get_all_albums()
|
||||
|
||||
for album in db_albums:
|
||||
albumhash = album[1]
|
||||
colors = json.loads(album[2])
|
||||
|
||||
for _al in cls.albums:
|
||||
if _al.albumhash == albumhash:
|
||||
_al.set_colors(colors)
|
||||
break
|
||||
# for _al in cls.albums:
|
||||
# if _al.albumhash == albumhash:
|
||||
# _al.set_colors(colors)
|
||||
# break
|
||||
|
||||
print("Done!")
|
||||
|
||||
@@ -98,9 +98,7 @@ class AlbumStore:
|
||||
Returns N albums by the given albumartist, excluding the specified album.
|
||||
"""
|
||||
|
||||
albums = [
|
||||
album for album in cls.albums if artisthash in album.albumartists_hashes
|
||||
]
|
||||
albums = [album for album in cls.albums if artisthash in album.artisthashes]
|
||||
|
||||
albums = [
|
||||
album
|
||||
@@ -126,25 +124,11 @@ class AlbumStore:
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def get_albums_by_hashes(cls, albumhashes: list[str]) -> list[Album]:
|
||||
def get_albums_by_hashes(cls, albumhashes: Iterable[str]) -> list[Album]:
|
||||
"""
|
||||
Returns albums by their hashes.
|
||||
"""
|
||||
albums_str = "-".join(albumhashes)
|
||||
albums = [a for a in cls.albums if a.albumhash in albums_str]
|
||||
|
||||
# sort albums by the order of the hashes
|
||||
albums.sort(key=lambda x: albumhashes.index(x.albumhash))
|
||||
return albums
|
||||
|
||||
@classmethod
|
||||
def get_albums_by_artisthash(cls, artisthash: str) -> list[Album]:
|
||||
"""
|
||||
Returns all albums by the given artist.
|
||||
"""
|
||||
return [
|
||||
album for album in cls.albums if artisthash in album.albumartists_hashes
|
||||
]
|
||||
return [cls.albummap[albumhash].album for albumhash in albumhashes]
|
||||
|
||||
@classmethod
|
||||
def count_albums_by_artisthash(cls, artisthash: str):
|
||||
@@ -174,3 +158,37 @@ class AlbumStore:
|
||||
Removes an album from the store.
|
||||
"""
|
||||
cls.albums = CustomList(a for a in cls.albums if a.albumhash != albumhash)
|
||||
|
||||
@classmethod
|
||||
def get_albums_by_artisthash(cls, hash: str):
|
||||
"""
|
||||
Returns all albums by the given artist hash.
|
||||
"""
|
||||
artist = ArtistStore.artistmap.get(hash)
|
||||
|
||||
if not artist:
|
||||
return []
|
||||
|
||||
return [cls.albummap[albumhash].album for albumhash in artist.albumhashes]
|
||||
|
||||
@classmethod
|
||||
def get_albums_by_artisthashes(cls, hashes: Iterable[str]):
|
||||
"""
|
||||
Returns all albums by the given artist hashes.
|
||||
"""
|
||||
albums = []
|
||||
for hash in hashes:
|
||||
albums.extend(cls.get_albums_by_artisthash(hash))
|
||||
|
||||
return albums
|
||||
|
||||
@classmethod
|
||||
def get_album_tracks(cls, albumhash: str) -> list[Track]:
|
||||
"""
|
||||
Returns all tracks for the given album hash.
|
||||
"""
|
||||
album = cls.albummap.get(albumhash)
|
||||
if not album:
|
||||
return []
|
||||
|
||||
return TrackStore.get_tracks_by_trackhashes(album.trackhashes)
|
||||
|
||||
+46
-16
@@ -1,19 +1,30 @@
|
||||
import json
|
||||
from typing import Iterable
|
||||
|
||||
from app.db.sqlite.artistcolors import SQLiteArtistMethods as ardb
|
||||
from app.lib.tagger import create_artists
|
||||
from app.models import Artist
|
||||
from app.utils.bisection import use_bisection
|
||||
from app.utils.customlist import CustomList
|
||||
from app.utils.progressbar import tqdm
|
||||
from .tracks import TrackStore
|
||||
|
||||
from .albums import AlbumStore
|
||||
# from .albums import AlbumStore
|
||||
from .tracks import TrackStore
|
||||
|
||||
ARTIST_LOAD_KEY = ""
|
||||
|
||||
|
||||
class ArtistMapEntry:
|
||||
def __init__(self, artist: Artist) -> None:
|
||||
self.artist = artist
|
||||
self.albumhashes: set[str] = set()
|
||||
self.trackhashes: set[str] = set()
|
||||
|
||||
|
||||
class ArtistStore:
|
||||
artists: list[Artist] = CustomList()
|
||||
artistmap: dict[str, ArtistMapEntry] = {}
|
||||
|
||||
@classmethod
|
||||
def load_artists(cls, instance_key: str):
|
||||
@@ -24,15 +35,27 @@ class ArtistStore:
|
||||
ARTIST_LOAD_KEY = instance_key
|
||||
|
||||
print("Loading artists... ", end="")
|
||||
cls.artists.clear()
|
||||
cls.artistmap.clear()
|
||||
|
||||
cls.artists.extend(get_all_artists(TrackStore.tracks, AlbumStore.albums))
|
||||
print("Done!")
|
||||
for artist in ardb.get_all_artists():
|
||||
cls.artistmap = {
|
||||
artist.artisthash: ArtistMapEntry(artist=artist)
|
||||
for artist in create_artists()
|
||||
}
|
||||
|
||||
for track in TrackStore.get_flat_list():
|
||||
if instance_key != ARTIST_LOAD_KEY:
|
||||
return
|
||||
|
||||
cls.map_artist_color(artist)
|
||||
for hash in track.artisthashes:
|
||||
cls.artistmap[hash].trackhashes.add(track.trackhash)
|
||||
cls.artistmap[hash].albumhashes.add(track.albumhash)
|
||||
|
||||
print("Done!")
|
||||
# for artist in ardb.get_all_artists():
|
||||
# if instance_key != ARTIST_LOAD_KEY:
|
||||
# return
|
||||
|
||||
# cls.map_artist_color(artist)
|
||||
|
||||
@classmethod
|
||||
def map_artist_color(cls, artist_tuple: tuple):
|
||||
@@ -65,24 +88,20 @@ class ArtistStore:
|
||||
cls.artists.append(artist)
|
||||
|
||||
@classmethod
|
||||
def get_artist_by_hash(cls, artisthash: str) -> Artist:
|
||||
def get_artist_by_hash(cls, artisthash: str):
|
||||
"""
|
||||
Returns an artist by its hash.P
|
||||
"""
|
||||
artists = sorted(cls.artists, key=lambda x: x.artisthash)
|
||||
try:
|
||||
artist = use_bisection(artists, "artisthash", [artisthash])[0]
|
||||
return artist
|
||||
except IndexError:
|
||||
return None
|
||||
entry = cls.artistmap.get(artisthash, None)
|
||||
if entry is not None:
|
||||
return entry.artist
|
||||
|
||||
@classmethod
|
||||
def get_artists_by_hashes(cls, artisthashes: list[str]) -> list[Artist]:
|
||||
def get_artists_by_hashes(cls, artisthashes: Iterable[str]):
|
||||
"""
|
||||
Returns artists by their hashes.
|
||||
"""
|
||||
artists = sorted(cls.artists, key=lambda x: x.artisthash)
|
||||
artists = use_bisection(artists, "artisthash", artisthashes)
|
||||
artists = [cls.get_artist_by_hash(hash) for hash in artisthashes]
|
||||
return [a for a in artists if a is not None]
|
||||
|
||||
@classmethod
|
||||
@@ -113,3 +132,14 @@ class ArtistStore:
|
||||
Removes an artist from the store.
|
||||
"""
|
||||
cls.artists = CustomList(a for a in cls.artists if a.artisthash != artisthash)
|
||||
|
||||
@classmethod
|
||||
def get_artist_tracks(cls, artisthash: str):
|
||||
"""
|
||||
Returns all tracks by the given artist hash.
|
||||
"""
|
||||
entry = cls.artistmap.get(artisthash)
|
||||
if entry is not None:
|
||||
return TrackStore.get_tracks_by_trackhashes(entry.trackhashes)
|
||||
|
||||
return []
|
||||
|
||||
+5
-4
@@ -1,8 +1,7 @@
|
||||
# from tqdm import tqdm
|
||||
|
||||
import itertools
|
||||
import sys
|
||||
from typing import Callable
|
||||
from typing import Callable, Iterable
|
||||
from flask_jwt_extended import current_user
|
||||
from app.db.libdata import TrackTable
|
||||
from app.db.sqlite.favorite import SQLiteFavoriteMethods as favdb
|
||||
@@ -243,7 +242,7 @@ class TrackStore:
|
||||
# ================================================
|
||||
|
||||
@classmethod
|
||||
def get_tracks_by_trackhashes(cls, trackhashes: list[str]) -> list[Track]:
|
||||
def get_tracks_by_trackhashes(cls, trackhashes: Iterable[str]) -> list[Track]:
|
||||
"""
|
||||
Returns a list of tracks by their hashes.
|
||||
"""
|
||||
@@ -259,7 +258,9 @@ class TrackStore:
|
||||
tracks.append(track)
|
||||
|
||||
# sort the tracks in the order of the given trackhashes
|
||||
tracks.sort(key=lambda t: trackhashes.index(t.trackhash))
|
||||
if type(trackhashes) == list:
|
||||
tracks.sort(key=lambda t: trackhashes.index(t.trackhash))
|
||||
|
||||
return tracks
|
||||
|
||||
@classmethod
|
||||
|
||||
Reference in New Issue
Block a user