DB-backed incremental redesign: find tracks missing locally
Replace the JSON-file pipeline with a SQLite/SQLAlchemy store (Postgres via DATABASE_URL) and an incremental, ID-based matcher whose goal is "which Spotify/iTunes tracks am I missing locally". - src/db: models (Track/MbCache/FileState/IngestState) + engine/session. - src/ingest: upsert prefetched spotify/itunes/local (work_dir/latest-1) by (source, source_id); pulls ISRC from raw Spotify; preserves enrichment. - src/enrich: MusicBrainz recording-MBID resolver. Resolves via NORMALIZED text search (so the same song across sources lands on one MBID + one cached lookup), ISRC as exact fallback. Only touches unresolved tracks, caches positive AND negative results, rate-limited and resumable. - src/match: normalization + cross-source matcher with tiers ISRC > MBID > normalized text; reports present/missing and the match method. - main.py: CLI (ingest / enrich / match). Gui.py: missing-tracks viewer. - README + requirements (SQLAlchemy, musicbrainzngs); ignore *.db. Incremental everywhere: re-running updates only what changed; enrich continues where it stopped. Live Spotify/local backends are left intact. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Шурупов Илья Викторович
parent
83f782d125
commit
0e64d439c4
16 changed files with 784 additions and 274 deletions
268
Gui.py
268
Gui.py
|
|
@ -1,208 +1,90 @@
|
|||
import time
|
||||
"""Streamlit GUI: what's in your Spotify / iTunes libraries that you do NOT
|
||||
have locally, decided by real ID matching (recording MBID / ISRC) with a
|
||||
normalized-text fallback.
|
||||
|
||||
import streamlit as st
|
||||
from st_aggrid import AgGrid, GridOptionsBuilder, GridUpdateMode
|
||||
Reads only the database (run `python main.py ingest` and `enrich` first).
|
||||
"""
|
||||
import pandas as pd
|
||||
import streamlit as st
|
||||
from st_aggrid import AgGrid, GridOptionsBuilder
|
||||
|
||||
import Flow
|
||||
from src.db.database import init_db
|
||||
from src.match.matcher import classify_all, summary
|
||||
|
||||
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) | {
|
||||
}
|
||||
st.set_page_config(layout="wide", page_title="MusicIndexer — Missing tracks")
|
||||
|
||||
|
||||
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,
|
||||
)
|
||||
@st.cache_data(ttl=30)
|
||||
def load_data():
|
||||
init_db()
|
||||
missing, present = classify_all()
|
||||
return missing, present, summary()
|
||||
|
||||
|
||||
def run():
|
||||
flow = Flow.Flow()
|
||||
def grid(df, key):
|
||||
gb = GridOptionsBuilder.from_dataframe(df)
|
||||
gb.configure_default_column(filter=True, sortable=True, resizable=True)
|
||||
gb.configure_pagination(paginationAutoPageSize=False, paginationPageSize=50)
|
||||
AgGrid(df, gridOptions=gb.build(), height=620, fit_columns_on_grid_load=True, key=key)
|
||||
|
||||
# flow.fetch_spotify()
|
||||
# flow.fetch_local()
|
||||
|
||||
# flow.parse_spotify_library()
|
||||
# flow.parse_local_library()
|
||||
def main():
|
||||
st.title("🎧 Tracks you're missing locally")
|
||||
|
||||
flow.load_libraries()
|
||||
if st.sidebar.button("🔄 Refresh from DB"):
|
||||
load_data.clear()
|
||||
|
||||
# flow.map_local_to_spotify()
|
||||
missing, present, summ = load_data()
|
||||
|
||||
# flow.log_stats()
|
||||
# --- summary metrics ---
|
||||
for source in ("spotify", "itunes"):
|
||||
s = summ.get(source, {})
|
||||
if not s:
|
||||
continue
|
||||
st.subheader(source.capitalize())
|
||||
c1, c2, c3, c4, c5 = st.columns(5)
|
||||
c1.metric("Total", s["total"])
|
||||
c2.metric("Missing locally", s["missing"])
|
||||
c3.metric("Have (MBID)", s["present_mbid"])
|
||||
c4.metric("Have (text)", s["present_text"])
|
||||
coverage = (s["has_mbid"] / s["total"] * 100) if s["total"] else 0
|
||||
c5.metric("MBID coverage", f"{coverage:.0f}%")
|
||||
|
||||
gui = LibraryView()
|
||||
gui.add_libraries([flow.spotify_library, flow.itunes_library, flow.local_library])
|
||||
gui.render()
|
||||
local = summ.get("local", {})
|
||||
st.caption(
|
||||
f"Local library: {local.get('total', 0)} tracks, "
|
||||
f"{local.get('has_mbid', 0)} with a resolved MBID. "
|
||||
"‘Have (MBID)’ = certain ID match; ‘Have (text)’ = normalized "
|
||||
"artist/title match (run `enrich` to upgrade text matches to ID)."
|
||||
)
|
||||
|
||||
run()
|
||||
st.divider()
|
||||
|
||||
# --- controls ---
|
||||
view = st.sidebar.radio("View", ["Missing", "Present", "All"])
|
||||
sources = st.sidebar.multiselect("Sources", ["spotify", "itunes"],
|
||||
default=["spotify", "itunes"])
|
||||
|
||||
if view == "Missing":
|
||||
rows = [dict(r, status="missing") for r in missing]
|
||||
elif view == "Present":
|
||||
rows = [dict(r, status="present") for r in present]
|
||||
else:
|
||||
rows = ([dict(r, status="missing") for r in missing]
|
||||
+ [dict(r, status="present") for r in present])
|
||||
|
||||
rows = [r for r in rows if r["source"] in sources]
|
||||
|
||||
if not rows:
|
||||
st.info("Nothing to show — run `python main.py ingest` (and `enrich`).")
|
||||
return
|
||||
|
||||
df = pd.DataFrame(rows)[
|
||||
["status", "source", "artist", "title", "album", "matched_by",
|
||||
"isrc", "mbid", "play_count", "spotify_id"]
|
||||
]
|
||||
st.write(f"**{len(df)}** tracks")
|
||||
grid(df, key=f"{view}-{'-'.join(sorted(sources))}")
|
||||
|
||||
|
||||
main()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue