diff --git a/.gitignore b/.gitignore index 880f9c3..2803b17 100644 --- a/.gitignore +++ b/.gitignore @@ -2,6 +2,3 @@ prefetched* __pycache__ .idea secret -work_dir -old -tmp \ No newline at end of file diff --git a/Flow.py b/Flow.py index fd10266..bb35731 100644 --- a/Flow.py +++ b/Flow.py @@ -1,31 +1,21 @@ -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.backends.spotify.SpotifyWebAPI import fetch_all -from src.backends.spotify.ParseSpotify import Parser +from src.spotify.SpotifyWebAPI import fetch_all +from src.spotify.ParseSpotify import Parser -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.local.LocalLibraryIndexer import update_local_library +from src.local.ParseLocal import LocalParser +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 : Library = None - local_library : Library = None - itunes_library : Library = None + spotify_library = None + local_library = None @staticmethod def fetch_spotify(self): @@ -48,90 +38,18 @@ 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): - 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) - + map_local_to_spotify(client_id, client_secret, self.local_library, self.spotify_library, work_dir) def run(self, arg): self.fetch_local() diff --git a/Gui.py b/Gui.py deleted file mode 100644 index 12d1144..0000000 --- a/Gui.py +++ /dev/null @@ -1,208 +0,0 @@ -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 deleted file mode 100644 index 7c830a2..0000000 --- a/error.log +++ /dev/null @@ -1,2 +0,0 @@ - -(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 deleted file mode 100755 index eb55739..0000000 --- a/gui.sh +++ /dev/null @@ -1 +0,0 @@ -streamlit run Gui.py diff --git a/main.py b/main.py index 05a781d..98b4efa 100644 --- a/main.py +++ b/main.py @@ -1,41 +1,17 @@ + import Flow - -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(): +if __name__ == '__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.gen_fake_libs() - # flow.find_mbids() - flow.map_local_to_spotify() - flow.save_libraries() - # flow.log_stats() - - # print("asd") - - -# test1() -main() + print("asd") diff --git a/old/env b/old/env new file mode 100755 index 0000000..8955572 --- /dev/null +++ b/old/env @@ -0,0 +1,36 @@ + +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 0b2a991..10d378a 100644 --- a/old/main.py +++ b/old/main.py @@ -1,3 +1,79 @@ +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 = [] @@ -38,4 +114,121 @@ 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 \ No newline at end of file + 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 diff --git a/src/Library.py b/src/Library.py index 246d581..c3287b8 100644 --- a/src/Library.py +++ b/src/Library.py @@ -4,125 +4,48 @@ from datetime import datetime from pathlib import Path from typing import List, Optional, Any - -class Entity: +class Artist: def __init__(self): - 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 + self.id : str = None + self.title : str = None + self.tracks : List[Track] = [] -class Artist(Entity): +class Track: def __init__(self): - super().__init__() - self.tracks: List[Track] = [] - self.albums: List[Album] = [] + 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 -class Track(Entity): +class Album: def __init__(self): - super().__init__() - - self.artists: List[Artist] = [] - self.album: Album = None - - self.has_lyrics: bool = False + self.id : str = None + self.title : str = None + self.release_date : datetime = None + self.tracks : List[Track] = [] + self.artists : List[Artist] = [] -class Album(Entity): +class Playlist: def __init__(self): - 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 + self.title : str = None + 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.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.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 278776a..bc968ea 100644 --- a/src/LibrarySerializer.py +++ b/src/LibrarySerializer.py @@ -26,57 +26,48 @@ class LibrarySaver: def __init__(self, library: Library): self.library = library - def _entity_to_dict(self, entity: Entity) -> dict[str, Any]: + @staticmethod + def _artist_to_dict(artist: Artist) -> dict[str, Any]: return { - "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, - } - - def _artist_to_dict(self, artist: Artist) -> dict[str, Any]: - return self._entity_to_dict(artist) | { + "id": artist.id, + "title": artist.title, "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) | { + @staticmethod + def _track_to_dict(track: Track) -> dict[str, Any]: + return { + "id": track.id, + "title": track.title, "artists": [artist.id for artist in track.artists], "album": track.album.id, "duration_ms": track.duration_ms, - "has_lyrics": track.has_lyrics, + "added_at": _dt_to_str(track.added_at), + "local_path": track.local_path, } - def _playlist_to_dict(self, playlist: Playlist) -> dict[str, Any]: - return self._entity_to_dict(playlist) | { + @staticmethod + def _playlist_to_dict(playlist: Playlist) -> dict[str, Any]: + return { + "title": playlist.title, "tracks": [track.id for track in playlist.tracks], + "created_at": _dt_to_str(playlist.created_at), "description": playlist.description, } - def _album_to_dict(self ,album: Album) -> dict[str, Any]: - return self._entity_to_dict(album) | { + @staticmethod + def _album_to_dict(album: Album) -> dict[str, Any]: + return { + "id": album.id, + "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] } 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], @@ -91,82 +82,40 @@ 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.get_track(data.get("id")) - self.load_entity(track, data) - + track = self.track_map.get(data.get("id"), Track()) + track.id = data.get("id") + track.title = data.get("title") track.duration_ms = data.get("duration_ms") - 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.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()) self.track_map[track.id] = track return track def load_artist(self, data: dict[str, Any]): - 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 [] + 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")] self.artist_map[artist.id] = artist return artist def load_album(self, data: dict[str, Any]): - 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")] + 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")] self.album_map[album.id] = album return album @@ -177,7 +126,6 @@ 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")] @@ -186,6 +134,4 @@ 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 new file mode 100644 index 0000000..84e37c5 --- /dev/null +++ b/src/MapLocalToSpotify.py @@ -0,0 +1,140 @@ + +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 7c17ab5..c40bebb 100644 --- a/src/StatGenerator.py +++ b/src/StatGenerator.py @@ -1,210 +1,42 @@ -from src.Library import * -from src.helpers import * -from pathlib import Path -import json - - -def is_resolved(track: Track): - return track.local_path is not None - - -def get_mapping(mapping_path) -> dict[str, Any]: - data = {} - - with Path(mapping_path).open("r", encoding="utf-8") as f: - data = json.load(f) - - return data["track_map"] - - -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") +def print_link_stats(link_pattern): + resolved_reversed = {} + resolved = [] + unresolved_local = [] + unresolved_remote = [] + + 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) + + 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)] + + 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 + + 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] + + 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 diff --git a/src/backends/itunes/ParseItunesXml.py b/src/backends/itunes/ParseItunesXml.py deleted file mode 100755 index 4b0f417..0000000 --- a/src/backends/itunes/ParseItunesXml.py +++ /dev/null @@ -1,194 +0,0 @@ -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 deleted file mode 100644 index 08629f7..0000000 --- a/src/backends/local/LocalLibraryIndexer.py +++ /dev/null @@ -1,183 +0,0 @@ -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/backends/mb/mbidFinder.py b/src/backends/mb/mbidFinder.py deleted file mode 100644 index 6ffcd09..0000000 --- a/src/backends/mb/mbidFinder.py +++ /dev/null @@ -1,54 +0,0 @@ -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 deleted file mode 100644 index c7d04f4..0000000 --- a/src/backends/navidrome/navidrome.py +++ /dev/null @@ -1,110 +0,0 @@ -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/helpers.py b/src/helpers.py index 3c7db3e..8f81bbf 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, ensure_ascii=False) + json.dump(data, file, indent=4) print(f"Data saved to {file_path}.") diff --git a/src/itunes/ParseItunesToJson.py b/src/itunes/ParseItunesToJson.py new file mode 100755 index 0000000..1505a36 --- /dev/null +++ b/src/itunes/ParseItunesToJson.py @@ -0,0 +1,101 @@ +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 new file mode 100644 index 0000000..d3a7ce2 --- /dev/null +++ b/src/local/LocalLibraryIndexer.py @@ -0,0 +1,78 @@ +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/backends/local/LocalPLaylistGenerator.py b/src/local/LocalPLaylistGenerator.py similarity index 93% rename from src/backends/local/LocalPLaylistGenerator.py rename to src/local/LocalPLaylistGenerator.py index 1c0f1e1..0299498 100644 --- a/src/backends/local/LocalPLaylistGenerator.py +++ b/src/local/LocalPLaylistGenerator.py @@ -1,7 +1,7 @@ from pathlib import Path -def (): +def generate_playlists(local_library, library_path, remote_playlists, link_pattern, output_dir): output_dir = Path(output_dir) output_playlists = [] diff --git a/src/backends/local/ParseLocal.py b/src/local/ParseLocal.py similarity index 59% rename from src/backends/local/ParseLocal.py rename to src/local/ParseLocal.py index b1fbba3..602e89e 100644 --- a/src/backends/local/ParseLocal.py +++ b/src/local/ParseLocal.py @@ -10,6 +10,46 @@ 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] = {} @@ -20,7 +60,9 @@ class LocalParser: def parse_artists(self, data: str) -> List[Artist]: if not data: - raise "No artist found" + data = "unknown" + + data = split_artists(data) artists = [] @@ -45,7 +87,7 @@ class LocalParser: def parse_album(self, data: str) -> Album: if not data: - raise "album is none" + data = "unknown" name = data album_id = name @@ -62,6 +104,15 @@ 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()) @@ -71,26 +122,14 @@ class LocalParser: track.id = track_id track.title = data.get("name") - track.artists = self.parse_artists(data.get("artists")) + track.artists = self.parse_artists(data.get("artist")) 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) - 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" + self.add_tracks_to_albums_and_artists() return self.library diff --git a/src/lyrics/lyrics.py b/src/lyrics/lyrics.py deleted file mode 100644 index ae6ed6d..0000000 --- a/src/lyrics/lyrics.py +++ /dev/null @@ -1,113 +0,0 @@ -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 deleted file mode 100644 index 7aaa544..0000000 --- a/src/mappers/DummyLibGenerator.py +++ /dev/null @@ -1,46 +0,0 @@ -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 deleted file mode 100644 index 1dbbf75..0000000 --- a/src/mappers/FuzzyMapper.py +++ /dev/null @@ -1,130 +0,0 @@ - -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 deleted file mode 100644 index 7342848..0000000 --- a/src/mappers/RemoteLibraryResolver.py +++ /dev/null @@ -1,9 +0,0 @@ - -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 deleted file mode 100644 index e8878e9..0000000 --- a/src/mappers/SpotifySearchMapping.py +++ /dev/null @@ -1,56 +0,0 @@ - -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 diff --git a/src/backends/spotify/ParseSpotify.py b/src/spotify/ParseSpotify.py similarity index 87% rename from src/backends/spotify/ParseSpotify.py rename to src/spotify/ParseSpotify.py index fa5d80c..49882cb 100644 --- a/src/backends/spotify/ParseSpotify.py +++ b/src/spotify/ParseSpotify.py @@ -29,14 +29,19 @@ class Parser: self.parse_top_artists(json.loads(top_artists.read_text())) self.parse_top_tracks(json.loads(top_tracks.read_text())) - self.library.update_cache() - self.library.name = "Spotify" - - self.score_tracks() - self.library.auto_score() + self.add_tracks_to_albums_and_artists() 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(): @@ -79,7 +84,7 @@ class Parser: track.id = track_id track.title = data.get("name") - track.artists = [] + self.parse_album(data.get("album")).artists + track.artists = [self.parse_artist(artist) for artist in data.get("artists")] track.album = self.parse_album(data.get("album")) track.duration_ms = data.get("duration_ms") @@ -118,14 +123,6 @@ 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/backends/spotify/SpotifyAuthenticator.py b/src/spotify/SpotifyAuthenticator.py similarity index 100% rename from src/backends/spotify/SpotifyAuthenticator.py rename to src/spotify/SpotifyAuthenticator.py diff --git a/src/backends/spotify/SpotifyWebAPI.py b/src/spotify/SpotifyWebAPI.py similarity index 98% rename from src/backends/spotify/SpotifyWebAPI.py rename to src/spotify/SpotifyWebAPI.py index 8f4807f..5ddc12f 100644 --- a/src/backends/spotify/SpotifyWebAPI.py +++ b/src/spotify/SpotifyWebAPI.py @@ -2,7 +2,7 @@ import requests from PIL import Image from io import BytesIO -from src.backends.spotify.SpotifyAuthenticator import authenticate +from src.spotify.SpotifyAuthenticator import authenticate from src.helpers import *