ilusha/MusicIndexer
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: ilusha:main
Branches Tags
ilusha:main
ilusha:db-redesign
ilusha:mbid-finder-untested
ilusha:tmp3
ilusha:tmp2
ilusha:tmp
ilusha:master
..
compare: ilusha:tmp
Branches Tags
ilusha:db-redesign
ilusha:main
ilusha:mbid-finder-untested
ilusha:tmp3
ilusha:tmp2
ilusha:tmp
ilusha:master

2 commits
main ... tmp

Author SHA1 Message Date
IlyaShurupov
366860a1ad atmop 2024-07-08 23:24:51 +03:00
Ilya Shurupov
bb5453b59e initial 2024-07-08 23:20:28 +03:00
31 changed files with 347 additions and 2411 deletions
Download patch file Download diff file Expand all files Collapse all files

9
.env.example
View file

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

5
.gitignore vendored
View file

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

13
Env.py
View file

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

136
Flow.py
View file

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

208
Gui.py
View file

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

103
README.md
View file

@ -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<br>
client secret - ***REDACTED***

7
TODO
View file

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

25
common.py Normal file
View file

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

2
error.log
View file

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

5
gui.sh
View file

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

45
loadArtwork.py Normal file
View file

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

140
main.py
View file

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

41
old/main.py
View file

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

9
requirements.txt
View file

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

37
src/backends/spotify/SpotifyAuthenticator.py → spotifyAuth.py
View file

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

147
spotifyFetch.py Normal file
View file

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

128
src/Library.py
View file

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

191
src/LibrarySerializer.py
View file

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

210
src/StatGenerator.py
View file

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

192
src/backends/itunes/ParseItunesXml.py
View file

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

171
src/backends/local/LocalLibraryIndexer.py
View file

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

50
src/backends/local/LocalPLaylistGenerator.py
View file

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

100
src/backends/local/ParseLocal.py
View file

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

110
src/backends/navidrome/navidrome.py
View file

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

131
src/backends/spotify/ParseSpotify.py
View file

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

198
src/backends/spotify/SpotifyWebAPI.py
View file

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

55
src/helpers.py
View file

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

113
src/lyrics/lyrics.py
View file

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

112
src/mappers/FuzzyMapper.py
View file

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

9
src/mappers/RemoteLibraryResolver.py
View file

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

56
src/mappers/SpotifySearchMapping.py
View file

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