Tighten metadata and import safety
- Normalize album import track display handling so queue labels and match rows stay consistent - Bound MusicBrainz caches and avoid caching transient lookup failures - Stop swallowing programmer errors in source enrichment helpers - Restore import config test seams without reintroducing lazy imports - Guard task completion calls and fix the Windows path test expectation - Keep file lock tracking from growing without bound
This commit is contained in:
Antti Kettunen
parent
9315e74bea
commit
02305096a3
10 changed files with 477 additions and 298 deletions
|
|
@ -9,14 +9,11 @@ import subprocess
|
|||
import time
|
||||
from pathlib import Path
|
||||
|
||||
from config.settings import config_manager
|
||||
|
||||
logger = logging.getLogger("imports.file_ops")
|
||||
|
||||
|
||||
def _get_config_manager():
|
||||
from config.settings import config_manager
|
||||
return config_manager
|
||||
|
||||
|
||||
def safe_move_file(src, dst):
|
||||
"""Move a file safely across filesystems."""
|
||||
src = Path(src)
|
||||
|
|
@ -176,7 +173,6 @@ def downsample_hires_flac(final_path, context):
|
|||
"""Downsample a hi-res FLAC to 16-bit/44.1kHz if enabled."""
|
||||
from mutagen.flac import FLAC
|
||||
|
||||
config_manager = _get_config_manager()
|
||||
if not config_manager.get("lossy_copy.downsample_hires", False):
|
||||
return None
|
||||
|
||||
|
|
@ -287,7 +283,6 @@ def create_lossy_copy(final_path):
|
|||
"""Convert a FLAC file to a lossy copy using the configured codec."""
|
||||
from mutagen.flac import FLAC
|
||||
|
||||
config_manager = _get_config_manager()
|
||||
if not config_manager.get("lossy_copy.enabled", False):
|
||||
return None
|
||||
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ from datetime import datetime
|
|||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from config.settings import config_manager
|
||||
from core.imports.context import (
|
||||
get_import_clean_artist,
|
||||
get_import_clean_title,
|
||||
|
|
@ -25,15 +26,12 @@ logger = get_logger("imports.guards")
|
|||
|
||||
|
||||
def _get_config_manager():
|
||||
from config.settings import config_manager
|
||||
|
||||
return config_manager
|
||||
|
||||
|
||||
def move_to_quarantine(file_path: str, context: dict, reason: str, automation_engine=None) -> str:
|
||||
"""Move a file to the quarantine folder and write a metadata sidecar."""
|
||||
config_manager = _get_config_manager()
|
||||
download_dir = config_manager.get("soulseek.download_path", "./downloads")
|
||||
download_dir = _get_config_manager().get("soulseek.download_path", "./downloads")
|
||||
quarantine_dir = Path(download_dir) / "ss_quarantine"
|
||||
quarantine_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
|
@ -95,7 +93,6 @@ def check_flac_bit_depth(file_path: str, context: dict) -> Optional[str]:
|
|||
if not context.get("_audio_quality", "").startswith("FLAC"):
|
||||
return None
|
||||
|
||||
config_manager = _get_config_manager()
|
||||
quality_profile = MusicDatabase().get_quality_profile()
|
||||
flac_config = quality_profile.get("qualities", {}).get("flac", {})
|
||||
flac_pref = flac_config.get("bit_depth", "any")
|
||||
|
|
@ -107,7 +104,7 @@ def check_flac_bit_depth(file_path: str, context: dict) -> Optional[str]:
|
|||
return None
|
||||
|
||||
flac_fallback = flac_config.get("bit_depth_fallback", True)
|
||||
downsample_enabled = config_manager.get("lossy_copy.downsample_hires", False)
|
||||
downsample_enabled = _get_config_manager().get("lossy_copy.downsample_hires", False)
|
||||
track_info = context.get("track_info", {})
|
||||
track_name = track_info.get("name", os.path.basename(file_path))
|
||||
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ import json
|
|||
import os
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from config.settings import config_manager
|
||||
from core.imports.context import (
|
||||
extract_artist_name,
|
||||
get_import_clean_album,
|
||||
|
|
@ -31,8 +32,6 @@ logger = get_logger("imports.side_effects")
|
|||
|
||||
|
||||
def _get_config_manager():
|
||||
from config.settings import config_manager
|
||||
|
||||
return config_manager
|
||||
|
||||
|
||||
|
|
@ -222,8 +221,7 @@ def record_download_provenance(context: Dict[str, Any]) -> None:
|
|||
def record_soulsync_library_entry(context: Dict[str, Any], artist_context: Dict[str, Any], album_info: Dict[str, Any]) -> None:
|
||||
"""Write imported media to the SoulSync library tables when the active server is SoulSync."""
|
||||
try:
|
||||
config_manager = _get_config_manager()
|
||||
if config_manager.get_active_media_server() != "soulsync":
|
||||
if _get_config_manager().get_active_media_server() != "soulsync":
|
||||
return
|
||||
|
||||
context = normalize_import_context(context)
|
||||
|
|
@ -288,7 +286,7 @@ def record_soulsync_library_entry(context: Dict[str, Any], artist_context: Dict[
|
|||
if genres:
|
||||
from core.genre_filter import filter_genres as _filter_genres
|
||||
|
||||
genres = _filter_genres(genres, config_manager)
|
||||
genres = _filter_genres(genres, _get_config_manager())
|
||||
genres_json = json.dumps(genres) if genres else ""
|
||||
|
||||
bitrate = 0
|
||||
|
|
|
|||
|
|
@ -4,8 +4,9 @@ from __future__ import annotations
|
|||
|
||||
import os
|
||||
import threading
|
||||
import weakref
|
||||
from types import SimpleNamespace
|
||||
from typing import Any, Dict
|
||||
from typing import Any
|
||||
|
||||
from utils.logging_config import get_logger as _create_logger
|
||||
|
||||
|
|
@ -26,7 +27,7 @@ __all__ = [
|
|||
"wipe_source_tags",
|
||||
]
|
||||
|
||||
_FILE_LOCKS: Dict[str, threading.Lock] = {}
|
||||
_FILE_LOCKS: "weakref.WeakValueDictionary[str, threading.Lock]" = weakref.WeakValueDictionary()
|
||||
_FILE_LOCKS_LOCK = threading.Lock()
|
||||
|
||||
|
||||
|
|
@ -117,6 +118,8 @@ def get_mutagen_symbols():
|
|||
|
||||
|
||||
def get_file_lock(file_path: str) -> threading.Lock:
|
||||
# Keep a per-path lock while it is actively referenced, but let it
|
||||
# fall out of the cache once nobody is using it anymore.
|
||||
with _FILE_LOCKS_LOCK:
|
||||
lock = _FILE_LOCKS.get(file_path)
|
||||
if lock is None:
|
||||
|
|
|
|||
|
|
@ -3,10 +3,14 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import socket
|
||||
import threading
|
||||
import time
|
||||
from collections import OrderedDict
|
||||
from typing import Any, Dict
|
||||
|
||||
import requests
|
||||
|
||||
from core.imports.context import (
|
||||
extract_artist_name,
|
||||
get_import_clean_artist,
|
||||
|
|
@ -38,12 +42,17 @@ __all__ = [
|
|||
"mb_release_detail_cache_lock",
|
||||
]
|
||||
|
||||
mb_release_cache: Dict[tuple, str] = {}
|
||||
_MB_RELEASE_CACHE_MAX_ENTRIES = 4096
|
||||
_MB_RELEASE_DETAIL_CACHE_MAX_ENTRIES = 4096
|
||||
|
||||
mb_release_cache: "OrderedDict[tuple, str]" = OrderedDict()
|
||||
mb_release_cache_lock = threading.RLock()
|
||||
mb_release_detail_cache: Dict[str, Dict[str, Any]] = {}
|
||||
mb_release_detail_cache: "OrderedDict[str, Dict[str, Any]]" = OrderedDict()
|
||||
mb_release_detail_cache_lock = threading.RLock()
|
||||
logger = _create_logger("metadata.source")
|
||||
|
||||
_SOURCE_NETWORK_EXCEPTIONS = (requests.RequestException, socket.timeout, TimeoutError)
|
||||
|
||||
_EDITION_PAREN_RE = re.compile(
|
||||
r'\s*[\(\[]\s*(?:deluxe|expanded|remaster(?:ed)?|anniversary|special|collector|'
|
||||
r'limited|bonus|platinum|gold|super\s*deluxe|standard)'
|
||||
|
|
@ -64,6 +73,29 @@ def normalize_album_cache_key(album_name: str) -> str:
|
|||
return result.lower().strip()
|
||||
|
||||
|
||||
def _bounded_cache_get(cache, key):
|
||||
value = cache.get(key)
|
||||
if value is not None and hasattr(cache, "move_to_end"):
|
||||
cache.move_to_end(key)
|
||||
return value
|
||||
|
||||
|
||||
def _bounded_cache_set(cache, key, value, max_entries: int) -> None:
|
||||
cache[key] = value
|
||||
if hasattr(cache, "move_to_end"):
|
||||
cache.move_to_end(key)
|
||||
while len(cache) > max_entries:
|
||||
cache.popitem(last=False)
|
||||
|
||||
|
||||
def _call_source_lookup(label: str, func, *args, **kwargs):
|
||||
try:
|
||||
return func(*args, **kwargs)
|
||||
except _SOURCE_NETWORK_EXCEPTIONS as exc:
|
||||
logger.warning("%s lookup failed (network): %s", label, exc)
|
||||
return None
|
||||
|
||||
|
||||
SOURCE_TAG_CONFIG = {
|
||||
"SPOTIFY_TRACK_ID": "spotify.tags.track_id",
|
||||
"SPOTIFY_ARTIST_ID": "spotify.tags.artist_id",
|
||||
|
|
@ -194,119 +226,124 @@ def _process_musicbrainz_source(pp: dict, metadata: dict, cfg, runtime, track_ti
|
|||
if not mb_service:
|
||||
return
|
||||
|
||||
try:
|
||||
result = mb_service.match_recording(track_title, artist_name)
|
||||
if result and result.get("mbid"):
|
||||
pp["recording_mbid"] = result["mbid"]
|
||||
pp["id_tags"]["MUSICBRAINZ_RECORDING_ID"] = pp["recording_mbid"]
|
||||
details = mb_service.mb_client.get_recording(pp["recording_mbid"], includes=["isrcs", "genres"])
|
||||
if details:
|
||||
isrcs = details.get("isrcs", [])
|
||||
if isrcs:
|
||||
pp["isrc"] = isrcs[0]
|
||||
pp["mb_genres"] = [g["name"] for g in sorted(details.get("genres", []), key=lambda x: x.get("count", 0), reverse=True)]
|
||||
result = _call_source_lookup("MusicBrainz recording", mb_service.match_recording, track_title, artist_name)
|
||||
if result and result.get("mbid"):
|
||||
pp["recording_mbid"] = result["mbid"]
|
||||
pp["id_tags"]["MUSICBRAINZ_RECORDING_ID"] = pp["recording_mbid"]
|
||||
details = _call_source_lookup(
|
||||
"MusicBrainz recording details",
|
||||
mb_service.mb_client.get_recording,
|
||||
pp["recording_mbid"],
|
||||
includes=["isrcs", "genres"],
|
||||
)
|
||||
if details:
|
||||
isrcs = details.get("isrcs", [])
|
||||
if isrcs:
|
||||
pp["isrc"] = isrcs[0]
|
||||
pp["mb_genres"] = [g["name"] for g in sorted(details.get("genres", []), key=lambda x: x.get("count", 0), reverse=True)]
|
||||
|
||||
track_artist_name = metadata.get("artist", "") or artist_name
|
||||
if ", " in track_artist_name:
|
||||
track_artist_name = track_artist_name.split(", ")[0]
|
||||
artist_result = mb_service.match_artist(track_artist_name)
|
||||
if artist_result and artist_result.get("mbid"):
|
||||
pp["artist_mbid"] = artist_result["mbid"]
|
||||
pp["id_tags"]["MUSICBRAINZ_ARTIST_ID"] = pp["artist_mbid"]
|
||||
|
||||
album_name_for_mb = metadata.get("album", "")
|
||||
if album_name_for_mb:
|
||||
artist_key = (pp.get("batch_artist_name") or artist_name).lower().strip()
|
||||
rc_key_norm = (normalize_album_cache_key(album_name_for_mb), artist_key)
|
||||
rc_key_exact = (album_name_for_mb.lower().strip(), artist_key)
|
||||
with mb_release_cache_lock:
|
||||
cached = mb_release_cache.get(rc_key_norm)
|
||||
if cached is None:
|
||||
cached = mb_release_cache.get(rc_key_exact)
|
||||
if cached is not None:
|
||||
pp["release_mbid"] = cached
|
||||
else:
|
||||
try:
|
||||
rc_result = mb_service.match_release(album_name_for_mb, artist_name)
|
||||
pp["release_mbid"] = rc_result.get("mbid", "") if rc_result else ""
|
||||
except Exception:
|
||||
pp["release_mbid"] = ""
|
||||
mb_release_cache[rc_key_norm] = pp["release_mbid"]
|
||||
mb_release_cache[rc_key_exact] = pp["release_mbid"]
|
||||
if pp["release_mbid"]:
|
||||
pp["id_tags"]["MUSICBRAINZ_RELEASE_ID"] = pp["release_mbid"]
|
||||
track_artist_name = metadata.get("artist", "") or artist_name
|
||||
if ", " in track_artist_name:
|
||||
track_artist_name = track_artist_name.split(", ")[0]
|
||||
artist_result = _call_source_lookup("MusicBrainz artist", mb_service.match_artist, track_artist_name)
|
||||
if artist_result and artist_result.get("mbid"):
|
||||
pp["artist_mbid"] = artist_result["mbid"]
|
||||
pp["id_tags"]["MUSICBRAINZ_ARTIST_ID"] = pp["artist_mbid"]
|
||||
|
||||
album_name_for_mb = metadata.get("album", "")
|
||||
if album_name_for_mb:
|
||||
artist_key = (pp.get("batch_artist_name") or artist_name).lower().strip()
|
||||
rc_key_norm = (normalize_album_cache_key(album_name_for_mb), artist_key)
|
||||
rc_key_exact = (album_name_for_mb.lower().strip(), artist_key)
|
||||
release_mbid = None
|
||||
with mb_release_cache_lock:
|
||||
cached = _bounded_cache_get(mb_release_cache, rc_key_norm)
|
||||
if cached is None:
|
||||
cached = _bounded_cache_get(mb_release_cache, rc_key_exact)
|
||||
if cached:
|
||||
release_mbid = cached
|
||||
else:
|
||||
rc_result = _call_source_lookup("MusicBrainz release", mb_service.match_release, album_name_for_mb, artist_name)
|
||||
if rc_result and rc_result.get("mbid"):
|
||||
release_mbid = rc_result["mbid"]
|
||||
if release_mbid:
|
||||
_bounded_cache_set(mb_release_cache, rc_key_norm, release_mbid, _MB_RELEASE_CACHE_MAX_ENTRIES)
|
||||
_bounded_cache_set(mb_release_cache, rc_key_exact, release_mbid, _MB_RELEASE_CACHE_MAX_ENTRIES)
|
||||
pp["release_mbid"] = release_mbid or ""
|
||||
if pp["release_mbid"]:
|
||||
pp["id_tags"]["MUSICBRAINZ_RELEASE_ID"] = pp["release_mbid"]
|
||||
|
||||
if pp["release_mbid"]:
|
||||
with mb_release_detail_cache_lock:
|
||||
release_detail = _bounded_cache_get(mb_release_detail_cache, pp["release_mbid"])
|
||||
if release_detail is None:
|
||||
release_detail = _call_source_lookup(
|
||||
"MusicBrainz release details",
|
||||
mb_service.mb_client.get_release,
|
||||
pp["release_mbid"],
|
||||
includes=["release-groups", "labels", "media", "artist-credits", "recordings"],
|
||||
) or {}
|
||||
with mb_release_detail_cache_lock:
|
||||
release_detail = mb_release_detail_cache.get(pp["release_mbid"])
|
||||
if release_detail is None:
|
||||
release_detail = mb_service.mb_client.get_release(
|
||||
pp["release_mbid"],
|
||||
includes=["release-groups", "labels", "media", "artist-credits", "recordings"],
|
||||
) or {}
|
||||
with mb_release_detail_cache_lock:
|
||||
mb_release_detail_cache[pp["release_mbid"]] = release_detail
|
||||
if release_detail:
|
||||
rg = release_detail.get("release-group", {})
|
||||
if rg.get("id"):
|
||||
pp["id_tags"]["MUSICBRAINZ_RELEASEGROUPID"] = rg["id"]
|
||||
ac = release_detail.get("artist-credit", [])
|
||||
if ac and isinstance(ac[0], dict):
|
||||
aa = ac[0].get("artist", {})
|
||||
if aa.get("id"):
|
||||
pp["id_tags"]["MUSICBRAINZ_ALBUMARTISTID"] = aa["id"]
|
||||
if rg.get("primary-type"):
|
||||
pp["id_tags"]["RELEASETYPE"] = rg["primary-type"]
|
||||
if rg.get("first-release-date"):
|
||||
pp["id_tags"]["ORIGINALDATE"] = rg["first-release-date"]
|
||||
if not pp["release_year"] and len(rg["first-release-date"]) >= 4:
|
||||
year = rg["first-release-date"][:4]
|
||||
if year.isdigit():
|
||||
pp["release_year"] = year
|
||||
if release_detail.get("status"):
|
||||
pp["id_tags"]["RELEASESTATUS"] = release_detail["status"]
|
||||
if release_detail.get("country"):
|
||||
pp["id_tags"]["RELEASECOUNTRY"] = release_detail["country"]
|
||||
if release_detail.get("barcode"):
|
||||
pp["id_tags"]["BARCODE"] = release_detail["barcode"]
|
||||
media_list = release_detail.get("media", [])
|
||||
if media_list:
|
||||
fmt = media_list[0].get("format", "")
|
||||
if fmt:
|
||||
pp["id_tags"]["MEDIA"] = fmt
|
||||
pp["id_tags"]["TOTALDISCS"] = str(len(media_list))
|
||||
label_info = release_detail.get("label-info", [])
|
||||
if label_info and isinstance(label_info[0], dict):
|
||||
cat = label_info[0].get("catalog-number", "")
|
||||
if cat:
|
||||
pp["id_tags"]["CATALOGNUMBER"] = cat
|
||||
text_rep = release_detail.get("text-representation", {})
|
||||
if isinstance(text_rep, dict) and text_rep.get("script"):
|
||||
pp["id_tags"]["SCRIPT"] = text_rep["script"]
|
||||
if release_detail.get("asin"):
|
||||
pp["id_tags"]["ASIN"] = release_detail["asin"]
|
||||
track_num = metadata.get("track_number")
|
||||
disc_num = metadata.get("disc_number") or 1
|
||||
if track_num and media_list:
|
||||
try:
|
||||
track_num_int = int(track_num)
|
||||
disc_num_int = int(disc_num)
|
||||
for medium in media_list:
|
||||
if medium.get("position", 1) == disc_num_int:
|
||||
for mtrack in (medium.get("tracks") or medium.get("track-list", [])):
|
||||
if mtrack.get("position") == track_num_int:
|
||||
if mtrack.get("id"):
|
||||
pp["id_tags"]["MUSICBRAINZ_RELEASETRACKID"] = mtrack["id"]
|
||||
release_recording = mtrack.get("recording", {})
|
||||
if release_recording.get("id"):
|
||||
pp["recording_mbid"] = release_recording["id"]
|
||||
pp["id_tags"]["MUSICBRAINZ_RECORDING_ID"] = release_recording["id"]
|
||||
break
|
||||
break
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
except Exception as exc:
|
||||
logger.error("MusicBrainz lookup failed (non-fatal): %s", exc)
|
||||
_bounded_cache_set(mb_release_detail_cache, pp["release_mbid"], release_detail, _MB_RELEASE_DETAIL_CACHE_MAX_ENTRIES)
|
||||
if release_detail:
|
||||
rg = release_detail.get("release-group", {})
|
||||
if rg.get("id"):
|
||||
pp["id_tags"]["MUSICBRAINZ_RELEASEGROUPID"] = rg["id"]
|
||||
ac = release_detail.get("artist-credit", [])
|
||||
if ac and isinstance(ac[0], dict):
|
||||
aa = ac[0].get("artist", {})
|
||||
if aa.get("id"):
|
||||
pp["id_tags"]["MUSICBRAINZ_ALBUMARTISTID"] = aa["id"]
|
||||
if rg.get("primary-type"):
|
||||
pp["id_tags"]["RELEASETYPE"] = rg["primary-type"]
|
||||
if rg.get("first-release-date"):
|
||||
pp["id_tags"]["ORIGINALDATE"] = rg["first-release-date"]
|
||||
if not pp["release_year"] and len(rg["first-release-date"]) >= 4:
|
||||
year = rg["first-release-date"][:4]
|
||||
if year.isdigit():
|
||||
pp["release_year"] = year
|
||||
if release_detail.get("status"):
|
||||
pp["id_tags"]["RELEASESTATUS"] = release_detail["status"]
|
||||
if release_detail.get("country"):
|
||||
pp["id_tags"]["RELEASECOUNTRY"] = release_detail["country"]
|
||||
if release_detail.get("barcode"):
|
||||
pp["id_tags"]["BARCODE"] = release_detail["barcode"]
|
||||
media_list = release_detail.get("media", [])
|
||||
if media_list:
|
||||
fmt = media_list[0].get("format", "")
|
||||
if fmt:
|
||||
pp["id_tags"]["MEDIA"] = fmt
|
||||
pp["id_tags"]["TOTALDISCS"] = str(len(media_list))
|
||||
label_info = release_detail.get("label-info", [])
|
||||
if label_info and isinstance(label_info[0], dict):
|
||||
cat = label_info[0].get("catalog-number", "")
|
||||
if cat:
|
||||
pp["id_tags"]["CATALOGNUMBER"] = cat
|
||||
text_rep = release_detail.get("text-representation", {})
|
||||
if isinstance(text_rep, dict) and text_rep.get("script"):
|
||||
pp["id_tags"]["SCRIPT"] = text_rep["script"]
|
||||
if release_detail.get("asin"):
|
||||
pp["id_tags"]["ASIN"] = release_detail["asin"]
|
||||
track_num = metadata.get("track_number")
|
||||
disc_num = metadata.get("disc_number") or 1
|
||||
if track_num and media_list:
|
||||
try:
|
||||
track_num_int = int(track_num)
|
||||
disc_num_int = int(disc_num)
|
||||
for medium in media_list:
|
||||
if medium.get("position", 1) == disc_num_int:
|
||||
for mtrack in (medium.get("tracks") or medium.get("track-list", [])):
|
||||
if mtrack.get("position") == track_num_int:
|
||||
if mtrack.get("id"):
|
||||
pp["id_tags"]["MUSICBRAINZ_RELEASETRACKID"] = mtrack["id"]
|
||||
release_recording = mtrack.get("recording", {})
|
||||
if release_recording.get("id"):
|
||||
pp["recording_mbid"] = release_recording["id"]
|
||||
pp["id_tags"]["MUSICBRAINZ_RECORDING_ID"] = release_recording["id"]
|
||||
break
|
||||
break
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
|
||||
def _process_deezer_source(pp: dict, metadata: dict, cfg, runtime, track_title: str, artist_name: str) -> None:
|
||||
|
|
@ -315,33 +352,30 @@ def _process_deezer_source(pp: dict, metadata: dict, cfg, runtime, track_title:
|
|||
if not track_title or not artist_name:
|
||||
return
|
||||
|
||||
try:
|
||||
deezer_worker = getattr(runtime, "deezer_worker", None)
|
||||
dz_client = deezer_worker.client if deezer_worker else None
|
||||
if not dz_client:
|
||||
return
|
||||
dz_result = dz_client.search_track(artist_name, track_title)
|
||||
if dz_result and _names_match(dz_result.get("title", ""), track_title) and _names_match(dz_result.get("artist", {}).get("name", ""), artist_name):
|
||||
dz_track_id = dz_result["id"]
|
||||
pp["id_tags"]["DEEZER_TRACK_ID"] = str(dz_track_id)
|
||||
dz_artist_id = dz_result.get("artist", {}).get("id")
|
||||
if dz_artist_id:
|
||||
pp["id_tags"]["DEEZER_ARTIST_ID"] = str(dz_artist_id)
|
||||
dz_details = dz_client.get_track_details(dz_track_id)
|
||||
if dz_details:
|
||||
bpm_val = dz_details.get("bpm")
|
||||
if bpm_val and bpm_val > 0:
|
||||
pp["deezer_bpm"] = bpm_val
|
||||
dz_isrc = dz_details.get("isrc")
|
||||
if dz_isrc:
|
||||
pp["deezer_isrc"] = dz_isrc
|
||||
if not pp["release_year"]:
|
||||
dz_album = dz_result.get("album", {})
|
||||
dz_release = (dz_album.get("release_date", "") if isinstance(dz_album, dict) else "") or ""
|
||||
if len(dz_release) >= 4 and dz_release[:4].isdigit():
|
||||
pp["release_year"] = dz_release[:4]
|
||||
except Exception as exc:
|
||||
logger.error("Deezer lookup failed (non-fatal): %s", exc)
|
||||
deezer_worker = getattr(runtime, "deezer_worker", None)
|
||||
dz_client = deezer_worker.client if deezer_worker else None
|
||||
if not dz_client:
|
||||
return
|
||||
dz_result = _call_source_lookup("Deezer track", dz_client.search_track, artist_name, track_title)
|
||||
if dz_result and _names_match(dz_result.get("title", ""), track_title) and _names_match(dz_result.get("artist", {}).get("name", ""), artist_name):
|
||||
dz_track_id = dz_result["id"]
|
||||
pp["id_tags"]["DEEZER_TRACK_ID"] = str(dz_track_id)
|
||||
dz_artist_id = dz_result.get("artist", {}).get("id")
|
||||
if dz_artist_id:
|
||||
pp["id_tags"]["DEEZER_ARTIST_ID"] = str(dz_artist_id)
|
||||
dz_details = _call_source_lookup("Deezer track details", dz_client.get_track_details, dz_track_id)
|
||||
if dz_details:
|
||||
bpm_val = dz_details.get("bpm")
|
||||
if bpm_val and bpm_val > 0:
|
||||
pp["deezer_bpm"] = bpm_val
|
||||
dz_isrc = dz_details.get("isrc")
|
||||
if dz_isrc:
|
||||
pp["deezer_isrc"] = dz_isrc
|
||||
if not pp["release_year"]:
|
||||
dz_album = dz_result.get("album", {})
|
||||
dz_release = (dz_album.get("release_date", "") if isinstance(dz_album, dict) else "") or ""
|
||||
if len(dz_release) >= 4 and dz_release[:4].isdigit():
|
||||
pp["release_year"] = dz_release[:4]
|
||||
|
||||
|
||||
def _process_audiodb_source(pp: dict, metadata: dict, cfg, runtime, track_title: str, artist_name: str) -> None:
|
||||
|
|
@ -350,29 +384,26 @@ def _process_audiodb_source(pp: dict, metadata: dict, cfg, runtime, track_title:
|
|||
if not track_title or not artist_name:
|
||||
return
|
||||
|
||||
try:
|
||||
audiodb_worker = getattr(runtime, "audiodb_worker", None)
|
||||
adb_client = audiodb_worker.client if audiodb_worker else None
|
||||
if not adb_client:
|
||||
return
|
||||
adb_result = adb_client.search_track(artist_name, track_title)
|
||||
if adb_result and _names_match(adb_result.get("strTrack", ""), track_title) and _names_match(adb_result.get("strArtist", ""), artist_name):
|
||||
adb_track_id = adb_result.get("idTrack")
|
||||
if adb_track_id:
|
||||
pp["id_tags"]["AUDIODB_TRACK_ID"] = str(adb_track_id)
|
||||
adb_mb_track = adb_result.get("strMusicBrainzID")
|
||||
if adb_mb_track and "MUSICBRAINZ_RECORDING_ID" not in pp["id_tags"]:
|
||||
pp["id_tags"]["MUSICBRAINZ_RECORDING_ID"] = adb_mb_track
|
||||
pp["recording_mbid"] = adb_mb_track
|
||||
adb_mb_artist = adb_result.get("strMusicBrainzArtistID")
|
||||
if adb_mb_artist and "MUSICBRAINZ_ARTIST_ID" not in pp["id_tags"]:
|
||||
pp["id_tags"]["MUSICBRAINZ_ARTIST_ID"] = adb_mb_artist
|
||||
pp["artist_mbid"] = adb_mb_artist
|
||||
pp["audiodb_mood"] = adb_result.get("strMood") or None
|
||||
pp["audiodb_style"] = adb_result.get("strStyle") or None
|
||||
pp["audiodb_genre"] = adb_result.get("strGenre") or None
|
||||
except Exception as exc:
|
||||
logger.error("AudioDB lookup failed (non-fatal): %s", exc)
|
||||
audiodb_worker = getattr(runtime, "audiodb_worker", None)
|
||||
adb_client = audiodb_worker.client if audiodb_worker else None
|
||||
if not adb_client:
|
||||
return
|
||||
adb_result = _call_source_lookup("AudioDB track", adb_client.search_track, artist_name, track_title)
|
||||
if adb_result and _names_match(adb_result.get("strTrack", ""), track_title) and _names_match(adb_result.get("strArtist", ""), artist_name):
|
||||
adb_track_id = adb_result.get("idTrack")
|
||||
if adb_track_id:
|
||||
pp["id_tags"]["AUDIODB_TRACK_ID"] = str(adb_track_id)
|
||||
adb_mb_track = adb_result.get("strMusicBrainzID")
|
||||
if adb_mb_track and "MUSICBRAINZ_RECORDING_ID" not in pp["id_tags"]:
|
||||
pp["id_tags"]["MUSICBRAINZ_RECORDING_ID"] = adb_mb_track
|
||||
pp["recording_mbid"] = adb_mb_track
|
||||
adb_mb_artist = adb_result.get("strMusicBrainzArtistID")
|
||||
if adb_mb_artist and "MUSICBRAINZ_ARTIST_ID" not in pp["id_tags"]:
|
||||
pp["id_tags"]["MUSICBRAINZ_ARTIST_ID"] = adb_mb_artist
|
||||
pp["artist_mbid"] = adb_mb_artist
|
||||
pp["audiodb_mood"] = adb_result.get("strMood") or None
|
||||
pp["audiodb_style"] = adb_result.get("strStyle") or None
|
||||
pp["audiodb_genre"] = adb_result.get("strGenre") or None
|
||||
|
||||
|
||||
def _process_tidal_source(pp: dict, metadata: dict, cfg, runtime, track_title: str, artist_name: str) -> None:
|
||||
|
|
@ -381,35 +412,32 @@ def _process_tidal_source(pp: dict, metadata: dict, cfg, runtime, track_title: s
|
|||
if not track_title or not artist_name:
|
||||
return
|
||||
|
||||
try:
|
||||
tidal_client = getattr(runtime, "tidal_client", None)
|
||||
if not (tidal_client and tidal_client.is_authenticated()):
|
||||
return
|
||||
td_result = tidal_client.search_track(artist_name, track_title)
|
||||
if td_result and _names_match(td_result.get("title", ""), track_title):
|
||||
td_track_id = td_result.get("id")
|
||||
if td_track_id:
|
||||
pp["id_tags"]["TIDAL_TRACK_ID"] = str(td_track_id)
|
||||
td_artist = td_result.get("artist", {})
|
||||
if isinstance(td_artist, dict) and td_artist.get("id"):
|
||||
pp["id_tags"]["TIDAL_ARTIST_ID"] = str(td_artist["id"])
|
||||
if td_track_id:
|
||||
td_details = tidal_client.get_track(str(td_track_id))
|
||||
if td_details:
|
||||
pp["tidal_isrc"] = td_details.get("isrc")
|
||||
td_copyright = td_details.get("copyright")
|
||||
if isinstance(td_copyright, dict):
|
||||
td_copyright = td_copyright.get("text", td_copyright.get("name", ""))
|
||||
pp["tidal_copyright"] = td_copyright or None
|
||||
if not pp["release_year"]:
|
||||
td_album = td_result.get("album", {})
|
||||
td_release = ""
|
||||
if isinstance(td_album, dict):
|
||||
td_release = str(td_album.get("release_date", "") or td_album.get("releaseDate", "") or "")
|
||||
if len(td_release) >= 4 and td_release[:4].isdigit():
|
||||
pp["release_year"] = td_release[:4]
|
||||
except Exception as exc:
|
||||
logger.error("Tidal lookup failed (non-fatal): %s", exc)
|
||||
tidal_client = getattr(runtime, "tidal_client", None)
|
||||
if not (tidal_client and tidal_client.is_authenticated()):
|
||||
return
|
||||
td_result = _call_source_lookup("Tidal track", tidal_client.search_track, artist_name, track_title)
|
||||
if td_result and _names_match(td_result.get("title", ""), track_title):
|
||||
td_track_id = td_result.get("id")
|
||||
if td_track_id:
|
||||
pp["id_tags"]["TIDAL_TRACK_ID"] = str(td_track_id)
|
||||
td_artist = td_result.get("artist", {})
|
||||
if isinstance(td_artist, dict) and td_artist.get("id"):
|
||||
pp["id_tags"]["TIDAL_ARTIST_ID"] = str(td_artist["id"])
|
||||
if td_track_id:
|
||||
td_details = _call_source_lookup("Tidal track details", tidal_client.get_track, str(td_track_id))
|
||||
if td_details:
|
||||
pp["tidal_isrc"] = td_details.get("isrc")
|
||||
td_copyright = td_details.get("copyright")
|
||||
if isinstance(td_copyright, dict):
|
||||
td_copyright = td_copyright.get("text", td_copyright.get("name", ""))
|
||||
pp["tidal_copyright"] = td_copyright or None
|
||||
if not pp["release_year"]:
|
||||
td_album = td_result.get("album", {})
|
||||
td_release = ""
|
||||
if isinstance(td_album, dict):
|
||||
td_release = str(td_album.get("release_date", "") or td_album.get("releaseDate", "") or "")
|
||||
if len(td_release) >= 4 and td_release[:4].isdigit():
|
||||
pp["release_year"] = td_release[:4]
|
||||
|
||||
|
||||
def _process_qobuz_source(pp: dict, metadata: dict, cfg, runtime, track_title: str, artist_name: str) -> None:
|
||||
|
|
@ -418,50 +446,47 @@ def _process_qobuz_source(pp: dict, metadata: dict, cfg, runtime, track_title: s
|
|||
if not track_title or not artist_name:
|
||||
return
|
||||
|
||||
try:
|
||||
qobuz_worker = getattr(runtime, "qobuz_enrichment_worker", None)
|
||||
qz_client = qobuz_worker.client if qobuz_worker else None
|
||||
if not (qz_client and qz_client.is_authenticated()):
|
||||
return
|
||||
qz_result = qz_client.search_track(artist_name, track_title)
|
||||
if qz_result:
|
||||
qz_performer = qz_result.get("performer") or {}
|
||||
if not isinstance(qz_performer, dict):
|
||||
qz_performer = {}
|
||||
qz_artist_name = qz_performer.get("name", "")
|
||||
if _names_match(qz_result.get("title", ""), track_title) and _names_match(qz_artist_name, artist_name):
|
||||
qz_track_id = qz_result.get("id")
|
||||
if qz_track_id:
|
||||
pp["id_tags"]["QOBUZ_TRACK_ID"] = str(qz_track_id)
|
||||
if qz_performer.get("id"):
|
||||
pp["id_tags"]["QOBUZ_ARTIST_ID"] = str(qz_performer["id"])
|
||||
qz_isrc = qz_result.get("isrc")
|
||||
if isinstance(qz_isrc, dict):
|
||||
qz_isrc = qz_isrc.get("value", qz_isrc.get("id", ""))
|
||||
if qz_isrc:
|
||||
pp["qobuz_isrc"] = qz_isrc
|
||||
qz_copyright = qz_result.get("copyright")
|
||||
if isinstance(qz_copyright, dict):
|
||||
qz_copyright = qz_copyright.get("text", qz_copyright.get("name", ""))
|
||||
if isinstance(qz_copyright, str):
|
||||
pp["qobuz_copyright"] = qz_copyright
|
||||
qz_album = qz_result.get("album", {})
|
||||
if isinstance(qz_album, dict):
|
||||
qz_label_info = qz_album.get("label", {})
|
||||
if isinstance(qz_label_info, dict) and qz_label_info.get("name"):
|
||||
pp["qobuz_label"] = qz_label_info["name"]
|
||||
if not pp["release_year"]:
|
||||
qz_release = str(qz_album.get("release_date_original", "") or "")
|
||||
if not qz_release:
|
||||
qz_ts = qz_album.get("released_at")
|
||||
if qz_ts and isinstance(qz_ts, (int, float)) and qz_ts > 0:
|
||||
import datetime as _dt
|
||||
qobuz_worker = getattr(runtime, "qobuz_enrichment_worker", None)
|
||||
qz_client = qobuz_worker.client if qobuz_worker else None
|
||||
if not (qz_client and qz_client.is_authenticated()):
|
||||
return
|
||||
qz_result = _call_source_lookup("Qobuz track", qz_client.search_track, artist_name, track_title)
|
||||
if qz_result:
|
||||
qz_performer = qz_result.get("performer") or {}
|
||||
if not isinstance(qz_performer, dict):
|
||||
qz_performer = {}
|
||||
qz_artist_name = qz_performer.get("name", "")
|
||||
if _names_match(qz_result.get("title", ""), track_title) and _names_match(qz_artist_name, artist_name):
|
||||
qz_track_id = qz_result.get("id")
|
||||
if qz_track_id:
|
||||
pp["id_tags"]["QOBUZ_TRACK_ID"] = str(qz_track_id)
|
||||
if qz_performer.get("id"):
|
||||
pp["id_tags"]["QOBUZ_ARTIST_ID"] = str(qz_performer["id"])
|
||||
qz_isrc = qz_result.get("isrc")
|
||||
if isinstance(qz_isrc, dict):
|
||||
qz_isrc = qz_isrc.get("value", qz_isrc.get("id", ""))
|
||||
if qz_isrc:
|
||||
pp["qobuz_isrc"] = qz_isrc
|
||||
qz_copyright = qz_result.get("copyright")
|
||||
if isinstance(qz_copyright, dict):
|
||||
qz_copyright = qz_copyright.get("text", qz_copyright.get("name", ""))
|
||||
if isinstance(qz_copyright, str):
|
||||
pp["qobuz_copyright"] = qz_copyright
|
||||
qz_album = qz_result.get("album", {})
|
||||
if isinstance(qz_album, dict):
|
||||
qz_label_info = qz_album.get("label", {})
|
||||
if isinstance(qz_label_info, dict) and qz_label_info.get("name"):
|
||||
pp["qobuz_label"] = qz_label_info["name"]
|
||||
if not pp["release_year"]:
|
||||
qz_release = str(qz_album.get("release_date_original", "") or "")
|
||||
if not qz_release:
|
||||
qz_ts = qz_album.get("released_at")
|
||||
if qz_ts and isinstance(qz_ts, (int, float)) and qz_ts > 0:
|
||||
import datetime as _dt
|
||||
|
||||
qz_release = str(_dt.datetime.utcfromtimestamp(qz_ts).year)
|
||||
if len(qz_release) >= 4 and qz_release[:4].isdigit():
|
||||
pp["release_year"] = qz_release[:4]
|
||||
except Exception as exc:
|
||||
logger.error("Qobuz lookup failed (non-fatal): %s", exc)
|
||||
qz_release = str(_dt.datetime.utcfromtimestamp(qz_ts).year)
|
||||
if len(qz_release) >= 4 and qz_release[:4].isdigit():
|
||||
pp["release_year"] = qz_release[:4]
|
||||
|
||||
|
||||
def _process_lastfm_source(pp: dict, metadata: dict, cfg, runtime, track_title: str, artist_name: str) -> None:
|
||||
|
|
@ -470,25 +495,22 @@ def _process_lastfm_source(pp: dict, metadata: dict, cfg, runtime, track_title:
|
|||
if not track_title or not artist_name:
|
||||
return
|
||||
|
||||
try:
|
||||
lastfm_worker = getattr(runtime, "lastfm_worker", None)
|
||||
lf_client = lastfm_worker.client if lastfm_worker else None
|
||||
if not lf_client:
|
||||
return
|
||||
lf_result = lf_client.get_track_info(artist_name, track_title)
|
||||
if lf_result:
|
||||
lf_url = lf_result.get("url")
|
||||
if lf_url:
|
||||
pp["lastfm_url"] = lf_url
|
||||
lf_toptags = lf_result.get("toptags", {})
|
||||
if isinstance(lf_toptags, dict):
|
||||
tag_list = lf_toptags.get("tag", [])
|
||||
if isinstance(tag_list, list):
|
||||
pp["lastfm_tags"] = [tag.get("name", "") for tag in tag_list if isinstance(tag, dict) and tag.get("name")]
|
||||
elif isinstance(tag_list, dict) and tag_list.get("name"):
|
||||
pp["lastfm_tags"] = [tag_list["name"]]
|
||||
except Exception as exc:
|
||||
logger.error("Last.fm lookup failed (non-fatal): %s", exc)
|
||||
lastfm_worker = getattr(runtime, "lastfm_worker", None)
|
||||
lf_client = lastfm_worker.client if lastfm_worker else None
|
||||
if not lf_client:
|
||||
return
|
||||
lf_result = _call_source_lookup("Last.fm track", lf_client.get_track_info, artist_name, track_title)
|
||||
if lf_result:
|
||||
lf_url = lf_result.get("url")
|
||||
if lf_url:
|
||||
pp["lastfm_url"] = lf_url
|
||||
lf_toptags = lf_result.get("toptags", {})
|
||||
if isinstance(lf_toptags, dict):
|
||||
tag_list = lf_toptags.get("tag", [])
|
||||
if isinstance(tag_list, list):
|
||||
pp["lastfm_tags"] = [tag.get("name", "") for tag in tag_list if isinstance(tag, dict) and tag.get("name")]
|
||||
elif isinstance(tag_list, dict) and tag_list.get("name"):
|
||||
pp["lastfm_tags"] = [tag_list["name"]]
|
||||
|
||||
|
||||
def _process_genius_source(pp: dict, metadata: dict, cfg, runtime, track_title: str, artist_name: str) -> None:
|
||||
|
|
@ -497,26 +519,23 @@ def _process_genius_source(pp: dict, metadata: dict, cfg, runtime, track_title:
|
|||
if not track_title or not artist_name:
|
||||
return
|
||||
|
||||
try:
|
||||
import core.genius_client as _genius_module
|
||||
import core.genius_client as _genius_module
|
||||
|
||||
if time.time() < _genius_module._rate_limit_until:
|
||||
logger.info("Genius rate-limited, skipping (non-blocking)")
|
||||
return
|
||||
genius_worker = getattr(runtime, "genius_worker", None)
|
||||
g_client = genius_worker.client if genius_worker else None
|
||||
if not g_client:
|
||||
return
|
||||
g_result = g_client.search_song(artist_name, track_title)
|
||||
if g_result:
|
||||
g_id = g_result.get("id")
|
||||
if g_id:
|
||||
pp["id_tags"]["GENIUS_TRACK_ID"] = str(g_id)
|
||||
g_url = g_result.get("url")
|
||||
if g_url:
|
||||
pp["genius_url"] = g_url
|
||||
except Exception as exc:
|
||||
logger.error("Genius lookup failed (non-fatal): %s", exc)
|
||||
if time.time() < _genius_module._rate_limit_until:
|
||||
logger.info("Genius rate-limited, skipping (non-blocking)")
|
||||
return
|
||||
genius_worker = getattr(runtime, "genius_worker", None)
|
||||
g_client = genius_worker.client if genius_worker else None
|
||||
if not g_client:
|
||||
return
|
||||
g_result = _call_source_lookup("Genius track", g_client.search_song, artist_name, track_title)
|
||||
if g_result:
|
||||
g_id = g_result.get("id")
|
||||
if g_id:
|
||||
pp["id_tags"]["GENIUS_TRACK_ID"] = str(g_id)
|
||||
g_url = g_result.get("url")
|
||||
if g_url:
|
||||
pp["genius_url"] = g_url
|
||||
|
||||
|
||||
def _process_source_enrichment(source_name: str, pp: dict, metadata: dict, cfg, runtime, track_title: str, artist_name: str) -> None:
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ from __future__ import annotations
|
|||
|
||||
import threading
|
||||
import time
|
||||
from functools import wraps
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
matched_context_lock = threading.Lock()
|
||||
|
|
@ -20,6 +21,17 @@ activity_feed_lock = threading.Lock()
|
|||
_activity_toast_emitter = None
|
||||
|
||||
|
||||
def caller_must_hold_tasks_lock(func):
|
||||
"""Best-effort guard for helpers that mutate download_tasks in place."""
|
||||
@wraps(func)
|
||||
def wrapper(*args, **kwargs):
|
||||
if not tasks_lock.locked():
|
||||
raise RuntimeError(f"{func.__name__}() requires tasks_lock to be held by the caller")
|
||||
return func(*args, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
def set_activity_toast_emitter(emitter) -> None:
|
||||
"""Set the WebSocket-style emitter used by add_activity_item."""
|
||||
global _activity_toast_emitter
|
||||
|
|
@ -50,6 +62,7 @@ def add_activity_item(icon, title, subtitle, time_ago="Now", show_toast=True):
|
|||
return activity_item
|
||||
|
||||
|
||||
@caller_must_hold_tasks_lock
|
||||
def mark_task_completed(task_id: str, track_info: Optional[Dict[str, Any]] = None) -> bool:
|
||||
"""Mark a download task as completed.
|
||||
|
||||
|
|
|
|||
|
|
@ -1,3 +1,5 @@
|
|||
import os
|
||||
|
||||
import core.imports.album_naming as album_naming
|
||||
import core.imports.paths as import_paths
|
||||
|
||||
|
|
@ -69,7 +71,7 @@ def test_get_file_path_from_template_raw_handles_quality_and_disc_placeholders(m
|
|||
},
|
||||
)
|
||||
|
||||
assert folder_path == "Artist One/Album One"
|
||||
assert folder_path == os.path.join("Artist One", "Album One")
|
||||
assert filename == "3 - Song One [FLAC 16bit]"
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,10 @@
|
|||
from collections import OrderedDict
|
||||
import types
|
||||
import sqlite3
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from core.metadata import enrichment as me
|
||||
from core.metadata import artwork as ma
|
||||
from core.metadata import source as ms
|
||||
|
|
@ -350,6 +354,109 @@ def test_embed_source_ids_writes_musicbrainz_release_year_and_updates_album_year
|
|||
assert row["year"] == 2021
|
||||
|
||||
|
||||
def test_musicbrainz_release_lookup_failure_does_not_poison_cache(monkeypatch):
|
||||
class _FakeMBClient:
|
||||
def get_release(self, mbid, includes=None):
|
||||
return {}
|
||||
|
||||
class _FakeMBService:
|
||||
def __init__(self):
|
||||
self.release_calls = 0
|
||||
self.mb_client = _FakeMBClient()
|
||||
|
||||
def match_recording(self, title, artist):
|
||||
return None
|
||||
|
||||
def match_artist(self, artist):
|
||||
return {"mbid": "artist-mbid"}
|
||||
|
||||
def match_release(self, album, artist):
|
||||
self.release_calls += 1
|
||||
if self.release_calls == 1:
|
||||
raise requests.RequestException("temporary MusicBrainz outage")
|
||||
return {"mbid": "release-mbid"}
|
||||
|
||||
monkeypatch.setattr(ms, "get_config_manager", lambda: _Config({"musicbrainz.embed_tags": True}))
|
||||
monkeypatch.setattr(ms, "mb_release_cache", {})
|
||||
monkeypatch.setattr(ms, "mb_release_detail_cache", {})
|
||||
|
||||
service = _FakeMBService()
|
||||
runtime = types.SimpleNamespace(mb_worker=types.SimpleNamespace(mb_service=service))
|
||||
pp = {
|
||||
"id_tags": {},
|
||||
"track_title": "Song One",
|
||||
"artist_name": "Artist One",
|
||||
"batch_artist_name": "Artist One",
|
||||
"metadata": {"album": "Album One"},
|
||||
"recording_mbid": None,
|
||||
"artist_mbid": None,
|
||||
"release_mbid": "",
|
||||
"mb_genres": [],
|
||||
"isrc": None,
|
||||
"deezer_bpm": None,
|
||||
"deezer_isrc": None,
|
||||
"audiodb_mood": None,
|
||||
"audiodb_style": None,
|
||||
"audiodb_genre": None,
|
||||
"tidal_isrc": None,
|
||||
"tidal_copyright": None,
|
||||
"qobuz_isrc": None,
|
||||
"qobuz_copyright": None,
|
||||
"qobuz_label": None,
|
||||
"lastfm_tags": [],
|
||||
"lastfm_url": None,
|
||||
"genius_url": None,
|
||||
"release_year": None,
|
||||
}
|
||||
metadata = {"album": "Album One", "artist": "Artist One"}
|
||||
|
||||
ms._process_musicbrainz_source(pp, metadata, _Config({"musicbrainz.embed_tags": True}), runtime, "Song One", "Artist One")
|
||||
assert service.release_calls == 1
|
||||
assert ms.mb_release_cache == {}
|
||||
|
||||
poisoned_norm_key = (ms.normalize_album_cache_key("Album One"), "artist one")
|
||||
poisoned_exact_key = ("album one", "artist one")
|
||||
ms.mb_release_cache[poisoned_norm_key] = ""
|
||||
ms.mb_release_cache[poisoned_exact_key] = ""
|
||||
|
||||
ms._process_musicbrainz_source(pp, metadata, _Config({"musicbrainz.embed_tags": True}), runtime, "Song One", "Artist One")
|
||||
assert service.release_calls == 2
|
||||
|
||||
|
||||
def test_source_processors_do_not_swallow_programmer_errors():
|
||||
class _BoomClient:
|
||||
def search_track(self, artist_name, track_title):
|
||||
raise ValueError("boom")
|
||||
|
||||
runtime = types.SimpleNamespace(deezer_worker=types.SimpleNamespace(client=_BoomClient()))
|
||||
pp = {
|
||||
"id_tags": {},
|
||||
"batch_artist_name": "Artist One",
|
||||
"release_year": None,
|
||||
}
|
||||
metadata = {"album": "Album One"}
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
ms._process_deezer_source(pp, metadata, _Config({"deezer.embed_tags": True}), runtime, "Song One", "Artist One")
|
||||
|
||||
|
||||
def test_musicbrainz_caches_evict_oldest_entries():
|
||||
release_cache = OrderedDict()
|
||||
detail_cache = OrderedDict()
|
||||
|
||||
ms._bounded_cache_set(release_cache, ("album-1", "artist"), "release-1", 2)
|
||||
ms._bounded_cache_set(release_cache, ("album-2", "artist"), "release-2", 2)
|
||||
assert list(release_cache.keys()) == [("album-1", "artist"), ("album-2", "artist")]
|
||||
|
||||
assert ms._bounded_cache_get(release_cache, ("album-1", "artist")) == "release-1"
|
||||
ms._bounded_cache_set(release_cache, ("album-3", "artist"), "release-3", 2)
|
||||
assert list(release_cache.keys()) == [("album-1", "artist"), ("album-3", "artist")]
|
||||
|
||||
ms._bounded_cache_set(detail_cache, "release-1", {"title": "One"}, 1)
|
||||
ms._bounded_cache_set(detail_cache, "release-2", {"title": "Two"}, 1)
|
||||
assert list(detail_cache.keys()) == ["release-2"]
|
||||
|
||||
|
||||
def test_enhance_file_metadata_forwards_runtime_to_source_embedding(monkeypatch):
|
||||
audio = _FakeAudio()
|
||||
symbols = _fake_symbols(audio)
|
||||
|
|
|
|||
32
tests/test_runtime_state.py
Normal file
32
tests/test_runtime_state.py
Normal file
|
|
@ -0,0 +1,32 @@
|
|||
import pytest
|
||||
|
||||
import core.runtime_state as runtime_state
|
||||
|
||||
|
||||
def test_mark_task_completed_requires_tasks_lock():
|
||||
original_tasks = dict(runtime_state.download_tasks)
|
||||
runtime_state.download_tasks.clear()
|
||||
runtime_state.download_tasks["task-1"] = {"status": "running", "stream_processed": False}
|
||||
|
||||
try:
|
||||
with pytest.raises(RuntimeError, match="tasks_lock"):
|
||||
runtime_state.mark_task_completed("task-1")
|
||||
finally:
|
||||
runtime_state.download_tasks.clear()
|
||||
runtime_state.download_tasks.update(original_tasks)
|
||||
|
||||
|
||||
def test_mark_task_completed_succeeds_when_lock_held():
|
||||
original_tasks = dict(runtime_state.download_tasks)
|
||||
runtime_state.download_tasks.clear()
|
||||
runtime_state.download_tasks["task-1"] = {"status": "running", "stream_processed": False}
|
||||
|
||||
try:
|
||||
with runtime_state.tasks_lock:
|
||||
assert runtime_state.mark_task_completed("task-1", {"name": "Song One"}) is True
|
||||
assert runtime_state.download_tasks["task-1"]["status"] == "completed"
|
||||
assert runtime_state.download_tasks["task-1"]["stream_processed"] is True
|
||||
assert runtime_state.download_tasks["task-1"]["track_info"] == {"name": "Song One"}
|
||||
finally:
|
||||
runtime_state.download_tasks.clear()
|
||||
runtime_state.download_tasks.update(original_tasks)
|
||||
|
|
@ -1214,9 +1214,7 @@ function importPageRenderMatchList() {
|
|||
// Build rows
|
||||
let matchedCount = 0;
|
||||
const rows = data.matches.map((m, idx) => {
|
||||
const track = m.track || m.spotify_track || {};
|
||||
const trackNumber = track.track_number ?? track.trackNumber;
|
||||
const displayTrackNumber = trackNumber ? trackNumber : (idx + 1);
|
||||
const trackInfo = _importPageGetTrackDisplayInfo(m, idx);
|
||||
let file = null;
|
||||
let confidence = m.confidence;
|
||||
let isOverride = false;
|
||||
|
|
@ -1256,8 +1254,8 @@ function importPageRenderMatchList() {
|
|||
<div class="import-page-match-row ${file ? 'matched' : ''}"
|
||||
ondragover="importPageHandleDragOver(event)" ondragleave="this.classList.remove('drag-over')" ondrop="importPageHandleDrop(event, ${idx})"
|
||||
onclick="importPageTapAssign(${idx})">
|
||||
<span class="import-page-match-num">${displayTrackNumber}</span>
|
||||
<span class="import-page-match-track">${_esc(track.name || track.title || 'Unknown Track')}</span>
|
||||
<span class="import-page-match-num">${trackInfo.displayTrackNumber}</span>
|
||||
<span class="import-page-match-track">${_esc(trackInfo.name)}</span>
|
||||
<span class="import-page-match-file ${file ? 'has-file' : ''}">
|
||||
${file
|
||||
? `<span class="import-page-match-file-name">${_esc(file.filename)}</span>
|
||||
|
|
@ -1307,6 +1305,21 @@ function importPageRenderMatchList() {
|
|||
processBtn.textContent = `Process ${matchedCount} Track${matchedCount !== 1 ? 's' : ''}`;
|
||||
}
|
||||
|
||||
function _importPageGetTrackDisplayInfo(item, index) {
|
||||
const track = item?.track || item?.spotify_track || {};
|
||||
const rawTrackNumber = track.track_number ?? track.trackNumber ?? null;
|
||||
const trackNumber = rawTrackNumber === null || rawTrackNumber === undefined || rawTrackNumber === ''
|
||||
? null
|
||||
: String(rawTrackNumber).split('/')[0].trim();
|
||||
|
||||
return {
|
||||
track,
|
||||
name: track.name || track.title || `Track ${index + 1}`,
|
||||
trackNumber,
|
||||
displayTrackNumber: trackNumber || String(index + 1),
|
||||
};
|
||||
}
|
||||
|
||||
// --- Album Tab: Drag and Drop ---
|
||||
|
||||
function importPageStartDrag(event, stagingFileIndex) {
|
||||
|
|
@ -1672,7 +1685,7 @@ function _importQueueAdd(job) {
|
|||
async function _importQueueRunJob(entry, job) {
|
||||
for (let i = 0; i < job.items.length; i++) {
|
||||
const itemName = job.type === 'album'
|
||||
? (job.items[i].spotify_track?.name || `Track ${i + 1}`)
|
||||
? _importPageGetTrackDisplayInfo(job.items[i], i).name
|
||||
: (job.items[i].title || job.items[i].filename || `File ${i + 1}`);
|
||||
|
||||
// Update status with current track info
|
||||
|
|
|
|||
Loading…
Reference in a new issue