diff --git a/Flow.py b/Flow.py index e1ac33d..5e813e0 100644 --- a/Flow.py +++ b/Flow.py @@ -96,6 +96,10 @@ class Flow: self.spotify_library = LibraryLoader().load(work_dir / "spotify_library.json") self.local_library = LibraryLoader().load(work_dir / "local_library.json") + def save_libraries(self): + LibrarySaver(self.spotify_library).save(work_dir / "spotify_library.json") + LibrarySaver(self.local_library).save(work_dir / "local_library.json") + def map_local_to_spotify(self): resolve_remote_tracks(client_id, client_secret, self.local_library, self.spotify_library, work_dir) diff --git a/Gui.py b/Gui.py new file mode 100644 index 0000000..85fd6e3 --- /dev/null +++ b/Gui.py @@ -0,0 +1,206 @@ +import time + +import streamlit as st +from st_aggrid import AgGrid, GridOptionsBuilder, GridUpdateMode +import pandas as pd + +import Flow + +from src.Library import Library, Artist, Album, Track, Playlist + +def track_to_row(track: Track): + return { + "Title": track.title, + "Album": track.album.title if track.album else "", + "Artists": ", ".join(a.title if a.title else "no_arist_name" for a in track.artists), + "Lyrics": track.has_lyrics, + "AutoScore": track.auto_score, + "Added": track.added_at.strftime("%Y-%m-%d") if track.added_at else "", + "LocalId": str(track.local_id) if track.local_id else "", + "Path": str(track.local_path) if track.local_path else "", + } + +def artist_to_row(artist: Artist): + return { + "Title": artist.title, + "AutoScore": artist.auto_score, + "Resolved": int(artist.resolved_percentage * 100), + "LocalId": str(artist.local_id) if artist.local_id else "", + } + +def album_to_row(album: Album): + return { + "Title": album.title, + "AutoScore": album.auto_score, + "Resolved": int(album.resolved_percentage * 100), + "LocalId": str(album.local_id) if album.local_id else "", + } + + +class LibraryView: + def __init__(self): + self.lib : Library = None + self.libs : list[Library] = None + self.view = None + self.lyrics_only = False + self.query = None + + def add_libraries(self, libs : list[Library]): + self.libs = libs + + def render(self): + st.set_page_config(layout="wide") + + with st.sidebar: + lib = st.selectbox( + "Library", + options=[lib.name for lib in self.libs], + ) + + for iter in self.libs: + if iter.name == lib: + self.lib = iter + + st.write(f"Showing library '{self.lib.name}'") + + self.view = st.radio( + "View", + ["Tree"] + ) + + if self.view == "Tree": + self.render_artists_albums_tracks() + # elif self.view == "Tracks": + # self.render_tracks() + # elif self.view == "Playlists": + # self.render_playlists() + + def track_visible( + self, + track: Track, + artist: Artist | None, + album: Album | None, + ): + if self.lyrics_only and not track.has_lyrics: + return False + + if self.query and self.query.lower() not in track.title.lower(): + return False + + if artist and all(a.id != artist.id for a in track.artists): + return False + + if album and (not track.album or track.album.id != album.id): + return False + + return True + + def render_artists_albums_tracks(self): + # --- ARTISTS TABLE --- + st.subheader("Artists") + artist_rows = [artist_to_row(a) for a in self.lib.artists] + df_artists = pd.DataFrame(artist_rows) + + gb_artists = GridOptionsBuilder.from_dataframe(df_artists) + gb_artists.configure_selection(selection_mode="multiple", use_checkbox=True) + gb_artists.configure_column("Title", filter="agTextColumnFilter") + gb_artists.configure_column("AutoScore", sort="desc") + grid_options_artists = gb_artists.build() + + grid_response_artists = AgGrid( + df_artists, + gridOptions=grid_options_artists, + update_mode=GridUpdateMode.SELECTION_CHANGED, + allow_unsafe_jscode=True, + enable_enterprise_modules=False, + fit_columns_on_grid_load=True, + ) + + selected_rows_df = grid_response_artists.get("selected_rows") + selected_artist_titles = selected_rows_df["Title"].tolist() if selected_rows_df is not None and not selected_rows_df.empty else [] + selected_artists = [a for a in self.lib.artists if a.title in selected_artist_titles] + + # --- ALBUMS TABLE --- + st.subheader("Albums") + if selected_artists: + albums_list = [] + for artist in selected_artists: + albums_list.extend(artist.albums) + albums_list = list({al.id: al for al in albums_list}.values()) + else: + albums_list = self.lib.albums + + album_rows = [album_to_row(a) for a in albums_list] + df_albums = pd.DataFrame(album_rows) + + gb_albums = GridOptionsBuilder.from_dataframe(df_albums) + gb_albums.configure_selection(selection_mode="multiple", use_checkbox=True) + gb_albums.configure_column("Title", filter="agTextColumnFilter") + gb_albums.configure_column("AutoScore", sort="desc") + grid_options_albums = gb_albums.build() + + grid_response_albums = AgGrid( + df_albums, + gridOptions=grid_options_albums, + update_mode=GridUpdateMode.SELECTION_CHANGED, + allow_unsafe_jscode=True, + enable_enterprise_modules=False, + fit_columns_on_grid_load=True, + ) + + selected_rows_df = grid_response_albums.get("selected_rows") + selected_album_titles = selected_rows_df["Title"].tolist() if selected_rows_df is not None and not selected_rows_df.empty else [] + selected_albums = [al for al in albums_list if al.title in selected_album_titles] + + # --- TRACKS TABLE --- + st.subheader("Tracks") + if selected_albums: + tracks_list = [] + for album in selected_albums: + tracks_list.extend(album.tracks) + elif selected_artists: + tracks_list = [] + for artist in selected_artists: + tracks_list.extend([t for t in self.lib.tracks if artist in t.artists]) + else: + tracks_list = self.lib.tracks + + track_rows = [track_to_row(t) for t in tracks_list] + df_tracks = pd.DataFrame(track_rows) + + gb_tracks = GridOptionsBuilder.from_dataframe(df_tracks) + gb_tracks.configure_selection(selection_mode="multiple", use_checkbox=True) + gb_tracks.configure_column("Title", filter="agTextColumnFilter") + gb_tracks.configure_column("AutoScore", sort="desc") + grid_options_tracks = gb_tracks.build() + + AgGrid( + df_tracks, + gridOptions=grid_options_tracks, + update_mode=GridUpdateMode.SELECTION_CHANGED, + allow_unsafe_jscode=True, + enable_enterprise_modules=False, + fit_columns_on_grid_load=True, + ) + + +def run(): + flow = Flow.Flow() + + # flow.fetch_spotify() + # flow.fetch_local() + + # flow.parse_spotify_library() + # flow.parse_local_library() + + flow.load_libraries() + + # flow.map_local_to_spotify() + + # flow.log_stats() + + gui = LibraryView() + gui.add_libraries([flow.spotify_library, flow.local_library]) + gui.render() + +run() \ No newline at end of file diff --git a/error.log b/error.log new file mode 100644 index 0000000..7c830a2 --- /dev/null +++ b/error.log @@ -0,0 +1,2 @@ + +(process:38049): GLib-GIO-CRITICAL **: 14:06:16.561: g_dbus_connection_emit_signal: assertion 'G_IS_DBUS_CONNECTION (connection)' failed diff --git a/gui.sh b/gui.sh new file mode 100755 index 0000000..eb55739 --- /dev/null +++ b/gui.sh @@ -0,0 +1 @@ +streamlit run Gui.py diff --git a/main.py b/main.py index af9ed31..646539a 100644 --- a/main.py +++ b/main.py @@ -5,15 +5,14 @@ import Flow def test1(): flow = Flow.Flow() - flow.fetch_local() + # flow.fetch_local() flow.parse_local_library() flow.load_libraries() # flow.map_local_to_spotify() # flow.log_stats() - flow.fetch_and_save_lyrics() - + # flow.fetch_and_save_lyrics() def main(): flow = Flow.Flow() @@ -27,11 +26,12 @@ def main(): flow.load_libraries() flow.map_local_to_spotify() + flow.save_libraries() - flow.log_stats() + # flow.log_stats() - print("asd") + # print("asd") -test1() -# main() +# test1() +main() diff --git a/src/Library.py b/src/Library.py index 0de72ce..46d7566 100644 --- a/src/Library.py +++ b/src/Library.py @@ -7,64 +7,86 @@ from typing import List, Optional, Any class Artist: def __init__(self): - self.id : str = None - self.title : str = None - self.tracks : List[Track] = [] - self.albums : List[Album] = [] + self.id: str = None + self.title: str = None + self.tracks: List[Track] = [] + self.albums: List[Album] = [] + self.local_id: str = None + self.auto_score: float = 0 + self.fav: bool = False + self.rating: int = 0 + self.resolved_percentage : float = 0 class Track: def __init__(self): - self.id : str = None - self.title : str = None - self.artists : List[Artist] = [] - self.album : Album = None - self.duration_ms : int = None - self.added_at : datetime = None - self.local_path : Path = None - self.has_lyrics : bool = False + self.id: str = None + self.title: str = None + self.artists: List[Artist] = [] + self.album: Album = None + self.duration_ms: int = None + self.added_at: datetime = None + self.local_path: Path = None + self.has_lyrics: bool = False + self.local_id: str = None + self.auto_score: float = 0 + self.fav: bool = False + self.rating: int = 0 class Album: def __init__(self): - self.id : str = None - self.title : str = None - self.release_date : datetime = None - self.tracks : List[Track] = [] - self.artists : List[Artist] = [] + self.id: str = None + self.title: str = None + self.release_date: datetime = None + self.tracks: List[Track] = [] + self.artists: List[Artist] = [] + self.local_id: str = None + self.auto_score: float = 0 + self.fav: bool = False + self.rating: int = 0 + self.resolved_percentage : float = 0 class Playlist: def __init__(self): - self.title : str = None - self.tracks : List[Track] = [] - self.created_at : datetime = None - self.description : str = None + self.title: str = None + self.tracks: List[Track] = [] + self.created_at: datetime = None + self.description: str = None + self.auto_score: float = 0 + self.fav: bool = False + self.rating: int = 0 + self.resolved_percentage : float = 0 class Library: def __init__(self): - self.artists : List[Artist] = [] - self.playlists : List[Playlist] = [] - self.tracks : List[Track] = [] - self.albums : List[Album] = [] - self.liked_tracks : List[Track] = [] - self.top_tracks : List[Track]= [] - self.top_artists : List[Artist] = [] + self.artists: List[Artist] = [] + self.playlists: List[Playlist] = [] + self.tracks: List[Track] = [] + self.albums: List[Album] = [] - self.artist_map : dict[str, Artist] = None - self.album_map : dict[str, Album] = None - self.track_map : dict[str, Track] = None + self.liked_tracks: List[Track] = [] + self.top_tracks: List[Track] = [] + self.top_artists: List[Artist] = [] + self.artist_map: dict[str, Artist] = None + self.album_map: dict[str, Album] = None + self.track_map: dict[str, Track] = None + + self.name = "Unnamed" def update_cache(self): self.update_id_maps() self.create_reverse_links() + self.score_albums() + self.calc_resolved_percentage() def update_id_maps(self): - self.artist_map : dict[str, Artist] = { artist.id : artist for artist in self.artists } - self.album_map : dict[str, Album] = { album.id : album for album in self.albums } - self.track_map : dict[str, Track] = { track.id : track for track in self.tracks } + self.artist_map: dict[str, Artist] = {artist.id: artist for artist in self.artists} + self.album_map: dict[str, Album] = {album.id: album for album in self.albums} + self.track_map: dict[str, Track] = {track.id: track for track in self.tracks} def create_reverse_links(self): for track_id, track in self.track_map.items(): @@ -79,3 +101,36 @@ class Library: for artist in album.artists: if album not in artist.albums: artist.albums.append(album) + + def auto_score(self): + self.score_tracks() + self.score_artists() + self.score_albums() + + def score_tracks(self): + for idx, track in enumerate(self.top_tracks): + score = len(self.top_tracks) - idx + track.auto_score = score + + def score_artists(self): + for track in self.tracks: + for artist in track.artists: + artist.auto_score += track.auto_score + + def score_albums(self): + for track in self.tracks: + track.album.auto_score += track.auto_score + + def calc_resolved_percentage(self): + for artist in self.artists: + artist_count_resolved = 0 + artist_count = 0 + for album in artist.albums: + album_count_resolved = 0 + for track in album.tracks: + if track.local_id: + album_count_resolved += 1 + album.resolved_percentage = album_count_resolved / len(album.tracks) + artist_count_resolved += album_count_resolved + artist_count += len(album.tracks) + artist.resolved_percentage = artist_count_resolved / artist_count \ No newline at end of file diff --git a/src/LibrarySerializer.py b/src/LibrarySerializer.py index 84f6a46..f331824 100644 --- a/src/LibrarySerializer.py +++ b/src/LibrarySerializer.py @@ -33,6 +33,10 @@ class LibrarySaver: "title": artist.title, "tracks": [track.id for track in artist.tracks], "albums": [album.id for album in artist.albums], + "local_id": artist.local_id, + "auto_score": artist.auto_score, + "fav" : artist.fav, + "rating" : artist.rating, } @staticmethod @@ -46,6 +50,10 @@ class LibrarySaver: "added_at": _dt_to_str(track.added_at), "local_path": track.local_path, "has_lyrics": track.has_lyrics, + "local_id": track.local_id, + "auto_score": track.auto_score, + "fav" : track.fav, + "rating" : track.rating, } @staticmethod @@ -55,6 +63,9 @@ class LibrarySaver: "tracks": [track.id for track in playlist.tracks], "created_at": _dt_to_str(playlist.created_at), "description": playlist.description, + "auto_score": playlist.auto_score, + "fav" : playlist.fav, + "rating" : playlist.rating, } @staticmethod @@ -64,12 +75,17 @@ class LibrarySaver: "title": album.title, "release_date": album.release_date, "tracks": [track.id for track in album.tracks], - "artists": [artist.id for artist in album.artists] + "artists": [artist.id for artist in album.artists], + "local_id": album.local_id, + "auto_score": album.auto_score, + "fav" : album.fav, + "rating" : album.rating, } def save(self, path: Path) -> None: data = { + "name": self.library.name, "tracks": [self._track_to_dict(track) for track in self.library.tracks], "albums": [self._album_to_dict(album) for album in self.library.albums], "artists": [self._artist_to_dict(artist) for artist in self.library.artists], @@ -121,6 +137,10 @@ class LibraryLoader: track.artists = [self.get_artist(artist_id) for artist_id in data.get("artists")] track.album = self.get_album(data.get("album")) track.has_lyrics = data.get("has_lyrics") + track.local_id = data.get("local_id") + track.auto_score = data.get("auto_score") + track.fav = data.get("fav") + track.rating = data.get("rating") self.track_map[track.id] = track return track @@ -131,6 +151,10 @@ class LibraryLoader: artist.title = data.get("title") artist.tracks = [self.get_track(track_id) for track_id in data.get("tracks")] artist.albums = [self.get_album(album_id) for album_id in data.get("albums")] if data.get("albums") else [] + artist.local_id = data.get("local_id") + artist.auto_score = data.get("auto_score") + artist.fav = data.get("fav") + artist.rating = data.get("rating") self.artist_map[artist.id] = artist return artist @@ -141,6 +165,10 @@ class LibraryLoader: album.tracks = [self.get_track(track_id) for track_id in data.get("tracks")] album.title = data.get("title") album.artists = [self.get_artist(artist_id) for artist_id in data.get("artists")] + album.local_id = data.get("local_id") + album.auto_score = data.get("auto_score") + album.fav = data.get("fav") + album.rating = data.get("rating") self.album_map[album.id] = album return album @@ -151,6 +179,7 @@ class LibraryLoader: library = Library() + library.name = data.get("name") library.tracks = [self.load_track(track) for track in data.get("tracks")] library.albums = [self.load_album(album) for album in data.get("albums")] library.artists = [self.load_artist(artist) for artist in data.get("artists")] diff --git a/src/StatGenerator.py b/src/StatGenerator.py index ac16622..7c17ab5 100644 --- a/src/StatGenerator.py +++ b/src/StatGenerator.py @@ -15,13 +15,12 @@ def get_mapping(mapping_path) -> dict[str, Any]: with Path(mapping_path).open("r", encoding="utf-8") as f: data = json.load(f) - return data + return data["track_map"] class AlbumStat: def __init__(self): self.album: Album = Album() - self.score: int = 0 self.resolved_precent: float = 0 def calc_resolved_precent(self): @@ -37,7 +36,6 @@ class AlbumStat: class ArtistStat: def __init__(self): self.artist: Artist = Artist() - self.score: int = 0 self.resolved_precent: float = 0 self.albums_stats: list[AlbumStat] = [] @@ -87,18 +85,22 @@ class LibraryStats: for remote, local in self.mapping.items(): self.remote_id_map[remote].local_path = self.local_id_map[local].local_path + def score_tracks(self): + for idx, track in enumerate(self.remote_lib.top_tracks): + score = len(self.remote_lib.top_tracks) - idx + track.auto_score = score + def score_artists(self): for artist in self.remote_lib.artists: artist_stat = ArtistStat() artist_stat.artist = artist - artist_stat.score = 0 + artist_stat.artist.auto_score = 0 self.artists_map[artist_stat.artist.id] = artist_stat self.artists.append(artist_stat) - for idx, track in enumerate(self.remote_lib.top_tracks): - score = len(self.remote_lib.top_tracks) - idx + for track in self.remote_lib.tracks: for artist in track.artists: - self.artists_map[artist.id].score += score + self.artists_map[artist.id].artist.auto_score += track.auto_score for artist_stat in self.artists: artist_stat.calc_resolved_precent() @@ -111,9 +113,8 @@ class LibraryStats: stat.album = album self.albums_map[album.id] = stat - for idx, track in enumerate(self.remote_lib.top_tracks): - score = len(self.remote_lib.top_tracks) - idx - self.albums_map[track.album.id].score += score + for track in self.remote_lib.tracks: + self.albums_map[track.album.id].album.auto_score += track.auto_score for _, album in self.albums_map.items(): album.calc_resolved_precent() diff --git a/src/backends/local/LocalLibraryIndexer.py b/src/backends/local/LocalLibraryIndexer.py index 1a8eaab..6f132d4 100644 --- a/src/backends/local/LocalLibraryIndexer.py +++ b/src/backends/local/LocalLibraryIndexer.py @@ -1,64 +1,97 @@ +import os import fnmatch -import mutagen -import eyed3 -from mutagen.flac import FLAC from mutagen import File as MutagenFile +from mutagen.id3 import ID3, ID3NoHeaderError +from mutagen.flac import FLAC +from mutagen.mp4 import MP4 +from mutagen.oggvorbis import OggVorbis +from tqdm import tqdm from src.helpers import * - music_extensions = ('*.mp3', '*.flac', '*.wav', '*.aac', '*.ogg', '*.m4a') exclude_directories = ("*mary--*", "*example-word*") -from mutagen.id3 import ID3 -from mutagen.id3 import ID3NoHeaderError - - def detect_lyrics(abs_path, ext): ext = ext.lower() - # ---------- MP3 ---------- - if ext == ".mp3": - try: - tags = ID3(abs_path) - except ID3NoHeaderError: - return False - - return bool( - tags.getall("SYLT") or - tags.getall("USLT") - ) - - # ---------- FLAC / OGG / OPUS ---------- - if ext in {".flac", ".ogg", ".opus"}: - try: - audio = MutagenFile(abs_path) - if not audio or not audio.tags: - return False - - for key in audio.tags.keys(): - k = key.upper() - if k in {"LYRICS", "LRC", "SYNCEDLYRICS"}: - if audio.tags[key]: - return True - except Exception: - return False - - # ---------- Fallback ---------- try: audio = MutagenFile(abs_path) - if audio and audio.tags: - for key in audio.tags.keys(): - if "LYRIC" in key.upper(): - return True + if not audio or not audio.tags: + return False + + # MP3-specific SYLT/USLT detection + if ext == ".mp3" and isinstance(audio, ID3): + return bool(audio.getall("SYLT") or audio.getall("USLT")) + + # FLAC/OGG/OPUS lyrics detection + for key, value in audio.tags.items(): + k = key.upper() + if k in {"LYRICS", "LRC", "SYNCEDLYRICS"} and value: + return True + + # Generic fallback + for key in audio.tags.keys(): + if "LYRIC" in key.upper(): + return True + except Exception: - pass + return False return False -def update_local_library(root_dir, output_dir): +def get_artist(tags, fallback=None) -> list : + # FLAC / OGG / general multiple artist fields + artist_fields = ["ARTISTS", "TXXX:ARTISTS", "©ART", "artist"] + for field in artist_fields: + if field in tags and tags[field]: + values = tags.get(field) + if not isinstance(values, list): + return values.text + return values + + # M4A + if '\xa9ART' in tags and tags['\xa9ART']: + values = tags['\xa9ART'] + return values + + if tags.get("TPE1"): + text = getattr(tags.get("TPE1"), "text", None) + if text: + return text + + return fallback + +def safe_tag_get(tags, key, index=0): + try: + value = tags.get(key, None) + if isinstance(value, list): + # filter out None/empty, convert to str + value = [str(v) for v in value if v] + return value[index] if len(value) > index else fallback + return str(value) + except (KeyError, ValueError, TypeError): + return None + +def get_title(tags): + ids = ["TIT2", "TITLE", "title", "\xa9nam"] + for key in ids: + val = safe_tag_get(tags, key) + if val and val != "None": + return val + return None + +def get_album(tags): + ids = ["TALB", "album", "\xa9alb"] + for key in ids: + val = safe_tag_get(tags, key) + if val and val != "None": + return val + return None + +def update_local_library(root_dir, output_dir): set_workdir(output_dir / "local") old_library = get_data("local_library") @@ -91,61 +124,48 @@ def update_local_library(root_dir, output_dir): song_id = get_new_song_id(song_id, relative_path) new_library[song_id] = { - "name": song_name, + "name": None, "path": relative_path, "type": track_type, - "artist": "", - "album": "", + "artists": None, + "album": None, "duration": None } - for track_id, track in new_library.items(): - abs_path = os.path.join(root_dir, track['path']) + to_remove = [] - track_abs_path = os.path.join(root_dir, track["path"]) - track["has_lyrics"] = detect_lyrics(track_abs_path, track['type'].lower()) + with tqdm(total=len(new_library.items()), desc='Indexing local library') as process_bar: + for track_id, track in new_library.items(): + process_bar.update(1) + + abs_path = os.path.join(root_dir, track['path']) + track["has_lyrics"] = detect_lyrics(abs_path, track['type'].lower()) + + try: + audio = MutagenFile(abs_path) + if audio and audio.tags: + tags = audio.tags + + track['artists'] = get_artist(tags) + track['name'] = get_title(tags) + track['album'] = get_album(tags) + + if not len(track['artists']) or not track['name'] or not track['album']: + print(track) + raise "Cannot index track" - match track['type'].lower(): - case ".mp3": - mp3track = eyed3.load(abs_path) - if not mp3track or not mp3track.info: - print(f"Failed to load mp3 {abs_path}") - track['name'] = track['path'] else: - track['duration'] = mp3track.info.time_secs + print(track) + to_remove.append(track_id) + # raise "Cannot index track" - tag = mp3track.tag - if tag: - track['artist'] = tag.artist - track['name'] = tag.title - track['album'] = tag.album + except Exception as e: + print(e) + print(abs_path) + raise e - case ".flac": - try: - audio = FLAC(abs_path) - if audio.info: - track['duration'] = audio.info.length - - metadata = audio.tags - if metadata: - if 'artist' in metadata: - track['artist'] = " ".join(metadata['artist']) - if 'TITLE' in metadata: - track['name'] = " ".join(metadata['TITLE']) - if 'album' in metadata: - track['album'] = " ".join(metadata['album']) - - except mutagen.MutagenError: - print(f"Invalid flac header {abs_path}") - track['name'] = track['path'] - - case _: - try: - audio = MutagenFile(abs_path) - if audio and audio.info: - track['duration'] = audio.info.length - except Exception: - pass + for rem in to_remove: + new_library.pop(rem) save_data(new_library, "local_library") diff --git a/src/backends/local/ParseLocal.py b/src/backends/local/ParseLocal.py index 49b3f55..b1fbba3 100644 --- a/src/backends/local/ParseLocal.py +++ b/src/backends/local/ParseLocal.py @@ -10,46 +10,6 @@ PROTECTED_ARTISTS = { "The Good, The Bad & The Queen", } -def split_artists(artist_tag: str) -> list[str]: - if not artist_tag: - return [] - - placeholder_comma = "§COMMA§" - placeholder_amp = "§AMP§" - - protected_map = {} - - # Protect known artist names - for artist in PROTECTED_ARTISTS: - protected = ( - artist - .replace(",", placeholder_comma) - .replace("&", placeholder_amp) - ) - protected_map[protected] = artist - artist_tag = artist_tag.replace(artist, protected) - - # Split remaining separators - parts = re.split(r"[,&]", artist_tag) - - # Restore protected names and trim spaces - result = [] - for part in parts: - part = part.strip() - if not part: - continue - - part = ( - part - .replace(placeholder_comma, ",") - .replace(placeholder_amp, "&") - ) - - result.append(part) - - return result - - class LocalParser: def __init__(self): self.track_map: dict[str, Track] = {} @@ -60,9 +20,7 @@ class LocalParser: def parse_artists(self, data: str) -> List[Artist]: if not data: - data = "unknown" - - data = split_artists(data) + raise "No artist found" artists = [] @@ -87,7 +45,7 @@ class LocalParser: def parse_album(self, data: str) -> Album: if not data: - data = "unknown" + raise "album is none" name = data album_id = name @@ -113,12 +71,15 @@ class LocalParser: track.id = track_id track.title = data.get("name") - track.artists = self.parse_artists(data.get("artist")) + track.artists = self.parse_artists(data.get("artists")) track.album = self.parse_album(data.get("album")) track.local_path = data.get("path") track.duration_ms = data.get("duration") track.has_lyrics = data.get("has_lyrics") + if not track.album or not len(track.artists) or not track.title: + raise "error parsing" + self.track_map[track.id] = track self.library.tracks.append(track) @@ -129,6 +90,8 @@ class LocalParser: self.library.update_cache() + self.library.name = "Local" + return self.library diff --git a/src/backends/navidrome/navidrome.py b/src/backends/navidrome/navidrome.py new file mode 100644 index 0000000..c7d04f4 --- /dev/null +++ b/src/backends/navidrome/navidrome.py @@ -0,0 +1,110 @@ +import sqlite3 +import os +from dataclasses import dataclass, field +from typing import List, Optional + +@dataclass +class Track: + id: int + title: str + path: str + favorite: bool + rating: Optional[int] + play_count: int + last_played: Optional[str] + playlists: List[str] = field(default_factory=list) + +@dataclass +class Album: + id: int + title: str + tracks: List[Track] = field(default_factory=list) + +@dataclass +class Artist: + id: str + name: str + albums: List[Album] = field(default_factory=list) + +class NavidromeDB: + def __init__(self, db_path: str): + if not os.path.isfile(db_path): + raise FileNotFoundError(f"Navidrome DB not found: {db_path}") + self.conn = sqlite3.connect(db_path) + self.conn.row_factory = sqlite3.Row + + def close(self): + self.conn.close() + + def get_artists(self) -> List[Artist]: + artists = [] + c = self.conn.cursor() + c.execute("SELECT id, name FROM artist ORDER BY name") + for row in c.fetchall(): + artist = Artist(id=row["id"], name=row["name"]) + artist.albums = self.get_albums(artist.id) + artists.append(artist) + return artists + + def get_albums(self, artist_id: str) -> List[Album]: + albums = [] + c = self.conn.cursor() + c.execute( + "SELECT id, name FROM album WHERE album_artist_id=? ORDER BY name", + (artist_id,) + ) + for row in c.fetchall(): + album = Album(id=row["id"], title=row["name"]) + album.tracks = self.get_tracks(album.id) + albums.append(album) + return albums + + def get_tracks(self, album_id: int) -> List[Track]: + tracks = [] + c = self.conn.cursor() + c.execute( + "SELECT t.id, t.name AS Title, t.path, ut.favorite, ut.rating, ut.play_count AS PlayCount, ut.last_played AS LastPlayed " + "FROM track t " + "LEFT JOIN usertrack ut ON ut.track_id = t.id " + "WHERE t.album_id=? ORDER BY t.track_number", + (album_id,) + ) + for row in c.fetchall(): + track = Track( + id=row["id"], + title=row["Title"], + path=row["path"], + favorite=bool(row["favorite"]), + rating=row["rating"], + play_count=row["PlayCount"] or 0, + last_played=row["LastPlayed"], + playlists=self.get_playlists_for_track(row["id"]) + ) + tracks.append(track) + return tracks + + def get_playlists_for_track(self, track_id: int) -> List[str]: + c = self.conn.cursor() + c.execute( + "SELECT p.name FROM playlist p " + "JOIN playlisttrack pt ON pt.playlist_id=p.id " + "WHERE pt.track_id=?", + (track_id,) + ) + return [row["name"] for row in c.fetchall()] + +# Example usage +if __name__ == "__main__": + db_path = "/mnt/main/data/music/navidrome.db" + navidb = NavidromeDB(db_path) + + try: + artists = navidb.get_artists() + for artist in artists: + print(f"Artist: {artist.name}") + for album in artist.albums: + print(f" Album: {album.title}") + for track in album.tracks: + print(f" Track: {track.title} | Path: {track.path} | Fav: {track.favorite} | Plays: {track.play_count}") + finally: + navidb.close() diff --git a/src/backends/spotify/ParseSpotify.py b/src/backends/spotify/ParseSpotify.py index 50806bf..29c24e7 100644 --- a/src/backends/spotify/ParseSpotify.py +++ b/src/backends/spotify/ParseSpotify.py @@ -30,6 +30,8 @@ class Parser: self.parse_top_tracks(json.loads(top_tracks.read_text())) self.library.update_cache() + self.library.name = "Spotify" + self.library.auto_score() return self.library @@ -75,7 +77,7 @@ class Parser: track.id = track_id track.title = data.get("name") - track.artists = [self.parse_artist(artist) for artist in data.get("artists")] + track.artists = [] + self.parse_album(data.get("album")).artists track.album = self.parse_album(data.get("album")) track.duration_ms = data.get("duration_ms") @@ -114,6 +116,9 @@ class Parser: artist.id = data.get("id") artist.title = data.get("name") + if artist.title is None: + artist.title = "err_no_title" + self.library.artists.append(artist) return artist diff --git a/src/mappers/FuzzyMapper.py b/src/mappers/FuzzyMapper.py index ca78db9..c6413d0 100644 --- a/src/mappers/FuzzyMapper.py +++ b/src/mappers/FuzzyMapper.py @@ -5,12 +5,46 @@ from tqdm import tqdm from src.Library import * from src.helpers import * +import unicodedata + +user_def_artist_map = { + "donda" : "kanye west", +} + +def normalize(s): + return unicodedata.normalize("NFKC", s).replace("‐", "-") def get_score(first: str, second: str) -> float: + first_norm = normalize(first).lower() + second_norm = normalize(second).lower() + return float(first_norm == second_norm) * 100 return fuzz.token_set_ratio(first, second) +def get_album_score(first: str, second: str) -> float: + return get_score(first, second) + +def get_track_score(first: str, second: str) -> float: + return get_score(first, second) + +def get_artist_score(first: str, second: str) -> float: + if not first or not second: + return 0.0 + + first_lower = first.lower() + second_lower = second.lower() + + for key, value in user_def_artist_map.items(): + key_lower = key.lower() + value_lower = value.lower() + if (first_lower == key_lower and second_lower == value_lower) or \ + (first_lower == value_lower and second_lower == key_lower): + return 100.0 + + return get_score(first, second) + + def is_close_enough(first: float) -> bool: - return first > 50 + return first > 90 def generate_map(local_lib : Library, remote_lib : Library, output_dir : Path): @@ -21,12 +55,13 @@ def generate_map(local_lib : Library, remote_lib : Library, output_dir : Path): with (tqdm(total=len(remote_lib.artists), desc='searching remote artists in local library') as process_bar): for remote_artist in remote_lib.artists: for local_artist in local_lib.artists: - score = get_score(remote_artist.title, local_artist.title) + score = get_artist_score(remote_artist.title, local_artist.title) if not is_close_enough(score): continue if remote_artist.id not in artist_map or score > artist_map[remote_artist.id][1]: artist_map[remote_artist.id] = (local_artist.id, score) + remote_artist.local_id = local_artist.id process_bar.update(1) @@ -38,12 +73,13 @@ def generate_map(local_lib : Library, remote_lib : Library, output_dir : Path): for remote_album in remote_artist.albums: for local_album in local_artist.albums: - score = get_score(remote_album.title, local_album.title) + score = get_album_score(remote_album.title, local_album.title) if not is_close_enough(score): continue if remote_album.id not in artist_map or score > album_map[remote_album.id][1]: album_map[remote_album.id] = (local_album.id, score) + remote_album.local_id = local_album.id process_bar.update(1) @@ -54,15 +90,23 @@ def generate_map(local_lib : Library, remote_lib : Library, output_dir : Path): for remote_track in remote_album.tracks: for local_track in local_album.tracks: - score = get_score(remote_track.title, local_track.title) + score = get_track_score(remote_track.title, local_track.title) if not is_close_enough(score): continue if remote_track.id not in artist_map or score > track_map[remote_track.id][1]: track_map[remote_track.id] = (local_track.id, score) + remote_track.local_id = local_track.id process_bar.update(1) - out = { remote_track_id : local_track_id for remote_track_id, (local_track_id, _) in track_map.items() } + remote_lib.calc_resolved_percentage() + + out = { + "artist_map" : { remote_track_id : local_track_id for remote_track_id, (local_track_id, _) in artist_map.items() }, + "album_map" : { remote_track_id : local_track_id for remote_track_id, (local_track_id, _) in album_map.items() }, + "track_map" : { remote_track_id : local_track_id for remote_track_id, (local_track_id, _) in track_map.items() }, + } + set_workdir(output_dir) save_data(out, "remote_to_local_map")