diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..27f5a90 --- /dev/null +++ b/.env.example @@ -0,0 +1,9 @@ +# 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 diff --git a/.gitignore b/.gitignore index 2803b17..7b75b45 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,8 @@ prefetched* __pycache__ .idea secret +.env +.venv +work_dir +old +tmp \ No newline at end of file diff --git a/Env.py b/Env.py new file mode 100644 index 0000000..c609347 --- /dev/null +++ b/Env.py @@ -0,0 +1,13 @@ +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") diff --git a/Flow.py b/Flow.py new file mode 100644 index 0000000..9ced6ae --- /dev/null +++ b/Flow.py @@ -0,0 +1,136 @@ + +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 \ No newline at end of file diff --git a/Gui.py b/Gui.py new file mode 100644 index 0000000..3c0ecc8 --- /dev/null +++ b/Gui.py @@ -0,0 +1,208 @@ +import time + +import streamlit as st +from st_aggrid import AgGrid, GridOptionsBuilder, GridUpdateMode +import pandas as pd + +import Flow + +from src.Library import * + +def entity_to_row(entity: Entity): + return { + "id": entity.id, + "mbid": entity.mbid, + "Title": entity.title, + "PLay Count": entity.play_count, + "Resolved": int(entity.resolved_percentage * 100), + "AutoScore": entity.auto_score, + "Added": entity.date_added.strftime("%Y-%m-%d") if entity.date_added else None, + "Played": entity.date_last_play.strftime("%Y-%m-%d") if entity.date_last_play else None, + "Released": entity.date_released.strftime("%Y-%m-%d") if entity.date_released else None, + "LocalId": str(entity.local_id) if entity.local_id else None, + "Path": str(entity.local_path) if entity.local_path else None, + } + +def track_to_row(track: Track): + return entity_to_row(track) | { + "Album": track.album.title if track.album else "", + "Artists": ", ".join(a.title if a.title else "no_arist_name" for a in track.artists), + "Lyrics": track.has_lyrics, + } + +def artist_to_row(artist: Artist): + return entity_to_row(artist) | { + } + +def album_to_row(album: Album): + return entity_to_row(album) | { + } + + +class LibraryView: + def __init__(self): + self.lib : Library = None + self.libs : list[Library] = None + self.view = None + self.lyrics_only = False + self.query = None + + def add_libraries(self, libs : list[Library]): + self.libs = libs + + def render(self): + st.set_page_config(layout="wide") + + with st.sidebar: + lib = st.selectbox( + "Library", + options=[lib.name for lib in self.libs], + ) + + for iter in self.libs: + if iter.name == lib: + self.lib = iter + + st.write(f"Showing library '{self.lib.name}'") + + self.view = st.radio( + "View", + ["Tree"] + ) + + if self.view == "Tree": + self.render_artists_albums_tracks() + # elif self.view == "Tracks": + # self.render_tracks() + # elif self.view == "Playlists": + # self.render_playlists() + + def track_visible( + self, + track: Track, + artist: Artist | None, + album: Album | None, + ): + if self.lyrics_only and not track.has_lyrics: + return False + + if self.query and self.query.lower() not in track.title.lower(): + return False + + if artist and all(a.id != artist.id for a in track.artists): + return False + + if album and (not track.album or track.album.id != album.id): + return False + + return True + + def render_artists_albums_tracks(self): + # --- ARTISTS TABLE --- + st.subheader("Artists") + artist_rows = [artist_to_row(a) for a in self.lib.artists] + df_artists = pd.DataFrame(artist_rows) + + gb_artists = GridOptionsBuilder.from_dataframe(df_artists) + gb_artists.configure_selection(selection_mode="multiple", use_checkbox=True) + gb_artists.configure_column("Title", filter="agTextColumnFilter") + gb_artists.configure_column("AutoScore", sort="desc") + grid_options_artists = gb_artists.build() + + grid_response_artists = AgGrid( + df_artists, + gridOptions=grid_options_artists, + update_mode=GridUpdateMode.SELECTION_CHANGED, + allow_unsafe_jscode=True, + enable_enterprise_modules=False, + fit_columns_on_grid_load=True, + ) + + selected_rows_df = grid_response_artists.get("selected_rows") + selected_artist_titles = selected_rows_df["Title"].tolist() if selected_rows_df is not None and not selected_rows_df.empty else [] + selected_artists = [a for a in self.lib.artists if a.title in selected_artist_titles] + + # --- ALBUMS TABLE --- + st.subheader("Albums") + if selected_artists: + albums_list = [] + for artist in selected_artists: + albums_list.extend(artist.albums) + albums_list = list({al.id: al for al in albums_list}.values()) + else: + albums_list = self.lib.albums + + album_rows = [album_to_row(a) for a in albums_list] + df_albums = pd.DataFrame(album_rows) + + gb_albums = GridOptionsBuilder.from_dataframe(df_albums) + gb_albums.configure_selection(selection_mode="multiple", use_checkbox=True) + gb_albums.configure_column("Title", filter="agTextColumnFilter") + gb_albums.configure_column("AutoScore", sort="desc") + grid_options_albums = gb_albums.build() + + grid_response_albums = AgGrid( + df_albums, + gridOptions=grid_options_albums, + update_mode=GridUpdateMode.SELECTION_CHANGED, + allow_unsafe_jscode=True, + enable_enterprise_modules=False, + fit_columns_on_grid_load=True, + ) + + selected_rows_df = grid_response_albums.get("selected_rows") + selected_album_titles = selected_rows_df["Title"].tolist() if selected_rows_df is not None and not selected_rows_df.empty else [] + selected_albums = [al for al in albums_list if al.title in selected_album_titles] + + # --- TRACKS TABLE --- + st.subheader("Tracks") + if selected_albums: + tracks_list = [] + for album in selected_albums: + tracks_list.extend(album.tracks) + elif selected_artists: + tracks_list = [] + for artist in selected_artists: + tracks_list.extend([t for t in self.lib.tracks if artist in t.artists]) + else: + tracks_list = self.lib.tracks + + track_rows = [track_to_row(t) for t in tracks_list] + df_tracks = pd.DataFrame(track_rows) + + gb_tracks = GridOptionsBuilder.from_dataframe(df_tracks) + gb_tracks.configure_selection(selection_mode="multiple", use_checkbox=True) + gb_tracks.configure_column("Title", filter="agTextColumnFilter") + gb_tracks.configure_column("AutoScore", sort="desc") + grid_options_tracks = gb_tracks.build() + + AgGrid( + df_tracks, + gridOptions=grid_options_tracks, + update_mode=GridUpdateMode.SELECTION_CHANGED, + allow_unsafe_jscode=True, + enable_enterprise_modules=False, + fit_columns_on_grid_load=True, + ) + + +def run(): + flow = Flow.Flow() + + # flow.fetch_spotify() + # flow.fetch_local() + + # flow.parse_spotify_library() + # flow.parse_local_library() + + flow.load_libraries() + + # flow.map_local_to_spotify() + + # flow.log_stats() + + gui = LibraryView() + gui.add_libraries([flow.spotify_library, flow.itunes_library, flow.local_library]) + gui.render() + +run() \ No newline at end of file diff --git a/README.md b/README.md index 68b9ba5..62d8bd7 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,100 @@ -# spotifyFetcher -loads all data through web api
-client secret - ***REDACTED*** +# MusicIndexer + +Pull your Spotify library and listening stats via the Spotify Web API, index a +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. diff --git a/TODO b/TODO new file mode 100644 index 0000000..f285a22 --- /dev/null +++ b/TODO @@ -0,0 +1,7 @@ +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 diff --git a/common.py b/common.py deleted file mode 100644 index bb08d3e..0000000 --- a/common.py +++ /dev/null @@ -1,25 +0,0 @@ -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." \ No newline at end of file diff --git a/error.log b/error.log new file mode 100644 index 0000000..7c830a2 --- /dev/null +++ b/error.log @@ -0,0 +1,2 @@ + +(process:38049): GLib-GIO-CRITICAL **: 14:06:16.561: g_dbus_connection_emit_signal: assertion 'G_IS_DBUS_CONNECTION (connection)' failed diff --git a/gui.sh b/gui.sh new file mode 100755 index 0000000..9be5f9d --- /dev/null +++ b/gui.sh @@ -0,0 +1,5 @@ +#!/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 "$@" diff --git a/loadArtwork.py b/loadArtwork.py deleted file mode 100644 index 4d2dc38..0000000 --- a/loadArtwork.py +++ /dev/null @@ -1,45 +0,0 @@ -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() diff --git a/main.py b/main.py index 49af365..68707c7 100644 --- a/main.py +++ b/main.py @@ -1,114 +1,38 @@ -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') +import Flow -def get_playlist_names(): - return [name for name, items in playlists.items()] +def test1(): + flow = Flow.Flow() + + # flow.fetch_local() + flow.parse_local_library() + flow.load_libraries() + + # flow.map_local_to_spotify() + # flow.log_stats() + + # flow.fetch_and_save_lyrics() -def get_artists_str(artists_pack): - track_artists = [artist['name'] for artist in artists_pack] - artists = " ".join(track_artists) - return artists +def main(): + flow = Flow.Flow() + + flow.fetch_spotify() + # 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") -def print_playlist_tracks(name): - 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() - - 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_exit(self, arg): - """Exit the command loop: exit""" - print("Goodbye!") - return True - - -if __name__ == '__main__': - Interpreter().cmdloop() +# test1() +main() diff --git a/old/main.py b/old/main.py new file mode 100644 index 0000000..0b2a991 --- /dev/null +++ b/old/main.py @@ -0,0 +1,41 @@ +def print_link_stats(link_pattern): + resolved_reversed = {} + resolved = [] + unresolved_local = [] + unresolved_remote = [] + + for local_id, links in link_pattern.items(): + if len(links['items']): + resolved.append(local_id) + resolved_reversed[links['items'][0]] = local_id + else: + unresolved_local.append(local_id) + + for track in user_tracks: + track_id = track['track']['id'] + if track_id not in resolved_reversed: + unresolved_remote.append(track_id) + + print(f"Local Tracks: {len(local_library)}") + print(f"Remote Tracks: {len(user_tracks)}") + + unresolved_remote = sort_remote_tracks_by_popularity(unresolved_remote) + unresolved_remote = unresolved_remote[0:min(len(unresolved_remote), 100)] + + print(f"\nUnresolved tracks from the remote: {len(unresolved_remote)}") + index = 0 + for remote_id in unresolved_remote: + remote_track = user_tracks_by_id[remote_id] + print(f" {index}: '{remote_track['name']}' by '{get_artists_str(remote_track['artists'])}'") + index += 1 + + print(f"\nResolved tracks: {len(resolved)}") + index = 0 + for link in resolved: + remote_id = link_pattern[link]['items'][0] + local_track = local_library[link] + remote_track = user_tracks_by_id[remote_id] + + print(f" {index}: '{local_track['name']}' by '{local_track['artist']}'", end=' ----> ') + print(f"'{remote_track['name']}' by '{get_artists_str(remote_track['artists'])}'") + index += 1 \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..9a1d950 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,9 @@ +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 diff --git a/spotifyFetch.py b/spotifyFetch.py deleted file mode 100644 index ec590bc..0000000 --- a/spotifyFetch.py +++ /dev/null @@ -1,147 +0,0 @@ -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() diff --git a/src/Library.py b/src/Library.py new file mode 100644 index 0000000..246d581 --- /dev/null +++ b/src/Library.py @@ -0,0 +1,128 @@ +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 diff --git a/src/LibrarySerializer.py b/src/LibrarySerializer.py new file mode 100644 index 0000000..278776a --- /dev/null +++ b/src/LibrarySerializer.py @@ -0,0 +1,191 @@ + +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 diff --git a/src/StatGenerator.py b/src/StatGenerator.py new file mode 100644 index 0000000..7c17ab5 --- /dev/null +++ b/src/StatGenerator.py @@ -0,0 +1,210 @@ +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") diff --git a/src/backends/itunes/ParseItunesXml.py b/src/backends/itunes/ParseItunesXml.py new file mode 100755 index 0000000..71eb2d7 --- /dev/null +++ b/src/backends/itunes/ParseItunesXml.py @@ -0,0 +1,192 @@ +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 + + diff --git a/src/backends/local/LocalLibraryIndexer.py b/src/backends/local/LocalLibraryIndexer.py new file mode 100644 index 0000000..6f132d4 --- /dev/null +++ b/src/backends/local/LocalLibraryIndexer.py @@ -0,0 +1,171 @@ +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") diff --git a/src/backends/local/LocalPLaylistGenerator.py b/src/backends/local/LocalPLaylistGenerator.py new file mode 100644 index 0000000..1c0f1e1 --- /dev/null +++ b/src/backends/local/LocalPLaylistGenerator.py @@ -0,0 +1,50 @@ + +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") \ No newline at end of file diff --git a/src/backends/local/ParseLocal.py b/src/backends/local/ParseLocal.py new file mode 100644 index 0000000..b1fbba3 --- /dev/null +++ b/src/backends/local/ParseLocal.py @@ -0,0 +1,100 @@ +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 diff --git a/src/backends/navidrome/navidrome.py b/src/backends/navidrome/navidrome.py new file mode 100644 index 0000000..c7d04f4 --- /dev/null +++ b/src/backends/navidrome/navidrome.py @@ -0,0 +1,110 @@ +import sqlite3 +import os +from dataclasses import dataclass, field +from typing import List, Optional + +@dataclass +class Track: + id: int + title: str + path: str + favorite: bool + rating: Optional[int] + play_count: int + last_played: Optional[str] + playlists: List[str] = field(default_factory=list) + +@dataclass +class Album: + id: int + title: str + tracks: List[Track] = field(default_factory=list) + +@dataclass +class Artist: + id: str + name: str + albums: List[Album] = field(default_factory=list) + +class NavidromeDB: + def __init__(self, db_path: str): + if not os.path.isfile(db_path): + raise FileNotFoundError(f"Navidrome DB not found: {db_path}") + self.conn = sqlite3.connect(db_path) + self.conn.row_factory = sqlite3.Row + + def close(self): + self.conn.close() + + def get_artists(self) -> List[Artist]: + artists = [] + c = self.conn.cursor() + c.execute("SELECT id, name FROM artist ORDER BY name") + for row in c.fetchall(): + artist = Artist(id=row["id"], name=row["name"]) + artist.albums = self.get_albums(artist.id) + artists.append(artist) + return artists + + def get_albums(self, artist_id: str) -> List[Album]: + albums = [] + c = self.conn.cursor() + c.execute( + "SELECT id, name FROM album WHERE album_artist_id=? ORDER BY name", + (artist_id,) + ) + for row in c.fetchall(): + album = Album(id=row["id"], title=row["name"]) + album.tracks = self.get_tracks(album.id) + albums.append(album) + return albums + + def get_tracks(self, album_id: int) -> List[Track]: + tracks = [] + c = self.conn.cursor() + c.execute( + "SELECT t.id, t.name AS Title, t.path, ut.favorite, ut.rating, ut.play_count AS PlayCount, ut.last_played AS LastPlayed " + "FROM track t " + "LEFT JOIN usertrack ut ON ut.track_id = t.id " + "WHERE t.album_id=? ORDER BY t.track_number", + (album_id,) + ) + for row in c.fetchall(): + track = Track( + id=row["id"], + title=row["Title"], + path=row["path"], + favorite=bool(row["favorite"]), + rating=row["rating"], + play_count=row["PlayCount"] or 0, + last_played=row["LastPlayed"], + playlists=self.get_playlists_for_track(row["id"]) + ) + tracks.append(track) + return tracks + + def get_playlists_for_track(self, track_id: int) -> List[str]: + c = self.conn.cursor() + c.execute( + "SELECT p.name FROM playlist p " + "JOIN playlisttrack pt ON pt.playlist_id=p.id " + "WHERE pt.track_id=?", + (track_id,) + ) + return [row["name"] for row in c.fetchall()] + +# Example usage +if __name__ == "__main__": + db_path = "/mnt/main/data/music/navidrome.db" + navidb = NavidromeDB(db_path) + + try: + artists = navidb.get_artists() + for artist in artists: + print(f"Artist: {artist.name}") + for album in artist.albums: + print(f" Album: {album.title}") + for track in album.tracks: + print(f" Track: {track.title} | Path: {track.path} | Fav: {track.favorite} | Plays: {track.play_count}") + finally: + navidb.close() diff --git a/src/backends/spotify/ParseSpotify.py b/src/backends/spotify/ParseSpotify.py new file mode 100644 index 0000000..cfca39d --- /dev/null +++ b/src/backends/spotify/ParseSpotify.py @@ -0,0 +1,131 @@ +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 \ No newline at end of file diff --git a/spotifyAuth.py b/src/backends/spotify/SpotifyAuthenticator.py similarity index 78% rename from spotifyAuth.py rename to src/backends/spotify/SpotifyAuthenticator.py index 64a4b2c..28cce0a 100644 --- a/spotifyAuth.py +++ b/src/backends/spotify/SpotifyAuthenticator.py @@ -8,20 +8,10 @@ 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) +refresh_token = None def retrieve_web_api_token(authorization_code): @@ -34,7 +24,7 @@ def retrieve_web_api_token(authorization_code): "client_secret": client_secret, "client_id": client_id, "code": authorization_code, - 'redirect_uri': 'http://localhost:8888', + 'redirect_uri': 'http://127.0.0.1:8888', } print("\nAuthenticating WEB API\n") @@ -55,7 +45,10 @@ def retrieve_web_api_token(authorization_code): print(token_response.json()) raise ValueError - token = token_response.json()['access_token'] + global refresh_token + response_json = token_response.json() + token = response_json['access_token'] + refresh_token = response_json.get('refresh_token') class RequestHandler(BaseHTTPRequestHandler): @@ -84,7 +77,7 @@ class RequestHandler(BaseHTTPRequestHandler): def run_redirect_server(): - httpd = HTTPServer(('localhost', 8888), RequestHandler) + httpd = HTTPServer(('127.0.0.1', 8888), RequestHandler) print('Starting HTTP redirection server') httpd.serve_forever() @@ -95,16 +88,22 @@ def open_authenticate_page(): params = { 'client_id': client_id, 'response_type': 'code', - 'redirect_uri': 'http://localhost:8888', + 'redirect_uri': 'http://127.0.0.1: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}?client_id={params['client_id']}&response_type={params['response_type']}&redirect_uri={params['redirect_uri']}&scope={params['scope']}" + auth_url = f"{authorize_url}?{urllib.parse.urlencode(params)}" webbrowser.open(auth_url) + print(f"If the browser did not open, visit:\n{auth_url}") -def authenticate(): - resolve_client_secret() +def authenticate(id, secret): + + global client_secret + global client_id + + client_secret=secret + client_id=id redirect_server = threading.Thread(target=run_redirect_server) redirect_server.start() diff --git a/src/backends/spotify/SpotifyWebAPI.py b/src/backends/spotify/SpotifyWebAPI.py new file mode 100644 index 0000000..e795d8a --- /dev/null +++ b/src/backends/spotify/SpotifyWebAPI.py @@ -0,0 +1,198 @@ +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() diff --git a/src/helpers.py b/src/helpers.py new file mode 100644 index 0000000..3c7db3e --- /dev/null +++ b/src/helpers.py @@ -0,0 +1,55 @@ +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}.") + + + diff --git a/src/lyrics/lyrics.py b/src/lyrics/lyrics.py new file mode 100644 index 0000000..ae6ed6d --- /dev/null +++ b/src/lyrics/lyrics.py @@ -0,0 +1,113 @@ +import requests +from mutagen import File +from mutagen.id3 import ID3, USLT, ID3NoHeaderError +import os + +def fetch_lyrics(artist: str, album: str, track: str, duration: int = 0) -> str: + + skip_artists = [ + "shurupis", + "unknown", + "Unknown" + ] + + skip_album = [ + "unknown", + "Unknown" + ] + + if artist in skip_artists or album in skip_album: + return "" + + + url = "https://lrclib.net/api/get" + params = { + "artist_name": artist, + "track_name": track, + "album_name": album, + "duration": duration, + } + try: + resp = requests.get(url, params=params, timeout=10) + if resp.status_code != 200: + try: + err = resp.json() + print(err.get("error"), "-", err.get("message")) + except ValueError: + print("HTTPError", resp.status_code) + return "" + data = resp.json() + if "error" in data: + print(data.get("error"), "-", data.get("message")) + return "" + return data.get("syncedLyrics", "") + except requests.RequestException as e: + print(type(e).__name__, "-", str(e)) + return "" + except ValueError as e: + print(type(e).__name__, "-", str(e)) + return "" + + +def write_lyrics(file_path, lyrics_text, lang="eng"): + """ + Writes lyrics text into tags recognized by Navidrome. + Provides detailed logs and success/failure info. + """ + + ext = os.path.splitext(file_path)[1].lower() + print(f"[INFO] Processing file: {file_path} (extension: {ext})") + + try: + # ---------- MP3 ---------- + if ext == ".mp3": + print("[INFO] Detected MP3 file") + try: + tags = ID3(file_path) + print("[INFO] Existing ID3 tags loaded") + except ID3NoHeaderError: + tags = ID3() + print("[INFO] No ID3 header found, creating new tags") + + tags.delall("USLT") + print("[INFO] Removed existing USLT frames (lyrics)") + + tags.add( + USLT( + encoding=3, # UTF-8 + lang=lang, + desc="", + text=lyrics_text + ) + ) + tags.save(file_path) + print("[SUCCESS] Lyrics written to MP3 tags") + return + + # ---------- FLAC / OGG / OPUS ---------- + audio = File(file_path) + if audio is None: + print(f"[ERROR] Unsupported or unrecognized file format: {file_path}") + return + + if ext in {".flac", ".ogg", ".opus"}: + print(f"[INFO] Detected {ext.upper()} file") + audio["LRC"] = lyrics_text + audio["SYNCEDLYRICS"] = lyrics_text + audio.save() + print(f"[SUCCESS] Lyrics written to {ext.upper()} tags") + return + + # ---------- Fallback ---------- + if audio.tags is not None: + print("[INFO] Using fallback tag writing") + audio["LYRICS"] = lyrics_text + audio.save() + print("[SUCCESS] Lyrics written using fallback tags") + return + + print(f"[WARNING] No suitable tag frame found for {file_path}") + + except Exception as e: + print(f"[ERROR] Failed to write lyrics to {file_path}: {e}") + diff --git a/src/mappers/FuzzyMapper.py b/src/mappers/FuzzyMapper.py new file mode 100644 index 0000000..c6413d0 --- /dev/null +++ b/src/mappers/FuzzyMapper.py @@ -0,0 +1,112 @@ + +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") diff --git a/src/mappers/RemoteLibraryResolver.py b/src/mappers/RemoteLibraryResolver.py new file mode 100644 index 0000000..7342848 --- /dev/null +++ b/src/mappers/RemoteLibraryResolver.py @@ -0,0 +1,9 @@ + +import src.mappers.FuzzyMapper as Fuzzy +import src.mappers.SpotifySearchMapping as Spotify + +from src.Library import * + +def resolve_remote_tracks(client_id, client_secret, local_lib : Library, spotify_lib : Library, output_dir): + Fuzzy.generate_map(local_lib, spotify_lib, output_dir) + # Spotify.generate_map(client_id, client_secret, local_lib, spotify_lib, output_dir) \ No newline at end of file diff --git a/src/mappers/SpotifySearchMapping.py b/src/mappers/SpotifySearchMapping.py new file mode 100644 index 0000000..e8878e9 --- /dev/null +++ b/src/mappers/SpotifySearchMapping.py @@ -0,0 +1,56 @@ + +from tqdm import tqdm + +from src.Library import * +from src.helpers import * +from src.backends.spotify.SpotifyWebAPI import find_song, update_access_token + +def generate_map(client_id, client_secret, local_lib : Library, spotify_lib : Library, output_dir): + + set_workdir(output_dir) + + update_access_token(client_id, client_secret) + + def find_local_songs_on_spotify(): + search_log = [] + + total = len(local_lib.tracks) + found_tracks_map = {} + with tqdm(total=total, desc='Searching local songs on spotify') as pbar: + for track in local_lib.tracks: + + parts = [track.title, track.artists[0].title if len(track.artists) else "", track.album.title] + search_pattern = " ".join(p for p in parts if p) + + if search_pattern == "": + print(f"cannot generate search pattern for the track - {track.local_path}") + pbar.update(1) + continue + + found_tracks = find_song(search_pattern)[0:10] + found_tracks_map[track.id] = found_tracks + search_log.append({"path" : track.local_path, "search_pattern": search_pattern, "result" : found_tracks_map[track.id]}) + pbar.update(1) + + save_data(search_log, "spotify_search_log") + save_data(found_tracks_map, "spotify_found_map") + + return found_tracks_map + + spotify_found_tracks = find_local_songs_on_spotify() + + spotify_user_track = {track.id for track in spotify_lib.tracks} + + spotify_pattern = {} + for local_id, found_items in spotify_found_tracks.items(): + spotify_pattern[local_id] = {"score": 1.0, "items": []} + + if not len(found_items): + continue + + for found_item in found_items: + spotify_track_id = found_item['id'] + if spotify_track_id in spotify_user_track: + spotify_pattern[local_id]['items'].append(spotify_track_id) + + save_data(spotify_pattern, "local_to_spotify_id_map") \ No newline at end of file