refactor initial not fully recovered

Env.py

@@ -0,0 +1,7 @@
+from pathlib import Path
+
+local_library_path = Path("/home/auser/Music/")
+work_dir = Path("./work_dir/new")
+
+client_id = '***REDACTED***'
+client_secret = "***REDACTED***"

main.py

@@ -1,234 +1,77 @@
-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/"
+from Env import client_id, client_secret, local_library_path
-rel_autogen_dir = "0autogen"
+from Env import work_dir
 
-playlists = get_data('playlistTracks')
+from src.LibrarySerializer import LibrarySaver, LibraryLoader
-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}
 
+from src.spotify.SpotifyWebAPI import fetch_all
+from src.spotify.ParseSpotify import Parser
 
-def get_playlist_names(pl):
+from src.local.LocalLibraryIndexer import update_local_library
-    return [name for name, items in pl.items()]
+from src.local.ParseLocal import LocalParser
 
-
+from src.MapLocalToSpotify import map_local_to_spotify
-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 = []
-    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
 
 
 class Interpreter(cmd.Cmd):
     intro = "Welcome to the interactive command loop. Type help or ? to list commands.\n"
-    prompt = "(spotify) "
+    prompt = "(kuw) "
 
-    def do_list_stat_vs_types(self, arg):
+    spotify_library = None
-        """prints available linking patterns"""
+    local_library = None
-        print(f"0 - fuzzy tags")
-        print(f"1 - spotify")
-        print(f"2 - fuzzy")
 
-    def do_stat_local_library_vs_spotify(self, arg):
+    @staticmethod
-        """compares local library and remote with the link pattern generated by the generate_pattern_command"""
+    def do_fetch_spotify(self):
-        print_link_stats(link_pattern)
+        fetch_all(client_id, client_secret, work_dir)
-
-    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):
+    @staticmethod
-        """find local tracks and fetches the data from them for further analysis"""
+    def do_fetch_local():
-        global local_library
+        update_local_library(local_library_path, work_dir)
-        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
+    def do_parse_spotify_library(self):
-        run_pattern_generators()
+        parser = Parser()
-        link_pattern = get_data('link_pattern_spotify')
+        library_parsed = parser.parse(work_dir / "spotify" / "playlistTracks.json", work_dir / "spotify" / "tracks.json")
+        LibrarySaver(library_parsed).save(work_dir / "spotify_library.json")
 
-    def do_fetch(self, arg):
+    def do_parse_local_library(self):
-        """Fetch all spotify data"""
+        parser = LocalParser()
-        try:
+        library_parsed = parser.parse(work_dir / "local" / "local_library.json")
-            update_access_token()
+        LibrarySaver(library_parsed).save(work_dir / "local_library.json")
-            fetch_data()
 
-            global playlists, top_artists, top_tracks, user_tracks
+    def do_load_libraries(self):
+        self.spotify_library = LibraryLoader().load(work_dir / "spotify_library.json")
+        self.local_library = LibraryLoader().load(work_dir / "local_library.json")
 
-            playlists = get_data('playlistTracks')
+    def do_map_local_to_spotify(self):
-            top_artists = get_data('top_artists')
+        map_local_to_spotify(client_id, client_secret, self.local_library, self.spotify_library, work_dir)
-            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):
+    def do_run(self, arg):
-        """generates todos for the pirates"""
+        # self.do_fetch_local()
-        print("sorry not implemented yet.")
+        # self.do_fetch_spotify()
 
-    def do_run_all_default(self, arg):
+        # self.do_parse_local_library()
-        """runs general flow"""
+        # self.do_parse_spotify_library()
 
-        local_lib_path = "/home/auser/Music/"
+        self.do_load_libraries()
-        print("Fetching spotify data...")
-        self.do_fetch([])
 
-        print("Fetching local data...")
+        self.do_map_local_to_spotify()
-        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
 
+    @staticmethod
     def do_exit(self, arg):
-        """Exit the command loop: exit"""
-        print("Goodbye!")
         return True
 
 
 if __name__ == '__main__':
     try:
-        Interpreter().cmdloop()
+        Interpreter().do_run("")
+        # Interpreter().cmdloop()
+
     except KeyboardInterrupt as kb:
         print("process terminated")

old/main.py

@@ -0,0 +1,234 @@
+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 = []
+    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
+
+
+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")

src/Library.py

@@ -0,0 +1,159 @@
+from __future__ import annotations
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import List, Optional, Any
+
+
+# =========================
+# Core models
+# =========================
+
+class Track:
+    def __init__(
+        self,
+        *,
+        id: str,
+        title: str,
+        artists: List[str],
+        album: Optional[str] = None,
+        duration_ms: Optional[int] = None,
+        added_at: Optional[datetime] = None,
+        spotify: Optional["SpotifyData"] = None,
+        local: Optional["LocalData"] = None,
+    ):
+        self.id = id
+        self.title = title
+        self.artists = artists
+        self.album = album
+        self.duration_ms = duration_ms
+        self.added_at = added_at
+        self.spotify = spotify
+        self.local = local
+
+
+class Playlist:
+    def __init__(
+        self,
+        *,
+        name: str,
+        tracks: Optional[List[Track]] = None,
+        created_at: Optional[datetime] = None,
+        description: Optional[str] = None,
+        spotify: Optional["SpotifyData"] = None,
+    ):
+        self.name = name
+        self.tracks = tracks or []
+        self.created_at = created_at
+        self.description = description
+        self.spotify = spotify
+
+
+class Album:
+    def __init__(
+        self,
+        *,
+        title: str,
+        release_date: Optional[str] = None,
+        tracks: Optional[List[Track]] = None,
+        spotify: Optional["SpotifyData"] = None,
+    ):
+        self.title = title
+        self.release_date = release_date
+        self.tracks = tracks or []
+        self.spotify = spotify
+
+
+class TrackList:
+    def __init__(self, *, tracks: Optional[List[Track]] = None):
+        self.tracks = tracks or []
+
+    def total_duration_ms(self) -> int:
+        return sum(t.duration_ms or 0 for t in self.tracks)
+
+
+# =========================
+# Spotify models
+# =========================
+
+class SpotifyArtist:
+    def __init__(self, *, id: str, name: str, uri: str, url: str):
+        self.id = id
+        self.name = name
+        self.uri = uri
+        self.url = url
+
+
+class SpotifyAlbum:
+    def __init__(
+        self,
+        *,
+        id: str,
+        name: str,
+        release_date: str,
+        total_tracks: int,
+        uri: str,
+        url: str,
+    ):
+        self.id = id
+        self.name = name
+        self.release_date = release_date
+        self.total_tracks = total_tracks
+        self.uri = uri
+        self.url = url
+
+
+class SpotifyData:
+    def __init__(
+        self,
+        *,
+        id: str,
+        uri: str,
+        url: str,
+        href: str,
+        file_path: Optional[Path] = None,
+        popularity: Optional[int] = None,
+        is_playable: Optional[bool] = None,
+        is_local: Optional[bool] = None,
+        artists: Optional[List[SpotifyArtist]] = None,
+        album: Optional[SpotifyAlbum] = None,
+        external_ids: Optional[dict[str, str]] = None,
+    ):
+        self.id = id
+        self.uri = uri
+        self.url = url
+        self.href = href
+        self.file_path = file_path
+        self.popularity = popularity
+        self.is_playable = is_playable
+        self.is_local = is_local
+        self.artists = artists or []
+        self.album = album
+        self.external_ids = external_ids or {}
+
+class LocalData:
+    def __init__(
+        self,
+        *,
+        path: str,
+    ):
+        self.path = path
+
+# =========================
+# Library container
+# =========================
+
+class Library:
+    def __init__(
+        self,
+        *,
+        playlists: Optional[List[Playlist]] = None,
+        tracks: Optional[List[Track]] = None,
+        tracklists: Optional[List[TrackList]] = None,
+        albums: Optional[List[Album]] = None,
+    ):
+        self.playlists = playlists or []
+        self.tracks = tracks or []
+        self.tracklists = tracklists or []
+        self.albums = albums or []
+

src/LibrarySerializer.py

@@ -0,0 +1,199 @@
+
+from src.Library import *
+
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import List, Optional, Any
+
+def _dt_to_str(dt: Optional[datetime]) -> Optional[str]:
+    return dt.isoformat() if dt else None
+
+
+def _str_to_dt(s: Optional[str]) -> Optional[datetime]:
+    return datetime.fromisoformat(s) if s else None
+
+
+def _path_to_str(p: Optional[Path]) -> Optional[str]:
+    return str(p) if p else None
+
+
+def _str_to_path(s: Optional[str]) -> Optional[Path]:
+    return Path(s) if s else None
+
+
+class LibrarySaver:
+    def __init__(self, library: Library):
+        self.library = library
+
+    def save(self, path: Path) -> None:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with path.open("w", encoding="utf-8") as f:
+            json.dump(self.to_dict(), f, indent=2, ensure_ascii=False)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "tracks": [self._track_to_dict(t) for t in self.library.tracks],
+            "playlists": [self._playlist_to_dict(p) for p in self.library.playlists],
+            "tracklists": [self._tracklist_to_dict(tl) for tl in self.library.tracklists],
+            "albums": [self._album_to_dict(a) for a in self.library.albums],
+        }
+
+    def _track_to_dict(self, t: Track) -> dict[str, Any]:
+        return {
+            "id": t.id,
+            "title": t.title,
+            "artists": t.artists,
+            "album": t.album,
+            "duration_ms": t.duration_ms,
+            "added_at": _dt_to_str(t.added_at),
+            "spotify": self._spotify_to_dict(t.spotify),
+            "local": self._local_to_dict(t.local),
+        }
+
+    def _playlist_to_dict(self, p: Playlist) -> dict[str, Any]:
+        return {
+            "name": p.name,
+            "tracks": [t.id for t in p.tracks],
+            "created_at": _dt_to_str(p.created_at),
+            "description": p.description,
+            "spotify": self._spotify_to_dict(p.spotify),
+        }
+
+    def _tracklist_to_dict(self, tl: TrackList) -> dict[str, Any]:
+        return {"tracks": [t.id for t in tl.tracks]}
+
+    def _album_to_dict(self, a: Album) -> dict[str, Any]:
+        return {
+            "title": a.title,
+            "release_date": a.release_date,
+            "tracks": [t.id for t in a.tracks],
+            "spotify": self._spotify_to_dict(a.spotify),
+        }
+
+    def _local_to_dict(self, local):
+        if not local:
+            return None
+        return {"path": local.path}
+
+    def _spotify_to_dict(self, s: Optional[SpotifyData]) -> Optional[dict[str, Any]]:
+        if not s:
+            return None
+        return {
+            "id": s.id,
+            "uri": s.uri,
+            "url": s.url,
+            "href": s.href,
+            "file_path": _path_to_str(s.file_path),
+            "popularity": s.popularity,
+            "is_playable": s.is_playable,
+            "is_local": s.is_local,
+            "external_ids": s.external_ids,
+            "artists": [{"id": a.id, "name": a.name, "uri": a.uri, "url": a.url} for a in s.artists],
+            "album": {
+                "id": s.album.id,
+                "name": s.album.name,
+                "release_date": s.album.release_date,
+                "total_tracks": s.album.total_tracks,
+                "uri": s.album.uri,
+                "url": s.album.url,
+            } if s.album else None,
+        }
+
+
+class LibraryLoader:
+    def __init__(self):
+        # Map of track id -> Track object
+        self.track_map: dict[str, Track] = {}
+
+    def load(self, path: Path) -> Library:
+        with path.open("r", encoding="utf-8") as f:
+            data = json.load(f)
+
+        # Load tracks first
+        tracks = []
+        for t in data.get("tracks", []):
+            track = self._track_from_dict(t)
+            tracks.append(track)
+            self.track_map[track.id] = track
+
+        # Load playlists using track_map
+        playlists = []
+        for p in data.get("playlists", []):
+            pl_tracks = []
+            for tid in p.get("tracks", []):
+                if tid not in self.track_map:
+                    raise KeyError(f"Track ID '{tid}' not found in track map")
+                pl_tracks.append(self.track_map[tid])
+
+            playlists.append(Playlist(
+                name=p["name"],
+                tracks=pl_tracks,
+                created_at=_str_to_dt(p.get("created_at")),
+                description=p.get("description"),
+                spotify=self._spotify_from_dict(p.get("spotify")),
+            ))
+
+
+        # Load tracklists
+        tracklists = []
+        for tl in data.get("tracklists", []):
+            tl_tracks = []
+            for tid in tl.get("tracks", []):
+                if tid not in self.track_map:
+                    raise KeyError(f"Track ID '{tid}' not found in track map (tracklist)")
+                tl_tracks.append(self.track_map[tid])
+            tracklists.append(TrackList(tracks=tl_tracks))
+
+        # Load albums
+        albums = []
+        for a in data.get("albums", []):
+            alb_tracks = []
+            for tid in a.get("tracks", []):
+                if tid not in self.track_map:
+                    raise KeyError(f"Track ID '{tid}' not found in track map (album '{a.get('title')}')")
+                alb_tracks.append(self.track_map[tid])
+            albums.append(Album(
+                title=a["title"],
+                release_date=a.get("release_date"),
+                tracks=alb_tracks,
+                spotify=self._spotify_from_dict(a.get("spotify")),
+            ))
+
+        return Library(playlists=playlists, tracks=tracks, tracklists=tracklists, albums=albums)
+
+    def _track_from_dict(self, d: dict[str, Any]) -> Track:
+        return Track(
+            id=d["id"],
+            title=d["title"],
+            artists=d["artists"],
+            album=d.get("album"),
+            duration_ms=d.get("duration_ms"),
+            added_at=_str_to_dt(d.get("added_at")),
+            spotify=self._spotify_from_dict(d.get("spotify")),
+            local=self._local_from_dict(d.get("local")),
+        )
+
+    def _local_from_dict(self, d):
+        if not d:
+            return None
+        return LocalData(
+            path=d["path"],
+        )
+
+    def _spotify_from_dict(self, d: Optional[dict[str, Any]]) -> Optional[SpotifyData]:
+        if not d:
+            return None
+        return SpotifyData(
+            id=d["id"],
+            uri=d["uri"],
+            url=d["url"],
+            href=d["href"],
+            file_path=_str_to_path(d.get("file_path")),
+            popularity=d.get("popularity"),
+            is_playable=d.get("is_playable"),
+            is_local=d.get("is_local"),
+            external_ids=d.get("external_ids"),
+            artists=[SpotifyArtist(**a) for a in d.get("artists", [])],
+            album=SpotifyAlbum(**d["album"]) if d.get("album") else None,
+        )

src/MapLocalToSpotify.py

@@ -1,47 +1,52 @@
-from DataBase import *
-from SpotifyWebAPI import find_song, update_access_token
 
 from rapidfuzz import fuzz
 from tqdm import tqdm
 import requests
 
-def spotify_pattern_generator():
+from src.Library import *
-    update_access_token()
+from src.helpers import *
-
+from src.spotify.SpotifyWebAPI import find_song, update_access_token
-    local_library = get_data("local_library")
-    spotify_library = get_data("tracks")
 
 
-    def find_local_songs_on_spotify(local_lib):
+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_library)
+        total = len(local_lib.tracks)
         found_tracks_map = {}
         with tqdm(total=total, desc='Searching local songs on spotify') as pbar:
-            for track_id, track in local_lib.items():
+            for track in local_lib.tracks:
-                parts = [track["name"], track["artist"], track.get("album")]
+
+                parts = [track.title, track.artists[0] if len(track.artists) else "", track.album]
                 search_pattern = " ".join(p for p in parts if p)
 
                 if search_pattern == "":
-                    print(f"cannot generate search pattern for the track - {track["path"]}")
+                    print(f"cannot generate search pattern for the track - {track.local.path}")
                     pbar.update(1)
                     continue
 
-                found_tracks = find_song(search_pattern)[0:1]
+                found_tracks = find_song(search_pattern)[0:10]
-                found_tracks_map[track_id] = found_tracks
+                found_tracks_map[track.id] = found_tracks
-                search_log.append({"path" : track["path"], "search_pattern": search_pattern, "result" : found_tracks_map[track_id]})
+                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(local_library)
+    spotify_found_tracks = find_local_songs_on_spotify()
-    spotify_user_track = {track['track']['id']: {"local_mappings": []} for track in spotify_library}
+
+    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": 0.0, "items": []}
+        spotify_pattern[local_id] = {"score": 1.0, "items": []}
 
         if not len(found_items):
             continue
@@ -50,14 +55,8 @@ def spotify_pattern_generator():
             spotify_track_id = found_item['id']
             if spotify_track_id in spotify_user_track:
                 spotify_pattern[local_id]['items'].append(spotify_track_id)
-                spotify_user_track[spotify_track_id]['local_mappings'].append(local_id)
 
-    for spotify_id, track_local_mappings in spotify_user_track.items():
+    save_data(spotify_pattern, "local_to_spotify_id_map")
-        score = len(track_local_mappings['local_mappings'])
-        for local_id in track_local_mappings['local_mappings']:
-            spotify_pattern[local_id]['score'] = 1 / score
-
-    save_data(spotify_pattern, "link_pattern_spotify")
 
 
 def fuzzy_pattern_generator():
@@ -126,7 +125,7 @@ def fuzzy_tag_pattern_generator():
                 ratios.append((fuzzy_tags_ratio(tag1, tag2), track['id']))
 
             filtered_ratios = []
-            threshold = 0.8
+            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]))
@@ -139,11 +138,3 @@ def fuzzy_tag_pattern_generator():
             pbar.update(1)
 
     save_data(mappings, "link_pattern_tags")
-
-
-def run_pattern_generators(type_id = 0):
-    spotify_pattern_generator()
-
-
-if __name__ == "__main__":
-    run_pattern_generators()

src/StatGenerator.py

@@ -0,0 +1,42 @@
+
+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

src/helpers.py

@@ -3,6 +3,13 @@ import os
 
 data_dir = 'prefetched'
 
+def set_workdir(dir):
+    global data_dir
+    data_dir = str(dir)
+
+def get_workdir():
+    global data_dir
+    return data_dir
 
 class SongTags:
     def __init__(self, title='', artist='', album=''):
@@ -40,10 +47,9 @@ 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)
-    return f"Data saved to {name}.json."
 
 
-def get_artists_str(artists_pack):
+    print(f"Data saved to {file_path}.")
-    track_artists = [artist['name'] for artist in artists_pack]
+
-    artists = " ".join(track_artists)
+
-    return artists
+

src/local/LocalLibraryIndexer.py

@@ -1,18 +1,20 @@
 import os
 import fnmatch
-
 import mutagen
-
-from DataBase import *
 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):
+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 = {}
@@ -74,7 +76,3 @@ def update_local_library(root_dir):
                     track['name'] = track['path']
 
     save_data(new_library, "local_library")
-
-
-if __name__ == "__main__":
-    update_local_library("/home/auser/Music/")

src/local/ParseLocal.py

@@ -0,0 +1,40 @@
+from datetime import datetime
+from typing import Any, Iterable
+from pathlib import Path
+import json
+
+from src.Library import *
+
+class LocalParser:
+    def __init__(self):
+        pass
+
+    def parse_track_item(self, id, track: dict[str, Any]) -> Track:
+        track = Track(
+            title=track["name"],
+            artists=[track.get("artist", [])],
+            album=track["album"] if track.get("album") else None,
+            duration_ms=None,
+            added_at=None,
+            spotify=None,
+            id=id,
+            local=LocalData(path=track["path"])
+        )
+        return track
+
+
+    def parse_all_tracks(self, items: dict[str, list[dict[str, Any]]]) -> list[Track]:
+        return [self.parse_track_item(id, track) for id, track in items.items()]
+
+    def build_library_from_tracks(self, tracks_content_json: dict[str, list[dict[str, Any]]]) -> Library:
+        library = Library()
+
+        tracks = self.parse_all_tracks(tracks_content_json)
+        library.tracklists.append(TrackList(tracks=list(tracks)))
+        library.tracks.extend(tracks)
+
+        return library
+
+    def parse(self, tracks_file: Path) -> Library:
+        tracks_content = json.loads(tracks_file.read_text())
+        return self.build_library_from_tracks(tracks_content)

src/spotify/ParseSpotify.py

@@ -0,0 +1,139 @@
+from datetime import datetime
+from typing import Any, Iterable
+from pathlib import Path
+import json
+
+from src.Library import *
+
+
+def get_artists_str(artists_pack):
+    track_artists = [artist['name'] for artist in artists_pack]
+    artists = " ".join(track_artists)
+    return artists
+
+class Parser:
+    def __init__(self):
+        # state: Spotify track ID -> Track object
+        self.track_map: dict[str, Track] = {}
+
+    # -------------------------
+    # Low-level helpers
+    # -------------------------
+    @staticmethod
+    def _parse_spotify_datetime(value: str | None) -> datetime | None:
+        if not value:
+            return None
+        return datetime.fromisoformat(value.replace("Z", "+00:00"))
+
+    @staticmethod
+    def _parse_spotify_artist(a: dict[str, Any]) -> SpotifyArtist:
+        return SpotifyArtist(
+            id=a["id"],
+            name=a["name"],
+            uri=a["uri"],
+            url=a["external_urls"]["spotify"],
+        )
+
+    @staticmethod
+    def _parse_spotify_album(a: dict[str, Any]) -> SpotifyAlbum:
+        return SpotifyAlbum(
+            id=a["id"],
+            name=a["name"],
+            release_date=a.get("release_date"),
+            total_tracks=a.get("total_tracks", 0),
+            uri=a["uri"],
+            url=a["external_urls"]["spotify"],
+        )
+
+    def _parse_spotify_data(self, track_json: dict[str, Any]) -> SpotifyData:
+        return SpotifyData(
+            id=track_json["id"],
+            uri=track_json["uri"],
+            url=track_json["external_urls"]["spotify"],
+            href=track_json["href"],
+            popularity=track_json.get("popularity"),
+            is_playable=track_json.get("is_playable"),
+            is_local=track_json.get("is_local"),
+            external_ids=track_json.get("external_ids", {}),
+            artists=[self._parse_spotify_artist(a) for a in track_json.get("artists", [])],
+            album=self._parse_spotify_album(track_json["album"])
+            if track_json.get("album")
+            else None,
+        )
+
+    # -------------------------
+    # Track parsers
+    # -------------------------
+    def parse_track_item(self, item: dict[str, Any]) -> Track:
+        """
+        Parses a single track item and returns Track object.
+        Reuses existing Track from track_map if already parsed.
+        """
+        t = item["track"]
+        track_id = t["id"]
+
+        # return existing track if already parsed
+        if track_id in self.track_map:
+            return self.track_map[track_id]
+
+        track = Track(
+            title=t["name"],
+            artists=[a["name"] for a in t.get("artists", [])],
+            album=t["album"]["name"] if t.get("album") else None,
+            duration_ms=t.get("duration_ms"),
+            added_at=self._parse_spotify_datetime(item.get("added_at")),
+            spotify=self._parse_spotify_data(t),
+            id=self._parse_spotify_data(t).id
+        )
+
+        self.track_map[track_id] = track
+        return track
+
+    def parse_track_items(self, items: Iterable[dict[str, Any]]) -> list[Track]:
+        return [self.parse_track_item(i) for i in items if i.get("track")]
+
+    # -------------------------
+    # Playlist parsers
+    # -------------------------
+    def parse_playlist_content(self, playlist_name: str, items: list[dict[str, Any]]) -> Playlist:
+        tracks = self.parse_track_items(items)
+        return Playlist(name=playlist_name, tracks=tracks)
+
+    def parse_all_playlist_content(self, data: dict[str, list[dict[str, Any]]]) -> list[Playlist]:
+        playlists: list[Playlist] = []
+        for name, items in data.items():
+            playlists.append(self.parse_playlist_content(name, items))
+        return playlists
+
+    def parse_all_tracks(self, items: dict[str, list[dict[str, Any]]]) -> list[Track]:
+        return [self.parse_track_item(i) for i in items]
+
+    # -------------------------
+    # Library builder
+    # -------------------------
+    def build_library_from_playlist_content(
+        self, playlist_content_json: dict[str, list[dict[str, Any]]],
+            tracks_content_json: dict[str, list[dict[str, Any]]]
+    ) -> Library:
+        library = Library()
+
+        self.parse_all_tracks(tracks_content_json)
+
+        library.tracklists.append(TrackList(tracks=list((self.track_map.values()))))
+
+        library.playlists = self.parse_all_playlist_content(playlist_content_json)
+        library.tracks.extend(self.track_map.values())
+
+        return library
+
+    # -------------------------
+    # Convenience method
+    # -------------------------
+    def parse(
+        self,
+        playlist_content_file: Path,
+        tracks_file: Path,
+    ) -> Library:
+        playlist_content = json.loads(playlist_content_file.read_text())
+        tracks_content = json.loads(tracks_file.read_text())
+        return self.build_library_from_playlist_content(playlist_content, tracks_content)

src/spotify/SpotifyAuthenticator.py

@@ -8,22 +8,11 @@ import threading
 import time
 import os.path
 
-client_id = '***REDACTED***'
+client_id = None
 client_secret = None
 token = None
 
 
-def resolve_client_secret():
-    global client_secret
-    if os.path.exists('secret'):
-        with open('secret', 'r') as file:
-            client_secret = file.readline().strip()
-    else:
-        client_secret = input("Enter the application secret: ")
-        with open('secret', 'w') as file:
-            file.write(client_secret)
-
-
 def retrieve_web_api_token(authorization_code):
     global token
 
@@ -103,8 +92,13 @@ def open_authenticate_page():
     webbrowser.open(auth_url)
 
 
-def authenticate():
+def authenticate(id, secret):
-    resolve_client_secret()
+
+    global client_secret
+    global client_id
+
+    client_secret=secret
+    client_id=id
 
     redirect_server = threading.Thread(target=run_redirect_server)
     redirect_server.start()

src/spotify/SpotifyWebAPI.py

@@ -1,8 +1,10 @@
 import requests
 from PIL import Image
 from io import BytesIO
-from SpotifyAuthenticator import authenticate
+
-from DataBase import *
+from src.spotify.SpotifyAuthenticator import authenticate
+
+from src.helpers import *
 
 FETCH_STEP = 50
 token = ''
@@ -142,7 +144,7 @@ def save_image_from_url(url, name, image_dir="playlist_covers"):
 
     image = Image.open(BytesIO(response.content))
 
-    directory = os.path.join(data_dir, image_dir)
+    directory = os.path.join(get_workdir(), image_dir)
     os.makedirs(directory, exist_ok=True)
     file_path = os.path.join(directory, f"{name}.jpg")
     print(f"Image saved {file_path}")
@@ -169,9 +171,9 @@ def load_artworks():
     load_playlist_covers()
 
 
-def update_access_token():
+def update_access_token(id, secret):
     global token
-    token = authenticate()
+    token = authenticate(id, secret)
 
 
 def fetch_data():
@@ -184,7 +186,11 @@ def fetch_data():
 
     load_artworks()
 
+def fetch_all(id, secret, workdir):
+    set_workdir(workdir / "spotify")
+    update_access_token(id, secret)
+    fetch_data()
 
 if __name__ == "__main__":
-    update_access_token()
+    update_access_token("", "")
     fetch_data()