Auto-reconcile embedded IDs for new tracks on library scans

Extends the manual "Import IDs from File Tags" backfill so newly-scanned
files get their embedded provider IDs pulled into the DB automatically —
no button press needed to keep up with new music.

How it works:
- insert_or_update_media_track now returns 'inserted' / 'updated' / False
  (truthy-compatible; existing `if track_success` callers unaffected) so
  the scan worker can tell a genuinely new row from an update.
- DatabaseUpdateWorker collects the ids it newly INSERTED this run
  (self._new_track_ids) across all insert paths (Plex/Jellyfin/deep).
- After run()/run_deep_scan(), web_server calls _reconcile_after_scan(),
  which gap-fills embedded IDs for just those new tracks. Runs as a
  post-scan pass (the scan loop itself is untouched/fast — the media
  server API never exposes these custom IDs, so the file must be read
  once regardless; batching at the end keeps it out of the hot loop and
  best-effort so it can never abort a scan). A progress phase ("Reading
  file tags for N new tracks…") surfaces the full-refresh tail.

Shared engine:
- New reconcile_library() in core does the paging + lazy parent-map
  loading (only loads albums/artists actually referenced — cheap when
  scoped to a few new tracks) + per-page commits. BOTH the manual button
  and the scan hook call it, so there's one tested orchestration, no
  duplication. The backfill job was refactored onto it.

Same hardened safety: gap-fill only, atomically guarded against
overwrite, schema-introspected, idempotent. Scoped to new arrivals for
incremental/deep; full refresh re-inserts everything as new (recovering
the IDs a full-refresh wipe destroys).

+10 reconcile tests (reconcile_library scope/idempotency/progress/stop +
the engine). Full suite clean (only pre-existing soundcloud /app env
failures remain).
This commit is contained in:
BoulderBadgeDad 2026-06-05 18:31:11 -07:00
parent e6d86dea26
commit 83c1cd92aa
5 changed files with 310 additions and 57 deletions

View file

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

View file

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

View file

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

View file

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

View file

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