stable stat

Flow.py

@@ -10,7 +10,7 @@ from src.spotify.ParseSpotify import Parser
 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
 
 
@@ -50,10 +50,10 @@ class Flow:
         self.local_library = LibraryLoader().load(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 / "local_to_spotify_id_map.json", work_dir / "stats.json", self.local_library, self.spotify_library)
+        log_stats(work_dir / "remote_to_local_map.json", work_dir / "stats.json", self.local_library, self.spotify_library)
 
 
     def run(self, arg):

main.py

@@ -7,12 +7,12 @@ if __name__ == '__main__':
     # flow.fetch_spotify()
     # flow.fetch_local()
 
-    # flow.parse_spotify_library()
-    # flow.parse_local_library()
+    flow.parse_spotify_library()
+    flow.parse_local_library()
 
     flow.load_libraries()
 
-    # flow.map_local_to_spotify()
+    flow.map_local_to_spotify()
 
     flow.log_stats()
 

src/Library.py

@@ -4,11 +4,13 @@ from datetime import datetime
 from pathlib import Path
 from typing import List, Optional, Any
 
+
 class Artist:
     def __init__(self):
         self.id : str = None
         self.title : str = None
         self.tracks : List[Track] = []
+        self.albums : List[Album] = []
 
 
 class Track:
@@ -49,3 +51,30 @@ class Library:
         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
+
+
+    def update_cache(self):
+        self.update_id_maps()
+        self.create_reverse_links()
+
+    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)

src/LibrarySerializer.py

@@ -32,6 +32,7 @@ class LibrarySaver:
             "id": artist.id,
             "title": artist.title,
             "tracks": [track.id for track in artist.tracks],
+            "albums": [album.id for album in artist.albums],
         }
 
     @staticmethod
@@ -127,6 +128,7 @@ class LibraryLoader:
         artist.id = data.get("id")
         artist.title = data.get("title")
         artist.tracks = [self.get_track(track_id) for track_id in data.get("tracks")]
+        artist.albums = [self.get_album(album_id) for album_id in data.get("albums")] if data.get("albums") else []
 
         self.artist_map[artist.id] = artist
         return artist
@@ -155,4 +157,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

src/MapLocalToSpotify.py

@@ -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")

src/StatGenerator.py

@@ -14,13 +14,7 @@ def get_mapping(mapping_path) -> dict[str, Any]:
     with Path(mapping_path).open("r", encoding="utf-8") as f:
         data = json.load(f)
 
-    out = {}
-
-    for local, remotes in data.items():
-        if len(remotes["items"]):
-            out[local] = remotes["items"][0]
-
-    return out
+    return data
 
 
 class AlbumStat:
@@ -99,7 +93,7 @@ class StatGenerator:
         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 local, remote in self.mapping.items():
+        for remote, local in self.mapping.items():
             self.remote_id_map[remote].local_path = self.local_id_map[local].local_path
 
     def score_artists(self):

src/local/ParseLocal.py

@@ -104,15 +104,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())
@@ -129,7 +120,12 @@ class LocalParser:
             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()
 
         return self.library
 

src/mappers/FuzzyMapper.py

@@ -0,0 +1,68 @@
+
+from rapidfuzz import fuzz
+from tqdm import tqdm
+
+from src.Library import *
+from src.helpers import *
+
+
+def get_score(first: str, second: str) -> float:
+    return fuzz.token_set_ratio(first, second)
+
+def is_close_enough(first: float) -> bool:
+    return first > 50
+
+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_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)
+
+            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_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)
+
+            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_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)
+
+            process_bar.update(1)
+
+    out = { 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")

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)

src/mappers/SpotifySearchMapping.py

@@ -0,0 +1,56 @@
+
+from tqdm import tqdm
+
+from src.Library import *
+from src.helpers import *
+from src.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")

src/spotify/ParseSpotify.py

@@ -29,19 +29,10 @@ 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()
 
         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():