danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

import page: share ONE staging scan across files/groups/hints (#935)

Browse source 
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.
This commit is contained in:
BoulderBadgeDad 2026-06-28 09:49:03 -07:00
parent 77e3673c9c
commit bcf99d76d3
2 changed files with 160 additions and 74 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

178
core/imports/routes.py
View file

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

56
tests/imports/test_import_routes.py
View file

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