diff --git a/core/import_album.py b/core/import_album.py index 953e091e..1f439f1b 100644 --- a/core/import_album.py +++ b/core/import_album.py @@ -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() diff --git a/core/import_album_naming.py b/core/import_album_naming.py new file mode 100644 index 00000000..338f326d --- /dev/null +++ b/core/import_album_naming.py @@ -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 diff --git a/core/import_file_ops.py b/core/import_file_ops.py index 261687a8..21cf9318 100644 --- a/core/import_file_ops.py +++ b/core/import_file_ops.py @@ -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: diff --git a/core/import_filename.py b/core/import_filename.py index 9298750d..92ff026a 100644 --- a/core/import_filename.py +++ b/core/import_filename.py @@ -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 "") diff --git a/core/import_paths.py b/core/import_paths.py index 7848425c..39e280c8 100644 --- a/core/import_paths.py +++ b/core/import_paths.py @@ -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): diff --git a/core/import_pipeline.py b/core/import_pipeline.py index ede3ae18..1574fe69 100644 --- a/core/import_pipeline.py +++ b/core/import_pipeline.py @@ -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 diff --git a/core/import_staging.py b/core/import_staging.py index da93580a..8f8fc6f0 100644 --- a/core/import_staging.py +++ b/core/import_staging.py @@ -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 diff --git a/tests/test_import_album.py b/tests/test_import_album.py index 81ab6fb6..45706da3 100644 --- a/tests/test_import_album.py +++ b/tests/test_import_album.py @@ -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, diff --git a/tests/test_import_file_ops.py b/tests/test_import_file_ops.py index e9d72901..056c95bf 100644 --- a/tests/test_import_file_ops.py +++ b/tests/test_import_file_ops.py @@ -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(): diff --git a/tests/test_import_paths.py b/tests/test_import_paths.py index f9a39fab..a3e766f9 100644 --- a/tests/test_import_paths.py +++ b/tests/test_import_paths.py @@ -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"}