From bcf99d76d3f6a2f912708686c2ec1293fba047f7 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Sun, 28 Jun 2026 09:49:03 -0700 Subject: [PATCH] import page: share ONE staging scan across files/groups/hints (#935) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ramonskie's thread finding: opening the Import page fires staging files + groups + hints together, and each one independently os.walk'd the whole staging folder AND mutagen-read every file's tags — 3x the directory walk and 3x the per-file tag I/O on every page open (the import scan storm + memory spike on large staging folders). they all need the same per-file tag data, so scan ONCE: _scan_staging_records walks staging and reads each file's metadata a single time, returning per-file records that files/groups/hints all derive from in-memory. a short TTL (6s) + a lock means the three near-simultaneous page-open requests share one scan instead of each kicking off a full re-read; the lock also prevents concurrent full scans. hints now derives from the same read_staging_file_metadata the other two use (same underlying tags) instead of a separate read_tags pass. album/singles process drop the cache on completion so the list updates immediately after files leave staging. net: 1 walk + 1 tag-read-per-file on page open instead of 3. 2 tests (shared-scan: 3 endpoints = 1 read per file, not 3; hints updated to the shared reader) + autouse cache-clear fixture; 695 import/staging tests green. --- core/imports/routes.py | 178 +++++++++++++++++----------- tests/imports/test_import_routes.py | 56 ++++++++- 2 files changed, 160 insertions(+), 74 deletions(-) diff --git a/core/imports/routes.py b/core/imports/routes.py index 39674bba..09043aed 100644 --- a/core/imports/routes.py +++ b/core/imports/routes.py @@ -3,6 +3,8 @@ from __future__ import annotations import os +import threading +import time import uuid from concurrent.futures import as_completed from dataclasses import dataclass @@ -71,36 +73,86 @@ class ImportRouteRuntime: logger: Any = module_logger +# ── Shared staging scan ────────────────────────────────────────────────────── +# Opening the Import page fires staging files/groups/hints together; each used to +# os.walk the whole staging folder AND mutagen-read every file independently — 3× +# the directory walk + 3× the tag I/O on every page open (the import-page scan +# storm + memory spike, issue #935). They all need the same per-file tag data, so +# scan ONCE and let all three derive their views in-memory. A short TTL + a lock +# means the three near-simultaneous page-open requests (and any concurrent caller) +# share a single scan instead of each kicking off a full re-read. +_STAGING_SCAN_LOCK = threading.Lock() +_STAGING_SCAN_TTL = 6.0 # seconds — covers the page-open burst; re-scans after +_staging_scan_cache: Dict[str, Any] = {"path": None, "ts": 0.0, "records": None} + + +def _scan_staging_records(runtime: ImportRouteRuntime, staging_path: str) -> list[Dict[str, Any]]: + """Walk staging + read each audio file's tags ONCE, returning per-file records + that staging files/groups/hints all derive from. Briefly cached + locked so the + page-open trio shares a single scan rather than each re-walking and re-reading.""" + now = time.time() + cached = _staging_scan_cache + if (cached["records"] is not None and cached["path"] == staging_path + and (now - cached["ts"]) < _STAGING_SCAN_TTL): + return cached["records"] + + with _STAGING_SCAN_LOCK: + # Double-check: another request may have filled the cache while we waited. + now = time.time() + if (cached["records"] is not None and cached["path"] == staging_path + and (now - cached["ts"]) < _STAGING_SCAN_TTL): + return cached["records"] + + records: list[Dict[str, Any]] = [] + if os.path.isdir(staging_path): + for root, _dirs, filenames in os.walk(staging_path): + rel_dir = os.path.relpath(root, staging_path) + top_folder = rel_dir.split(os.sep)[0] if rel_dir != "." else None + for fname in filenames: + ext = os.path.splitext(fname)[1].lower() + if ext not in AUDIO_EXTENSIONS: + continue + full_path = os.path.join(root, fname) + rel_path = os.path.relpath(full_path, staging_path) + meta = runtime.read_staging_file_metadata(full_path, rel_path) + records.append({ + "filename": fname, "rel_path": rel_path, "full_path": full_path, + "extension": ext, "title": meta["title"], "album": meta["album"], + "artist": meta["artist"], "albumartist": meta["albumartist"], + "track_number": meta["track_number"], "disc_number": meta["disc_number"], + "top_folder": top_folder, + }) + + _staging_scan_cache.update({"path": staging_path, "ts": time.time(), "records": records}) + return records + + +def invalidate_staging_scan_cache() -> None: + """Drop the cached staging scan (call after an import moves/removes files so the + next files/groups/hints request reflects the new state immediately).""" + _staging_scan_cache.update({"path": None, "ts": 0.0, "records": None}) + + def staging_files(runtime: ImportRouteRuntime) -> tuple[Dict[str, Any], int]: """Scan the staging folder and return audio files with tag metadata.""" try: staging_path = runtime.get_staging_path() os.makedirs(staging_path, exist_ok=True) - files = [] - for root, _dirs, filenames in os.walk(staging_path): - for fname in filenames: - ext = os.path.splitext(fname)[1].lower() - if ext not in AUDIO_EXTENSIONS: - continue - full_path = os.path.join(root, fname) - rel_path = os.path.relpath(full_path, staging_path) - - meta = runtime.read_staging_file_metadata(full_path, rel_path) - - files.append( - { - "filename": fname, - "rel_path": rel_path, - "full_path": full_path, - "title": meta["title"], - "artist": meta["albumartist"] or meta["artist"] or "Unknown Artist", - "album": meta["album"], - "track_number": meta["track_number"], - "disc_number": meta["disc_number"], - "extension": ext, - } - ) + files = [ + { + "filename": r["filename"], + "rel_path": r["rel_path"], + "full_path": r["full_path"], + "title": r["title"], + "artist": r["albumartist"] or r["artist"] or "Unknown Artist", + "album": r["album"], + "track_number": r["track_number"], + "disc_number": r["disc_number"], + "extension": r["extension"], + } + for r in _scan_staging_records(runtime, staging_path) + ] files.sort(key=lambda f: f["filename"].lower()) return {"success": True, "files": files, "staging_path": staging_path}, 200 @@ -117,31 +169,23 @@ def staging_groups(runtime: ImportRouteRuntime) -> tuple[Dict[str, Any], int]: return {"success": True, "groups": []}, 200 album_groups = {} - for root, _dirs, filenames in os.walk(staging_path): - for fname in filenames: - ext = os.path.splitext(fname)[1].lower() - if ext not in AUDIO_EXTENSIONS: - continue - full_path = os.path.join(root, fname) - rel_path = os.path.relpath(full_path, staging_path) + for r in _scan_staging_records(runtime, staging_path): + album = r["album"] + artist = r["albumartist"] or r["artist"] + if not album or not artist: + continue - meta = runtime.read_staging_file_metadata(full_path, rel_path) - album = meta["album"] - artist = meta["albumartist"] or meta["artist"] - if not album or not artist: - continue - - key = (album.lower().strip(), artist.lower().strip()) - if key not in album_groups: - album_groups[key] = {"album": album.strip(), "artist": artist.strip(), "files": []} - album_groups[key]["files"].append( - { - "filename": fname, - "full_path": full_path, - "title": meta["title"], - "track_number": meta["track_number"], - } - ) + key = (album.lower().strip(), artist.lower().strip()) + if key not in album_groups: + album_groups[key] = {"album": album.strip(), "artist": artist.strip(), "files": []} + album_groups[key]["files"].append( + { + "filename": r["filename"], + "full_path": r["full_path"], + "title": r["title"], + "track_number": r["track_number"], + } + ) groups = [] for group in album_groups.values(): @@ -173,28 +217,15 @@ def staging_hints(runtime: ImportRouteRuntime) -> tuple[Dict[str, Any], int]: tag_albums = {} folder_hints = {} - for root, _dirs, filenames in os.walk(staging_path): - audio_files = [f for f in filenames if os.path.splitext(f)[1].lower() in AUDIO_EXTENSIONS] - if not audio_files: - continue + for r in _scan_staging_records(runtime, staging_path): + if r["top_folder"]: + folder_hints[r["top_folder"]] = folder_hints.get(r["top_folder"], 0) + 1 - rel_dir = os.path.relpath(root, staging_path) - if rel_dir != ".": - top_folder = rel_dir.split(os.sep)[0] - folder_hints[top_folder] = folder_hints.get(top_folder, 0) + len(audio_files) - - for fname in audio_files: - full_path = os.path.join(root, fname) - try: - tags = runtime.read_tags(full_path) - if tags: - album = (tags.get("album") or [None])[0] - artist = (tags.get("artist") or (tags.get("albumartist") or [None]))[0] - if album: - key = (album.strip(), (artist or "").strip()) - tag_albums[key] = tag_albums.get(key, 0) + 1 - except Exception as exc: - runtime.logger.debug("tag read failed: %s", exc) + album = r["album"] + artist = r["artist"] or r["albumartist"] + if album: + key = (album.strip(), (artist or "").strip()) + tag_albums[key] = tag_albums.get(key, 0) + 1 queries = [] seen_queries_lower = set() @@ -371,6 +402,11 @@ def album_process(runtime: ImportRouteRuntime, data: Dict[str, Any]) -> tuple[Di ) runtime.refresh_import_suggestions_cache() + # Files just left staging — drop the shared scan so the next files/groups/hints + # reflects reality immediately instead of waiting out the cache TTL. + if processed > 0: + invalidate_staging_scan_cache() + return {"success": True, "processed": processed, "total": len(matches), "errors": errors}, 200 except Exception as exc: runtime.logger.error("Error processing album import: %s", exc) @@ -506,6 +542,10 @@ def singles_process(runtime: ImportRouteRuntime, files: list[Dict[str, Any]]) -> ) runtime.refresh_import_suggestions_cache() + # Files just left staging — drop the shared scan so the list updates immediately. + if processed > 0: + invalidate_staging_scan_cache() + return {"success": True, "processed": processed, "total": len(files), "errors": errors}, 200 except Exception as exc: runtime.logger.error("Error processing singles import: %s", exc) diff --git a/tests/imports/test_import_routes.py b/tests/imports/test_import_routes.py index 34409e9c..d4b6823a 100644 --- a/tests/imports/test_import_routes.py +++ b/tests/imports/test_import_routes.py @@ -1,6 +1,8 @@ import os from concurrent.futures import Future +import pytest + import core.imports.routes as import_routes from core.imports.routes import ( ImportRouteRuntime, @@ -17,6 +19,15 @@ from core.imports.routes import ( ) +@pytest.fixture(autouse=True) +def _clear_staging_scan_cache(): + # The shared staging scan is cached at module level; clear it between tests so + # one test's scan can't satisfy another within the TTL. + import_routes.invalidate_staging_scan_cache() + yield + import_routes.invalidate_staging_scan_cache() + + class _FakeLogger: def __init__(self): self.debug_messages = [] @@ -181,14 +192,21 @@ def test_staging_hints_prefers_tag_queries_then_folder_queries(tmp_path): _touch(tmp_path / "Folder_Album" / "02.mp3") _touch(tmp_path / "Loose" / "track.flac") - def _read_tags(file_path): - if file_path.endswith("01.mp3") or file_path.endswith("02.mp3"): - return {"album": ["Tagged Album"], "artist": ["Tagged Artist"]} - return {} + def _empty(artist="", album="", track_number=0): + return {"title": "", "artist": artist, "albumartist": "", + "album": album, "track_number": track_number, "disc_number": 1} + + # hints now derives from the shared staging scan (read_staging_file_metadata), + # the same reader files/groups use — not a separate read_tags pass. + metadata = { + os.path.join("Folder_Album", "01.mp3"): _empty(artist="Tagged Artist", album="Tagged Album", track_number=1), + os.path.join("Folder_Album", "02.mp3"): _empty(artist="Tagged Artist", album="Tagged Album", track_number=2), + os.path.join("Loose", "track.flac"): _empty(), + } runtime = ImportRouteRuntime( get_staging_path=lambda: str(tmp_path), - read_tags=_read_tags, + read_staging_file_metadata=_metadata_for(metadata), logger=_FakeLogger(), ) @@ -606,3 +624,31 @@ def test_singles_process_requires_files(): assert status == 400 assert payload == {"success": False, "error": "No files provided"} + + +def test_staging_scan_is_shared_across_files_groups_hints(tmp_path): + """#935: opening Import fires files+groups+hints together; they must share ONE + staging scan (one walk + one tag read per file), not re-read every file 3×.""" + _touch(tmp_path / "Album" / "01.mp3") + _touch(tmp_path / "Album" / "02.mp3") + + reads = [] + + def _meta(full_path, rel_path): + reads.append(rel_path) + return {"title": "T", "artist": "Artist", "albumartist": "Artist", + "album": "Album", "track_number": 1, "disc_number": 1} + + runtime = ImportRouteRuntime( + get_staging_path=lambda: str(tmp_path), + read_staging_file_metadata=_meta, + logger=_FakeLogger(), + ) + + # All three page-open endpoints, back to back (within the cache TTL). + staging_files(runtime) + staging_groups(runtime) + staging_hints(runtime) + + # 2 files × ONE shared scan = 2 reads — not 6 (which is 2 files × 3 endpoints). + assert sorted(reads) == [os.path.join("Album", "01.mp3"), os.path.join("Album", "02.mp3")]