diff --git a/.gitignore b/.gitignore index 2803b17..880f9c3 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,6 @@ prefetched* __pycache__ .idea secret +work_dir +old +tmp \ No newline at end of file diff --git a/Flow.py b/Flow.py index bb35731..fd10266 100644 --- a/Flow.py +++ b/Flow.py @@ -1,21 +1,31 @@ +from tqdm import tqdm +import src.lyrics.lyrics as Lyrics + from Env import client_id, client_secret, local_library_path from Env import work_dir +from src.Library import * from src.LibrarySerializer import LibrarySaver, LibraryLoader -from src.spotify.SpotifyWebAPI import fetch_all -from src.spotify.ParseSpotify import Parser +from src.backends.spotify.SpotifyWebAPI import fetch_all +from src.backends.spotify.ParseSpotify import Parser -from src.local.LocalLibraryIndexer import update_local_library -from src.local.ParseLocal import LocalParser +from src.backends.local.LocalLibraryIndexer import update_local_library +from src.backends.local.ParseLocal import LocalParser +from src.backends.itunes.ParseItunesXml import ParseItunesXml -from src.MapLocalToSpotify import map_local_to_spotify +from src.mappers.RemoteLibraryResolver import resolve_remote_tracks +from src.StatGenerator import log_stats + +import src.backends.mb.mbidFinder as mb +import src.mappers.DummyLibGenerator as libgen class Flow: - spotify_library = None - local_library = None + spotify_library : Library = None + local_library : Library = None + itunes_library : Library = None @staticmethod def fetch_spotify(self): @@ -38,18 +48,90 @@ class Flow: LibrarySaver(library_parsed).save(work_dir / "spotify_library.json") + @staticmethod + def parse_itunes_library(): + parser = ParseItunesXml() + + library_parsed = parser.parse( + work_dir / "itunes" / "Library.xml", + work_dir / "itunes" / "Library.json" + ) + + LibrarySaver(library_parsed).save(work_dir / "itunes_library.json") + @staticmethod def parse_local_library(): parser = LocalParser() library_parsed = parser.parse(work_dir / "local" / "local_library.json") LibrarySaver(library_parsed).save(work_dir / "local_library.json") + def find_mbids(self): + mb.find_mbids(self.spotify_library) + mb.find_mbids(self.itunes_library) + self.save_libraries() + + def fetch_and_save_lyrics(self): + """ + Fetches and writes lyrics for tracks that don't have lyrics yet. + Displays stats and uses a progress bar. + """ + + # ---------- Step 1: Collect tracks without lyrics ---------- + tracks_missing_lyrics = [ + track for track in self.local_library.tracks + if len(track.artists) > 0 and not track.has_lyrics + ] + + total_tracks = len(self.local_library.tracks) + missing_count = len(tracks_missing_lyrics) + + print(f"[INFO] Total tracks in library: {total_tracks}") + print(f"[INFO] Tracks missing lyrics: {missing_count}") + + if missing_count == 0: + print("[INFO] No tracks need lyrics. Exiting.") + return + + # ---------- Step 2: Fetch and write lyrics with progress bar ---------- + for track in tqdm(tracks_missing_lyrics, desc="Fetching lyrics", unit="track"): + artist_name = track.artists[0].title + album_title = track.album.title if track.album else "Unknown Album" + print(f"\n[INFO] Processing: '{track.title}' by '{artist_name}' on '{album_title}'") + + lyrics_text = Lyrics.fetch_lyrics(artist_name, album_title, track.title, track.duration_ms) + + if lyrics_text: + print("[SUCCESS] Lyrics found, writing to file...") + file_path = local_library_path / track.local_path + try: + Lyrics.write_lyrics(file_path, lyrics_text) + print(f"[SUCCESS] Lyrics written for '{track.title}'") + except Exception as e: + print(f"[ERROR] Failed to write lyrics for '{track.title}': {e}") + else: + print(f"[WARNING] Lyrics not found for '{track.title}'") + + def load_libraries(self): self.spotify_library = LibraryLoader().load(work_dir / "spotify_library.json") + self.itunes_library = LibraryLoader().load(work_dir / "itunes_library.json") self.local_library = LibraryLoader().load(work_dir / "local_library.json") + def gen_fake_libs(self): + libgen.generate_dummy_library(self.spotify_library, work_dir / "spotify" / "dummylib") + libgen.generate_dummy_library(self.itunes_library, work_dir / "itunes" / "dummylib") + + + 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): - map_local_to_spotify(client_id, client_secret, self.local_library, self.spotify_library, work_dir) + resolve_remote_tracks(client_id, client_secret, self.local_library, self.spotify_library, work_dir) + + def log_stats(self): + log_stats(work_dir / "remote_to_local_map.json", work_dir / "stats", self.local_library, self.spotify_library) + def run(self, arg): self.fetch_local() diff --git a/Gui.py b/Gui.py new file mode 100644 index 0000000..12d1144 --- /dev/null +++ b/Gui.py @@ -0,0 +1,208 @@ +import time + +import streamlit as st +from st_aggrid import AgGrid, GridOptionsBuilder, GridUpdateMode +import pandas as pd + +import Flow + +from src.Library import * + +def entity_to_row(entity: Entity): + return { + "id": entity.id, + "mbid": entity.mbid, + "Title": entity.title, + "PLay Count": entity.play_count, + "Resolved": int(entity.resolved_percentage * 100), + "AutoScore": entity.auto_score, + # "Added": entity.date_added.strftime("%Y-%m-%d") if entity.date_added else None, + # "Played": entity.date_last_play.strftime("%Y-%m-%d") if entity.date_last_play else None, + # "Released": entity.date_released.strftime("%Y-%m-%d") if entity.date_released else None, + "LocalId": str(entity.local_id) if entity.local_id else None, + "Path": str(entity.local_path) if entity.local_path else None, + } + +def track_to_row(track: Track): + return entity_to_row(track) | { + "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, + } + +def artist_to_row(artist: Artist): + return entity_to_row(artist) | { + } + +def album_to_row(album: Album): + return entity_to_row(album) | { + } + + +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.itunes_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 98b4efa..05a781d 100644 --- a/main.py +++ b/main.py @@ -1,17 +1,41 @@ - import Flow -if __name__ == '__main__': + +def test1(): + flow = Flow.Flow() + + # flow.fetch_local() + flow.parse_local_library() + flow.load_libraries() + + # flow.map_local_to_spotify() + # flow.log_stats() + + # flow.fetch_and_save_lyrics() + + +def main(): flow = Flow.Flow() # flow.fetch_spotify() - flow.fetch_local() + # flow.fetch_local() flow.parse_spotify_library() + flow.parse_itunes_library() flow.parse_local_library() flow.load_libraries() - flow.map_local_to_spotify() + # flow.gen_fake_libs() + # flow.find_mbids() - print("asd") + flow.map_local_to_spotify() + flow.save_libraries() + + # flow.log_stats() + + # print("asd") + + +# test1() +main() diff --git a/old/env b/old/env deleted file mode 100755 index 8955572..0000000 --- a/old/env +++ /dev/null @@ -1,36 +0,0 @@ - -PS1=" > " - -#export PATH="/usr/local/bin/:$PATH" -export PATH="$HOME/bin/scripts:$HOME/bin/:$PATH" -#export PATH="$HOME/home/auser/.local/bin:$PATH" - -export SIP="185.238.170.251" - -alias v=nvim -alias ll="ls -l -a" -alias gl="git log --oneline" -alias ssh_server="ssh auser@185.238.170.251" - -#eval "$(zoxide init bash --cmd zd)" - -#source $PMAIN/.scripts/bashmarks.sh - -fe() { - local result=$(command tere "$@") - [ -n "$result" ] && cd -- "$result" -} - -vim_configure() { - pwd=$(pwd) - cd ~/.config/nvim/ - nvim - cd $pwd -} - -pyenv() { - source ~/bin/python/env311/bin/activate -} - -source ~/src/scripts/remotes.sh -source /usr/share/fzf/key-bindings.bash diff --git a/old/main.py b/old/main.py index 10d378a..0b2a991 100644 --- a/old/main.py +++ b/old/main.py @@ -1,79 +1,3 @@ -from DataBase import * -from LocalLibraryIndexer import update_local_library -from LinkerPatternGenerator import run_pattern_generators -from LocalPLaylistGenerator import generate_playlists - -from SpotifyWebAPI import fetch_data, update_access_token -import cmd - -local_library_path = "/home/auser/Music/" -rel_autogen_dir = "0autogen" - -playlists = get_data('playlistTracks') -top_artists = get_data('top_artists') -top_tracks = get_data('top_tracks') -user_tracks = get_data('tracks') -link_pattern = get_data('link_pattern_spotify') -local_library = get_data("local_library") -user_tracks_by_id = {track['track']['id']: track['track'] for track in user_tracks} - - -def get_playlist_names(pl): - return [name for name, items in pl.items()] - - -def print_playlist_tracks(pl_id: int): - pls = list(playlists) - name = pls[pl_id] - pl = playlists[name] - print(name) - for track in pl: - track_data = track['track'] - track_name = track_data['name'] - artists = get_artists_str(track_data['artists']) - print(f" '{track_name}' - '{artists}' ") - - -def print_top_artists(): - for artist in top_artists: - name = artist['name'] - print(f" '{name}'") - - -def print_top_tracks(): - for track in top_tracks: - track_name = track['name'] - artists = get_artists_str(track['artists']) - print(f" '{track_name}' - '{artists}' ") - - -def print_stats(): - print(f" tracks - {len(user_tracks)}") - print(f" playlists - {len(playlists)}") - - for name, tracks in playlists.items(): - print(f" '{name}' - {len(tracks)}") - - print(f" top tracks - {len(top_tracks)}") - print(f" top artists - {len(top_artists)}") - - -def print_tracks(): - track_idx = 0 - for trackItem in user_tracks: - track = trackItem['track'] - print(f" {track_idx}: '{track['name']}' by '{get_artists_str(track['artists'])}'") - track_idx += 1 - - -def sort_remote_tracks_by_popularity(track_ids): - out = [] - for top_track in top_tracks: - if top_track['id'] in track_ids: - out.append(top_track['id']) - return out - - def print_link_stats(link_pattern): resolved_reversed = {} resolved = [] @@ -114,121 +38,4 @@ def print_link_stats(link_pattern): print(f" {index}: '{local_track['name']}' by '{local_track['artist']}'", end=' ----> ') print(f"'{remote_track['name']}' by '{get_artists_str(remote_track['artists'])}'") - index += 1 - - -class Interpreter(cmd.Cmd): - intro = "Welcome to the interactive command loop. Type help or ? to list commands.\n" - prompt = "(spotify) " - - def do_list_stat_vs_types(self, arg): - """prints available linking patterns""" - print(f"0 - fuzzy tags") - print(f"1 - spotify") - print(f"2 - fuzzy") - - def do_stat_local_library_vs_spotify(self, arg): - """compares local library and remote with the link pattern generated by the generate_pattern_command""" - print_link_stats(link_pattern) - - def do_playlists(self, arg): - """prints user playlists""" - pl_id = 0 - for pl in get_playlist_names(playlists): - print(f"{pl_id} - {pl}") - pl_id += 1 - - def do_playlist_tracks(self, arg): - """prints playlist [name]""" - try: - pl_id = int(arg.split()[0]) - print_playlist_tracks(pl_id) - except IndexError: - print("Invalid input") - - def do_top_artists(self, arg): - """Top artists""" - print_top_artists() - - def do_top_tracks(self, arg): - """Top tracks""" - print_top_tracks() - - def do_stat(self, arg): - """Print stats""" - print_stats() - - def do_tracks(self, arg): - """Print tracks""" - print_tracks() - - def do_generate_local_playlists(self, arg): - """generates local playlists based on the vs comparison type""" - generate_playlists(local_library, local_library_path, playlists, link_pattern, local_library_path + rel_autogen_dir) - - - def do_index_local_library(self, arg): - """find local tracks and fetches the data from them for further analysis""" - global local_library - update_local_library(local_library_path) - local_library = get_data("local_library") - - def do_generate_vs_stat(self, arg): - """compares local and remote library and generates the vs stats""" - - global link_pattern - run_pattern_generators() - link_pattern = get_data('link_pattern_spotify') - - def do_fetch(self, arg): - """Fetch all spotify data""" - try: - update_access_token() - fetch_data() - - global playlists, top_artists, top_tracks, user_tracks - - playlists = get_data('playlistTracks') - top_artists = get_data('top_artists') - top_tracks = get_data('top_tracks') - user_tracks = get_data('tracks') - - except ValueError: - print("Can not fetch the data.") - - def do_generate_todo(self, arg): - """generates todos for the pirates""" - print("sorry not implemented yet.") - - def do_run_all_default(self, arg): - """runs general flow""" - - local_lib_path = "/home/auser/Music/" - print("Fetching spotify data...") - self.do_fetch([]) - - print("Fetching local data...") - self.do_index_local_library(f"{local_lib_path}/organized") - - print("Generating vs patterns...") - self.do_generate_vs_stat("0") - - print("Generating todos...") - self.do_generate_todo(f"{local_lib_path}") - - print("Generating local playlists...") - self.do_generate_local_playlists(f"{local_lib_path}") - - return True - - def do_exit(self, arg): - """Exit the command loop: exit""" - print("Goodbye!") - return True - - -if __name__ == '__main__': - try: - Interpreter().cmdloop() - except KeyboardInterrupt as kb: - print("process terminated") \ No newline at end of file + index += 1 \ No newline at end of file diff --git a/src/Library.py b/src/Library.py index c3287b8..246d581 100644 --- a/src/Library.py +++ b/src/Library.py @@ -4,48 +4,125 @@ from datetime import datetime from pathlib import Path from typing import List, Optional, Any -class Artist: + +class Entity: def __init__(self): - self.id : str = None - self.title : str = None - self.tracks : List[Track] = [] + self.id: str = None + self.mbid: str = None + + self.title: str = None + self.duration_ms: int = None + + self.fav: bool = False + self.rating: int = 0 + self.play_count: int = 0 + self.date_added: datetime = None + self.date_released: datetime = None + self.date_last_play: datetime = None + self.auto_score: float = 0 + + self.local_id: str = None + self.local_path: Path = None + self.resolved_percentage: float = 0 -class Track: +class Artist(Entity): 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 + super().__init__() + self.tracks: List[Track] = [] + self.albums: List[Album] = [] -class Album: +class Track(Entity): def __init__(self): - self.id : str = None - self.title : str = None - self.release_date : datetime = None - self.tracks : List[Track] = [] - self.artists : List[Artist] = [] + super().__init__() + + self.artists: List[Artist] = [] + self.album: Album = None + + self.has_lyrics: bool = False -class Playlist: +class Album(Entity): def __init__(self): - self.title : str = None - self.tracks : List[Track] = [] - self.created_at : datetime = None - self.description : str = None + super().__init__() + self.tracks: List[Track] = [] + self.artists: List[Artist] = [] + + +class Playlist(Entity): + def __init__(self): + super().__init__() + self.tracks: List[Track] = [] + self.created_at: datetime = None + self.description: str = None 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.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} + + def create_reverse_links(self): + for track_id, track in self.track_map.items(): + for artist in track.artists: + if track not in artist.tracks: + artist.tracks.append(track) + + if track not in track.album.tracks: + track.album.tracks.append(track) + + for album_id, album in self.album_map.items(): + for artist in album.artists: + if album not in artist.albums: + artist.albums.append(album) + + def auto_score(self): + self.score_albums() + self.score_artists() + + 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 diff --git a/src/LibrarySerializer.py b/src/LibrarySerializer.py index bc968ea..278776a 100644 --- a/src/LibrarySerializer.py +++ b/src/LibrarySerializer.py @@ -26,48 +26,57 @@ class LibrarySaver: def __init__(self, library: Library): self.library = library - @staticmethod - def _artist_to_dict(artist: Artist) -> dict[str, Any]: + def _entity_to_dict(self, entity: Entity) -> dict[str, Any]: return { - "id": artist.id, - "title": artist.title, - "tracks": [track.id for track in artist.tracks], + "id": entity.id, + "mbid": entity.mbid, + + "title": entity.title, + "duration": entity.duration_ms, + + "fav" : entity.fav, + "rating" : entity.rating, + "play_count": entity.play_count, + "date_added": entity.date_added, + "date_released": entity.date_released, + "date_last_play": entity.date_last_play, + "auto_score": entity.auto_score, + + "local_id": entity.local_id, + "local_path": entity.local_path, + "resolved_percentage": entity.resolved_percentage, } - @staticmethod - def _track_to_dict(track: Track) -> dict[str, Any]: - return { - "id": track.id, - "title": track.title, + def _artist_to_dict(self, artist: Artist) -> dict[str, Any]: + return self._entity_to_dict(artist) | { + "tracks": [track.id for track in artist.tracks], + "albums": [album.id for album in artist.albums], + } + + def _track_to_dict(self, track: Track) -> dict[str, Any]: + return self._entity_to_dict(track) | { "artists": [artist.id for artist in track.artists], "album": track.album.id, "duration_ms": track.duration_ms, - "added_at": _dt_to_str(track.added_at), - "local_path": track.local_path, + "has_lyrics": track.has_lyrics, } - @staticmethod - def _playlist_to_dict(playlist: Playlist) -> dict[str, Any]: - return { - "title": playlist.title, + def _playlist_to_dict(self, playlist: Playlist) -> dict[str, Any]: + return self._entity_to_dict(playlist) | { "tracks": [track.id for track in playlist.tracks], - "created_at": _dt_to_str(playlist.created_at), "description": playlist.description, } - @staticmethod - def _album_to_dict(album: Album) -> dict[str, Any]: - return { - "id": album.id, - "title": album.title, - "release_date": album.release_date, + def _album_to_dict(self ,album: Album) -> dict[str, Any]: + return self._entity_to_dict(album) | { "tracks": [track.id for track in album.tracks], - "artists": [artist.id for artist in album.artists] + "artists": [artist.id for artist in album.artists], } 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], @@ -82,40 +91,82 @@ class LibrarySaver: json.dump(data, f, indent=2, ensure_ascii=False) + class LibraryLoader: def __init__(self): self.track_map: dict[str, Track] = {} self.album_map: dict[str, Album] = {} self.artist_map: dict[str, Artist] = {} + def get_artist(self, artist_id) -> Artist: + artist = self.artist_map.get(artist_id) + if artist is None: + artist = Artist() + self.artist_map[artist_id] = artist + return artist + + def get_album(self, id) -> Album: + item = self.album_map.get(id) + if item is None: + item = Album() + self.album_map[id] = item + return item + + def get_track(self, id) -> Track: + item = self.track_map.get(id) + if item is None: + item = Track() + self.track_map[id] = item + return item + + def load_entity(self, entity : Entity, data: dict[str, Any]): + entity.id = data.get("id") + entity.mbid = data.get("mbid") + + entity.title = data.get("title") + entity.duration_ms = data.get("duration_ms") + + entity.fav = data.get("fav") + entity.rating = data.get("rating") + entity.play_count = data.get("play_count") + entity.date_added = _str_to_dt(data.get("date_added")) + entity.date_released = _str_to_dt(data.get("date_released")) + entity.date_last_play = _str_to_dt(data.get("date_last_play")) + entity.auto_score = data.get("auto_score") + + entity.local_id = data.get("local_id") + entity.local_path = data.get("local_path") + entity.resolved_percentage = data.get("resolved_percentage") + + def load_track(self, data: dict[str, Any]) -> Track: - track = self.track_map.get(data.get("id"), Track()) - track.id = data.get("id") - track.title = data.get("title") + track = self.get_track(data.get("id")) + self.load_entity(track, data) + track.duration_ms = data.get("duration_ms") - track.added_at = _str_to_dt(data.get("added_at")) - track.local_path = data.get("local_path") - track.artists = [self.artist_map.get(artist_id, Artist()) for artist_id in data.get("artists")] - track.album = self.album_map.get(data.get("album"), Album()) + 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") self.track_map[track.id] = track return track def load_artist(self, data: dict[str, Any]): - artist = self.artist_map.get(data.get("id"), Artist()) - artist.id = data.get("id") - artist.title = data.get("title") - artist.tracks = [self.track_map.get(track_id, Track()) for track_id in data.get("tracks")] + artist = self.get_artist(data.get("id")) + self.load_entity(artist, data) + + 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 [] self.artist_map[artist.id] = artist return artist def load_album(self, data: dict[str, Any]): - album = self.album_map.get(data.get("id"), Album()) - album.id = data.get("id") - album.tracks = [self.track_map.get(track_id, Track()) for track_id in data.get("tracks")] - album.title = data.get("title") - album.artists = [self.artist_map.get(artist_id, Artist()) for artist_id in data.get("artists")] + album = self.get_album(data.get("id")) + self.load_entity(album, data) + + album.tracks = [self.get_track(track_id) for track_id in data.get("tracks")] + album.artists = [self.get_artist(artist_id) for artist_id in data.get("artists")] self.album_map[album.id] = album return album @@ -126,6 +177,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")] @@ -134,4 +186,6 @@ class LibraryLoader: library.top_tracks = [self.track_map[track] for track in data.get("top_tracks")] library.liked_tracks = [self.track_map[track] for track in data.get("liked_tracks")] + library.update_cache() + return library diff --git a/src/MapLocalToSpotify.py b/src/MapLocalToSpotify.py deleted file mode 100644 index 84e37c5..0000000 --- a/src/MapLocalToSpotify.py +++ /dev/null @@ -1,140 +0,0 @@ - -from rapidfuzz import fuzz -from tqdm import tqdm -import requests - -from src.Library import * -from src.helpers import * -from src.spotify.SpotifyWebAPI import find_song, update_access_token - - -def map_local_to_spotify(client_id, client_secret, local_lib : Library, spotify_lib : Library, output_dir): - - set_workdir(output_dir) - - update_access_token(client_id, client_secret) - - def find_local_songs_on_spotify(): - search_log = [] - - total = len(local_lib.tracks) - found_tracks_map = {} - with tqdm(total=total, desc='Searching local songs on spotify') as pbar: - for track in local_lib.tracks: - - parts = [track.title, track.artists[0].title if len(track.artists) else "", track.album.title] - search_pattern = " ".join(p for p in parts if p) - - if search_pattern == "": - print(f"cannot generate search pattern for the track - {track.local_path}") - pbar.update(1) - continue - - found_tracks = find_song(search_pattern)[0:10] - found_tracks_map[track.id] = found_tracks - search_log.append({"path" : track.local_path, "search_pattern": search_pattern, "result" : found_tracks_map[track.id]}) - pbar.update(1) - - save_data(search_log, "spotify_search_log") - save_data(found_tracks_map, "spotify_found_map") - - return found_tracks_map - - spotify_found_tracks = find_local_songs_on_spotify() - - spotify_user_track = {track.id for track in spotify_lib.tracks} - - spotify_pattern = {} - for local_id, found_items in spotify_found_tracks.items(): - spotify_pattern[local_id] = {"score": 1.0, "items": []} - - if not len(found_items): - continue - - for found_item in found_items: - spotify_track_id = found_item['id'] - if spotify_track_id in spotify_user_track: - spotify_pattern[local_id]['items'].append(spotify_track_id) - - save_data(spotify_pattern, "local_to_spotify_id_map") - - -def fuzzy_pattern_generator(): - local_library = get_data("local_library") - track_descriptions = get_data("tracks") - - def get_matched(src_pattern, patterns): - result = {} - for target_pattern in patterns: - similarity_score = fuzz.token_set_ratio(src_pattern, target_pattern) - result[target_pattern] = similarity_score / 100 - - sorted_results = sorted(result.items(), key=lambda item: item[1], reverse=True) - return sorted_results - - def resolve_tracks(s_patterns, t_patterns): - - pattern_map = {} - - total = len(s_patterns) - with tqdm(total=total, desc='Progress') as pbar: - for s_pattern in s_patterns: - res = get_matched(s_pattern, t_patterns) - matched = {"score": 0, "items": []} - if len(res): - matched['items'] = res[0:min(len(res), 4)] - matched['score'] = res[0][1] - pattern_map[s_pattern] = matched - pbar.update(1) - - return pattern_map - - source_patterns = [] - for track in track_descriptions: - artists = get_artists_str(track['track']['artists']) - pattern = f"{track['track']['name']} {artists}" - source_patterns.append(pattern) - - target_patterns = [local['path'] for _, local in local_library.items()] - - fuzzy_pattern = resolve_tracks(source_patterns, target_patterns) - save_data(fuzzy_pattern, "link_pattern_fuzzy") - - -def fuzzy_tag_pattern_generator(): - local_library = get_data("local_library") - spotify_library = get_data("tracks") - mappings = {} - - def fuzzy_tags_ratio(first: SongTags, second: SongTags): - title_ratio = fuzz.token_set_ratio(first.title, second.title) / 100 - artist_ratio = fuzz.token_set_ratio(first.artist, second.artist) / 100 - album_ratio = fuzz.token_set_ratio(first.album, second.album) / 100 - return title_ratio, artist_ratio, album_ratio - - with tqdm(total=len(local_library), desc='fuzzy_tag_pattern_generator') as pbar: - for local_track_id, local_track in local_library.items(): - mapping = mappings.get(local_track_id, {"score": 1.0, "items": []}) - ratios = [] - for remote_track in spotify_library: - track = remote_track['track'] - - tag1 = SongTags(local_track['name'], local_track['artist'], local_track['album']) - tag2 = SongTags(track['name'], get_artists_str(track['artists']), track['album']['name']) - - ratios.append((fuzzy_tags_ratio(tag1, tag2), track['id'])) - - filtered_ratios = [] - threshold = 0.81 - for ratio in ratios: - if all([val > threshold for val in ratio[0]]): - filtered_ratios.append((sum(i for i in ratio[0]) / len(ratio[0]), ratio[1])) - - filtered_ratios.sort(reverse=True, key=lambda x: x[0]) - if len(filtered_ratios): - mapping['items'] = [filtered_ratios[0][1]] - - mappings[local_track_id] = mapping - pbar.update(1) - - save_data(mappings, "link_pattern_tags") \ No newline at end of file diff --git a/src/StatGenerator.py b/src/StatGenerator.py index c40bebb..7c17ab5 100644 --- a/src/StatGenerator.py +++ b/src/StatGenerator.py @@ -1,42 +1,210 @@ +from src.Library import * +from src.helpers import * -def print_link_stats(link_pattern): - resolved_reversed = {} - resolved = [] - unresolved_local = [] - unresolved_remote = [] +from pathlib import Path +import json - for local_id, links in link_pattern.items(): - if len(links['items']): - resolved.append(local_id) - resolved_reversed[links['items'][0]] = local_id - else: - unresolved_local.append(local_id) - for track in user_tracks: - track_id = track['track']['id'] - if track_id not in resolved_reversed: - unresolved_remote.append(track_id) +def is_resolved(track: Track): + return track.local_path is not None - print(f"Local Tracks: {len(local_library)}") - print(f"Remote Tracks: {len(user_tracks)}") - unresolved_remote = sort_remote_tracks_by_popularity(unresolved_remote) - unresolved_remote = unresolved_remote[0:min(len(unresolved_remote), 100)] +def get_mapping(mapping_path) -> dict[str, Any]: + data = {} - print(f"\nUnresolved tracks from the remote: {len(unresolved_remote)}") - index = 0 - for remote_id in unresolved_remote: - remote_track = user_tracks_by_id[remote_id] - print(f" {index}: '{remote_track['name']}' by '{get_artists_str(remote_track['artists'])}'") - index += 1 + with Path(mapping_path).open("r", encoding="utf-8") as f: + data = json.load(f) - print(f"\nResolved tracks: {len(resolved)}") - index = 0 - for link in resolved: - remote_id = link_pattern[link]['items'][0] - local_track = local_library[link] - remote_track = user_tracks_by_id[remote_id] + return data["track_map"] - print(f" {index}: '{local_track['name']}' by '{local_track['artist']}'", end=' ----> ') - print(f"'{remote_track['name']}' by '{get_artists_str(remote_track['artists'])}'") - index += 1 \ No newline at end of file + +class AlbumStat: + def __init__(self): + self.album: Album = Album() + self.resolved_precent: float = 0 + + def calc_resolved_precent(self): + self.resolved_precent = (sum(1 for track in self.album.tracks if is_resolved(track)) / len( + self.album.tracks)) * 100 + + def get_str(self): + return (f"{str(int(self.resolved_precent))} % - " + f"{self.album.title} - " + f"{str(self.album.artists[0].title) if len(self.album.artists) else " "}") + + +class ArtistStat: + def __init__(self): + self.artist: Artist = Artist() + self.resolved_precent: float = 0 + self.albums_stats: list[AlbumStat] = [] + + def calc_resolved_precent(self): + if len(self.artist.tracks) == 0: + self.resolved_precent = 0 + return + + self.resolved_precent = (sum(1 for track in self.artist.tracks if is_resolved(track)) / len( + self.artist.tracks)) * 100 + + def get_str(self): + return (f"{str(int(self.resolved_precent))} % - " + f"{self.artist.title}") + +class LibraryStats: + def __init__(self, mapping_path : Path, local_lib: Library, remote_lib: Library): + self.max_popular_albums = 100 + self.max_popular_artists = 100 + + self.max_unresolved_tracks = 100 + self.max_unresolved_albums = 20 + self.resolved_percentage_threshold = 30 + + self.albums: List[AlbumStat] = [] + self.artists: List[ArtistStat] = [] + + self.albums_map: dict[str, AlbumStat] = {} + self.artists_map: dict[str, ArtistStat] = {} + + self.local_lib: Library = local_lib + self.remote_lib: Library = remote_lib + + self.mapping: dict[str, Any] = get_mapping(mapping_path) + + self.remote_id_map: dict[str, Track] = {} + self.local_id_map: dict[str, Track] = {} + + self.find_add_local_paths() + self.score_artists() + self.score_albums() + + def find_add_local_paths(self): + self.remote_id_map = {track.id: track for track in self.remote_lib.tracks} + self.local_id_map = {track.id: track for track in self.local_lib.tracks} + + 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.artist.auto_score = 0 + self.artists_map[artist_stat.artist.id] = artist_stat + self.artists.append(artist_stat) + + for track in self.remote_lib.tracks: + for artist in track.artists: + self.artists_map[artist.id].artist.auto_score += track.auto_score + + for artist_stat in self.artists: + artist_stat.calc_resolved_precent() + + self.artists = sorted(self.artists, key=lambda item: item.score, reverse=True) + + def score_albums(self): + for album in self.remote_lib.albums: + stat = AlbumStat() + stat.album = album + self.albums_map[album.id] = stat + + 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() + + self.albums = [item[1] for item in + sorted(self.albums_map.items(), key=lambda item: item[1].score, reverse=True)] + + for album_stat in self.albums: + for artist in album_stat.album.artists: + self.artists_map[artist.id].albums_stats.append(album_stat) + + + def gen_album(self, album_stat, condition): + tracks = [] + + for track in album_stat.album.tracks: + if condition.should_add_track(track): + tracks.append(f"{"✔" if is_resolved(track) else "𐄂"} {track.title} -> {track.local_path}") + + return { + "album" : album_stat.get_str(), + "tracks" : tracks + } + + def gen_artist(self, artist_stat, condition): + albums = [] + + for album_stat in artist_stat.albums_stats: + if condition.should_add_album(album_stat): + albums.append(self.gen_album(album_stat, condition)) + + return { + "artist" : artist_stat.get_str(), + "albums" : albums, + } + + def generate_stat_tree(self, condition): + artists = [] + + for artist_stat in self.artists: + if condition.should_add_artist(artist_stat): + artists.append(self.gen_artist(artist_stat, condition)) + + return { + "artists" : artists + } + + def save_tree(self, output_path, condition): + save_data(self.generate_stat_tree(condition), output_path) + + def save_all(self, output_path): + + class Condition: + def should_add_artist(self, item): + return True + + def should_add_album(self, item): + return True + + def should_add_track(self, item): + return True + + + self.save_tree(output_path, Condition()) + + + def save_threshold(self, output_path, threshold = 80): + + class Condition: + def should_add_artist(self, item : ArtistStat): + for album in item.albums_stats: + if self.should_add_album(album): + return True + + return False + + def should_add_album(self, item: AlbumStat): + return item.resolved_precent < threshold and len(item.album.tracks) > 3 + + def should_add_track(self, item: Track): + return not is_resolved(item) + + self.save_tree(output_path, Condition()) + + + +def log_stats(mapping_path: Path, output_dir: Path, local_lib: Library, remote_lib: Library): + set_workdir(output_dir) + + stat = LibraryStats(mapping_path, local_lib, remote_lib) + + stat.save_all("stats_all") + stat.save_threshold("stats_missing") diff --git a/src/backends/itunes/ParseItunesXml.py b/src/backends/itunes/ParseItunesXml.py new file mode 100755 index 0000000..4b0f417 --- /dev/null +++ b/src/backends/itunes/ParseItunesXml.py @@ -0,0 +1,194 @@ +import xmltodict +import json +import xmltodict +import json +import re + +from src.Library import * + +class XmlToJson: + def flatten_dict(self, d): + + out = [] + for song in d: + newSong = {} + + strCount = 0 + intCount = 0 + dateCount = 0 + + def getStr(id): + nonlocal strCount + if id not in song["key"]: + return "undef" + if len(song["string"]) <= strCount: + return "error" + strCount += 1 + return song["string"][strCount - 1] + + def getInt(id): + nonlocal intCount + if id not in song["key"]: + return -1 + if len(song["integer"]) <= intCount: + return -1 + intCount += 1 + return song["integer"][intCount - 1] + + def getDate(id): + nonlocal dateCount + if id not in song["key"]: + return "undef" + if len(song["date"]) <= dateCount: + return "error" + dateCount += 1 + return song["date"][dateCount - 1] + + newSong["Track ID"] = getInt("Track ID") + newSong["Name"] = getStr("Name") + newSong["Artist"] = getStr("Artist") + newSong["Album Artist"] = getStr("Album Artist") + newSong["Composer"] = getStr("Composer") + newSong["Album"] = getStr("Album") + newSong["Genre"] = getStr("Genre") + newSong["Kind"] = getStr("Kind") + newSong["Size"] = getInt("Size") + newSong["Total Time"] = getInt("Total Time") + newSong["Disc Number"] = getInt("Disc Number") + newSong["Disc Count"] = getInt("Disc Count") + newSong["Track Number"] = getInt("Track Number") + newSong["Track Count"] = getInt("Track Count") + newSong["Year"] = getInt("Year") + newSong["Date Modified"] = getDate("Date Modified") + newSong["Date Added"] = getDate("Date Added") + newSong["Bit Rate"] = getInt("Bit Rate") + newSong["Sample Rate"] = getInt("Sample Rate") + newSong["Play Count"] = getInt("Play Count") + newSong["Play Date"] = getInt("Play Date") + newSong["Play Date UTC"] = getDate("Play Date UTC") + newSong["Skip Count"] = getInt("Skip Count") + newSong["Skip Date"] = getDate("Skip Date") + newSong["Release Date"] = getDate("Release Date") + newSong["Album Rating"] = getInt("Album Rating") + newSong["Album Rating Computed"] = "Album Rating Computed" in song["key"] + newSong["Loved"] = "Loved" in song["key"] + newSong["Album Loved"] = "Album Loved" in song["key"] + newSong["Explicit"] = "Explicit" in song["key"] + newSong["Compilation"] = "Compilation" in song["key"] + newSong["Artwork Count"] = getInt("Artwork Count") + newSong["Sort Album"] = getStr("Sort Album") + newSong["Sort Artist"] = getStr("Sort Artist") + newSong["Sort Name"] = getStr("Sort Name") + newSong["Persistent ID"] = getStr("Persistent ID") + newSong["Track Type"] = getStr("Track Type") + + out.append(newSong) + + return out + + + def convert(self, filename, out_path): + with open(filename, 'r', encoding='utf-8') as xml_file: + data_dict = xmltodict.parse(xml_file.read()) + + # Extract the "Tracks" dictionary to be flattened + tracks_dict = data_dict['plist']['dict']['dict'] + + flat_tracks_dict = self.flatten_dict(tracks_dict["dict"]) + + json_data = json.dumps(flat_tracks_dict, indent=2) + + with open(out_path, 'w', encoding='utf-8') as json_file: + json_file.write(json_data) + + +class ParseItunesXml: + def __init__(self): + self.track_map: dict[str, Track] = {} + self.artist_map: dict[str, Artist] = {} + self.album_map: dict[str, Album] = {} + self.library = Library() + + + def parse_artists(self, data: list) -> List[Artist]: + if not data: + raise "No artist found" + + artists = [] + + for artist_data in data: + name = artist_data + artist_id = name + + if artist_id in self.artist_map: + artists.append(self.artist_map[artist_id]) + continue + + artist = Artist() + artist.id = artist_id + artist.title = name + + self.artist_map[artist_id] = artist + self.library.artists.append(artist) + + artists.append(artist) + + return artists + + def parse_album(self, data: str) -> Album: + if not data: + raise "album is none" + + name = data + album_id = name + + if album_id in self.album_map: + return self.album_map[album_id] + + album = Album() + album.id = album_id + album.title = name + + self.album_map[album_id] = album + self.library.albums.append(album) + + return album + + + def parse(self, tracks_file_xml: Path, tracks_file_json: Path) -> Library: + XmlToJson().convert(tracks_file_xml, tracks_file_json) + + tracks_content = json.loads(tracks_file_json.read_text()) + + for data in tracks_content: + track = Track() + + track.id = data.get("Track ID") + track.title = data.get("Name") + track.artists = self.parse_artists([data.get("Artist")]) + track.album = self.parse_album(data.get("Album")) + track.play_count = int(data.get("Play Count")) + + 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) + + for track_id, track in self.track_map.items(): + for artist in track.artists: + if artist not in track.album.artists: + track.album.artists.append(artist) + + self.library.update_cache() + + self.library.name = "Itunes" + + for track in self.library.tracks: + track.auto_score = track.play_count + + self.library.auto_score() + + return self.library + + diff --git a/src/backends/local/LocalLibraryIndexer.py b/src/backends/local/LocalLibraryIndexer.py new file mode 100644 index 0000000..08629f7 --- /dev/null +++ b/src/backends/local/LocalLibraryIndexer.py @@ -0,0 +1,183 @@ +import os +import fnmatch +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*") + + +def detect_lyrics(abs_path, ext): + ext = ext.lower() + + try: + audio = MutagenFile(abs_path) + 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: + return False + + return False + + +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 get_mb_artist(tags, fallback=None) -> list : + artist_fields = ["TXXX:MusicBrainz Artist Id", "----:com.apple.iTunes:MusicBrainz Artist Id", "MUSICBRAINZ_ARTISTID"] + 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 + + 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") + known_paths = {track['path']: track_id for track_id, track in old_library.items()} + new_library = {} + + song_id = '0' + + def get_new_song_id(prev_id, song_path): + if song_path in known_paths: + return int(known_paths[song_path]) + while (prev_id in old_library) or (prev_id in new_library): + prev_id = str(int(prev_id) + 1) + return prev_id + + for dir_path, dir_names, filenames in os.walk(root_dir): + dir_names[:] = [ + d for d in dir_names + if not any(fnmatch.fnmatch(d, exclude) for exclude in exclude_directories) + ] + + filtered_filenames = [ + filename for filename in filenames + if any(fnmatch.fnmatch(filename, ext) for ext in music_extensions) + ] + + for filename in filtered_filenames: + song_name, track_type = os.path.splitext(filename) + relative_path = os.path.relpath(os.path.join(dir_path, filename), root_dir) + song_id = get_new_song_id(song_id, relative_path) + + new_library[song_id] = { + "name": None, + "path": relative_path, + "type": track_type, + "artists": None, + "album": None, + "duration": None + } + + to_remove = [] + + 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['artists_mbid'] = get_mb_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" + + else: + print(track) + to_remove.append(track_id) + # raise "Cannot index track" + + except Exception as e: + print(e) + print(abs_path) + raise e + + + for rem in to_remove: + new_library.pop(rem) + + save_data(new_library, "local_library") diff --git a/src/local/LocalPLaylistGenerator.py b/src/backends/local/LocalPLaylistGenerator.py similarity index 93% rename from src/local/LocalPLaylistGenerator.py rename to src/backends/local/LocalPLaylistGenerator.py index 0299498..1c0f1e1 100644 --- a/src/local/LocalPLaylistGenerator.py +++ b/src/backends/local/LocalPLaylistGenerator.py @@ -1,7 +1,7 @@ from pathlib import Path -def generate_playlists(local_library, library_path, remote_playlists, link_pattern, output_dir): +def (): output_dir = Path(output_dir) output_playlists = [] diff --git a/src/local/ParseLocal.py b/src/backends/local/ParseLocal.py similarity index 59% rename from src/local/ParseLocal.py rename to src/backends/local/ParseLocal.py index 602e89e..b1fbba3 100644 --- a/src/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 @@ -104,15 +62,6 @@ class LocalParser: return album - def add_tracks_to_albums_and_artists(self): - for track_id, track in self.track_map.items(): - for artist in track.artists: - if track not in artist.tracks: - artist.tracks.append(track) - - if track not in track.album.tracks: - track.album.tracks.append(track) - def parse(self, tracks_file: Path) -> Library: tracks_content = json.loads(tracks_file.read_text()) @@ -122,14 +71,26 @@ 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) - self.add_tracks_to_albums_and_artists() + for track_id, track in self.track_map.items(): + for artist in track.artists: + if artist not in track.album.artists: + track.album.artists.append(artist) + + self.library.update_cache() + + self.library.name = "Local" return self.library diff --git a/src/backends/mb/mbidFinder.py b/src/backends/mb/mbidFinder.py new file mode 100644 index 0000000..6ffcd09 --- /dev/null +++ b/src/backends/mb/mbidFinder.py @@ -0,0 +1,54 @@ +from src.Library import * + +import musicbrainzngs as mb +from tqdm import tqdm + +mb.set_useragent( + "resolve_spotify", + "1.0", + "ilusha.main@gmail.com" +) + +def find_mbid_by_tags(track: Track) -> str | None: + if not track.title or not track.artists: + return None + + artist_names = [a.title for a in track.artists if a.title] + if not artist_names: + return None + + query_parts = [ + f'recording:"{track.title}"', + f'artist:"{artist_names[0]}"' + ] + + query = " AND ".join(query_parts) + + try: + result = mb.search_recordings(query=query, limit=5) + except Exception: + return None + + recordings = result.get("recording-list", []) + if not recordings: + return None + + # Pick best-scored result + best = max( + recordings, + key=lambda r: int(r.get("score", 0)) + ) + + return best.get("id") + + +def find_mbids(library: Library): + tracks = library.tracks + + for track in tqdm(tracks, desc="Resolving MBIDs of " + library.name + " library", unit="track"): + if track.mbid: + continue + + track.mbid = find_mbid_by_tags(track) + if not track.mbid: + print(f"not found for track {track.title} {track.artists[0].title}") 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/spotify/ParseSpotify.py b/src/backends/spotify/ParseSpotify.py similarity index 87% rename from src/spotify/ParseSpotify.py rename to src/backends/spotify/ParseSpotify.py index 49882cb..fa5d80c 100644 --- a/src/spotify/ParseSpotify.py +++ b/src/backends/spotify/ParseSpotify.py @@ -29,19 +29,14 @@ class Parser: self.parse_top_artists(json.loads(top_artists.read_text())) self.parse_top_tracks(json.loads(top_tracks.read_text())) - self.add_tracks_to_albums_and_artists() + self.library.update_cache() + self.library.name = "Spotify" + + self.score_tracks() + self.library.auto_score() return self.library - def add_tracks_to_albums_and_artists(self): - for track_id, track in self.track_map.items(): - for artist in track.artists: - if track not in artist.tracks: - artist.tracks.append(track) - - if track not in track.album.tracks: - track.album.tracks.append(track) - def parse_playlists(self, data): for name, tracks in data.items(): @@ -84,7 +79,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") @@ -123,6 +118,14 @@ 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 + + def score_tracks(self): + for idx, track in enumerate(self.library.top_tracks): + score = len(self.library.top_tracks) - idx + track.auto_score = score \ No newline at end of file diff --git a/src/spotify/SpotifyAuthenticator.py b/src/backends/spotify/SpotifyAuthenticator.py similarity index 100% rename from src/spotify/SpotifyAuthenticator.py rename to src/backends/spotify/SpotifyAuthenticator.py diff --git a/src/spotify/SpotifyWebAPI.py b/src/backends/spotify/SpotifyWebAPI.py similarity index 98% rename from src/spotify/SpotifyWebAPI.py rename to src/backends/spotify/SpotifyWebAPI.py index 5ddc12f..8f4807f 100644 --- a/src/spotify/SpotifyWebAPI.py +++ b/src/backends/spotify/SpotifyWebAPI.py @@ -2,7 +2,7 @@ import requests from PIL import Image from io import BytesIO -from src.spotify.SpotifyAuthenticator import authenticate +from src.backends.spotify.SpotifyAuthenticator import authenticate from src.helpers import * diff --git a/src/helpers.py b/src/helpers.py index 8f81bbf..3c7db3e 100644 --- a/src/helpers.py +++ b/src/helpers.py @@ -46,7 +46,7 @@ def save_data(data, name): file_path = os.path.join(data_dir, f"{name}.json") with open(file_path, 'w') as file: - json.dump(data, file, indent=4) + json.dump(data, file, indent=4, ensure_ascii=False) print(f"Data saved to {file_path}.") diff --git a/src/itunes/ParseItunesToJson.py b/src/itunes/ParseItunesToJson.py deleted file mode 100755 index 1505a36..0000000 --- a/src/itunes/ParseItunesToJson.py +++ /dev/null @@ -1,101 +0,0 @@ -import xmltodict -import json - - -def flatten_dict(d): - - out = [] - for song in d: - newSong = {} - - strCount = 0 - intCount = 0 - dateCount = 0 - - def getStr(id): - nonlocal strCount - if id not in song["key"]: - return "undef" - if len(song["string"]) <= strCount: - return "error" - strCount += 1 - return song["string"][strCount - 1] - - def getInt(id): - nonlocal intCount - if id not in song["key"]: - return -1 - if len(song["integer"]) <= intCount: - return -1 - intCount += 1 - return song["integer"][intCount - 1] - - def getDate(id): - nonlocal dateCount - if id not in song["key"]: - return "undef" - if len(song["date"]) <= dateCount: - return "error" - dateCount += 1 - return song["date"][dateCount - 1] - - newSong["Track ID"] = getInt("Track ID") - newSong["Name"] = getStr("Name") - newSong["Artist"] = getStr("Artist") - newSong["Album Artist"] = getStr("Album Artist") - newSong["Composer"] = getStr("Composer") - newSong["Album"] = getStr("Album") - newSong["Genre"] = getStr("Genre") - newSong["Kind"] = getStr("Kind") - newSong["Size"] = getInt("Size") - newSong["Total Time"] = getInt("Total Time") - newSong["Disc Number"] = getInt("Disc Number") - newSong["Disc Count"] = getInt("Disc Count") - newSong["Track Number"] = getInt("Track Number") - newSong["Track Count"] = getInt("Track Count") - newSong["Year"] = getInt("Year") - newSong["Date Modified"] = getDate("Date Modified") - newSong["Date Added"] = getDate("Date Added") - newSong["Bit Rate"] = getInt("Bit Rate") - newSong["Sample Rate"] = getInt("Sample Rate") - newSong["Play Count"] = getInt("Play Count") - newSong["Play Date"] = getInt("Play Date") - newSong["Play Date UTC"] = getDate("Play Date UTC") - newSong["Skip Count"] = getInt("Skip Count") - newSong["Skip Date"] = getDate("Skip Date") - newSong["Release Date"] = getDate("Release Date") - newSong["Album Rating"] = getInt("Album Rating") - newSong["Album Rating Computed"] = "Album Rating Computed" in song["key"] - newSong["Loved"] = "Loved" in song["key"] - newSong["Album Loved"] = "Album Loved" in song["key"] - newSong["Explicit"] = "Explicit" in song["key"] - newSong["Compilation"] = "Compilation" in song["key"] - newSong["Artwork Count"] = getInt("Artwork Count") - newSong["Sort Album"] = getStr("Sort Album") - newSong["Sort Artist"] = getStr("Sort Artist") - newSong["Sort Name"] = getStr("Sort Name") - newSong["Persistent ID"] = getStr("Persistent ID") - newSong["Track Type"] = getStr("Track Type") - - out.append(newSong) - - return out - - -def convert(filename, out_path): - with open(filename, 'r', encoding='utf-8') as xml_file: - data_dict = xmltodict.parse(xml_file.read()) - - # Extract the "Tracks" dictionary to be flattened - tracks_dict = data_dict['plist']['dict']['dict'] - - flat_tracks_dict = flatten_dict(tracks_dict["dict"]) - - json_data = json.dumps(flat_tracks_dict, indent=2) - - with open(out_path, 'w', encoding='utf-8') as json_file: - json_file.write(json_data) - - -if __name__ == "__main__": - convert('./prefetched/ItunesLibrary.xml', './prefetched/ItunesLibrary.json') diff --git a/src/local/LocalLibraryIndexer.py b/src/local/LocalLibraryIndexer.py deleted file mode 100644 index d3a7ce2..0000000 --- a/src/local/LocalLibraryIndexer.py +++ /dev/null @@ -1,78 +0,0 @@ -import os -import fnmatch -import mutagen -import eyed3 -from mutagen.flac import FLAC - -from src.helpers import * - - -music_extensions = ('*.mp3', '*.flac', '*.wav', '*.aac', '*.ogg', '*.m4a') -exclude_directories = ("*mary--*", "*example-word*") - - -def update_local_library(root_dir, output_dir): - - set_workdir(output_dir / "local") - - old_library = get_data("local_library") - known_paths = {track['path']: track_id for track_id, track in old_library.items()} - new_library = {} - - song_id = '0' - - def get_new_song_id(prev_id, song_path): - if song_path in known_paths: - return int(known_paths[song_path]) - while (prev_id in old_library) or (prev_id in new_library): - prev_id = str(int(prev_id) + 1) - return prev_id - - for dir_path, dir_names, filenames in os.walk(root_dir): - dir_names[:] = [d for d in dir_names if not any(fnmatch.fnmatch(d, exclude) for exclude in exclude_directories)] - filtered_filenames = [filename for filename in filenames if any(fnmatch.fnmatch(filename, ext) for ext in music_extensions)] - - for filename in filtered_filenames: - song_name, track_type = os.path.splitext(filename) - relative_path = os.path.relpath(os.path.join(dir_path, filename), root_dir) - song_id = get_new_song_id(song_id, relative_path) - new_library[song_id] = { - "name": song_name, - "path": relative_path, - "type": track_type, - "artist": "", - "album": "" - } - - for track_id, track in new_library.items(): - abs_path = os.path.join(root_dir, track['path']) - match track['type']: - case ".mp3": - mp3track = eyed3.load(abs_path) - if not mp3track: - print(f"Failed to load mp3 {abs_path} using just name of the file as track name") - track['name'] = track['path'] - else: - tag = mp3track.tag - if not tag: - print(f"Invalid mp3 header {abs_path} using just name of the file as track name") - track['name'] = track['path'] - else: - track['artist'] = tag.artist - track['name'] = tag.title - track['album'] = tag.album - - case ".flac": - try: - metadata = FLAC(abs_path).tags - 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} using just name of the file as track name") - track['name'] = track['path'] - - save_data(new_library, "local_library") diff --git a/src/lyrics/lyrics.py b/src/lyrics/lyrics.py new file mode 100644 index 0000000..ae6ed6d --- /dev/null +++ b/src/lyrics/lyrics.py @@ -0,0 +1,113 @@ +import requests +from mutagen import File +from mutagen.id3 import ID3, USLT, ID3NoHeaderError +import os + +def fetch_lyrics(artist: str, album: str, track: str, duration: int = 0) -> str: + + skip_artists = [ + "shurupis", + "unknown", + "Unknown" + ] + + skip_album = [ + "unknown", + "Unknown" + ] + + if artist in skip_artists or album in skip_album: + return "" + + + url = "https://lrclib.net/api/get" + params = { + "artist_name": artist, + "track_name": track, + "album_name": album, + "duration": duration, + } + try: + resp = requests.get(url, params=params, timeout=10) + if resp.status_code != 200: + try: + err = resp.json() + print(err.get("error"), "-", err.get("message")) + except ValueError: + print("HTTPError", resp.status_code) + return "" + data = resp.json() + if "error" in data: + print(data.get("error"), "-", data.get("message")) + return "" + return data.get("syncedLyrics", "") + except requests.RequestException as e: + print(type(e).__name__, "-", str(e)) + return "" + except ValueError as e: + print(type(e).__name__, "-", str(e)) + return "" + + +def write_lyrics(file_path, lyrics_text, lang="eng"): + """ + Writes lyrics text into tags recognized by Navidrome. + Provides detailed logs and success/failure info. + """ + + ext = os.path.splitext(file_path)[1].lower() + print(f"[INFO] Processing file: {file_path} (extension: {ext})") + + try: + # ---------- MP3 ---------- + if ext == ".mp3": + print("[INFO] Detected MP3 file") + try: + tags = ID3(file_path) + print("[INFO] Existing ID3 tags loaded") + except ID3NoHeaderError: + tags = ID3() + print("[INFO] No ID3 header found, creating new tags") + + tags.delall("USLT") + print("[INFO] Removed existing USLT frames (lyrics)") + + tags.add( + USLT( + encoding=3, # UTF-8 + lang=lang, + desc="", + text=lyrics_text + ) + ) + tags.save(file_path) + print("[SUCCESS] Lyrics written to MP3 tags") + return + + # ---------- FLAC / OGG / OPUS ---------- + audio = File(file_path) + if audio is None: + print(f"[ERROR] Unsupported or unrecognized file format: {file_path}") + return + + if ext in {".flac", ".ogg", ".opus"}: + print(f"[INFO] Detected {ext.upper()} file") + audio["LRC"] = lyrics_text + audio["SYNCEDLYRICS"] = lyrics_text + audio.save() + print(f"[SUCCESS] Lyrics written to {ext.upper()} tags") + return + + # ---------- Fallback ---------- + if audio.tags is not None: + print("[INFO] Using fallback tag writing") + audio["LYRICS"] = lyrics_text + audio.save() + print("[SUCCESS] Lyrics written using fallback tags") + return + + print(f"[WARNING] No suitable tag frame found for {file_path}") + + except Exception as e: + print(f"[ERROR] Failed to write lyrics to {file_path}: {e}") + diff --git a/src/mappers/DummyLibGenerator.py b/src/mappers/DummyLibGenerator.py new file mode 100644 index 0000000..7aaa544 --- /dev/null +++ b/src/mappers/DummyLibGenerator.py @@ -0,0 +1,46 @@ +from mutagen.easyid3 import EasyID3 +from pydub import AudioSegment +from mutagen.easyid3 import EasyID3 + +from src.Library import * + +def safe_name(s: str) -> str: + return ( + s.replace("/", "_") + .replace("\\", "_") + .replace(":", "_") + .replace("*", "_") + .replace("?", "_") + .replace('"', "_") + .replace("<", "_") + .replace(">", "_") + .replace("|", "_") + .strip() + ) + +def generate_dummy_library(library: Library, root_path: Path): + root_path.mkdir(parents=True, exist_ok=True) + + for artist in library.artists: + artist_path = root_path / safe_name(artist.title or f"artist_{artist.id}") + artist_path.mkdir(exist_ok=True) + + for album in artist.albums: + album_path = artist_path / safe_name(album.title or f"album_{album.id}") + album_path.mkdir(exist_ok=True) + + for track in album.tracks: + track_title = track.title or f"track_{track.id}" + track_file = album_path / f"{safe_name(track_title)}.mp3" + + # duration_ms = max(track.duration_ms or 100, 100) + duration_ms = 10 + silence = AudioSegment.silent(duration=duration_ms) + silence.export(track_file, format="mp3") + + audio = EasyID3(track_file) + audio["title"] = track.title or "" + audio["artist"] = ", ".join([a.title or "" for a in track.artists]) + audio["album"] = album.title or "" + audio["tracknumber"] = str(album.tracks.index(track) + 1) + audio.save() diff --git a/src/mappers/FuzzyMapper.py b/src/mappers/FuzzyMapper.py new file mode 100644 index 0000000..1dbbf75 --- /dev/null +++ b/src/mappers/FuzzyMapper.py @@ -0,0 +1,130 @@ + +from rapidfuzz import fuzz +from tqdm import tqdm + +from src.Library import * +from src.helpers import * +import re +import string + +import unicodedata + +user_def_artist_map = { + "donda" : "kanye west", +} + +user_def_album_map = { + "music" : "music sorry 4 da wait", +} + +def get_score(first: str, second: str) -> float: + return float(first == second) * 100 + +def clean_title(title: str) -> str: + title = unicodedata.normalize("NFKC", title).replace("‐", "-") + title = unicodedata.normalize("NFKD", title) + title = re.sub(r"\([^)]*\)", "", title) + title = "".join( + c for c in title + if not unicodedata.category(c).startswith("P") + ) + title = title.translate(str.maketrans("", "", string.punctuation)) + title = title.lower() + return " ".join(title.split()) + +def find_mapping(first, second, mapping): + for key, value in mapping.items(): + if (first == key and second == value) or (first == value and second == key): + return True + return False + +def get_album_score(first: str, second: str) -> float: + first = clean_title(first) + second = clean_title(second) + + if find_mapping(first, second, user_def_album_map): + return 100 + + return get_score(first, second) + +def get_track_score(first: str, second: str) -> float: + return get_score(clean_title(first), clean_title(second)) + +def get_artist_score(first: str, second: str) -> float: + first = clean_title(first) + second = clean_title(second) + + if find_mapping(first, second, user_def_artist_map): + return 100 + + return get_score(first, second) + + +def is_close_enough(first: float) -> bool: + return first > 90 + +def generate_map(local_lib : Library, remote_lib : Library, output_dir : Path): + + artist_map : dict[str, tuple[str, float]] = {} + album_map : dict[str, tuple[str, float]] = {} + track_map : dict[str, tuple[str, float]] = {} + + 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_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) + + + with tqdm(total=len(artist_map), desc='searching remote albums for each artist in local library') as process_bar: + for remote_artist_id, (local_artist_id, _) in artist_map.items(): + remote_artist = remote_lib.artist_map[remote_artist_id] + local_artist = local_lib.artist_map[local_artist_id] + + for remote_album in remote_artist.albums: + for local_album in local_artist.albums: + 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) + + with tqdm(total=len(album_map), desc='searching remote tracks for each album in local library') as process_bar: + for remote_album_id, (local_album_id, _) in album_map.items(): + remote_album = remote_lib.album_map[remote_album_id] + local_album = local_lib.album_map[local_album_id] + + for remote_track in remote_album.tracks: + for local_track in local_album.tracks: + 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 + remote_track.local_path = local_track.local_path + + process_bar.update(1) + + 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") diff --git a/src/mappers/RemoteLibraryResolver.py b/src/mappers/RemoteLibraryResolver.py new file mode 100644 index 0000000..7342848 --- /dev/null +++ b/src/mappers/RemoteLibraryResolver.py @@ -0,0 +1,9 @@ + +import src.mappers.FuzzyMapper as Fuzzy +import src.mappers.SpotifySearchMapping as Spotify + +from src.Library import * + +def resolve_remote_tracks(client_id, client_secret, local_lib : Library, spotify_lib : Library, output_dir): + Fuzzy.generate_map(local_lib, spotify_lib, output_dir) + # Spotify.generate_map(client_id, client_secret, local_lib, spotify_lib, output_dir) \ No newline at end of file diff --git a/src/mappers/SpotifySearchMapping.py b/src/mappers/SpotifySearchMapping.py new file mode 100644 index 0000000..e8878e9 --- /dev/null +++ b/src/mappers/SpotifySearchMapping.py @@ -0,0 +1,56 @@ + +from tqdm import tqdm + +from src.Library import * +from src.helpers import * +from src.backends.spotify.SpotifyWebAPI import find_song, update_access_token + +def generate_map(client_id, client_secret, local_lib : Library, spotify_lib : Library, output_dir): + + set_workdir(output_dir) + + update_access_token(client_id, client_secret) + + def find_local_songs_on_spotify(): + search_log = [] + + total = len(local_lib.tracks) + found_tracks_map = {} + with tqdm(total=total, desc='Searching local songs on spotify') as pbar: + for track in local_lib.tracks: + + parts = [track.title, track.artists[0].title if len(track.artists) else "", track.album.title] + search_pattern = " ".join(p for p in parts if p) + + if search_pattern == "": + print(f"cannot generate search pattern for the track - {track.local_path}") + pbar.update(1) + continue + + found_tracks = find_song(search_pattern)[0:10] + found_tracks_map[track.id] = found_tracks + search_log.append({"path" : track.local_path, "search_pattern": search_pattern, "result" : found_tracks_map[track.id]}) + pbar.update(1) + + save_data(search_log, "spotify_search_log") + save_data(found_tracks_map, "spotify_found_map") + + return found_tracks_map + + spotify_found_tracks = find_local_songs_on_spotify() + + spotify_user_track = {track.id for track in spotify_lib.tracks} + + spotify_pattern = {} + for local_id, found_items in spotify_found_tracks.items(): + spotify_pattern[local_id] = {"score": 1.0, "items": []} + + if not len(found_items): + continue + + for found_item in found_items: + spotify_track_id = found_item['id'] + if spotify_track_id in spotify_user_track: + spotify_pattern[local_id]['items'].append(spotify_track_id) + + save_data(spotify_pattern, "local_to_spotify_id_map") \ No newline at end of file