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