diff --git a/core/database_update_worker.py b/core/database_update_worker.py index 3262ee30..a862d24d 100644 --- a/core/database_update_worker.py +++ b/core/database_update_worker.py @@ -41,7 +41,13 @@ class DatabaseUpdateWorker: self.database_path = database_path self.full_refresh = full_refresh self.should_stop = False - + + # Track ids of rows newly INSERTED this run (not updates). The web + # layer reads this after the scan to gap-fill embedded provider IDs + # for the new files (auto-reconcile), so newly-added music contributes + # its Spotify/MusicBrainz/etc. ids without a manual backfill. + self._new_track_ids = set() + # Statistics tracking self.processed_artists = 0 self.processed_albums = 0 @@ -880,6 +886,8 @@ class DatabaseUpdateWorker: track_success = self.database.insert_or_update_media_track(track, album_id, artist_id, server_source=self.server_type) if track_success: total_processed_tracks += 1 + if track_success == 'inserted': + self._new_track_ids.add(str(track.ratingKey)) logger.debug(f"Processed new track: {track.title}") except Exception as e: logger.warning(f"Failed to process track '{getattr(track, 'title', 'Unknown')}': {e}") @@ -1344,6 +1352,8 @@ class DatabaseUpdateWorker: skipped_count += 1 elif track_success: track_count += 1 + if track_success == 'inserted': + self._new_track_ids.add(track_id_str) except Exception as e: logger.warning(f"Failed to process track '{getattr(track, 'title', 'Unknown')}': {e}") diff --git a/core/library/embedded_id_reconcile.py b/core/library/embedded_id_reconcile.py index 06f20711..b7bc5430 100644 --- a/core/library/embedded_id_reconcile.py +++ b/core/library/embedded_id_reconcile.py @@ -233,6 +233,122 @@ def reconcile_track_row( return TrackReconcileResult(applied, len(plan.conflicts), readable=True) +@dataclass +class ReconcileTotals: + """Accumulated counts over a reconcile run.""" + total: int = 0 + processed: int = 0 + ids_filled: int = 0 + entities_updated: int = 0 + conflicts: int = 0 + unreadable: int = 0 + + +def _load_missing_rows(cursor, ids, table, target: Dict[str, Dict[str, Any]]) -> None: + """Load any not-yet-cached entity rows for ``ids`` into ``target`` in place. + + Ids with no row get an empty dict so they're never re-queried. Chunked to + keep the IN clause bounded. + """ + missing = [i for i in {x for x in ids if x} if i not in target] + for start in range(0, len(missing), 500): + chunk = missing[start:start + 500] + ph = ','.join('?' * len(chunk)) + cursor.execute(f"SELECT * FROM {table} WHERE id IN ({ph})", chunk) + for r in cursor.fetchall(): + target[str(r['id'])] = dict(r) + for i in missing: + target.setdefault(i, {}) # mark absent → don't re-query + + +def reconcile_library( + conn, + read_tags, + track_ids=None, + page_size: int = 500, + on_progress=None, + should_stop=None, +) -> ReconcileTotals: + """Gap-fill embedded provider IDs into the DB for a set of tracks. + + Shared orchestration used by both the manual backfill job and the + auto-reconcile hook on library scans. Pages the track list (bounded + memory), lazily loads only the parent album/artist rows actually + referenced (cheap when scoped to a handful of new tracks), and commits + per page so concurrent enrichment workers aren't starved of the write + lock. + + Args: + conn: open DB connection; this function commits per page. + read_tags: callable ``(file_path) -> tags dict | None``. The caller + injects path resolution + ``read_embedded_tags`` so this module + stays free of Flask / docker-path concerns. ``None`` => unreadable. + track_ids: iterable of track ids to reconcile, or ``None`` for every + track that has a ``file_path``. + page_size: rows materialised per page. + on_progress: optional ``(totals, current_title) -> None`` after each + track (for live UI). + should_stop: optional ``() -> bool`` checked between tracks/pages to + abort early. + + Returns: + :class:`ReconcileTotals`. + """ + from utils.logging_config import get_logger + logger = get_logger("library.reconcile") + + totals = ReconcileTotals() + cur = conn.cursor() + + if track_ids is None: + cur.execute("SELECT id FROM tracks WHERE file_path IS NOT NULL AND TRIM(file_path) != ''") + ids = [str(r[0]) for r in cur.fetchall()] + else: + ids = [str(t) for t in track_ids if t is not None] + totals.total = len(ids) + + album_map: Dict[str, Dict[str, Any]] = {} + artist_map: Dict[str, Dict[str, Any]] = {} + + for start in range(0, len(ids), page_size): + if should_stop and should_stop(): + break + page = ids[start:start + page_size] + ph = ','.join('?' * len(page)) + cur.execute(f"SELECT * FROM tracks WHERE id IN ({ph})", page) + rows = [dict(r) for r in cur.fetchall()] + + _load_missing_rows(cur, [str(r['album_id']) for r in rows if r.get('album_id') is not None], + 'albums', album_map) + _load_missing_rows(cur, [str(r['artist_id']) for r in rows if r.get('artist_id') is not None], + 'artists', artist_map) + + for tr in rows: + if should_stop and should_stop(): + break + title = tr.get('title') or '?' + try: + tags = read_tags(tr.get('file_path')) + result = reconcile_track_row(cur, tr, album_map, artist_map, tags) + if not result.readable: + totals.unreadable += 1 + else: + totals.ids_filled += result.applied.ids_filled + totals.entities_updated += result.applied.rows_updated + totals.conflicts += result.conflicts + except Exception as e: + logger.debug("reconcile: skipped track %s: %s", tr.get('id'), e) + totals.unreadable += 1 + finally: + totals.processed += 1 + if on_progress: + on_progress(totals, title) + + conn.commit() + + return totals + + def _existing_columns(cursor, table: str) -> set: """Return the set of column names on ``table`` (migration-safe guard).""" cursor.execute(f"PRAGMA table_info({table})") diff --git a/database/music_database.py b/database/music_database.py index e0fcdbe1..43935ead 100644 --- a/database/music_database.py +++ b/database/music_database.py @@ -6033,8 +6033,12 @@ class MusicDatabase: except Exception as e: logger.debug("history logging: %s", e) - return True - + # Truthy on success (existing `if track_success` callers keep + # working); the specific value lets the scan worker tell a + # genuinely new row from an updated one so it can reconcile + # embedded IDs only for new arrivals. + return 'inserted' if is_new_track else 'updated' + except Exception as e: error_text = str(e).lower() if ( diff --git a/tests/library/test_embedded_id_reconcile.py b/tests/library/test_embedded_id_reconcile.py index 9c95b896..b60f1643 100644 --- a/tests/library/test_embedded_id_reconcile.py +++ b/tests/library/test_embedded_id_reconcile.py @@ -19,8 +19,10 @@ from core.library.embedded_id_reconcile import ( Fill, ReconcileApplied, ReconcilePlan, + ReconcileTotals, apply_reconcile_plan, plan_reconcile, + reconcile_library, reconcile_track_row, ) @@ -271,3 +273,91 @@ def test_reconcile_track_row_handles_null_parent_ids(): assert result.applied.ids_filled == 1 # album fill has no album id to land on cur.execute("SELECT spotify_track_id FROM tracks WHERE id='t1'") assert cur.fetchone()[0] == 'TRK' + + +# --------------------------------------------------------------------------- +# reconcile_library — the shared orchestration (paging, lazy maps, scope) +# --------------------------------------------------------------------------- + +def _make_library_db(): + conn = sqlite3.connect(':memory:') + conn.row_factory = sqlite3.Row + cur = conn.cursor() + cur.execute("""CREATE TABLE tracks (id TEXT PRIMARY KEY, album_id TEXT, artist_id TEXT, + file_path TEXT, title TEXT, spotify_track_id TEXT, spotify_match_status TEXT, + spotify_last_attempted TIMESTAMP)""") + cur.execute("""CREATE TABLE albums (id TEXT PRIMARY KEY, spotify_album_id TEXT, + spotify_match_status TEXT, spotify_last_attempted TIMESTAMP)""") + cur.execute("""CREATE TABLE artists (id TEXT PRIMARY KEY, spotify_artist_id TEXT, + spotify_match_status TEXT, spotify_last_attempted TIMESTAMP)""") + # Two tracks on the same album/artist, one orphan track with no file. + cur.execute("INSERT INTO artists (id) VALUES ('ar1')") + cur.execute("INSERT INTO albums (id) VALUES ('al1')") + cur.execute("INSERT INTO tracks (id, album_id, artist_id, file_path, title) VALUES ('t1','al1','ar1','/a.flac','A')") + cur.execute("INSERT INTO tracks (id, album_id, artist_id, file_path, title) VALUES ('t2','al1','ar1','/b.flac','B')") + cur.execute("INSERT INTO tracks (id, album_id, artist_id, file_path, title) VALUES ('t3','al1','ar1','','NoFile')") + conn.commit() + return conn, cur + + +def _reader(mapping): + """read_tags stub: file_path -> tags dict (or None).""" + return lambda fp: mapping.get(fp) + + +def test_reconcile_library_whole_library_fills_all(): + conn, cur = _make_library_db() + tags = { + '/a.flac': {'spotify_track_id': 'TA', 'spotify_album_id': 'ALB', 'spotify_artist_id': 'ART'}, + '/b.flac': {'spotify_track_id': 'TB', 'spotify_album_id': 'ALB', 'spotify_artist_id': 'ART'}, + } + totals = reconcile_library(conn, _reader(tags)) + assert isinstance(totals, ReconcileTotals) + # t3 has empty file_path -> excluded from the whole-library SELECT entirely. + assert totals.total == 2 and totals.processed == 2 + # t1: track+album+artist (3); t2: only its own track id (parents already filled) (1) + assert totals.ids_filled == 4 + cur.execute("SELECT spotify_track_id FROM tracks WHERE id='t2'") + assert cur.fetchone()[0] == 'TB' + cur.execute("SELECT spotify_album_id FROM albums WHERE id='al1'") + assert cur.fetchone()[0] == 'ALB' + + +def test_reconcile_library_scoped_to_given_ids(): + conn, cur = _make_library_db() + tags = {'/b.flac': {'spotify_track_id': 'TB'}, '/a.flac': {'spotify_track_id': 'TA'}} + totals = reconcile_library(conn, _reader(tags), track_ids=['t2']) + assert totals.total == 1 and totals.ids_filled == 1 + cur.execute("SELECT spotify_track_id FROM tracks WHERE id='t2'") + assert cur.fetchone()[0] == 'TB' + cur.execute("SELECT spotify_track_id FROM tracks WHERE id='t1'") + assert cur.fetchone()[0] is None # not in scope + + +def test_reconcile_library_unreadable_counted(): + conn, cur = _make_library_db() + totals = reconcile_library(conn, _reader({}), track_ids=['t1', 't2']) # reader returns None + assert totals.unreadable == 2 and totals.ids_filled == 0 + + +def test_reconcile_library_is_idempotent(): + conn, cur = _make_library_db() + tags = {'/a.flac': {'spotify_track_id': 'TA', 'spotify_album_id': 'ALB', 'spotify_artist_id': 'ART'}} + first = reconcile_library(conn, _reader(tags), track_ids=['t1']) + second = reconcile_library(conn, _reader(tags), track_ids=['t1']) + assert first.ids_filled == 3 + assert second.ids_filled == 0 # nothing left to fill + + +def test_reconcile_library_progress_and_stop(): + conn, cur = _make_library_db() + seen = [] + reconcile_library(conn, _reader({'/a.flac': {'spotify_track_id': 'TA'}}), + track_ids=['t1', 't2'], + on_progress=lambda totals, title: seen.append(title)) + assert seen == ['A', 'B'] + + # should_stop halts before processing anything + totals = reconcile_library(conn, _reader({}), track_ids=['t1', 't2'], + should_stop=lambda: True) + assert totals.processed == 0 diff --git a/web_server.py b/web_server.py index 4b7d4b24..2cf730c8 100644 --- a/web_server.py +++ b/web_server.py @@ -9837,6 +9837,71 @@ def get_write_tags_batch_status(): # the API lookup entirely — large API savings on an already-tagged library. # Gap-fill only: an existing id is never overwritten (see # core/library/embedded_id_reconcile.py). +def _reconcile_library_tracks(conn, track_ids=None, on_progress=None, should_stop=None): + """Run the embedded-ID reconcile with web-server path resolution injected. + + Thin wrapper over core.library.embedded_id_reconcile.reconcile_library that + supplies the read_tags callable (docker/library path resolution + mutagen + read). Used by the manual backfill job and available for any scoped + reconcile. ``track_ids=None`` => whole library. + """ + from core.library.file_tags import read_embedded_tags + from core.library.embedded_id_reconcile import reconcile_library + + def _read_tags(file_path): + resolved = _resolve_library_file_path(file_path) + if not resolved: + return None + info = read_embedded_tags(resolved) + return info.get('tags') if info.get('available') else None + + return reconcile_library(conn, _read_tags, track_ids=track_ids, + on_progress=on_progress, should_stop=should_stop) + + +def _reconcile_after_scan(worker): + """Gap-fill embedded provider IDs for tracks a scan newly inserted. + + Runs after a library scan/deep-scan completes, scoped to the rows the + worker actually INSERTED this run (``worker._new_track_ids``). New files + are written with empty provider-id columns, so this reads their tags and + fills any IDs present — keeping the DB current without a manual backfill. + Best-effort: never raises into the scan flow. + """ + try: + new_ids = list(getattr(worker, '_new_track_ids', None) or []) + if not new_ids: + return + n = len(new_ids) + try: + _db_update_phase_callback( + f"Reading file tags for {n} new track{'s' if n != 1 else ''}…") + except Exception: # noqa: S110 — best-effort UI phase, never block the reconcile + pass + + def _on_progress(totals, title): + try: + pct = (totals.processed / totals.total * 100) if totals.total else 100 + _db_update_progress_callback(title, totals.processed, totals.total, pct) + except Exception: # noqa: S110 — best-effort UI progress tick + pass + + database = get_database() + conn = database._get_connection() + try: + totals = _reconcile_library_tracks(conn, track_ids=new_ids, on_progress=_on_progress) + logger.info( + "[Reconcile] Post-scan: filled %d id(s) across %d row(s) from %d new " + "track(s) (%d unreadable, %d conflicts)", + totals.ids_filled, totals.entities_updated, totals.processed, + totals.unreadable, totals.conflicts, + ) + finally: + conn.close() + except Exception as e: + logger.warning("[Reconcile] Post-scan reconcile failed (non-fatal): %s", e) + + _reconcile_ids_state = { 'status': 'idle', # idle | running | done 'total': 0, @@ -9868,71 +9933,32 @@ def reconcile_embedded_ids(): database = get_database() def _run(): - from core.library.file_tags import read_embedded_tags - from core.library.embedded_id_reconcile import reconcile_track_row conn = None try: conn = database._get_connection() - cur = conn.cursor() - # Parent IDs in memory (these tables are far smaller than tracks). - cur.execute("SELECT * FROM albums") - album_map = {str(r['id']): dict(r) for r in cur.fetchall()} - cur.execute("SELECT * FROM artists") - artist_map = {str(r['id']): dict(r) for r in cur.fetchall()} - # Track IDs only first (light); rows are pulled per page below so - # memory stays bounded on large libraries. Each page's SELECT is - # fully fetched before any UPDATE, so reusing one cursor is safe. - cur.execute("SELECT id FROM tracks WHERE file_path IS NOT NULL AND TRIM(file_path) != ''") - track_ids = [str(r['id']) for r in cur.fetchall()] - with _reconcile_ids_lock: - _reconcile_ids_state['total'] = len(track_ids) + def _on_progress(totals, title): + with _reconcile_ids_lock: + _reconcile_ids_state.update({ + 'total': totals.total, + 'processed': totals.processed, + 'entities_updated': totals.entities_updated, + 'ids_filled': totals.ids_filled, + 'conflicts': totals.conflicts, + 'unreadable': totals.unreadable, + 'current': title, + }) - PAGE = 500 - for start in range(0, len(track_ids), PAGE): - page = track_ids[start:start + PAGE] - ph = ','.join('?' * len(page)) - cur.execute(f"SELECT * FROM tracks WHERE id IN ({ph})", page) - rows = [dict(r) for r in cur.fetchall()] - - for tr in rows: - title = tr.get('title') or '?' - with _reconcile_ids_lock: - _reconcile_ids_state['current'] = title - - # One bad file must never abort the whole library scan. - try: - resolved = _resolve_library_file_path(tr.get('file_path')) - info = read_embedded_tags(resolved) if resolved else {'available': False} - tags = info.get('tags') if info.get('available') else None - - result = reconcile_track_row(cur, tr, album_map, artist_map, tags) - - with _reconcile_ids_lock: - if not result.readable: - _reconcile_ids_state['unreadable'] += 1 - else: - _reconcile_ids_state['entities_updated'] += result.applied.rows_updated - _reconcile_ids_state['ids_filled'] += result.applied.ids_filled - _reconcile_ids_state['conflicts'] += result.conflicts - except Exception as _te: - logger.debug("reconcile: skipped track %s: %s", tr.get('id'), _te) - with _reconcile_ids_lock: - _reconcile_ids_state['unreadable'] += 1 - finally: - with _reconcile_ids_lock: - _reconcile_ids_state['processed'] += 1 - - # Commit per page — releases the write lock so concurrent - # enrichment workers aren't starved during a long scan. - conn.commit() + # Whole-library backfill (track_ids=None). Shares the exact + # orchestration used by the on-scan auto-reconcile hook. + _reconcile_library_tracks(conn, on_progress=_on_progress) except Exception as e: logger.error(f"Reconcile embedded IDs background error: {e}") finally: if conn is not None: try: conn.close() - except Exception: + except Exception: # noqa: S110 — connection cleanup, nothing to recover pass with _reconcile_ids_lock: _reconcile_ids_state['status'] = 'done' @@ -15193,6 +15219,10 @@ def _run_db_update_task(full_refresh, server_type): # This is a blocking call that runs the worker logic db_update_worker.run() + # Auto-reconcile: pull embedded provider IDs from newly-added files into + # the DB so enrichment workers skip those API lookups (#tag-id-backfill). + _reconcile_after_scan(db_update_worker) + def _run_deep_scan_task(server_type): """Run a deep library scan in the background thread.""" @@ -15241,6 +15271,9 @@ def _run_deep_scan_task(server_type): # Run deep scan instead of normal run() db_update_worker.run_deep_scan() + # Auto-reconcile newly-inserted files' embedded provider IDs into the DB. + _reconcile_after_scan(db_update_worker) + @app.route('/api/database/stats', methods=['GET']) def get_database_stats():