142a1aaf38
Sokhi (continued from #806): volume-numbered series ('B小町 …キャラクター ソングCD Vol.2' / 'Vol.2.5' / 'Vol.4' / 'Vol.4.5') got each other's art from both normal downloads and the retag tool. Two distinct holes, one principle: 1. The art picker's _album_matches validates by significant-token SUBSET — built to tolerate '(Deluxe)'/'- Remastered' suffixes. CJK strips out of the normalizer entirely, so Vol.4 → {b,tv,cd,vol,4}, a clean subset of Vol.4.5's {b,tv,cd,vol,4,5}: the wrong volume validated as "the same album with a suffix". Affected every fuzzy art source (iTunes, Deezer, AudioDB, Spotify) in downloads, retag, and the missing-art repair. 2. MusicBrainz match_release scores by string similarity — Vol.4 vs Vol.4.5 is 0.973, so the wrong volume could win the match outright, and its MBID then feeds Cover Art Archive with NO downstream validation (CAA is MBID-keyed, trusted by design). With Sokhi's MB metadata source this is the likely path in his logs (his release-group 404s push re-matching). The shared rule (core.text.title_match.numeric_tokens_differ): digit-bearing tokens must be IDENTICAL between the two titles. A number on one side only — volume, part, sequel, remaster year — is a different release, never a suffix. '1989' vs '1989 (Deluxe)' still matches (digits shared); 'Album' vs 'Album 2' now rejects (sequels!). Art picker rejects outright (falls through to next source / the download's own art — the designed cost of a false reject); MB matcher halves the candidate's confidence, landing it below the 70 gate while the exact-volume result is untouched. Tests: helper truth table, the exact reported pairs through _album_matches, and match_release end-to-end (wrong volume alone → no match beats a wrong MBID; exact volume beats near-identical wrong one despite lower MB score). 828 matching/metadata + 301 musicbrainz/retag/artwork tests pass.
764 lines
32 KiB
Python
764 lines
32 KiB
Python
from typing import Optional, Dict, Any
|
|
import json
|
|
import re
|
|
from datetime import datetime, timedelta
|
|
from difflib import SequenceMatcher
|
|
from utils.logging_config import get_logger
|
|
from core.musicbrainz_client import MusicBrainzClient
|
|
from database.music_database import MusicDatabase
|
|
|
|
logger = get_logger("musicbrainz_service")
|
|
|
|
class MusicBrainzService:
|
|
"""Service layer for MusicBrainz integration with caching and matching logic"""
|
|
|
|
def __init__(self, database: MusicDatabase, app_name: str = "SoulSync", app_version: str = "1.0", contact_email: str = ""):
|
|
self.db = database
|
|
self.mb_client = MusicBrainzClient(app_name, app_version, contact_email)
|
|
self.retry_days = 30 # Retry 'not_found' items after 30 days
|
|
|
|
def _calculate_similarity(self, str1: str, str2: str) -> float:
|
|
"""Calculate string similarity score (0.0 to 1.0)"""
|
|
if not str1 or not str2:
|
|
return 0.0
|
|
|
|
# Normalize for comparison
|
|
s1 = str1.lower().strip()
|
|
s2 = str2.lower().strip()
|
|
|
|
if s1 == s2:
|
|
return 1.0
|
|
|
|
return SequenceMatcher(None, s1, s2).ratio()
|
|
|
|
def _check_cache(self, entity_type: str, entity_name: str, artist_name: Optional[str] = None) -> Optional[Dict[str, Any]]:
|
|
"""Check if we have a cached MusicBrainz result"""
|
|
conn = None
|
|
try:
|
|
conn = self.db._get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
# Fix: Match exact artist_name (not OR artist_name IS NULL)
|
|
# This prevents getting wrong cached results
|
|
if artist_name is not None:
|
|
cursor.execute("""
|
|
SELECT musicbrainz_id, metadata_json, match_confidence, last_updated
|
|
FROM musicbrainz_cache
|
|
WHERE entity_type = ? AND entity_name = ? AND artist_name = ?
|
|
ORDER BY last_updated DESC
|
|
LIMIT 1
|
|
""", (entity_type, entity_name, artist_name))
|
|
else:
|
|
cursor.execute("""
|
|
SELECT musicbrainz_id, metadata_json, match_confidence, last_updated
|
|
FROM musicbrainz_cache
|
|
WHERE entity_type = ? AND entity_name = ? AND artist_name IS NULL
|
|
ORDER BY last_updated DESC
|
|
LIMIT 1
|
|
""", (entity_type, entity_name))
|
|
|
|
row = cursor.fetchone()
|
|
|
|
if row:
|
|
# Shorter TTL for null results (failed lookups) so they get retried sooner
|
|
last_updated = datetime.fromisoformat(row[3]) if row[3] else None
|
|
ttl_days = 30 if row[0] is None else 90 # row[0] is musicbrainz_id
|
|
if last_updated and (datetime.now() - last_updated).days > ttl_days:
|
|
logger.debug(f"Cache entry for {entity_type} '{entity_name}' is stale (> {ttl_days} days)")
|
|
return None
|
|
|
|
# Parse JSON with error handling
|
|
try:
|
|
metadata = json.loads(row[1]) if row[1] else None
|
|
except json.JSONDecodeError:
|
|
logger.warning(f"Invalid JSON in cache for {entity_type} '{entity_name}', ignoring")
|
|
metadata = None
|
|
|
|
return {
|
|
'musicbrainz_id': row[0],
|
|
'metadata': metadata,
|
|
'confidence': row[2]
|
|
}
|
|
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error checking cache: {e}")
|
|
return None
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
def _save_to_cache(self, entity_type: str, entity_name: str, artist_name: Optional[str],
|
|
musicbrainz_id: Optional[str], metadata: Optional[Dict], confidence: int):
|
|
"""Save MusicBrainz result to cache"""
|
|
conn = None
|
|
try:
|
|
conn = self.db._get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
metadata_json = json.dumps(metadata) if metadata else None
|
|
|
|
cursor.execute("""
|
|
INSERT OR REPLACE INTO musicbrainz_cache
|
|
(entity_type, entity_name, artist_name, musicbrainz_id, metadata_json, match_confidence, last_updated)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""", (entity_type, entity_name, artist_name, musicbrainz_id, metadata_json, confidence, datetime.now()))
|
|
|
|
conn.commit()
|
|
|
|
logger.debug(f"Cached {entity_type} '{entity_name}' (MBID: {musicbrainz_id}, confidence: {confidence})")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error saving to cache: {e}")
|
|
if conn:
|
|
conn.rollback()
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
def match_artist(self, artist_name: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Match an artist by name to MusicBrainz
|
|
|
|
Returns:
|
|
Dict with 'mbid', 'name', 'confidence' or None if no good match
|
|
"""
|
|
# Check cache first
|
|
cached = self._check_cache('artist', artist_name)
|
|
if cached:
|
|
logger.debug(f"Cache hit for artist '{artist_name}'")
|
|
return {
|
|
'mbid': cached['musicbrainz_id'],
|
|
'name': artist_name,
|
|
'confidence': cached['confidence'],
|
|
'cached': True
|
|
}
|
|
|
|
# Search MusicBrainz
|
|
try:
|
|
results = self.mb_client.search_artist(artist_name, limit=5)
|
|
|
|
if not results:
|
|
logger.info(f"No MusicBrainz results for artist '{artist_name}'")
|
|
self._save_to_cache('artist', artist_name, None, None, None, 0)
|
|
return None
|
|
|
|
# Find best match
|
|
best_match = None
|
|
best_confidence = 0
|
|
|
|
for result in results:
|
|
mb_name = result.get('name', '')
|
|
mb_score = result.get('score', 0) # MusicBrainz search score
|
|
|
|
# Calculate our own similarity
|
|
similarity = self._calculate_similarity(artist_name, mb_name)
|
|
|
|
# Combine MusicBrainz score with our similarity (weighted)
|
|
# Cap at 100 to prevent edge cases where MB score > 100
|
|
confidence = min(100, int((similarity * 60) + (mb_score / 100 * 40)))
|
|
|
|
if confidence > best_confidence:
|
|
best_confidence = confidence
|
|
best_match = result
|
|
|
|
# Only return matches with confidence >= 70%
|
|
if best_match and best_confidence >= 70:
|
|
mbid = best_match.get('id')
|
|
mb_name = best_match.get('name')
|
|
|
|
# Save to cache
|
|
self._save_to_cache('artist', artist_name, None, mbid, best_match, best_confidence)
|
|
|
|
logger.info(f"Matched artist '{artist_name}' → '{mb_name}' (MBID: {mbid}, confidence: {best_confidence})")
|
|
|
|
return {
|
|
'mbid': mbid,
|
|
'name': mb_name,
|
|
'confidence': best_confidence,
|
|
'cached': False
|
|
}
|
|
else:
|
|
logger.info(f"Low confidence match for artist '{artist_name}' (best: {best_confidence})")
|
|
self._save_to_cache('artist', artist_name, None, None, None, best_confidence)
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error matching artist '{artist_name}': {e}")
|
|
return None
|
|
|
|
# Version qualifiers that distinguish releases (Deluxe, Remastered, etc.)
|
|
_VERSION_QUALIFIERS = re.compile(
|
|
r'\b(deluxe|expanded|remaster(?:ed)?|anniversary|special|collector|'
|
|
r'limited|bonus|platinum|gold|super\s*deluxe|standard)\b',
|
|
re.IGNORECASE
|
|
)
|
|
|
|
def _extract_version_qualifier(self, title: str) -> str:
|
|
"""Extract version qualifiers from an album title, normalized and sorted."""
|
|
qualifiers = sorted(set(q.lower() for q in self._VERSION_QUALIFIERS.findall(title)))
|
|
return ' '.join(qualifiers)
|
|
|
|
def match_release(self, album_name: str, artist_name: Optional[str] = None) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Match a release (album) by name to MusicBrainz
|
|
|
|
Returns:
|
|
Dict with 'mbid', 'title', 'confidence' or None if no good match
|
|
"""
|
|
# Check cache first
|
|
cached = self._check_cache('release', album_name, artist_name)
|
|
if cached:
|
|
logger.debug(f"Cache hit for release '{album_name}'")
|
|
return {
|
|
'mbid': cached['musicbrainz_id'],
|
|
'title': album_name,
|
|
'confidence': cached['confidence'],
|
|
'cached': True
|
|
}
|
|
|
|
# Search MusicBrainz
|
|
try:
|
|
results = self.mb_client.search_release(album_name, artist_name, limit=5)
|
|
|
|
if not results:
|
|
logger.info(f"No MusicBrainz results for release '{album_name}'")
|
|
self._save_to_cache('release', album_name, artist_name, None, None, 0)
|
|
return None
|
|
|
|
# Extract version qualifier from search query for preference matching
|
|
query_qualifier = self._extract_version_qualifier(album_name)
|
|
|
|
# Find best match
|
|
best_match = None
|
|
best_confidence = 0
|
|
|
|
for result in results:
|
|
mb_title = result.get('title', '')
|
|
mb_score = result.get('score', 0)
|
|
|
|
# Calculate title similarity
|
|
title_similarity = self._calculate_similarity(album_name, mb_title)
|
|
|
|
# If we have artist info, check artist match too
|
|
artist_bonus = 0
|
|
if artist_name and 'artist-credit' in result:
|
|
artist_credits = result['artist-credit']
|
|
for credit in artist_credits:
|
|
if isinstance(credit, dict) and 'artist' in credit:
|
|
mb_artist = credit['artist'].get('name', '')
|
|
artist_similarity = self._calculate_similarity(artist_name, mb_artist)
|
|
if artist_similarity > 0.7:
|
|
artist_bonus = 20
|
|
break
|
|
|
|
# Version qualifier matching: prefer releases with the same
|
|
# edition qualifier (Deluxe, Remastered, etc.) as the query.
|
|
# This prevents "Playing the Angel (Deluxe)" from matching the
|
|
# standard "Playing the Angel" release.
|
|
version_bonus = 0
|
|
if query_qualifier:
|
|
mb_qualifier = self._extract_version_qualifier(mb_title)
|
|
if query_qualifier == mb_qualifier:
|
|
version_bonus = 10 # Same edition — strong preference
|
|
elif mb_qualifier and mb_qualifier in query_qualifier:
|
|
version_bonus = 5 # Partial match (e.g. "deluxe" in "super deluxe")
|
|
elif not mb_qualifier:
|
|
version_bonus = -5 # Query has qualifier but result doesn't — penalize
|
|
|
|
# Combine scores - cap at 100
|
|
confidence = min(100, int((title_similarity * 50) + (mb_score / 100 * 30) + artist_bonus + version_bonus))
|
|
|
|
# Numeric difference = different release. 'Vol.4' vs 'Vol.4.5'
|
|
# scores 0.97 string similarity, so a near-identical wrong
|
|
# volume could win and its MBID then feeds CAA art with NO
|
|
# downstream validation (CAA is MBID-keyed — Sokhi's wrong
|
|
# covers). Halving lands any such candidate below the 70 gate
|
|
# while leaving the exact-volume result untouched.
|
|
from core.text.title_match import numeric_tokens_differ
|
|
if numeric_tokens_differ(album_name, mb_title):
|
|
confidence = int(confidence * 0.5)
|
|
|
|
if confidence > best_confidence:
|
|
best_confidence = confidence
|
|
best_match = result
|
|
|
|
# Only return matches with confidence >= 70%
|
|
if best_match and best_confidence >= 70:
|
|
mbid = best_match.get('id')
|
|
mb_title = best_match.get('title')
|
|
|
|
# Save to cache
|
|
self._save_to_cache('release', album_name, artist_name, mbid, best_match, best_confidence)
|
|
|
|
logger.info(f"Matched release '{album_name}' → '{mb_title}' (MBID: {mbid}, confidence: {best_confidence})")
|
|
|
|
return {
|
|
'mbid': mbid,
|
|
'title': mb_title,
|
|
'confidence': best_confidence,
|
|
'cached': False
|
|
}
|
|
else:
|
|
logger.info(f"Low confidence match for release '{album_name}' (best: {best_confidence})")
|
|
self._save_to_cache('release', album_name, artist_name, None, None, best_confidence)
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error matching release '{album_name}': {e}")
|
|
return None
|
|
|
|
def match_recording(self, track_name: str, artist_name: Optional[str] = None) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Match a recording (track) by name to MusicBrainz
|
|
|
|
Returns:
|
|
Dict with 'mbid', 'title', 'confidence' or None if no good match
|
|
"""
|
|
# Check cache first
|
|
cached = self._check_cache('recording', track_name, artist_name)
|
|
if cached:
|
|
logger.debug(f"Cache hit for recording '{track_name}'")
|
|
return {
|
|
'mbid': cached['musicbrainz_id'],
|
|
'title': track_name,
|
|
'confidence': cached['confidence'],
|
|
'cached': True
|
|
}
|
|
|
|
# Search MusicBrainz
|
|
try:
|
|
results = self.mb_client.search_recording(track_name, artist_name, limit=5)
|
|
|
|
if not results:
|
|
logger.info(f"No MusicBrainz results for recording '{track_name}'")
|
|
self._save_to_cache('recording', track_name, artist_name, None, None, 0)
|
|
return None
|
|
|
|
# Find best match
|
|
best_match = None
|
|
best_confidence = 0
|
|
|
|
for result in results:
|
|
mb_title = result.get('title', '')
|
|
mb_score = result.get('score', 0)
|
|
|
|
# Calculate title similarity
|
|
title_similarity = self._calculate_similarity(track_name, mb_title)
|
|
|
|
# Hard gate: title must be at least 60% similar.
|
|
# Without this, artist bonus + MB score can push totally
|
|
# different titles (e.g. "Sweet Surrender" → "Answers")
|
|
# past the confidence threshold.
|
|
if title_similarity < 0.6:
|
|
continue
|
|
|
|
# If we have artist info, check artist match too
|
|
artist_bonus = 0
|
|
if artist_name and 'artist-credit' in result:
|
|
artist_credits = result['artist-credit']
|
|
for credit in artist_credits:
|
|
if isinstance(credit, dict) and 'artist' in credit:
|
|
mb_artist = credit['artist'].get('name', '')
|
|
artist_similarity = self._calculate_similarity(artist_name, mb_artist)
|
|
if artist_similarity > 0.7:
|
|
artist_bonus = 20
|
|
break
|
|
|
|
# Combine scores - cap at 100
|
|
confidence = min(100, int((title_similarity * 50) + (mb_score / 100 * 30) + artist_bonus))
|
|
|
|
if confidence > best_confidence:
|
|
best_confidence = confidence
|
|
best_match = result
|
|
|
|
# Only return matches with confidence >= 70%
|
|
if best_match and best_confidence >= 70:
|
|
mbid = best_match.get('id')
|
|
mb_title = best_match.get('title')
|
|
|
|
# Save to cache
|
|
self._save_to_cache('recording', track_name, artist_name, mbid, best_match, best_confidence)
|
|
|
|
logger.info(f"Matched recording '{track_name}' → '{mb_title}' (MBID: {mbid}, confidence: {best_confidence})")
|
|
|
|
return {
|
|
'mbid': mbid,
|
|
'title': mb_title,
|
|
'confidence': best_confidence,
|
|
'cached': False
|
|
}
|
|
else:
|
|
logger.info(f"Low confidence match for recording '{track_name}' (best: {best_confidence})")
|
|
self._save_to_cache('recording', track_name, artist_name, None, None, best_confidence)
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error matching recording '{track_name}': {e}")
|
|
return None
|
|
|
|
def lookup_artist_aliases(self, artist_name: str) -> list:
|
|
"""Find alternate-spelling aliases for an artist by NAME.
|
|
|
|
Multi-tier resolution:
|
|
1. Library DB row (`artists.aliases` populated by the MB
|
|
worker when the artist was enriched). Fast path — no
|
|
network.
|
|
2. Existing musicbrainz_cache entry (entity_type='artist_aliases')
|
|
— caches a prior live MB lookup for this name.
|
|
3. Live MB lookup: search artist → fetch aliases for the best
|
|
MBID → cache the result.
|
|
|
|
Always returns a list (possibly empty) — never raises. Empty
|
|
result on any tier means "no alternate spellings found, fall
|
|
back to direct match" which is identical to pre-fix behaviour.
|
|
|
|
Used by the AcoustID verifier when an artist comparison fails
|
|
the direct similarity check. Caching means each unique artist
|
|
name only hits MB once per cache TTL even if 100 download
|
|
candidates fail verification with that artist.
|
|
"""
|
|
if not artist_name:
|
|
return []
|
|
|
|
# Tier 1: library DB
|
|
library = self.get_artist_aliases(artist_name)
|
|
if library:
|
|
return library
|
|
|
|
# Tier 2: cached live lookup (re-uses musicbrainz_cache table)
|
|
cached = self._check_cache('artist_aliases', artist_name)
|
|
if cached:
|
|
metadata = cached.get('metadata') or {}
|
|
aliases = metadata.get('aliases') if isinstance(metadata, dict) else None
|
|
if isinstance(aliases, list):
|
|
return [str(x).strip() for x in aliases if x]
|
|
# Cache hit with empty result — respect it (don't re-query)
|
|
return []
|
|
|
|
# Tier 3: live MB lookup. Search → fetch by MBID → cache.
|
|
# Issue #586 — strict search queries `artist:"..."` only and
|
|
# MISSES alias / sortname indexes. When MB's canonical name is
|
|
# the non-Latin form (e.g. `Дмитрий Яблонский`), the user's
|
|
# Latin input ("Dmitry Yablonsky") finds nothing under strict.
|
|
# Fall back to non-strict (bare query, hits alias + sortname
|
|
# indexes) when strict returns empty OR all results fail the
|
|
# trust gate.
|
|
scored = self._search_and_score_artists(artist_name, strict=True)
|
|
if not scored or self._best_score(scored) < 0.85:
|
|
non_strict = self._search_and_score_artists(artist_name, strict=False)
|
|
if non_strict and (not scored or self._best_score(non_strict) > self._best_score(scored)):
|
|
scored = non_strict
|
|
|
|
if not scored:
|
|
self._save_to_cache('artist_aliases', artist_name, None, None, {'aliases': []}, 0)
|
|
return []
|
|
|
|
scored.sort(key=lambda x: -x[0])
|
|
best_score, best_mbid, best_mb_score = scored[0]
|
|
|
|
# The genuine cross-script match (romaji↔kanji, latin↔cyrillic)
|
|
# has near-zero LOCAL similarity, so its COMBINED score sinks
|
|
# below an unrelated same-script decoy — even though MB itself is
|
|
# certain. "Sawano Hiroyuki": a decoy entity led on combined
|
|
# (sim 0.82, mb_score 83, combined 0.82 — just under the 0.85 bar)
|
|
# while the real artist '澤野弘之' had mb_score 100 but combined
|
|
# 0.30, sorted last. So evaluate the MB-SCORE leader independently
|
|
# of the combined ranking for the mb-only escape, not scored[0].
|
|
mb_leader = max(scored, key=lambda x: x[2]) # (combined, mbid, raw_mb)
|
|
mb_scores_desc = sorted((x[2] for x in scored), reverse=True)
|
|
mb_unambiguous = len(mb_scores_desc) < 2 or (mb_scores_desc[0] - mb_scores_desc[1]) >= 5
|
|
|
|
# Trust gate. Two ways to pass:
|
|
# 1. Combined score >= 0.85 (the historical strict bar that
|
|
# catches same-script matches) → trust the combined leader.
|
|
# 2. MB's OWN score is very high (>= 95) AND that MB-score leader
|
|
# is unambiguous → trust IT. Bridges the cross-script case
|
|
# where local similarity is near zero ("Dmitry Yablonsky" vs
|
|
# "Дмитрий Яблонский" sim ~0) but MB's index is confident.
|
|
passes_combined = best_score >= 0.85
|
|
passes_mb_only = mb_leader[2] >= 95 and mb_unambiguous
|
|
if not (passes_combined or passes_mb_only):
|
|
logger.debug(
|
|
"lookup_artist_aliases: best match for %r below trust "
|
|
"threshold (combined=%.2f, best_mb=%d, leader_mb=%d)",
|
|
artist_name, best_score, best_mb_score, mb_leader[2],
|
|
)
|
|
self._save_to_cache('artist_aliases', artist_name, None, None, {'aliases': []}, 0)
|
|
return []
|
|
|
|
# Pick the entity to pull aliases from. Combined-strong matches use
|
|
# the combined leader; the mb-only escape uses the MB-score leader
|
|
# (which may differ from scored[0] in the cross-script case above).
|
|
if passes_combined:
|
|
chosen_mbid, chosen_conf = best_mbid, best_score
|
|
else:
|
|
chosen_mbid, chosen_conf = mb_leader[1], mb_leader[2] / 100.0
|
|
|
|
# Ambiguity detection: when 2+ results both score high (within
|
|
# 0.1 of the best combined), the search hit multiple distinct
|
|
# artists with similar names. Pulling aliases for one could
|
|
# produce wrong matches. Skip + cache empty. The unambiguous
|
|
# MB-score leader (passes_mb_only) is exempt — its decisiveness
|
|
# was already checked via mb_unambiguous.
|
|
if len(scored) >= 2 and (scored[0][0] - scored[1][0]) < 0.1 and not passes_mb_only:
|
|
logger.debug(
|
|
"lookup_artist_aliases: ambiguous match for %r — top "
|
|
"two results within 0.1 (%.2f / %.2f). Skipping alias lookup.",
|
|
artist_name, scored[0][0], scored[1][0],
|
|
)
|
|
self._save_to_cache('artist_aliases', artist_name, None, None, {'aliases': []}, 0)
|
|
return []
|
|
|
|
aliases = self.fetch_artist_aliases(chosen_mbid)
|
|
self._save_to_cache(
|
|
'artist_aliases', artist_name, None, chosen_mbid,
|
|
{'aliases': aliases}, int(chosen_conf * 100),
|
|
)
|
|
return aliases
|
|
|
|
def _search_and_score_artists(self, artist_name: str, strict: bool):
|
|
"""Search MB for an artist and score each result.
|
|
|
|
Returns a list of (combined_score, mbid, raw_mb_score) tuples.
|
|
Combined score: 70% local similarity + 30% MB's own relevance
|
|
score (0..1). raw_mb_score preserved separately so the trust
|
|
gate can prefer high-MB-score results in cross-script cases
|
|
where local similarity is near zero.
|
|
|
|
Returns empty list on any failure.
|
|
"""
|
|
try:
|
|
results = self.mb_client.search_artist(artist_name, limit=3, strict=strict)
|
|
except Exception as e:
|
|
logger.debug(
|
|
"lookup_artist_aliases: search_artist(%r, strict=%s) raised: %s",
|
|
artist_name, strict, e,
|
|
)
|
|
return []
|
|
scored = []
|
|
for result in results or []:
|
|
mb_name = result.get('name', '')
|
|
mb_score = result.get('score', 0)
|
|
sim = self._calculate_similarity(artist_name, mb_name)
|
|
combined = (sim * 0.7) + (mb_score / 100 * 0.3)
|
|
mbid = result.get('id')
|
|
if mbid:
|
|
scored.append((combined, mbid, mb_score))
|
|
return scored
|
|
|
|
@staticmethod
|
|
def _best_score(scored):
|
|
return max((s[0] for s in scored), default=0.0) if scored else 0.0
|
|
|
|
def fetch_artist_aliases(self, mbid: str) -> list:
|
|
"""Fetch the alias list for an artist from MusicBrainz.
|
|
|
|
Issue #442 — Japanese kanji / Cyrillic / etc. spellings of an
|
|
artist's name are stored as `aliases` on the MusicBrainz
|
|
artist record. Pull them so SoulSync can recognise that
|
|
`澤野弘之` and `Hiroyuki Sawano` refer to the same artist.
|
|
|
|
Issue #586 — for some artists MB's CANONICAL `name` is the
|
|
non-Latin spelling (e.g. `Дмитрий Яблонский`) while the
|
|
Latin spelling lives in `aliases` — but the inverse also
|
|
happens, where the Latin canonical name has the Cyrillic in
|
|
aliases. Either way the canonical `name` and `sort-name` are
|
|
themselves valid alternate spellings for matching purposes,
|
|
so include them alongside the explicit alias entries.
|
|
|
|
Returns the deduplicated list of alias `name` strings. Returns
|
|
empty list (NOT None) on any failure — caller should treat
|
|
empty as "no aliases available, fall back to direct match" so
|
|
a transient MB outage never causes a stricter verification
|
|
decision than today.
|
|
"""
|
|
if not mbid:
|
|
return []
|
|
try:
|
|
data = self.mb_client.get_artist(mbid, includes=['aliases'])
|
|
except Exception as e:
|
|
logger.debug("fetch_artist_aliases: get_artist(%s) raised: %s", mbid, e)
|
|
return []
|
|
if not data:
|
|
return []
|
|
|
|
seen = set()
|
|
cleaned = []
|
|
|
|
def _add(value):
|
|
if not isinstance(value, str):
|
|
return
|
|
text = value.strip()
|
|
if not text:
|
|
return
|
|
key = text.lower()
|
|
if key in seen:
|
|
return
|
|
seen.add(key)
|
|
cleaned.append(text)
|
|
|
|
# Canonical name + sort-name treated as aliases for matching —
|
|
# they're the strongest cross-script bridge when MB's
|
|
# canonical spelling differs from the user's input.
|
|
_add(data.get('name'))
|
|
_add(data.get('sort-name'))
|
|
|
|
# MB returns each alias as a dict with `name`, `sort-name`,
|
|
# `locale`, `primary`, `type`, etc. We only care about the
|
|
# display name — that's what `actual` artist strings will
|
|
# match against. Also pull alias sort-name when present
|
|
# (some entries have a different sortable form).
|
|
for entry in data.get('aliases') or []:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
_add(entry.get('name'))
|
|
_add(entry.get('sort-name'))
|
|
return cleaned
|
|
|
|
def update_artist_aliases(self, artist_id: int, aliases: list) -> None:
|
|
"""Persist the alias list to `artists.aliases` as a JSON array.
|
|
|
|
Idempotent — overwrites any existing value. Empty list
|
|
clears the column (caller may want this if MB has no aliases
|
|
for the artist anymore).
|
|
"""
|
|
if artist_id is None:
|
|
return
|
|
conn = None
|
|
try:
|
|
conn = self.db._get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"UPDATE artists SET aliases = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
|
|
(json.dumps(aliases) if aliases else None, artist_id),
|
|
)
|
|
conn.commit()
|
|
logger.debug("Updated artist %s aliases (%d entries)", artist_id, len(aliases or []))
|
|
except Exception as e:
|
|
logger.error(f"Error updating artist aliases for {artist_id}: {e}")
|
|
if conn:
|
|
conn.rollback()
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
def get_artist_aliases(self, artist_name: str) -> list:
|
|
"""Look up cached aliases for an artist by NAME (not id).
|
|
|
|
Used by the verifier where the expected artist comes from a
|
|
download's metadata-source data — we don't have a library
|
|
row's `id` to query, just the display name. Returns empty
|
|
list when the artist isn't in the library or has no aliases
|
|
recorded. The verifier falls back to live MB lookup in that
|
|
case.
|
|
"""
|
|
if not artist_name:
|
|
return []
|
|
conn = None
|
|
try:
|
|
conn = self.db._get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"SELECT aliases FROM artists WHERE name = ? COLLATE NOCASE LIMIT 1",
|
|
(artist_name,),
|
|
)
|
|
row = cursor.fetchone()
|
|
if not row or not row[0]:
|
|
return []
|
|
try:
|
|
parsed = json.loads(row[0])
|
|
except (TypeError, json.JSONDecodeError):
|
|
return []
|
|
if not isinstance(parsed, list):
|
|
return []
|
|
return [str(x).strip() for x in parsed if x]
|
|
except Exception as e:
|
|
logger.debug("get_artist_aliases lookup failed for %r: %s", artist_name, e)
|
|
return []
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
def update_artist_mbid(self, artist_id: int, mbid: Optional[str], status: str):
|
|
"""Update artist with MusicBrainz ID"""
|
|
conn = None
|
|
try:
|
|
conn = self.db._get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
UPDATE artists
|
|
SET musicbrainz_id = ?,
|
|
musicbrainz_last_attempted = ?,
|
|
musicbrainz_match_status = ?
|
|
WHERE id = ?
|
|
""", (mbid, datetime.now(), status, artist_id))
|
|
|
|
conn.commit()
|
|
|
|
logger.debug(f"Updated artist {artist_id} with MBID: {mbid}, status: {status}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error updating artist {artist_id}: {e}")
|
|
if conn:
|
|
conn.rollback()
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
def update_album_mbid(self, album_id: int, mbid: Optional[str], status: str):
|
|
"""Update album with MusicBrainz release ID"""
|
|
conn = None
|
|
try:
|
|
conn = self.db._get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
UPDATE albums
|
|
SET musicbrainz_release_id = ?,
|
|
musicbrainz_last_attempted = ?,
|
|
musicbrainz_match_status = ?
|
|
WHERE id = ?
|
|
""", (mbid, datetime.now(), status, album_id))
|
|
|
|
conn.commit()
|
|
|
|
logger.debug(f"Updated album {album_id} with MBID: {mbid}, status: {status}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error updating album {album_id}: {e}")
|
|
if conn:
|
|
conn.rollback()
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
def update_track_mbid(self, track_id: int, mbid: Optional[str], status: str):
|
|
"""Update track with MusicBrainz recording ID"""
|
|
conn = None
|
|
try:
|
|
conn = self.db._get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
UPDATE tracks
|
|
SET musicbrainz_recording_id = ?,
|
|
musicbrainz_last_attempted = ?,
|
|
musicbrainz_match_status = ?
|
|
WHERE id = ?
|
|
""", (mbid, datetime.now(), status, track_id))
|
|
|
|
conn.commit()
|
|
|
|
logger.debug(f"Updated track {track_id} with MBID: {mbid}, status: {status}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error updating track {track_id}: {e}")
|
|
if conn:
|
|
conn.rollback()
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|