tmp

.env.example

@@ -1,9 +0,0 @@
-# Copy to `.env` and fill in, then `source .env` before running.
-# Get these from https://developer.spotify.com/dashboard (your app's settings).
-# Register `http://127.0.0.1:8888` as a Redirect URI in that app.
-export SPOTIFY_CLIENT_ID=your_client_id_here
-export SPOTIFY_CLIENT_SECRET=your_client_secret_here
-
-# Optional overrides:
-# export MUSIC_LIBRARY_PATH=~/Music
-# export WORK_DIR=./work_dir/new

.gitignore

@@ -2,8 +2,3 @@ prefetched*
 __pycache__
 .idea
 secret
-.env
-.venv
-work_dir
-old
-tmp

Env.py

@@ -1,13 +0,0 @@
-import os
-from pathlib import Path
-
-# Local config. Paths can be overridden via environment variables.
-local_library_path = Path(os.environ.get("MUSIC_LIBRARY_PATH", "~/Music")).expanduser()
-work_dir = Path(os.environ.get("WORK_DIR", "./work_dir/new"))
-
-# Spotify app credentials. NEVER hardcode these here — set them in the
-# environment (see .env.example) so they don't end up in version control.
-#   export SPOTIFY_CLIENT_ID=...
-#   export SPOTIFY_CLIENT_SECRET=...
-client_id = os.environ.get("SPOTIFY_CLIENT_ID")
-client_secret = os.environ.get("SPOTIFY_CLIENT_SECRET")

Flow.py

@@ -1,136 +0,0 @@
-
-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.backends.local.LocalLibraryIndexer import update_local_library
-from src.backends.local.ParseLocal import LocalParser
-from src.backends.itunes.ParseItunesXml import ParseItunesXml
-
-
-from src.mappers.RemoteLibraryResolver import resolve_remote_tracks
-from src.StatGenerator import log_stats
-
-
-class Flow:
-    spotify_library : Library = None
-    local_library : Library = None
-    itunes_library : Library = None
-
-    @staticmethod
-    def fetch_spotify():
-        fetch_all(client_id, client_secret, work_dir)
-
-    @staticmethod
-    def fetch_local():
-        update_local_library(local_library_path, work_dir)
-
-    @staticmethod
-    def parse_spotify_library():
-        parser = Parser()
-
-        library_parsed = parser.parse(
-            work_dir / "spotify" / "playlistTracks.json",
-            work_dir / "spotify" / "tracks.json",
-            work_dir / "spotify" / "top_tracks.json",
-            work_dir / "spotify" / "top_artists.json",
-        )
-
-        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 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 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)
-
-
-    def run(self, arg):
-        self.fetch_local()
-        # self.do_fetch_spotify()
-
-        self.parse_local_library()
-        self.parse_spotify_library()
-
-        self.load_libraries()
-
-        # self.map_local_to_spotify()
-
-        return True

Gui.py

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

README.md

@@ -1,100 +1,3 @@
-# MusicIndexer
+# spotifyFetcher
-
+loads all data through web api<br>
-Pull your Spotify library and listening stats via the Spotify Web API, index a
+client secret - ***REDACTED***
-local music library (audio tags + iTunes XML), and match the two so you can see
-which tracks you own, what's missing, and your top artists/tracks/playlists — in
-a Streamlit GUI.
-
-## Features
-
-- **Spotify backend** — OAuth (Authorization Code) login, then fetch the user
-  profile, liked tracks, all playlists and their tracks, long-term top tracks
-  and top artists, and playlist/profile cover art. Resilient paginated fetching
-  (follows `next`, retries dropped connections, honours `429` rate limits) with
-  progress bars.
-- **Local backend** — index a folder of audio files via `mutagen` tags.
-- **iTunes backend** — parse an iTunes Library XML export.
-- **Matching** — map local tracks to Spotify tracks (fuzzy matcher; ISRC/MBID
-  matching planned).
-- **Stats & lyrics** — generate listening stats and fetch lyrics.
-- **GUI** — browse libraries as artist → album → track tables.
-
-## Setup
-
-Requires Python 3.13+.
-
-```bash
-python -m venv .venv
-source .venv/bin/activate
-pip install -r requirements.txt
-```
-
-### Credentials
-
-Create an app at https://developer.spotify.com/dashboard and add
-`http://127.0.0.1:8888` as a **Redirect URI**. Credentials are read from the
-environment — never commit them.
-
-```bash
-cp .env.example .env
-# edit .env with your client id/secret
-source .env
-```
-
-`Env.py` reads `SPOTIFY_CLIENT_ID` / `SPOTIFY_CLIENT_SECRET` (plus optional
-`MUSIC_LIBRARY_PATH` and `WORK_DIR`) from the environment.
-
-## Usage
-
-Edit `main.py` to uncomment the steps you want, then run:
-
-```bash
-python main.py
-```
-
-The flow steps (see `Flow.py`) include:
-
-| Step | What it does |
-|------|--------------|
-| `fetch_spotify()` | OAuth login + download all Spotify data into `work_dir/spotify/` |
-| `fetch_local()` | index local audio files into `work_dir/local/` |
-| `parse_spotify_library()` / `parse_itunes_library()` / `parse_local_library()` | build normalized libraries |
-| `load_libraries()` / `save_libraries()` | (de)serialize libraries to `work_dir` |
-| `map_local_to_spotify()` | match local tracks against the Spotify library |
-| `fetch_and_save_lyrics()` | fetch lyrics for tracks |
-| `log_stats()` | generate stats |
-
-On first `fetch_spotify()` a browser opens for the Spotify login; a local server
-on `127.0.0.1:8888` catches the redirect and exchanges the code for a token.
-
-### GUI
-
-```bash
-./gui.sh        # runs: .venv/bin/streamlit run Gui.py
-```
-
-## Layout
-
-```
-main.py                 entry point (CLI flow)
-Gui.py / gui.sh         Streamlit GUI
-Flow.py                 orchestrates fetch / parse / map / stats
-Env.py                  config (reads credentials from the environment)
-src/
-  Library.py            data model (Library / Artist / Album / Track)
-  LibrarySerializer.py  load/save libraries
-  backends/
-    spotify/            Web API client, OAuth, parser
-    local/              local-file indexer, parser, playlist generator
-    itunes/             iTunes XML parser
-    navidrome/          Navidrome backend
-  mappers/              local↔Spotify matching (fuzzy / search)
-  lyrics/               lyrics fetching
-work_dir/               fetched data and serialized libraries (gitignored)
-```
-
-## Notes
-
-- Data is cached as JSON under `work_dir/` (gitignored); re-running reuses it.
-- The deprecated Spotify endpoints (audio-features, recommendations, etc.) are
-  not used — only currently-supported endpoints.

TODO

@@ -1,7 +0,0 @@
-skip playlists that are not created by user
-
-create one big playlist with the order of the spotify and dont skip missing
-
-generate percentage like hierarchy of the missing items
-
-generate detailed log of the missing items and search results

common.py

@@ -0,0 +1,25 @@
+import json
+import os
+
+data_dir = 'prefetched'
+
+
+def get_data(name):
+    file_path = os.path.join(data_dir, f"{name}.json")
+    try:
+        with open(file_path, 'r') as file:
+            data = json.load(file)
+        return data
+    except FileNotFoundError:
+        return f"File {name}.json not found."
+    except json.JSONDecodeError:
+        return f"Error decoding JSON from {name}.json."
+
+
+def save_data(data, name):
+    os.makedirs(data_dir, exist_ok=True)
+
+    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."

error.log

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

gui.sh

@@ -1,5 +0,0 @@
-#!/usr/bin/env bash
-# Run from the script's directory using the project virtualenv,
-# so it works regardless of whether the venv is activated / on PATH.
-cd "$(dirname "$0")"
-exec .venv/bin/streamlit run Gui.py "$@"

loadArtwork.py

@@ -0,0 +1,45 @@
+import requests
+import os
+from PIL import Image
+from io import BytesIO
+
+from common import *
+
+
+def save_image_from_url(url, name, image_dir="playlist_covers"):
+    response = requests.get(url)
+
+    if response.status_code != 200:
+        raise "Cannot fetch the playlist cover"
+
+    image = Image.open(BytesIO(response.content))
+
+    directory = os.path.join(data_dir, image_dir)
+    os.makedirs(directory, exist_ok=True)
+    file_path = os.path.join(directory, f"{name}.jpg")
+    print(f"Image saved {file_path}")
+    image.save(file_path)
+
+
+def load_playlist_covers():
+    playlists = get_data('playlists')
+
+    for pl in playlists:
+        if len(pl['images']):
+            url = pl['images'][0]['url']
+            save_image_from_url(url, pl['name'])
+
+
+def load_user_cover():
+    user_data = get_data("user_data")
+    url = user_data['images'][1]['url']
+    save_image_from_url(url, "user", ".")
+
+
+def load():
+    load_user_cover()
+    load_playlist_covers()
+
+
+if __name__ == "__main__":
+    load()

main.py

@@ -1,38 +1,106 @@
-import Flow
+from common import *
+from spotifyFetch import fetch_data
+import cmd
+
+playlists = get_data('playlistTracks')
+top_artists = get_data('top_artists')
+top_tracks = get_data('top_tracks')
+user_tracks = get_data('tracks')
 
 
-def test1():
+def get_playlist_names():
-    flow = Flow.Flow()
+    return [name for name, items in playlists.items()]
-
-    # 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():
+def get_artists_str(artists_pack):
-    flow = Flow.Flow()
+    track_artists = [artist['name'] for artist in artists_pack]
-
+    artists = " ".join(track_artists)
-    flow.fetch_spotify()
+    return artists
-    # flow.fetch_local()
-
-    flow.parse_spotify_library()
-    #flow.parse_itunes_library()
-    #flow.parse_local_library()
-
-    #flow.load_libraries()
-
-    #flow.map_local_to_spotify()
-    #flow.save_libraries()
-
-    # flow.log_stats()
-
-    # print("asd")
 
 
-# test1()
+def print_playlist_tracks(name):
-main()
+    print(name)
+    for track in playlists[name]:
+        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():
+    for trackItem in user_tracks:
+        track = trackItem['track']
+        print(f"    '{track['name']}' - {get_artists_str(track['artists'])}")
+
+
+class Interpreter(cmd.Cmd):
+    intro = "Welcome to the interactive command loop. Type help or ? to list commands.\n"
+    prompt = "(spotify) "
+
+    def do_playlists(self, arg):
+        """prints user playlists"""
+        print("\n".join(get_playlist_names()))
+
+    def do_playlist_tracks(self, arg):
+        """prints playlist [name]"""
+        try:
+            name = arg.split()[0]
+            print_playlist_tracks(name)
+        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_fetch(self, arg):
+        """Fetch all spotify data"""
+        try:
+            fetch_data()
+        except ValueError:
+            print("Can not fetch the data.")
+
+    def do_exit(self, arg):
+        """Exit the command loop: exit"""
+        print("Goodbye!")
+        return True
+
+
+if __name__ == '__main__':
+    Interpreter().cmdloop()

old/main.py

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

prefetched_1/user_data.json

@@ -0,0 +1,26 @@
+{
+    "display_name": "ilusha",
+    "external_urls": {
+        "spotify": "https://open.spotify.com/user/31uszk7zb6zdzrzst4q3rvkfrfz4"
+    },
+    "href": "https://api.spotify.com/v1/users/31uszk7zb6zdzrzst4q3rvkfrfz4",
+    "id": "31uszk7zb6zdzrzst4q3rvkfrfz4",
+    "images": [
+        {
+            "url": "https://i.scdn.co/image/ab67757000003b82b4a21a7f0c35258b3fa4b79d",
+            "height": 64,
+            "width": 64
+        },
+        {
+            "url": "https://i.scdn.co/image/ab6775700000ee85b4a21a7f0c35258b3fa4b79d",
+            "height": 300,
+            "width": 300
+        }
+    ],
+    "type": "user",
+    "uri": "spotify:user:31uszk7zb6zdzrzst4q3rvkfrfz4",
+    "followers": {
+        "href": null,
+        "total": 7
+    }
+}

prefetched_1/user_profile.json

@@ -0,0 +1,33 @@
+{
+    "display_name": "ilusha",
+    "external_urls": {
+        "spotify": "https://open.spotify.com/user/31uszk7zb6zdzrzst4q3rvkfrfz4"
+    },
+    "href": "https://api.spotify.com/v1/users/31uszk7zb6zdzrzst4q3rvkfrfz4",
+    "id": "31uszk7zb6zdzrzst4q3rvkfrfz4",
+    "images": [
+        {
+            "url": "https://i.scdn.co/image/ab67757000003b82b4a21a7f0c35258b3fa4b79d",
+            "height": 64,
+            "width": 64
+        },
+        {
+            "url": "https://i.scdn.co/image/ab6775700000ee85b4a21a7f0c35258b3fa4b79d",
+            "height": 300,
+            "width": 300
+        }
+    ],
+    "type": "user",
+    "uri": "spotify:user:31uszk7zb6zdzrzst4q3rvkfrfz4",
+    "followers": {
+        "href": null,
+        "total": 7
+    },
+    "country": "US",
+    "product": "free",
+    "explicit_content": {
+        "filter_enabled": false,
+        "filter_locked": false
+    },
+    "email": "ilusha.basic@gmail.com"
+}

requirements.txt

@@ -1,9 +0,0 @@
-streamlit==1.58.0
-streamlit-aggrid==1.2.1.post2
-pandas==3.0.3
-Pillow==12.2.0
-mutagen==1.47.0
-rapidfuzz==3.14.5
-requests==2.34.2
-tqdm==4.68.2
-xmltodict==1.0.4

spotifyAuth.py

@@ -8,10 +8,20 @@ import threading
 import time
 import os.path
 
-client_id = None
+client_id = '***REDACTED***'
 client_secret = None
 token = None
-refresh_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):
@@ -24,7 +34,7 @@ def retrieve_web_api_token(authorization_code):
         "client_secret": client_secret,
         "client_id":     client_id,
         "code": authorization_code,
-        'redirect_uri': 'http://127.0.0.1:8888',
+        'redirect_uri': 'http://localhost:8888',
     }
 
     print("\nAuthenticating WEB API\n")
@@ -45,10 +55,7 @@ def retrieve_web_api_token(authorization_code):
         print(token_response.json())
         raise ValueError
 
-    global refresh_token
+    token = token_response.json()['access_token']
-    response_json = token_response.json()
-    token = response_json['access_token']
-    refresh_token = response_json.get('refresh_token')
 
 
 class RequestHandler(BaseHTTPRequestHandler):
@@ -77,7 +84,7 @@ class RequestHandler(BaseHTTPRequestHandler):
 
 
 def run_redirect_server():
-    httpd = HTTPServer(('127.0.0.1', 8888), RequestHandler)
+    httpd = HTTPServer(('localhost', 8888), RequestHandler)
     print('Starting HTTP redirection server')
     httpd.serve_forever()
 
@@ -88,22 +95,16 @@ def open_authenticate_page():
     params = {
         'client_id': client_id,
         'response_type': 'code',
-        'redirect_uri': 'http://127.0.0.1:8888',
+        'redirect_uri': 'http://localhost:8888',
         "scope": "user-top-read playlist-read-collaborative user-read-email user-library-read user-read-private playlist-read-private",
     }
 
-    auth_url = f"{authorize_url}?{urllib.parse.urlencode(params)}"
+    auth_url = f"{authorize_url}?client_id={params['client_id']}&response_type={params['response_type']}&redirect_uri={params['redirect_uri']}&scope={params['scope']}"
     webbrowser.open(auth_url)
-    print(f"If the browser did not open, visit:\n{auth_url}")
 
 
-def authenticate(id, secret):
+def authenticate():
-
+    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()

spotifyFetch.py

@@ -0,0 +1,147 @@
+import requests
+from loadArtwork import load
+from spotifyAuth import authenticate
+from common import *
+
+FETCH_STEP = 50
+token = ''
+
+
+def fetch(endpoint, method='GET', body=None):
+    url = f'https://api.spotify.com/{endpoint}'
+    headers = {
+        'Authorization': f'Bearer {token}',
+    }
+
+    response = requests.request(method, url, headers=headers, json=body)
+    response.raise_for_status()
+    return response.json()
+
+
+def fetch_playlists(user_id):
+    print("Fetching playlists")
+    endpoint = f"v1/me/playlists"
+    pl_total = fetch(f'{endpoint}?limit=1')['total']
+    pl_count = 0
+    playlists = []
+
+    while pl_count < pl_total:
+        res = fetch(f'{endpoint}?limit={FETCH_STEP}&offset={pl_count}')
+        playlists += res['items']
+        pl_count += FETCH_STEP
+
+    return playlists
+
+
+def fetch_playlist_items(playlist_id):
+    endpoint = f"v1/playlists/{playlist_id}/tracks"
+    items = fetch(f'{endpoint}?fields=total')['total']
+    loaded = 0
+    tracks = []
+
+    while loaded < items:
+        res = fetch(f'{endpoint}?limit={FETCH_STEP}&offset={loaded}')
+        
+        tracks += res['items']
+        loaded += FETCH_STEP
+
+    return tracks
+
+
+def fetch_tracks(user_id):
+    print("Fetching tracks")
+    endpoint = f"v1/me/tracks"
+    total = fetch(f"{endpoint}?limit=1&market=ES")['total']
+    current = 0
+    tracks = []
+
+    while total > current:
+        res = fetch(f"{endpoint}?limit={FETCH_STEP}&offset={current}&market=ES")
+        tracks += res['items']
+        current += FETCH_STEP
+
+    save_data(tracks, "tracks")
+    return tracks
+
+
+def fetch_tracks_top():
+    print("Fetching top tracks")
+    total = fetch(f'v1/me/top/tracks?fields=total')['total']
+    current = 0
+    tracks = []
+
+    while current < total:
+        res = fetch(f'v1/me/top/tracks?limit={FETCH_STEP}&offset={current}&time_range=long_term')
+        tracks += res['items']
+        current += FETCH_STEP
+
+    save_data(tracks, "top_tracks")
+    return tracks
+
+
+def fetch_artists_top():
+    print("Fetching top artists")
+
+    total = fetch(f'v1/me/top/artists?fields=total')['total']
+    current = 0
+    artists = []
+
+    while current < total:
+        res = fetch(f'v1/me/top/artists?limit={FETCH_STEP}&offset={current}&time_range=long_term')
+        artists += res['items']
+        current += FETCH_STEP
+
+    save_data(artists, "top_artists")
+    return artists
+
+
+def fetch_all_playlist_data(user_id):
+    playlists = fetch_playlists(user_id)
+
+    save_data(playlists, 'playlists')
+
+    print(len(playlists))
+
+    playlist_items = {}
+
+    print("Fetching playlists items")
+
+    index = 0
+    for pl in playlists:
+        pl_id = pl['id']
+        name = pl['name']
+        playlist_items[name] = fetch_playlist_items(pl_id)
+        print(f"{index} - {len(playlist_items[name])}")
+        index += 1
+
+    save_data(playlist_items, "playlistTracks")
+
+
+def get_user_data():
+    print("Fetching user data")
+
+    user_profile = fetch(f'v1/me')
+    user_id = user_profile['id']
+    user_data = fetch(f'v1/users/{user_id}')
+
+    save_data(user_profile, "user_profile")
+    save_data(user_data, "user_data")
+    return user_id
+
+
+def fetch_data():
+    global token
+    token = authenticate()
+
+    user_id = get_user_data()
+
+    fetch_tracks(user_id)
+    fetch_all_playlist_data(user_id)
+    fetch_tracks_top()
+    fetch_artists_top()
+
+    load()
+
+
+if __name__ == "__main__":
+    fetch_data()

src/Library.py

@@ -1,128 +0,0 @@
-from __future__ import annotations
-
-from datetime import datetime
-from pathlib import Path
-from typing import List, Optional, Any
-
-
-class Entity:
-    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
-
-
-class Artist(Entity):
-    def __init__(self):
-        super().__init__()
-        self.tracks: List[Track] = []
-        self.albums: List[Album] = []
-
-
-class Track(Entity):
-    def __init__(self):
-        super().__init__()
-
-        self.artists: List[Artist] = []
-        self.album: Album = None
-
-        self.has_lyrics: bool = False
-
-
-class Album(Entity):
-    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
-
-
-class Library:
-    def __init__(self):
-        self.artists: List[Artist] = []
-        self.playlists: List[Playlist] = []
-        self.tracks: List[Track] = []
-        self.albums: List[Album] = []
-
-        self.liked_tracks: List[Track] = []
-        self.top_tracks: List[Track] = []
-        self.top_artists: List[Artist] = []
-
-        self.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

src/LibrarySerializer.py

@@ -1,191 +0,0 @@
-
-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 _entity_to_dict(self, entity: Entity) -> 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) | {
-            "tracks": [track.id for track in artist.tracks],
-            "albums": [album.id for album in artist.albums],
-        }
-
-    def _track_to_dict(self, track: Track) -> dict[str, Any]:
-         return self._entity_to_dict(track) | {
-            "artists": [artist.id for artist in track.artists],
-            "album": track.album.id,
-            "duration_ms": track.duration_ms,
-            "has_lyrics": track.has_lyrics,
-        }
-
-    def _playlist_to_dict(self, playlist: Playlist) -> dict[str, Any]:
-        return self._entity_to_dict(playlist) | {
-            "tracks": [track.id for track in playlist.tracks],
-            "description": playlist.description,
-        }
-
-    def _album_to_dict(self ,album: Album) -> dict[str, Any]:
-        return self._entity_to_dict(album) | {
-            "tracks": [track.id for track in album.tracks],
-            "artists": [artist.id for artist in album.artists],
-        }
-
-    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],
-            "playlists": [self._playlist_to_dict(playlist) for playlist in self.library.playlists],
-            "top_tracks": [track.id for track in self.library.top_tracks],
-            "top_artists": [artist.id for artist in self.library.top_artists],
-            "liked_tracks": [track.id for track in self.library.liked_tracks],
-        }
-
-        path.parent.mkdir(parents=True, exist_ok=True)
-        with path.open("w", encoding="utf-8") as f:
-            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.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")
-
-        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 []
-
-        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")]
-
-        self.album_map[album.id] = album
-        return album
-
-    def load(self, path: Path) -> Library:
-        with path.open("r", encoding="utf-8") as f:
-            data = json.load(f)
-
-        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")]
-
-        library.top_artists = [self.artist_map[artist] for artist in data.get("top_artists")]
-        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/StatGenerator.py

@@ -1,210 +0,0 @@
-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")

src/backends/itunes/ParseItunesXml.py

@@ -1,192 +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
-
-        return self.library
-
-

src/backends/local/LocalLibraryIndexer.py

@@ -1,171 +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 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['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")

src/backends/local/LocalPLaylistGenerator.py

@@ -1,50 +0,0 @@
-
-from pathlib import Path
-
-def ():
-    output_dir = Path(output_dir)
-    output_playlists = []
-
-    remote_to_local_map = {}
-
-    for local_id, links in link_pattern.items():
-        if len(links['items']):
-            remote_to_local_map[links['items'][0]] = local_id
-
-    for name, tracks in remote_playlists.items():
-        local_tracks = []
-
-        for track in tracks:
-            track = track["track"]
-            path = "error"
-            error = ""
-            if track is None:
-                error = f"error: track is none"
-            elif track["id"] in remote_to_local_map:
-                path = local_library[remote_to_local_map[track["id"]]]["path"]
-            else:
-                error = f"error: not_found - {track["name"]}"
-
-            if error != "":
-                local_tracks.append("# " + error)
-            else:
-                path = (output_dir / ".." ).resolve() / Path(path)
-                local_tracks.append(path)
-
-        pl = {
-            "name": name,
-            "tracks": local_tracks,
-        }
-
-        output_playlists.append(pl)
-
-
-    for pl in output_playlists:
-        output_dir.mkdir(parents=True, exist_ok=True)
-
-        playlist_path = Path(output_dir) / f"{pl['name']}.m3u"
-
-        with (playlist_path.open("w", encoding="utf-8") as f):
-            for track in pl["tracks"]:
-                f.write(str(track))
-                f.write("\n")

src/backends/local/ParseLocal.py

@@ -1,100 +0,0 @@
-import json
-import re
-
-from src.Library import *
-
-import re
-
-PROTECTED_ARTISTS = {
-    "Tyler, The Creator",
-    "The Good, The Bad & The Queen",
-}
-
-class LocalParser:
-    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: str) -> 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: Path) -> Library:
-        tracks_content = json.loads(tracks_file.read_text())
-
-        for track_id, data in tracks_content.items():
-            track = Track()
-
-            track.id = track_id
-            track.title = data.get("name")
-            track.artists = self.parse_artists(data.get("artists"))
-            track.album = self.parse_album(data.get("album"))
-            track.local_path = data.get("path")
-            track.duration_ms = data.get("duration")
-            track.has_lyrics = data.get("has_lyrics")
-
-            if not track.album or not len(track.artists) or not track.title:
-                raise "error parsing"
-
-            self.track_map[track.id] = track
-            self.library.tracks.append(track)
-
-        for track_id, track in self.track_map.items():
-            for artist in track.artists:
-                if artist not in track.album.artists:
-                    track.album.artists.append(artist)
-
-        self.library.update_cache()
-
-        self.library.name = "Local"
-
-        return self.library
-
-
-    def parse_track_item(self, id, track: dict[str, Any]) -> Track:
-
-        return track

src/backends/navidrome/navidrome.py

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

src/backends/spotify/ParseSpotify.py

@@ -1,131 +0,0 @@
-from src.Library import *
-
-from typing import Any, Iterable
-import json
-
-from datetime import datetime
-from pathlib import Path
-
-
-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):
-        self.track_map: dict[str, Track] = {}
-        self.artist_map: dict[str, Artist] = {}
-        self.album_map: dict[str, Album] = {}
-        self.library = Library()
-
-    def parse(self, playlists: Path, tracks: Path, top_tracks: Path, top_artists: Path) -> Library:
-        self.library = Library()
-
-        self.parse_tracks(json.loads(tracks.read_text()))
-        self.parse_playlists(json.loads(playlists.read_text()))
-
-        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.library.auto_score()
-
-        self.score_tracks()
-
-        return self.library
-
-
-    def parse_playlists(self, data):
-        for name,  tracks in data.items():
-            playlist = Playlist()
-            playlist.title = name
-            for track in tracks:
-                track_data = track.get("track")
-                if track_data:
-                    track_id = track_data.get("id")
-                    playlist.tracks.append(self.parse_track(track_data, track.get("added_at")))
-
-            self.library.playlists.append(playlist)
-
-    def parse_top_tracks(self, data):
-        for track in data:
-            if track.get("id") in self.track_map:
-                self.library.top_tracks.append(self.track_map.get(track.get("id")))
-
-    def parse_top_artists(self, data):
-        for artist in data:
-            if artist.get("id") in self.artist_map:
-                self.library.top_artists.append(self.artist_map.get(artist.get("id")))
-
-    def parse_tracks(self, data):
-        for track in data:
-            self.parse_track(track.get("track"), track.get("added_at"))
-
-        for track in data:
-            self.library.liked_tracks.append(self.track_map.get(track.get("track").get("id")))
-
-    def parse_track(self, data: dict[str, Any], added_time : datetime) -> Track:
-        track_id = data.get("id")
-
-        if track_id in self.track_map:
-            return self.track_map[track_id]
-
-        track = Track()
-
-        self.track_map[track_id] = track
-
-        track.id = track_id
-        track.title = data.get("name")
-        track.artists = [] + self.parse_album(data.get("album")).artists
-        track.album = self.parse_album(data.get("album"))
-        track.duration_ms = data.get("duration_ms")
-
-        self.library.tracks.append(track)
-
-        return track
-
-    def parse_album(self, data):
-        album_id = data.get("id")
-
-        if album_id in self.album_map:
-            return self.album_map[album_id]
-
-        album = Album()
-
-        self.album_map[album_id] = album
-
-        album.id = data.get("id")
-        album.title = data.get("name")
-        album.artists = [self.parse_artist(artist) for artist in data.get("artists")]
-
-        self.library.albums.append(album)
-
-        return album
-
-    def parse_artist(self, data: dict[str, Any]) -> Artist:
-        artist_id = data.get("id")
-
-        if artist_id in self.artist_map:
-            return self.artist_map[artist_id]
-
-        artist = Artist()
-
-        self.artist_map[artist_id] = artist
-
-        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

src/backends/spotify/SpotifyWebAPI.py

@@ -1,198 +0,0 @@
-import requests
-import urllib.parse
-from requests.adapters import HTTPAdapter
-from urllib3.util.retry import Retry
-from PIL import Image
-from io import BytesIO
-from tqdm import tqdm
-
-from src.backends.spotify.SpotifyAuthenticator import authenticate
-
-from src.helpers import *
-
-FETCH_STEP = 50
-token = ''
-REQUEST_TIMEOUT = 30
-
-
-def _build_session():
-    """Session that retries transient failures (dropped connections, 5xx) and
-    honours Spotify's 429 Retry-After, so a single reset mid-fetch doesn't
-    abort the whole run."""
-    session = requests.Session()
-    retry = Retry(
-        total=6,
-        connect=6,
-        read=6,
-        backoff_factor=1.5,  # sleeps ~0, 1.5, 3, 6, 12, 24s between attempts
-        status_forcelist=(429, 500, 502, 503, 504),
-        allowed_methods=frozenset({"GET", "POST"}),
-        respect_retry_after_header=True,
-        raise_on_status=False,
-    )
-    adapter = HTTPAdapter(max_retries=retry)
-    session.mount("https://", adapter)
-    session.mount("http://", adapter)
-    return session
-
-
-_session = _build_session()
-
-
-def fetch(endpoint, method='GET', body=None):
-    url = endpoint if endpoint.startswith('http') else f'https://api.spotify.com/{endpoint}'
-    headers = {
-        'Authorization': f'Bearer {token}',
-    }
-
-    response = _session.request(method, url, headers=headers, json=body, timeout=REQUEST_TIMEOUT)
-    response.raise_for_status()
-    return response.json()
-
-
-def fetch_paginated(endpoint, desc="Fetching", leave=True):
-    """Collect every item of a paging object by following its `next` field,
-    which is Spotify's recommended way to page (more robust than tracking
-    offset/total manually). Shows a tqdm progress bar driven by the paging
-    object's `total`."""
-    items = []
-    page = fetch(endpoint)
-    with tqdm(total=page.get('total'), desc=desc, unit="item", leave=leave) as pbar:
-        while True:
-            batch = page.get('items', [])
-            items += batch
-            pbar.update(len(batch))
-            next_url = page.get('next')
-            if not next_url:
-                break
-            page = fetch(next_url)
-    return items
-
-
-def find_song(song_pattern):
-    query = urllib.parse.quote(song_pattern)
-    return fetch(f"v1/search?type=track&limit=10&q={query}")['tracks']['items']
-
-
-def fetch_playlists(user_id):
-    return fetch_paginated(f"v1/me/playlists?limit={FETCH_STEP}", desc="Fetching playlists")
-
-
-def fetch_playlist_items(playlist_id):
-    return fetch_paginated(
-        f"v1/playlists/{playlist_id}/tracks?limit={FETCH_STEP}",
-        desc="Fetching playlist tracks",
-        leave=False,
-    )
-
-
-def fetch_tracks(user_id):
-    tracks = fetch_paginated(f"v1/me/tracks?limit={FETCH_STEP}&market=ES", desc="Fetching liked tracks")
-    save_data(tracks, "tracks")
-    return tracks
-
-
-def fetch_tracks_top():
-    tracks = fetch_paginated(f"v1/me/top/tracks?limit={FETCH_STEP}&time_range=long_term", desc="Fetching top tracks")
-    save_data(tracks, "top_tracks")
-    return tracks
-
-
-def fetch_artists_top():
-    artists = fetch_paginated(f"v1/me/top/artists?limit={FETCH_STEP}&time_range=long_term", desc="Fetching top artists")
-    save_data(artists, "top_artists")
-    return artists
-
-
-def fetch_all_playlist_data(user_id):
-    playlists = fetch_playlists(user_id)
-
-    save_data(playlists, 'playlists')
-
-    print(len(playlists))
-
-    playlist_items = {}
-
-    print("Fetching playlists items")
-
-    index = 0
-    for pl in playlists:
-        pl_id = pl['id']
-        name = pl['name']
-        playlist_items[name] = fetch_playlist_items(pl_id)
-        print(f"{index} - {len(playlist_items[name])}")
-        index += 1
-
-    save_data(playlist_items, "playlistTracks")
-
-
-def get_user_data():
-    print("Fetching user data")
-
-    user_profile = fetch(f'v1/me')
-    user_id = user_profile['id']
-    user_data = fetch(f'v1/users/{user_id}')
-
-    save_data(user_profile, "user_profile")
-    save_data(user_data, "user_data")
-    return user_id
-
-
-def save_image_from_url(url, name, image_dir="playlist_covers"):
-    response = requests.get(url)
-
-    if response.status_code != 200:
-        raise RuntimeError("Cannot fetch the playlist cover")
-
-    image = Image.open(BytesIO(response.content))
-
-    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}")
-    image.save(file_path)
-
-
-def load_playlist_covers():
-    playlists = get_data('playlists')
-
-    for pl in playlists:
-        if len(pl['images']):
-            url = pl['images'][0]['url']
-            save_image_from_url(url, pl['name'])
-
-
-def load_user_cover():
-    user_data = get_data("user_data")
-    url = user_data['images'][1]['url']
-    save_image_from_url(url, "user", ".")
-
-
-def load_artworks():
-    load_user_cover()
-    load_playlist_covers()
-
-
-def update_access_token(id, secret):
-    global token
-    token = authenticate(id, secret)
-
-
-def fetch_data():
-    user_id = get_user_data()
-
-    fetch_tracks(user_id)
-    fetch_all_playlist_data(user_id)
-    fetch_tracks_top()
-    fetch_artists_top()
-
-    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("", "")
-    fetch_data()

src/helpers.py

@@ -1,55 +0,0 @@
-import json
-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=''):
-        self.artist = artist
-        self.title = title
-        self.album = album
-
-
-def get_data(name):
-    file_path = os.path.join(data_dir, f"{name}.json")
-    try:
-        with open(file_path, 'r') as file:
-            data = json.load(file)
-        return data
-    except FileNotFoundError:
-        print(f"File {name}.json not found.")
-        return {}
-    except json.JSONDecodeError:
-        print(f"Error decoding JSON from {name}.json.")
-        return {}
-
-def save_data(data, name):
-    def remove_available_markets(obj):
-        if isinstance(obj, dict):
-            return {key: remove_available_markets(value) for key, value in obj.items() if key != 'available_markets'}
-        elif isinstance(obj, list):
-            return [remove_available_markets(item) for item in obj]
-        else:
-            return obj
-
-    data = remove_available_markets(data)
-
-    os.makedirs(data_dir, exist_ok=True)
-
-    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)
-
-
-    print(f"Data saved to {file_path}.")
-
-
-

src/lyrics/lyrics.py

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

src/mappers/FuzzyMapper.py

@@ -1,112 +0,0 @@
-
-from rapidfuzz import fuzz
-from tqdm import tqdm
-
-from src.Library import *
-from src.helpers import *
-
-import unicodedata
-
-user_def_artist_map = {
-    "donda" : "kanye west",
-}
-
-def normalize(s):
-    return unicodedata.normalize("NFKC", s).replace("‐", "-")
-
-def get_score(first: str, second: str) -> float:
-    first_norm = normalize(first).lower()
-    second_norm = normalize(second).lower()
-    return float(first_norm == second_norm) * 100
-    return fuzz.token_set_ratio(first, second)
-
-def get_album_score(first: str, second: str) -> float:
-    return get_score(first, second)
-
-def get_track_score(first: str, second: str) -> float:
-    return get_score(first, second)
-
-def get_artist_score(first: str, second: str) -> float:
-    if not first or not second:
-        return 0.0
-
-    first_lower = first.lower()
-    second_lower = second.lower()
-
-    for key, value in user_def_artist_map.items():
-        key_lower = key.lower()
-        value_lower = value.lower()
-        if (first_lower == key_lower and second_lower == value_lower) or \
-           (first_lower == value_lower and second_lower == key_lower):
-            return 100.0
-
-    return get_score(first, second)
-
-
-def is_close_enough(first: float) -> bool:
-    return first > 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
-
-            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")

src/mappers/RemoteLibraryResolver.py

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

src/mappers/SpotifySearchMapping.py

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