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

Refine import module boundaries

Browse source 
- Move filename and staging helpers into their canonical modules
- Extract album naming and grouping from path handling
- Update import and test call sites to the new layout
This commit is contained in:
Antti Kettunen 2026-04-24 16:55:46 +03:00
parent 0bbf44809f
commit 6872e5080d
No known key found for this signature in database
GPG key ID: C6B2A3D250359BD7
10 changed files with 337 additions and 326 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

41
core/import_album.py
View file

@ -2,12 +2,10 @@
from __future__ import annotations
import os
from typing import Any, Dict, Iterable, List, Optional, Set
from core.import_context import normalize_import_context
from core.import_file_ops import read_staging_file_metadata
from core.import_staging import AUDIO_EXTENSIONS, get_staging_path
from core.import_staging import collect_staging_files
from utils.logging_config import get_logger
@ -138,41 +136,6 @@ def _coerce_track_int(value: Any, default: int = 1) -> int:
return default
def _collect_staging_files(file_paths: Optional[Iterable[str]] = None) -> List[Dict[str, Any]]:
staging_path = get_staging_path()
file_filter: Optional[Set[str]] = set(file_paths) if file_paths else None
staging_files: List[Dict[str, Any]] = []
if not os.path.isdir(staging_path):
return staging_files
for root, _dirs, filenames in os.walk(staging_path):
for filename in filenames:
ext = os.path.splitext(filename)[1].lower()
if ext not in AUDIO_EXTENSIONS:
continue
full_path = os.path.join(root, filename)
if file_filter is not None and full_path not in file_filter:
continue
meta = read_staging_file_metadata(full_path, filename)
staging_files.append(
{
"filename": filename,
"full_path": full_path,
"title": meta.get("title", ""),
"artist": meta.get("albumartist") or meta.get("artist") or "",
"album": meta.get("album", ""),
"albumartist": meta.get("albumartist") or meta.get("artist") or "",
"track_number": meta.get("track_number", 1),
"disc_number": meta.get("disc_number", 1),
}
)
return staging_files
def _normalize_match_track(track: Dict[str, Any], source: str, album: Dict[str, Any]) -> Dict[str, Any]:
track_album = track.get("album") if isinstance(track.get("album"), dict) else album
if isinstance(track_album, dict):
@ -472,7 +435,7 @@ def build_album_import_match_payload(
"resolved_album_id": album_response.get("resolved_album_id") or album_id,
}
staging_files = _collect_staging_files(file_paths)
staging_files = collect_staging_files(file_paths)
album_name_for_match = album.get("name") or album_name or ""
matches: List[Dict[str, Any]] = []
used_files: Set[int] = set()

184
core/import_album_naming.py Normal file
View file

@ -0,0 +1,184 @@
"""Album naming and grouping helpers used by import flows."""
from __future__ import annotations
import re
import threading
from typing import Any, Dict
from core.import_context import extract_artist_name
from utils.logging_config import get_logger
logger = get_logger("import_album_naming")
_album_cache_lock = threading.Lock()
_album_editions: dict[str, str] = {}
_album_name_cache: dict[str, str] = {}
def clear_album_grouping_cache() -> None:
"""Clear cached album grouping decisions.
Useful for tests and for any future config reload flows.
"""
with _album_cache_lock:
_album_editions.clear()
_album_name_cache.clear()
def get_base_album_name(album_name: str) -> str:
"""Extract the base album name without edition indicators."""
base_name = album_name or ""
base_name = re.sub(
r"\s*[\[\(][^)\]]*\b(deluxe|special|expanded|extended|bonus|remaster(?:ed)?|anniversary|collectors?|limited|silver|gold|platinum)\b[^)\]]*[\]\)]\s*$",
"",
base_name,
flags=re.IGNORECASE,
)
base_name = re.sub(r"\s*[\[\(][^)\]]*\bedition\b[^)\]]*[\]\)]\s*$", "", base_name, flags=re.IGNORECASE)
base_name = re.sub(
r"\s+(deluxe|special|expanded|extended|bonus|remastered|anniversary|collectors?|limited|silver|gold|platinum)\s*(edition)?\s*$",
"",
base_name,
flags=re.IGNORECASE,
)
return base_name.strip()
def detect_deluxe_edition(album_name: str) -> bool:
"""Detect if an album name indicates a deluxe/special edition."""
if not album_name:
return False
album_lower = album_name.lower()
deluxe_indicators = [
"deluxe",
"deluxe edition",
"special edition",
"expanded edition",
"extended edition",
"bonus",
"remastered",
"anniversary",
"collectors edition",
"limited edition",
"silver edition",
"gold edition",
"platinum edition",
]
for indicator in deluxe_indicators:
if indicator in album_lower:
logger.info("Detected deluxe edition: %r contains %r", album_name, indicator)
return True
return False
def normalize_base_album_name(base_album: str, artist_name: str) -> str:
"""Normalize the base album name to handle case variations and known corrections."""
normalized_lower = (base_album or "").lower().strip()
known_corrections = {
# Add specific album name corrections here as needed.
}
for variant, correction in known_corrections.items():
if normalized_lower == variant.lower():
logger.info("Album correction applied: %r -> %r", base_album, correction)
return correction
normalized = base_album or ""
normalized = re.sub(r"\s*&\s*", " & ", normalized)
normalized = re.sub(r"\s+", " ", normalized)
normalized = normalized.strip()
logger.info("Album variant normalization: %r -> %r", base_album, normalized)
return normalized
def clean_album_title(album_title: str, artist_name: str) -> str:
"""Clean up album title by removing common prefixes, suffixes, and artist redundancy."""
original = (album_title or "").strip()
cleaned = original
logger.info("Album Title Cleaning: %r (artist: %r)", original, artist_name)
cleaned = re.sub(r"^Album\s*-\s*", "", cleaned, flags=re.IGNORECASE)
artist_pattern = re.escape(artist_name or "") + r"\s*-\s*"
cleaned = re.sub(f"^{artist_pattern}", "", cleaned, flags=re.IGNORECASE)
cleaned = re.sub(r"\s*[\[\(]\d{4}[\]\)]\s*", " ", cleaned)
quality_patterns = [
r"\s*[\[\(].*?320.*?kbps.*?[\]\)]\s*",
r"\s*[\[\(].*?256.*?kbps.*?[\]\)]\s*",
r"\s*[\[\(].*?flac.*?[\]\)]\s*",
r"\s*[\[\(].*?mp3.*?[\]\)]\s*",
r"\s*[\[\(].*?itunes.*?[\]\)]\s*",
r"\s*[\[\(].*?web.*?[\]\)]\s*",
r"\s*[\[\(].*?cd.*?[\]\)]\s*",
]
for pattern in quality_patterns:
cleaned = re.sub(pattern, " ", cleaned, flags=re.IGNORECASE)
cleaned = re.sub(r"\s*[\[\(][^\]\)]*\b(deluxe|special|expanded|extended|bonus|remaster(?:ed)?|anniversary|collectors?|limited|silver|gold|platinum)\b[^\]\)]*[\]\)]\s*", " ", cleaned, flags=re.IGNORECASE)
cleaned = re.sub(r"\s*[\[\(][^\]\)]*\bedition\b[^\]\)]*[\]\)]\s*", " ", cleaned, flags=re.IGNORECASE)
cleaned = re.sub(r"\s*(deluxe|special|expanded|extended|bonus|remastered|anniversary|collectors?|limited|silver|gold|platinum)\s*(edition)?\s*$", "", cleaned, flags=re.IGNORECASE)
cleaned = re.sub(r"^[-\s\.]+", "", cleaned)
cleaned = re.sub(r"[-\s\.]+$", "", cleaned)
cleaned = re.sub(r"\s+", " ", cleaned).strip()
return cleaned if cleaned else original
def resolve_album_group(artist_context: dict, album_info: dict, original_album: str = None) -> str:
"""Smart album grouping: upgrade to deluxe if any track is deluxe."""
try:
with _album_cache_lock:
artist_name = extract_artist_name(artist_context)
detected_album = (album_info or {}).get("album_name", "")
if detected_album:
base_album = get_base_album_name(detected_album)
elif original_album:
cleaned_original = clean_album_title(original_album, artist_name)
base_album = get_base_album_name(cleaned_original)
else:
base_album = get_base_album_name(detected_album)
base_album = normalize_base_album_name(base_album, artist_name)
album_key = f"{artist_name}::{base_album}"
is_deluxe_track = False
if detected_album:
is_deluxe_track = detect_deluxe_edition(detected_album)
elif original_album:
is_deluxe_track = detect_deluxe_edition(original_album)
if album_key in _album_name_cache:
cached_name = _album_name_cache[album_key]
current_edition = _album_editions.get(album_key, "standard")
if is_deluxe_track and current_edition == "standard":
final_album_name = f"{base_album} (Deluxe Edition)"
_album_editions[album_key] = "deluxe"
_album_name_cache[album_key] = final_album_name
logger.info("Album cache upgrade: %r -> %r", album_key, final_album_name)
return final_album_name
logger.info("Using cached album name for %r: %r", album_key, cached_name)
return cached_name
logger.info("Album grouping - Key: %r, Detected: %r", album_key, detected_album)
current_edition = _album_editions.get(album_key, "standard")
if is_deluxe_track and current_edition == "standard":
logger.info("UPGRADE: Album %r upgraded from standard to deluxe!", base_album)
_album_editions[album_key] = "deluxe"
current_edition = "deluxe"
if current_edition == "deluxe":
final_album_name = f"{base_album} (Deluxe Edition)"
else:
final_album_name = base_album
_album_name_cache[album_key] = final_album_name
logger.info("Album resolution: %r -> %r (edition: %s)", detected_album, final_album_name, current_edition)
return final_album_name
except Exception as e:
logger.error("Error resolving album group: %s", e)
album_name = (album_info or {}).get("album_name", "Unknown Album")
return album_name

106
core/import_file_ops.py
View file

@ -4,13 +4,17 @@ from __future__ import annotations
import logging
import os
import re
import shutil
import subprocess
import time
from pathlib import Path
from typing import Any, Dict, Optional
# Backward-compatible re-exports; canonical homes are core.import_filename
# and core.import_staging.
from core.import_filename import extract_track_number_from_filename
from core.import_staging import read_staging_file_metadata
logger = logging.getLogger("import_file_ops")
@ -85,106 +89,6 @@ def safe_move_file(src, dst):
raise
def extract_track_number_from_filename(filename: str, title: str = None) -> int:
"""Extract track number from a filename. Returns 1 if not found."""
basename = os.path.splitext(os.path.basename(filename))[0].strip()
match = re.match(r"^\d[\-\.](\d{1,2})\s*[\-\.]\s*", basename)
if match:
num = int(match.group(1))
if 1 <= num <= 99:
return num
match = re.match(r"^\(?(\d{1,3})\)?\s*[\-\.)\]]\s*", basename)
if match:
num = int(match.group(1))
if 1 <= num <= 999:
return num
return 1
def _coerce_tag_number(value: Any, default: int = 1) -> int:
if value in (None, ""):
return default
if isinstance(value, (list, tuple)):
value = value[0] if value else None
if value in (None, ""):
return default
text = str(value).strip()
if not text:
return default
match = re.match(r"^(\d+)", text)
if match:
try:
return int(match.group(1))
except ValueError:
return default
try:
return int(text)
except (TypeError, ValueError):
return default
def read_staging_file_metadata(file_path: str, filename: Optional[str] = None) -> Dict[str, Any]:
"""Read common audio tag metadata from a staging file."""
try:
from mutagen import File as MutagenFile
tags = MutagenFile(file_path, easy=True)
except Exception:
tags = None
filename = filename or os.path.basename(file_path)
stem = os.path.splitext(os.path.basename(filename))[0]
def _first_tag(*keys: str) -> str:
if not tags:
return ""
for key in keys:
try:
value = tags.get(key) # type: ignore[attr-defined]
except Exception:
value = None
if value:
if isinstance(value, (list, tuple)):
value = value[0] if value else ""
text = str(value).strip()
if text:
return text
return ""
title = _first_tag("title")
artist = _first_tag("artist")
albumartist = _first_tag("albumartist")
album = _first_tag("album")
if not title:
title = stem
if not albumartist:
albumartist = artist
track_number = _coerce_tag_number(_first_tag("tracknumber", "track_number"), default=0)
if not track_number:
track_number = extract_track_number_from_filename(filename or file_path)
disc_number = _coerce_tag_number(_first_tag("discnumber", "disc_number"), default=1)
return {
"title": title,
"artist": artist,
"albumartist": albumartist,
"album": album,
"track_number": track_number,
"disc_number": disc_number,
}
def cleanup_empty_directories(download_path, moved_file_path):
"""Remove empty directories after a move, ignoring hidden files."""
try:

19
core/import_filename.py
View file

@ -14,6 +14,25 @@ _TRACK_PATTERNS = (
)
def extract_track_number_from_filename(filename: str, title: str = None) -> int:
"""Extract track number from a filename. Returns 1 if not found."""
basename = os.path.splitext(os.path.basename(filename))[0].strip()
match = re.match(r"^\d[\-\.](\d{1,2})\s*[\-\.]\s*", basename)
if match:
num = int(match.group(1))
if 1 <= num <= 99:
return num
match = re.match(r"^\(?(\d{1,3})\)?\s*[\-\.)\]]\s*", basename)
if match:
num = int(match.group(1))
if 1 <= num <= 999:
return num
return 1
def parse_filename_metadata(filename: str) -> Dict[str, Any]:
"""Extract artist/title/album hints from a loose filename."""
raw_path = str(filename or "")

174
core/import_paths.py
View file

@ -6,11 +6,14 @@ import json
import logging
import os
import re
import threading
from pathlib import Path
from typing import Any
# Album grouping lives in core.import_album_naming; this module keeps the
# imported helper because the path builder still needs it.
from core.import_album_naming import resolve_album_group
from core.import_context import (
extract_artist_name,
get_import_clean_title,
get_import_context_album,
get_import_original_search,
@ -21,10 +24,6 @@ from core.import_context import (
logger = logging.getLogger("import_paths")
_album_cache_lock = threading.Lock()
_album_editions: dict[str, str] = {}
_album_name_cache: dict[str, str] = {}
def _get_config_manager():
try:
@ -55,14 +54,6 @@ def _get_album_tracks_for_source(source: str, album_id: str):
return None
def _extract_artist_name(artist_context: Any) -> str:
if not artist_context:
return ""
if isinstance(artist_context, dict):
return str(artist_context.get("name", "") or "").strip()
return str(artist_context).strip()
def docker_resolve_path(path_str: str) -> str:
"""Resolve Docker-hosted Windows paths into container paths."""
if os.path.exists("/.dockerenv") and len(path_str) >= 3 and path_str[1] == ":" and path_str[0].isalpha():
@ -147,161 +138,6 @@ def clean_track_title(track_title: str, artist_name: str) -> str:
return cleaned if cleaned else original
def get_base_album_name(album_name: str) -> str:
"""Extract the base album name without edition indicators."""
base_name = album_name or ""
base_name = re.sub(
r"\s*[\[\(][^)\]]*\b(deluxe|special|expanded|extended|bonus|remaster(?:ed)?|anniversary|collectors?|limited|silver|gold|platinum)\b[^)\]]*[\]\)]\s*$",
"",
base_name,
flags=re.IGNORECASE,
)
base_name = re.sub(r"\s*[\[\(][^)\]]*\bedition\b[^)\]]*[\]\)]\s*$", "", base_name, flags=re.IGNORECASE)
base_name = re.sub(
r"\s+(deluxe|special|expanded|extended|bonus|remastered|anniversary|collectors?|limited|silver|gold|platinum)\s*(edition)?\s*$",
"",
base_name,
flags=re.IGNORECASE,
)
return base_name.strip()
def detect_deluxe_edition(album_name: str) -> bool:
"""Detect if an album name indicates a deluxe/special edition."""
if not album_name:
return False
album_lower = album_name.lower()
deluxe_indicators = [
"deluxe",
"deluxe edition",
"special edition",
"expanded edition",
"extended edition",
"bonus",
"remastered",
"anniversary",
"collectors edition",
"limited edition",
"silver edition",
"gold edition",
"platinum edition",
]
for indicator in deluxe_indicators:
if indicator in album_lower:
logger.info("Detected deluxe edition: %r contains %r", album_name, indicator)
return True
return False
def normalize_base_album_name(base_album: str, artist_name: str) -> str:
"""Normalize the base album name to handle case variations and known corrections."""
normalized_lower = (base_album or "").lower().strip()
known_corrections = {
# Add specific album name corrections here as needed.
}
for variant, correction in known_corrections.items():
if normalized_lower == variant.lower():
logger.info("Album correction applied: %r -> %r", base_album, correction)
return correction
normalized = base_album or ""
normalized = re.sub(r"\s*&\s*", " & ", normalized)
normalized = re.sub(r"\s+", " ", normalized)
normalized = normalized.strip()
logger.info("Album variant normalization: %r -> %r", base_album, normalized)
return normalized
def clean_album_title(album_title: str, artist_name: str) -> str:
"""Clean up album title by removing common prefixes, suffixes, and artist redundancy."""
original = (album_title or "").strip()
cleaned = original
logger.info("Album Title Cleaning: %r (artist: %r)", original, artist_name)
cleaned = re.sub(r"^Album\s*-\s*", "", cleaned, flags=re.IGNORECASE)
artist_pattern = re.escape(artist_name or "") + r"\s*-\s*"
cleaned = re.sub(f"^{artist_pattern}", "", cleaned, flags=re.IGNORECASE)
cleaned = re.sub(r"\s*[\[\(]\d{4}[\]\)]\s*", " ", cleaned)
quality_patterns = [
r"\s*[\[\(].*?320.*?kbps.*?[\]\)]\s*",
r"\s*[\[\(].*?256.*?kbps.*?[\]\)]\s*",
r"\s*[\[\(].*?flac.*?[\]\)]\s*",
r"\s*[\[\(].*?mp3.*?[\]\)]\s*",
r"\s*[\[\(].*?itunes.*?[\]\)]\s*",
r"\s*[\[\(].*?web.*?[\]\)]\s*",
r"\s*[\[\(].*?cd.*?[\]\)]\s*",
]
for pattern in quality_patterns:
cleaned = re.sub(pattern, " ", cleaned, flags=re.IGNORECASE)
cleaned = re.sub(r"\s*[\[\(][^\]\)]*\b(deluxe|special|expanded|extended|bonus|remaster(?:ed)?|anniversary|collectors?|limited|silver|gold|platinum)\b[^\]\)]*[\]\)]\s*", " ", cleaned, flags=re.IGNORECASE)
cleaned = re.sub(r"\s*[\[\(][^\]\)]*\bedition\b[^\]\)]*[\]\)]\s*", " ", cleaned, flags=re.IGNORECASE)
cleaned = re.sub(r"\s*(deluxe|special|expanded|extended|bonus|remastered|anniversary|collectors?|limited|silver|gold|platinum)\s*(edition)?\s*$", "", cleaned, flags=re.IGNORECASE)
cleaned = re.sub(r"^[-\s\.]+", "", cleaned)
cleaned = re.sub(r"[-\s\.]+$", "", cleaned)
cleaned = re.sub(r"\s+", " ", cleaned).strip()
return cleaned if cleaned else original
def resolve_album_group(artist_context: dict, album_info: dict, original_album: str = None) -> str:
"""Smart album grouping: upgrade to deluxe if any track is deluxe."""
try:
with _album_cache_lock:
artist_name = _extract_artist_name(artist_context)
detected_album = (album_info or {}).get("album_name", "")
if detected_album:
base_album = get_base_album_name(detected_album)
elif original_album:
cleaned_original = clean_album_title(original_album, artist_name)
base_album = get_base_album_name(cleaned_original)
else:
base_album = get_base_album_name(detected_album)
base_album = normalize_base_album_name(base_album, artist_name)
album_key = f"{artist_name}::{base_album}"
is_deluxe_track = False
if detected_album:
is_deluxe_track = detect_deluxe_edition(detected_album)
elif original_album:
is_deluxe_track = detect_deluxe_edition(original_album)
if album_key in _album_name_cache:
cached_name = _album_name_cache[album_key]
current_edition = _album_editions.get(album_key, "standard")
if is_deluxe_track and current_edition == "standard":
final_album_name = f"{base_album} (Deluxe Edition)"
_album_editions[album_key] = "deluxe"
_album_name_cache[album_key] = final_album_name
logger.info("Album cache upgrade: %r -> %r", album_key, final_album_name)
return final_album_name
logger.info("Using cached album name for %r: %r", album_key, cached_name)
return cached_name
logger.info("Album grouping - Key: %r, Detected: %r", album_key, detected_album)
current_edition = _album_editions.get(album_key, "standard")
if is_deluxe_track and current_edition == "standard":
logger.info("UPGRADE: Album %r upgraded from standard to deluxe!", base_album)
_album_editions[album_key] = "deluxe"
current_edition = "deluxe"
if current_edition == "deluxe":
final_album_name = f"{base_album} (Deluxe Edition)"
else:
final_album_name = base_album
_album_name_cache[album_key] = final_album_name
logger.info("Album resolution: %r -> %r (edition: %s)", detected_album, final_album_name, current_edition)
return final_album_name
except Exception as e:
logger.error("Error resolving album group: %s", e)
album_name = (album_info or {}).get("album_name", "Unknown Album")
return album_name
def get_album_type_display(raw_type, track_count) -> str:
@ -549,7 +385,7 @@ def build_final_path_for_track(context, artist_context, album_info, file_ext):
album_context = get_import_context_album(context)
source = get_import_source(context)
playlist_folder_mode = track_info.get("_playlist_folder_mode", False)
artist_name = _extract_artist_name(artist_context)
artist_name = extract_artist_name(artist_context)
source_info = track_info.get("source_info") or {}
if isinstance(source_info, str):

4
core/import_pipeline.py
View file

@ -12,7 +12,6 @@ from core.import_file_ops import (
cleanup_empty_directories,
create_lossy_copy,
downsample_hires_flac,
extract_track_number_from_filename,
get_audio_quality_string,
get_quality_tier_from_extension,
safe_move_file,
@ -29,6 +28,7 @@ from core.import_context import (
get_import_track_info,
normalize_import_context,
)
from core.import_filename import extract_track_number_from_filename
from core.import_guards import check_flac_bit_depth, move_to_quarantine
from core.import_side_effects import (
check_and_remove_from_wishlist,
@ -61,8 +61,8 @@ from core.import_paths import (
build_final_path_for_track,
build_simple_download_destination,
docker_resolve_path,
resolve_album_group,
)
from core.import_album_naming import resolve_album_group
from database.music_database import get_database
from utils.logging_config import get_logger

104
core/import_staging.py
View file

@ -4,9 +4,10 @@ from __future__ import annotations
import os
import threading
from typing import Any, Dict, List, Optional, Tuple
from typing import Any, Dict, Iterable, List, Optional, Tuple
from core.import_paths import docker_resolve_path
from core.import_filename import extract_track_number_from_filename
from utils.logging_config import get_logger
logger = get_logger("import_staging")
@ -64,6 +65,71 @@ def get_client_for_source(source: str):
return _get_client_for_source(source)
def read_staging_file_metadata(file_path: str, filename: Optional[str] = None) -> Dict[str, Any]:
"""Read common audio tag metadata from a staging file."""
try:
from mutagen import File as MutagenFile
tags = MutagenFile(file_path, easy=True)
except Exception:
tags = None
filename = filename or os.path.basename(file_path)
stem = os.path.splitext(os.path.basename(filename))[0]
def _first_tag(*keys: str) -> str:
if not tags:
return ""
for key in keys:
try:
value = tags.get(key) # type: ignore[attr-defined]
except Exception:
value = None
if value:
if isinstance(value, (list, tuple)):
value = value[0] if value else ""
text = str(value).strip()
if text:
return text
return ""
title = _first_tag("title")
artist = _first_tag("artist")
albumartist = _first_tag("albumartist")
album = _first_tag("album")
if not title:
title = stem
if not albumartist:
albumartist = artist
track_number = extract_track_number_from_filename(filename or file_path)
try:
# Preserve tag-based numbers when present, but still fall back to the filename parser.
tag_track_number = _first_tag("tracknumber", "track_number")
if tag_track_number:
track_number = int(str(tag_track_number).split("/")[0].strip() or track_number)
except (TypeError, ValueError):
pass
disc_number = 1
try:
tag_disc_number = _first_tag("discnumber", "disc_number")
if tag_disc_number:
disc_number = int(str(tag_disc_number).split("/")[0].strip() or 1)
except (TypeError, ValueError):
pass
return {
"title": title,
"artist": artist,
"albumartist": albumartist,
"album": album,
"track_number": track_number,
"disc_number": disc_number,
}
def _search_albums_for_source(source: str, client: Any, query: str, limit: int = 5):
from core.metadata_service import _search_albums_for_source as _metadata_search_albums_for_source
@ -452,3 +518,39 @@ def refresh_import_suggestions_cache():
with _import_suggestions_cache_lock:
_import_suggestions_cache["built"] = False
start_import_suggestions_cache()
def collect_staging_files(file_paths: Optional[Iterable[str]] = None) -> List[Dict[str, Any]]:
"""Collect audio files from the staging area with normalized metadata."""
staging_path = get_staging_path()
file_filter: Optional[set[str]] = set(file_paths) if file_paths else None
staging_files: List[Dict[str, Any]] = []
if not os.path.isdir(staging_path):
return staging_files
for root, _dirs, filenames in os.walk(staging_path):
for filename in filenames:
ext = os.path.splitext(filename)[1].lower()
if ext not in AUDIO_EXTENSIONS:
continue
full_path = os.path.join(root, filename)
if file_filter is not None and full_path not in file_filter:
continue
meta = read_staging_file_metadata(full_path, filename)
staging_files.append(
{
"filename": filename,
"full_path": full_path,
"title": meta.get("title", ""),
"artist": meta.get("albumartist") or meta.get("artist") or "",
"album": meta.get("album", ""),
"albumartist": meta.get("albumartist") or meta.get("artist") or "",
"track_number": meta.get("track_number", 1),
"disc_number": meta.get("disc_number", 1),
}
)
return staging_files

23
tests/test_import_album.py
View file

@ -87,19 +87,22 @@ def test_build_album_import_match_payload_uses_generic_track_keys(monkeypatch, t
staging_root.mkdir()
(staging_root / "Song One.flac").write_text("fake")
monkeypatch.setattr(import_album, "get_staging_path", lambda: str(staging_root))
monkeypatch.setattr(import_album, "_get_matching_engine", lambda: _FakeEngine())
monkeypatch.setattr(
import_album,
"read_staging_file_metadata",
lambda file_path, filename=None: {
"title": "Song One",
"artist": "Artist One",
"albumartist": "Artist One",
"album": "Album One",
"track_number": 1,
"disc_number": 1,
},
"collect_staging_files",
lambda file_paths=None: [
{
"filename": "Song One.flac",
"full_path": str(staging_root / "Song One.flac"),
"title": "Song One",
"artist": "Artist One",
"album": "Album One",
"albumartist": "Artist One",
"track_number": 1,
"disc_number": 1,
}
],
)
monkeypatch.setattr(
import_album,

4
tests/test_import_file_ops.py
View file

@ -3,10 +3,10 @@ import types
from core.import_file_ops import (
cleanup_empty_directories,
extract_track_number_from_filename,
safe_move_file,
read_staging_file_metadata,
)
from core.import_filename import extract_track_number_from_filename
from core.import_staging import read_staging_file_metadata
def test_extract_track_number_from_filename_handles_common_patterns():

4
tests/test_import_paths.py
View file

@ -1,3 +1,4 @@
import core.import_album_naming as album_naming
import core.import_paths as import_paths
@ -73,8 +74,7 @@ def test_get_file_path_from_template_raw_handles_quality_and_disc_placeholders(m
def test_resolve_album_group_upgrades_standard_to_deluxe():
import_paths._album_name_cache.clear()
import_paths._album_editions.clear()
album_naming.clear_album_grouping_cache()
artist_context = {"name": "Cache Artist"}
standard_album = {"album_name": "Cache Album"}