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

Merge pull request #582 from Nezreka/refactor/import-processing-routes

Browse source 
Extract manual import route handlers
This commit is contained in:
BoulderBadgeDad 2026-05-13 21:31:32 -07:00 committed by GitHub
commit b80567672e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 707 additions and 318 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

329
core/imports/routes.py
View file

@ -3,14 +3,23 @@
from __future__ import annotations
import os
import uuid
from concurrent.futures import as_completed
from dataclasses import dataclass
from typing import Any, Callable, Dict
from core.imports.album import build_album_import_context, build_album_import_match_payload, resolve_album_artist_context
from core.imports.context import get_import_context_artist, get_import_track_info, normalize_import_context
from core.imports.filename import parse_filename_metadata
from core.imports.staging import (
AUDIO_EXTENSIONS,
get_import_suggestions_cache,
get_primary_source as _get_primary_source,
get_staging_path as _get_staging_path,
read_staging_file_metadata as _read_staging_file_metadata,
refresh_import_suggestions_cache as _refresh_import_suggestions_cache,
search_import_albums as _search_import_albums,
search_import_tracks as _search_import_tracks,
)
from utils.logging_config import get_logger
@ -24,6 +33,12 @@ def _default_read_tags(file_path: str):
return MutagenFile(file_path, easy=True)
def _get_single_track_import_context(*args, **kwargs):
from core.imports.resolution import get_single_track_import_context
return get_single_track_import_context(*args, **kwargs)
@dataclass
class ImportRouteRuntime:
"""Dependencies needed to service import/staging HTTP endpoints."""
@ -31,6 +46,25 @@ class ImportRouteRuntime:
get_staging_path: Callable[[], str] = _get_staging_path
read_staging_file_metadata: Callable[[str, str], Dict[str, Any]] = _read_staging_file_metadata
read_tags: Callable[[str], Any] = _default_read_tags
get_primary_source: Callable[[], str] = _get_primary_source
search_import_albums: Callable[..., list] = _search_import_albums
search_import_tracks: Callable[..., list] = _search_import_tracks
build_album_import_match_payload: Callable[..., Dict[str, Any]] = build_album_import_match_payload
resolve_album_artist_context: Callable[..., Any] = resolve_album_artist_context
build_album_import_context: Callable[..., Dict[str, Any]] = build_album_import_context
get_single_track_import_context: Callable[..., Dict[str, Any]] = _get_single_track_import_context
parse_filename_metadata: Callable[[str], Dict[str, Any]] = parse_filename_metadata
normalize_import_context: Callable[[Dict[str, Any]], Dict[str, Any]] = normalize_import_context
get_import_context_artist: Callable[[Dict[str, Any]], Dict[str, Any]] = get_import_context_artist
get_import_track_info: Callable[[Dict[str, Any]], Dict[str, Any]] = get_import_track_info
process_single_import_file: Callable[["ImportRouteRuntime", Dict[str, Any]], tuple[str, str]] | None = None
post_process_matched_download: Callable[[str, Dict[str, Any], str], Any] | None = None
add_activity_item: Callable[[Any, Any, Any, Any], Any] | None = None
refresh_import_suggestions_cache: Callable[[], Any] = _refresh_import_suggestions_cache
automation_engine: Any = None
hydrabase_worker: Any = None
dev_mode_enabled: bool = False
import_singles_executor: Any = None
logger: Any = module_logger
@ -184,3 +218,298 @@ def staging_suggestions() -> tuple[Dict[str, Any], int]:
"""Return cached import suggestions and readiness state."""
cache = get_import_suggestions_cache()
return {"success": True, "suggestions": cache["suggestions"], "ready": cache["built"]}, 200
def search_albums(runtime: ImportRouteRuntime, query: str, limit: int = 12) -> tuple[Dict[str, Any], int]:
"""Search albums for manual import using the active metadata provider."""
try:
query = (query or "").strip()
if not query:
return {"success": False, "error": "Missing query parameter"}, 400
limit = min(int(limit), 50)
if runtime.get_primary_source() == "hydrabase" and runtime.hydrabase_worker and runtime.dev_mode_enabled:
runtime.hydrabase_worker.enqueue(query, "albums")
albums = runtime.search_import_albums(query, limit=limit)
return {"success": True, "albums": albums}, 200
except Exception as exc:
runtime.logger.error("Error searching albums for import: %s", exc)
return {"success": False, "error": str(exc)}, 500
def album_match(runtime: ImportRouteRuntime, data: Dict[str, Any]) -> tuple[Dict[str, Any], int]:
"""Match staging files to an album's tracklist."""
try:
data = data or {}
album_id = data.get("album_id")
album_name = data.get("album_name", "")
album_artist = data.get("album_artist", "")
source = str(data.get("source") or "").strip().lower()
filter_file_paths = set(data.get("file_paths", []))
if not album_id:
return {"success": False, "error": "Missing album_id"}, 400
if not source:
runtime.logger.warning(
"[Import Match] Missing 'source' on album_id=%s - lookup will "
"guess via primary-source priority chain. If this fires "
"consistently, a frontend caller is dropping source from "
"the match POST body.",
album_id,
)
payload = runtime.build_album_import_match_payload(
album_id,
album_name=album_name,
album_artist=album_artist,
file_paths=filter_file_paths,
source=source or None,
)
return payload, 200
except Exception as exc:
runtime.logger.error("Error matching album for import: %s", exc)
return {"success": False, "error": str(exc)}, 500
def album_process(runtime: ImportRouteRuntime, data: Dict[str, Any]) -> tuple[Dict[str, Any], int]:
"""Process matched album files through the post-processing pipeline."""
try:
data = data or {}
album = data.get("album", {})
matches = data.get("matches", [])
if not album or not matches:
return {"success": False, "error": "Missing album or matches data"}, 400
if runtime.post_process_matched_download is None:
return {"success": False, "error": "Import post-processing not available"}, 500
processed = 0
errors = []
album_name = album.get("name", album.get("album_name", "Unknown Album"))
artist_name = album.get("artist", album.get("artist_name", "Unknown Artist"))
album_id = album.get("id", album.get("album_id", ""))
source = str(album.get("source") or data.get("source") or "").strip().lower()
total_discs = max(
(
match.get("track", {}).get("disc_number", 1)
for match in matches
if match.get("track")
),
default=1,
)
artist_context = runtime.resolve_album_artist_context(album, source=source)
for match in matches:
staging_file = match.get("staging_file")
track = match.get("track") or {}
if not staging_file or not track:
continue
file_path = staging_file.get("full_path", "")
if not os.path.isfile(file_path):
errors.append(f"File not found: {staging_file.get('filename', '?')}")
continue
track_name = track.get("name", "Unknown Track")
track_number = track.get("track_number", 1)
context_key = f"import_album_{album_id}_{track_number}_{uuid.uuid4().hex[:8]}"
context = runtime.build_album_import_context(
album,
track,
artist_context=artist_context,
total_discs=total_discs,
source=source,
)
try:
runtime.post_process_matched_download(context_key, context, file_path)
processed += 1
runtime.logger.info("Import processed: %s. %s from %s", track_number, track_name, album_name)
except Exception as proc_err:
err_msg = f"{track_name}: {str(proc_err)}"
errors.append(err_msg)
runtime.logger.error("Import processing error: %s", err_msg)
if runtime.add_activity_item:
runtime.add_activity_item("", "Album Imported", f"{album_name} by {artist_name} ({processed}/{len(matches)} tracks)", "Now")
if processed > 0:
_emit_import_completed(
runtime,
track_count=processed,
album_name=album_name or "",
artist=artist_name or "",
playlist_name=f"Import: {album_name}" if album_name else "Import",
total_tracks=len(matches),
failed_tracks=len(errors),
log_label="album",
)
runtime.refresh_import_suggestions_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)
return {"success": False, "error": str(exc)}, 500
def search_tracks(runtime: ImportRouteRuntime, query: str, limit: int = 10) -> tuple[Dict[str, Any], int]:
"""Search tracks for manual single import using metadata source priority."""
try:
query = (query or "").strip()
if not query:
return {"success": False, "error": "Missing query parameter"}, 400
limit = min(int(limit), 30)
if runtime.get_primary_source() == "hydrabase" and runtime.hydrabase_worker and runtime.dev_mode_enabled:
runtime.hydrabase_worker.enqueue(query, "tracks")
tracks = runtime.search_import_tracks(query, limit=limit)
return {"success": True, "tracks": tracks}, 200
except Exception as exc:
runtime.logger.error("Error searching tracks for import: %s", exc)
return {"success": False, "error": str(exc)}, 500
def process_single_import_file(runtime: ImportRouteRuntime, file_info: Dict[str, Any]) -> tuple[str, str]:
"""Validate, resolve metadata, and post-process one single import file."""
file_path = file_info.get("full_path", "")
if not os.path.isfile(file_path):
return ("error", f"File not found: {file_info.get('filename', '?')}")
if runtime.post_process_matched_download is None:
return ("error", "Import post-processing not available")
title = file_info.get("title", "")
artist = file_info.get("artist", "")
manual_match = file_info.get("manual_match")
if manual_match is not None and not isinstance(manual_match, dict):
manual_match = None
manual_match_source = ""
manual_match_id = None
if manual_match:
manual_match_source = str(manual_match.get("source") or "").strip().lower()
manual_match_id = str(manual_match.get("id") or "").strip()
if not manual_match_id or not manual_match_source:
return ("error", f"Malformed manual match for file: {file_info.get('filename', '?')}")
if not title and not manual_match:
parsed = runtime.parse_filename_metadata(file_info.get("filename", ""))
title = parsed.get("title") or os.path.splitext(file_info.get("filename", "Unknown"))[0]
if not artist:
artist = parsed.get("artist", "")
try:
resolved = runtime.get_single_track_import_context(
title,
artist,
override_id=manual_match_id,
override_source=manual_match_source,
)
context = runtime.normalize_import_context(resolved["context"])
artist_data = runtime.get_import_context_artist(context)
track_data = runtime.get_import_track_info(context)
final_title = track_data.get("name", title)
final_artist = artist_data.get("name", artist)
context_key = f"import_single_{uuid.uuid4().hex[:8]}"
runtime.post_process_matched_download(context_key, context, file_path)
runtime.logger.info(
"Import single processed: %s by %s (source=%s)",
final_title,
final_artist,
resolved.get("source") or "local",
)
return ("ok", final_title)
except Exception as proc_err:
err_msg = f"{title}: {str(proc_err)}"
runtime.logger.error("Import single processing error: %s", err_msg)
return ("error", err_msg)
def singles_process(runtime: ImportRouteRuntime, files: list[Dict[str, Any]]) -> tuple[Dict[str, Any], int]:
"""Process individual staging files as singles through the import pipeline."""
try:
files = files or []
if not files:
return {"success": False, "error": "No files provided"}, 400
if runtime.import_singles_executor is None:
return {"success": False, "error": "Import executor not available"}, 500
processed = 0
errors = []
process_file = runtime.process_single_import_file or process_single_import_file
future_to_filename = {
runtime.import_singles_executor.submit(process_file, runtime, file_info):
file_info.get("filename", "?")
for file_info in files
}
for future in as_completed(future_to_filename):
try:
outcome, payload = future.result()
except Exception as worker_err:
errors.append(f"{future_to_filename[future]}: worker crashed: {worker_err}")
continue
if outcome == "ok":
processed += 1
else:
errors.append(payload)
if runtime.add_activity_item:
runtime.add_activity_item("", "Singles Imported", f"{processed}/{len(files)} tracks processed", "Now")
if processed > 0:
_emit_import_completed(
runtime,
track_count=processed,
album_name="",
artist="Various",
playlist_name="Import: Singles",
total_tracks=len(files),
failed_tracks=len(errors),
log_label="singles",
)
runtime.refresh_import_suggestions_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)
return {"success": False, "error": str(exc)}, 500
def _emit_import_completed(
runtime: ImportRouteRuntime,
*,
track_count: int,
album_name: str,
artist: str,
playlist_name: str,
total_tracks: int,
failed_tracks: int,
log_label: str,
) -> None:
# Keep import automation on the same chain as download batches:
# batch_complete -> auto-scan -> library_scan_completed -> auto-update DB.
try:
if runtime.automation_engine:
runtime.automation_engine.emit(
"import_completed",
{
"track_count": str(track_count),
"album_name": album_name,
"artist": artist,
},
)
runtime.automation_engine.emit(
"batch_complete",
{
"playlist_name": playlist_name,
"total_tracks": str(total_tracks),
"completed_tracks": str(track_count),
"failed_tracks": str(failed_tracks),
},
)
except Exception as exc:
runtime.logger.debug("%s import automation emit failed: %s", log_label, exc)

340
tests/imports/test_import_routes.py
View file

@ -1,13 +1,28 @@
import os
from concurrent.futures import Future
import core.imports.routes as import_routes
from core.imports.routes import ImportRouteRuntime, staging_files, staging_groups, staging_hints, staging_suggestions
from core.imports.routes import (
ImportRouteRuntime,
album_match,
album_process,
process_single_import_file,
search_albums,
search_tracks,
singles_process,
staging_files,
staging_groups,
staging_hints,
staging_suggestions,
)
class _FakeLogger:
def __init__(self):
self.debug_messages = []
self.error_messages = []
self.info_messages = []
self.warning_messages = []
def debug(self, msg, *args):
self.debug_messages.append(msg % args if args else msg)
@ -15,6 +30,50 @@ class _FakeLogger:
def error(self, msg, *args):
self.error_messages.append(msg % args if args else msg)
def info(self, msg, *args):
self.info_messages.append(msg % args if args else msg)
def warning(self, msg, *args):
self.warning_messages.append(msg % args if args else msg)
class _FakeHydrabaseWorker:
def __init__(self):
self.enqueued = []
def enqueue(self, query, kind):
self.enqueued.append((query, kind))
class _FakeAutomationEngine:
def __init__(self, fail=False):
self.events = []
self.fail = fail
def emit(self, name, payload):
if self.fail:
raise RuntimeError("emit boom")
self.events.append((name, payload))
class _FakeExecutor:
def __init__(self, outcomes=None):
self.outcomes = list(outcomes or [])
self.calls = []
def submit(self, fn, *args):
self.calls.append((fn, args))
future = Future()
if self.outcomes:
outcome = self.outcomes.pop(0)
if isinstance(outcome, BaseException):
future.set_exception(outcome)
else:
future.set_result(outcome)
else:
future.set_result(fn(*args))
return future
def _touch(path):
path.parent.mkdir(parents=True, exist_ok=True)
@ -228,3 +287,282 @@ def test_staging_hints_returns_error_when_path_resolution_fails():
assert payload["success"] is False
assert payload["error"] == "hint boom"
assert logger.error_messages == ["Error getting staging hints: hint boom"]
def test_search_albums_enqueues_hydrabase_and_caps_limit():
worker = _FakeHydrabaseWorker()
calls = []
runtime = ImportRouteRuntime(
get_primary_source=lambda: "hydrabase",
hydrabase_worker=worker,
dev_mode_enabled=True,
search_import_albums=lambda query, limit: calls.append((query, limit)) or [{"id": "album-1"}],
logger=_FakeLogger(),
)
payload, status = search_albums(runtime, " Album ", 99)
assert status == 200
assert payload == {"success": True, "albums": [{"id": "album-1"}]}
assert worker.enqueued == [("Album", "albums")]
assert calls == [("Album", 50)]
def test_search_albums_requires_query():
payload, status = search_albums(ImportRouteRuntime(logger=_FakeLogger()), "", 12)
assert status == 400
assert payload == {"success": False, "error": "Missing query parameter"}
def test_search_tracks_enqueues_hydrabase_and_caps_limit():
worker = _FakeHydrabaseWorker()
calls = []
runtime = ImportRouteRuntime(
get_primary_source=lambda: "hydrabase",
hydrabase_worker=worker,
dev_mode_enabled=True,
search_import_tracks=lambda query, limit: calls.append((query, limit)) or [{"id": "track-1"}],
logger=_FakeLogger(),
)
payload, status = search_tracks(runtime, " Track ", 99)
assert status == 200
assert payload == {"success": True, "tracks": [{"id": "track-1"}]}
assert worker.enqueued == [("Track", "tracks")]
assert calls == [("Track", 30)]
def test_album_match_warns_without_source_and_passes_file_filter():
logger = _FakeLogger()
calls = []
runtime = ImportRouteRuntime(
build_album_import_match_payload=lambda *args, **kwargs: calls.append((args, kwargs)) or {"success": True, "matches": []},
logger=logger,
)
payload, status = album_match(
runtime,
{
"album_id": "album-1",
"album_name": "Album",
"album_artist": "Artist",
"file_paths": ["a.flac", "b.flac"],
},
)
assert status == 200
assert payload == {"success": True, "matches": []}
assert calls == [
(
("album-1",),
{
"album_name": "Album",
"album_artist": "Artist",
"file_paths": {"a.flac", "b.flac"},
"source": None,
},
)
]
assert len(logger.warning_messages) == 1
assert "Missing 'source'" in logger.warning_messages[0]
def test_album_match_requires_album_id():
payload, status = album_match(ImportRouteRuntime(logger=_FakeLogger()), {})
assert status == 400
assert payload == {"success": False, "error": "Missing album_id"}
def test_album_process_posts_valid_files_and_records_side_effects(tmp_path):
good_file = tmp_path / "good.flac"
_touch(good_file)
processed_contexts = []
activity = []
refresh_calls = []
automation = _FakeAutomationEngine()
runtime = ImportRouteRuntime(
resolve_album_artist_context=lambda album, source="": {"name": "Artist"},
build_album_import_context=lambda album, track, **kwargs: {"album": album, "track": track, **kwargs},
post_process_matched_download=lambda key, context, path: processed_contexts.append((key, context, path)),
add_activity_item=lambda *args: activity.append(args),
refresh_import_suggestions_cache=lambda: refresh_calls.append("refresh"),
automation_engine=automation,
logger=_FakeLogger(),
)
payload, status = album_process(
runtime,
{
"album": {"id": "album-1", "name": "Album", "artist": "Artist", "source": "deezer"},
"matches": [
{
"staging_file": {"full_path": str(good_file), "filename": "good.flac"},
"track": {"name": "Good Track", "track_number": 1, "disc_number": 2},
},
{
"staging_file": {"full_path": str(tmp_path / "missing.flac"), "filename": "missing.flac"},
"track": {"name": "Missing Track", "track_number": 2},
},
],
},
)
assert status == 200
assert payload == {
"success": True,
"processed": 1,
"total": 2,
"errors": ["File not found: missing.flac"],
}
assert len(processed_contexts) == 1
key, context, path = processed_contexts[0]
assert key.startswith("import_album_album-1_1_")
assert context["artist_context"] == {"name": "Artist"}
assert context["total_discs"] == 2
assert context["source"] == "deezer"
assert path == str(good_file)
assert activity == [("", "Album Imported", "Album by Artist (1/2 tracks)", "Now")]
assert refresh_calls == ["refresh"]
assert automation.events == [
(
"import_completed",
{"track_count": "1", "album_name": "Album", "artist": "Artist"},
),
(
"batch_complete",
{
"playlist_name": "Import: Album",
"total_tracks": "2",
"completed_tracks": "1",
"failed_tracks": "1",
},
),
]
def test_album_process_requires_album_and_matches():
payload, status = album_process(ImportRouteRuntime(logger=_FakeLogger()), {"album": {}, "matches": []})
assert status == 400
assert payload == {"success": False, "error": "Missing album or matches data"}
def test_process_single_import_file_resolves_and_posts_context(tmp_path):
audio_file = tmp_path / "Artist - Song.flac"
_touch(audio_file)
post_calls = []
runtime = ImportRouteRuntime(
parse_filename_metadata=lambda filename: {"title": "Song", "artist": "Artist"},
get_single_track_import_context=lambda title, artist, **kwargs: {
"source": "deezer",
"context": {"track": {"name": title}, "artist": {"name": artist}},
},
normalize_import_context=lambda context: context,
get_import_context_artist=lambda context: context["artist"],
get_import_track_info=lambda context: context["track"],
post_process_matched_download=lambda key, context, path: post_calls.append((key, context, path)),
logger=_FakeLogger(),
)
outcome = process_single_import_file(runtime, {"full_path": str(audio_file), "filename": audio_file.name})
assert outcome == ("ok", "Song")
assert len(post_calls) == 1
assert post_calls[0][0].startswith("import_single_")
assert post_calls[0][1] == {"track": {"name": "Song"}, "artist": {"name": "Artist"}}
assert post_calls[0][2] == str(audio_file)
def test_process_single_import_file_rejects_malformed_manual_match(tmp_path):
audio_file = tmp_path / "Song.flac"
_touch(audio_file)
runtime = ImportRouteRuntime(post_process_matched_download=lambda *_args: None, logger=_FakeLogger())
outcome = process_single_import_file(
runtime,
{"full_path": str(audio_file), "filename": audio_file.name, "manual_match": {"id": "track-1"}},
)
assert outcome == ("error", "Malformed manual match for file: Song.flac")
def test_singles_process_aggregates_worker_results_and_side_effects():
activity = []
refresh_calls = []
automation = _FakeAutomationEngine()
executor = _FakeExecutor(outcomes=[("ok", "Song"), ("error", "Bad Song")])
runtime = ImportRouteRuntime(
import_singles_executor=executor,
add_activity_item=lambda *args: activity.append(args),
refresh_import_suggestions_cache=lambda: refresh_calls.append("refresh"),
automation_engine=automation,
logger=_FakeLogger(),
)
payload, status = singles_process(
runtime,
[{"filename": "a.flac"}, {"filename": "b.flac"}],
)
assert status == 200
assert payload == {
"success": True,
"processed": 1,
"total": 2,
"errors": ["Bad Song"],
}
assert len(executor.calls) == 2
assert activity == [("", "Singles Imported", "1/2 tracks processed", "Now")]
assert refresh_calls == ["refresh"]
assert automation.events == [
(
"import_completed",
{"track_count": "1", "album_name": "", "artist": "Various"},
),
(
"batch_complete",
{
"playlist_name": "Import: Singles",
"total_tracks": "2",
"completed_tracks": "1",
"failed_tracks": "1",
},
),
]
def test_singles_process_uses_injected_single_file_worker():
calls = []
executor = _FakeExecutor()
def fake_process_file(runtime, file_info):
calls.append((runtime, file_info))
return ("ok", file_info["filename"])
runtime = ImportRouteRuntime(
import_singles_executor=executor,
process_single_import_file=fake_process_file,
logger=_FakeLogger(),
)
payload, status = singles_process(runtime, [{"filename": "patched.flac"}])
assert status == 200
assert payload == {
"success": True,
"processed": 1,
"total": 1,
"errors": [],
}
assert calls == [(runtime, {"filename": "patched.flac"})]
assert executor.calls[0][0] is fake_process_file
def test_singles_process_requires_files():
payload, status = singles_process(ImportRouteRuntime(logger=_FakeLogger()), [])
assert status == 400
assert payload == {"success": False, "error": "No files provided"}

356
web_server.py
View file

@ -115,10 +115,7 @@ from core.imports.context import (
get_import_clean_album,
get_import_clean_title,
get_import_context_album,
get_import_context_artist,
get_import_original_search,
get_import_track_info,
normalize_import_context,
)
from core.wishlist.payloads import (
build_cancelled_task_wishlist_payload as _build_cancelled_task_wishlist_payload,
@ -161,23 +158,21 @@ from core.wishlist.state import (
is_wishlist_actually_processing as _is_wishlist_actually_processing,
reset_flag_if_stuck as _reset_wishlist_flag_if_stuck,
)
from core.imports.album import (
build_album_import_context,
build_album_import_match_payload,
resolve_album_artist_context,
)
from core.imports.album_naming import resolve_album_group as _resolve_album_group
from core.imports.album import build_album_import_match_payload
from core.imports.filename import extract_track_number_from_filename, parse_filename_metadata
from core.imports.staging import (
get_primary_source,
get_staging_path,
read_staging_file_metadata,
refresh_import_suggestions_cache,
search_import_albums,
search_import_tracks,
start_import_suggestions_cache,
)
from core.imports.routes import ImportRouteRuntime as _ImportRouteRuntime
from core.imports.routes import album_match as _import_album_match
from core.imports.routes import album_process as _import_album_process
from core.imports.routes import process_single_import_file as _import_process_single_import_file
from core.imports.routes import search_albums as _import_search_albums
from core.imports.routes import search_tracks as _import_search_tracks
from core.imports.routes import singles_process as _import_singles_process
from core.imports.routes import staging_files as _import_staging_files
from core.imports.routes import staging_groups as _import_staging_groups
from core.imports.routes import staging_hints as _import_staging_hints
@ -34269,7 +34264,17 @@ def repair_job_progress():
# ================================================================================================
def _build_import_route_runtime():
return _ImportRouteRuntime(logger=logger)
return _ImportRouteRuntime(
post_process_matched_download=_post_process_matched_download,
add_activity_item=add_activity_item,
automation_engine=automation_engine,
hydrabase_worker=hydrabase_worker,
dev_mode_enabled=dev_mode_enabled,
import_singles_executor=import_singles_executor,
build_album_import_match_payload=build_album_import_match_payload,
process_single_import_file=lambda runtime, file_info: _process_single_import_file(file_info),
logger=logger,
)
@app.route('/api/import/staging/files', methods=['GET'])
@ -34292,331 +34297,48 @@ def import_staging_hints():
@app.route('/api/import/search/albums', methods=['GET'])
def import_search_albums():
"""Search for albums using the active metadata provider."""
try:
query = request.args.get('q', '').strip()
if not query:
return jsonify({'success': False, 'error': 'Missing query parameter'}), 400
limit = min(int(request.args.get('limit', 12)), 50)
from core.metadata.registry import get_primary_source
if get_primary_source() == 'hydrabase' and hydrabase_worker and dev_mode_enabled:
hydrabase_worker.enqueue(query, 'albums')
albums = search_import_albums(query, limit=limit)
return jsonify({'success': True, 'albums': albums})
except Exception as e:
logger.error(f"Error searching albums for import: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
payload, status = _import_search_albums(
_build_import_route_runtime(),
request.args.get('q', ''),
request.args.get('limit', 12),
)
return jsonify(payload), status
@app.route('/api/import/album/match', methods=['POST'])
def import_album_match():
"""Match staging files to an album's tracklist."""
try:
data = request.get_json() or {}
album_id = data.get('album_id')
album_name = data.get('album_name', '')
album_artist = data.get('album_artist', '')
source = str(data.get('source') or '').strip().lower()
# Optional: only match specific files (from auto-group selection)
filter_file_paths = set(data.get('file_paths', []))
if not album_id:
return jsonify({'success': False, 'error': 'Missing album_id'}), 400
# Without `source`, the lookup chain has to guess which metadata
# source the album_id came from — and a Deezer numeric id will
# match nothing in Spotify/iTunes/Discogs/etc., resulting in the
# failure-fallback dict that github issue #524 surfaced as
# "Unknown Artist / album_id-as-title / 0 tracks / 1991". Frontend
# fix in the same PR populates source on every match POST; this
# log catches anything that still reaches us without it (curl,
# third-party, regression in another caller).
if not source:
logger.warning(
"[Import Match] Missing 'source' on album_id=%s — lookup will "
"guess via primary-source priority chain. If this fires "
"consistently, a frontend caller is dropping source from "
"the match POST body.",
album_id,
)
payload = build_album_import_match_payload(
album_id,
album_name=album_name,
album_artist=album_artist,
file_paths=filter_file_paths,
source=source or None,
)
return jsonify(payload)
except Exception as e:
logger.error(f"Error matching album for import: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
payload, status = _import_album_match(_build_import_route_runtime(), request.get_json() or {})
return jsonify(payload), status
@app.route('/api/import/album/process', methods=['POST'])
def import_album_process():
"""Process matched album files through the post-processing pipeline."""
try:
data = request.get_json() or {}
album = data.get('album', {})
matches = data.get('matches', [])
if not album or not matches:
return jsonify({'success': False, 'error': 'Missing album or matches data'}), 400
processed = 0
errors = []
album_name = album.get('name', album.get('album_name', 'Unknown Album'))
artist_name = album.get('artist', album.get('artist_name', 'Unknown Artist'))
album_id = album.get('id', album.get('album_id', ''))
source = str(album.get('source') or data.get('source') or '').strip().lower()
total_discs = max(
(
match.get('track', {}).get('disc_number', 1)
for match in matches
if match.get('track')
),
default=1,
)
artist_context = resolve_album_artist_context(album, source=source)
for match in matches:
staging_file = match.get('staging_file')
track = match.get('track') or {}
if not staging_file or not track:
continue
file_path = staging_file.get('full_path', '')
if not os.path.isfile(file_path):
errors.append(f"File not found: {staging_file.get('filename', '?')}")
continue
track_name = track.get('name', 'Unknown Track')
track_number = track.get('track_number', 1)
disc_number = track.get('disc_number', 1)
context_key = f"import_album_{album_id}_{track_number}_{uuid.uuid4().hex[:8]}"
context = build_album_import_context(
album,
track,
artist_context=artist_context,
total_discs=total_discs,
source=source,
)
try:
_post_process_matched_download(context_key, context, file_path)
processed += 1
logger.info(f"Import processed: {track_number}. {track_name} from {album_name}")
except Exception as proc_err:
err_msg = f"{track_name}: {str(proc_err)}"
errors.append(err_msg)
logger.error(f"Import processing error: {err_msg}")
add_activity_item("", "Album Imported", f"{album_name} by {artist_name} ({processed}/{len(matches)} tracks)", "Now")
# Emit events through automation engine — same chain as download batches
# batch_complete → auto-scan → library_scan_completed → auto-update DB
if processed > 0:
try:
if automation_engine:
automation_engine.emit('import_completed', {
'track_count': str(processed),
'album_name': album_name or '',
'artist': artist_name or '',
})
automation_engine.emit('batch_complete', {
'playlist_name': f"Import: {album_name}" if album_name else 'Import',
'total_tracks': str(len(matches)),
'completed_tracks': str(processed),
'failed_tracks': str(len(errors)),
})
except Exception as e:
logger.debug("album import automation emit failed: %s", e)
# Rebuild suggestions cache since staging contents changed
if processed > 0:
refresh_import_suggestions_cache()
return jsonify({
'success': True,
'processed': processed,
'total': len(matches),
'errors': errors
})
except Exception as e:
logger.error(f"Error processing album import: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
payload, status = _import_album_process(_build_import_route_runtime(), request.get_json() or {})
return jsonify(payload), status
@app.route('/api/import/search/tracks', methods=['GET'])
def import_search_tracks():
"""Search tracks using the configured metadata provider priority order."""
try:
query = request.args.get('q', '').strip()
if not query:
return jsonify({'success': False, 'error': 'Missing query parameter'}), 400
limit = min(int(request.args.get('limit', 10)), 30)
if get_primary_source() == 'hydrabase' and hydrabase_worker and dev_mode_enabled:
hydrabase_worker.enqueue(query, 'tracks')
tracks = search_import_tracks(query, limit=limit)
return jsonify({'success': True, 'tracks': tracks})
except Exception as e:
logger.error(f"Error searching tracks for import: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
payload, status = _import_search_tracks(
_build_import_route_runtime(),
request.args.get('q', ''),
request.args.get('limit', 10),
)
return jsonify(payload), status
def _process_single_import_file(file_info):
"""Worker function: validate, resolve metadata, post-process one file.
Returns ``("ok", title)`` on success, ``("error", message)`` on
failure, or ``("skip", reason)`` for files that need to be reported
but didn't actually run the pipeline. The caller aggregates these.
Designed to be safe to run concurrently from a ThreadPoolExecutor
— each file gets its own UUID context_key, downstream DB writes
serialize via SQLite's busy_timeout, and file-system ops touch
distinct destination paths.
"""
file_path = file_info.get('full_path', '')
if not os.path.isfile(file_path):
return ("error", f"File not found: {file_info.get('filename', '?')}")
title = file_info.get('title', '')
artist = file_info.get('artist', '')
manual_match = file_info.get('manual_match')
if manual_match is not None and not isinstance(manual_match, dict):
manual_match = None
manual_match_source = ''
manual_match_id = None
if manual_match:
manual_match_source = str(manual_match.get('source') or '').strip().lower()
manual_match_id = str(manual_match.get('id') or '').strip()
if not manual_match_id or not manual_match_source:
return ("error", f"Malformed manual match for file: {file_info.get('filename', '?')}")
if not title and not manual_match:
parsed = parse_filename_metadata(file_info.get('filename', ''))
title = parsed.get('title') or os.path.splitext(file_info.get('filename', 'Unknown'))[0]
if not artist:
artist = parsed.get('artist', '')
from core.imports.resolution import get_single_track_import_context
try:
resolved = get_single_track_import_context(
title,
artist,
override_id=manual_match_id,
override_source=manual_match_source,
)
context = normalize_import_context(resolved['context'])
artist_data = get_import_context_artist(context)
track_data = get_import_track_info(context)
final_title = track_data.get('name', title)
final_artist = artist_data.get('name', artist)
context_key = f"import_single_{uuid.uuid4().hex[:8]}"
_post_process_matched_download(context_key, context, file_path)
logger.info(
"Import single processed: %s by %s (source=%s)",
final_title,
final_artist,
resolved.get('source') or 'local',
)
return ("ok", final_title)
except Exception as proc_err:
err_msg = f"{title}: {str(proc_err)}"
logger.error(f"Import single processing error: {err_msg}")
return ("error", err_msg)
return _import_process_single_import_file(_build_import_route_runtime(), file_info)
@app.route('/api/import/singles/process', methods=['POST'])
def import_singles_process():
"""Process individual staging files as singles through the post-processing pipeline.
Files are processed in parallel through the
``import_singles_executor`` (3 workers). Per-file work is dominated
by metadata search round-trips, so parallelizing gives a near-
linear speedup without saturating any one provider's rate limits.
"""
try:
data = request.get_json()
files = data.get('files', [])
if not files:
return jsonify({'success': False, 'error': 'No files provided'}), 400
processed = 0
errors = []
# Submit all files at once so the executor pulls 3 at a time.
# as_completed yields in finish order; we don't need ordering
# because the caller just wants a count + error list.
future_to_filename = {
import_singles_executor.submit(_process_single_import_file, file_info):
file_info.get('filename', '?')
for file_info in files
}
for future in as_completed(future_to_filename):
try:
outcome, payload = future.result()
except Exception as worker_err:
# Catch-all for anything the worker itself didn't catch
# (shouldn't happen — _process_single_import_file wraps
# its own pipeline call — but defensive).
errors.append(
f"{future_to_filename[future]}: worker crashed: {worker_err}"
)
continue
if outcome == "ok":
processed += 1
else:
errors.append(payload)
add_activity_item("", "Singles Imported", f"{processed}/{len(files)} tracks processed", "Now")
# Emit events through automation engine — same chain as download batches
# batch_complete → auto-scan → library_scan_completed → auto-update DB
if processed > 0:
try:
if automation_engine:
automation_engine.emit('import_completed', {
'track_count': str(processed),
'album_name': '',
'artist': 'Various',
})
automation_engine.emit('batch_complete', {
'playlist_name': 'Import: Singles',
'total_tracks': str(len(files)),
'completed_tracks': str(processed),
'failed_tracks': str(len(errors)),
})
except Exception as e:
logger.debug("singles import automation emit failed: %s", e)
# Rebuild suggestions cache since staging contents changed
if processed > 0:
refresh_import_suggestions_cache()
return jsonify({
'success': True,
'processed': processed,
'total': len(files),
'errors': errors
})
except Exception as e:
logger.error(f"Error processing singles import: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
data = request.get_json() or {}
payload, status = _import_singles_process(_build_import_route_runtime(), data.get('files', []))
return jsonify(payload), status
# ── Auto-Import Worker ──
# Auto-Import Worker
auto_import_worker = None
try:
from core.auto_import_worker import AutoImportWorker