09b4638279
Adds strict checks to ensure Soulseek track versions (live, remix, acoustic, instrumental) are only matched if the Spotify track title contains corresponding version indicators. This prevents mismatched versions from being accepted and applies a reduced penalty for matching non-original versions.
824 lines
39 KiB
Python
824 lines
39 KiB
Python
from typing import List, Optional, Dict, Any, Tuple
|
||
import re
|
||
from dataclasses import dataclass
|
||
from difflib import SequenceMatcher
|
||
from unidecode import unidecode
|
||
from utils.logging_config import get_logger
|
||
from core.spotify_client import Track as SpotifyTrack
|
||
from core.plex_client import PlexTrackInfo
|
||
from core.soulseek_client import TrackResult
|
||
|
||
|
||
logger = get_logger("matching_engine")
|
||
|
||
@dataclass
|
||
class MatchResult:
|
||
spotify_track: SpotifyTrack
|
||
plex_track: Optional[PlexTrackInfo]
|
||
confidence: float
|
||
match_type: str
|
||
|
||
@property
|
||
def is_match(self) -> bool:
|
||
return self.plex_track is not None and self.confidence >= 0.8
|
||
|
||
class MusicMatchingEngine:
|
||
def __init__(self):
|
||
# Conservative title patterns - only remove clear noise, preserve meaningful differences like remixes
|
||
self.title_patterns = [
|
||
# Only remove explicit/clean markers - preserve remixes, versions, and content after hyphens
|
||
r'\s*\(explicit\)',
|
||
r'\s*\(clean\)',
|
||
# Remove featuring artists from the title itself
|
||
r'\sfeat\.?.*',
|
||
r'\sft\.?.*',
|
||
r'\sfeaturing.*'
|
||
]
|
||
|
||
self.artist_patterns = [
|
||
# Only remove featured artists, not parts of main artist names
|
||
r'\s*feat\..*',
|
||
r'\s*ft\..*',
|
||
r'\s*featuring.*',
|
||
# REMOVED: r'\s*&.*' - This breaks "Daryl Hall & John Oates", "Blood & Water"
|
||
# REMOVED: r'\s*and.*' - This breaks artist names with "and"
|
||
# REMOVED: r',.*' - This can break legitimate artist names with commas
|
||
]
|
||
|
||
def normalize_string(self, text: str) -> str:
|
||
"""
|
||
Normalizes string by handling common stylizations, converting to ASCII,
|
||
lowercasing, and replacing separators with spaces.
|
||
"""
|
||
if not text:
|
||
return ""
|
||
# Handle Korn/KoЯn variations - both uppercase Я (U+042F) and lowercase я (U+044F)
|
||
char_map = {
|
||
'Я': 'R', # Cyrillic 'Ya' to 'R'
|
||
'я': 'r', # Lowercase Cyrillic 'ya' to 'r'
|
||
}
|
||
|
||
# Apply the character replacements before other normalization steps
|
||
for original, replacement in char_map.items():
|
||
text = text.replace(original, replacement)
|
||
text = unidecode(text)
|
||
text = text.lower()
|
||
|
||
# Expand specific abbreviations for better matching
|
||
abbreviation_map = {
|
||
r'\bpt\.': 'part', # "pt." → "part"
|
||
r'\bvol\.': 'volume', # "vol." → "volume"
|
||
r'\bfeat\.': 'featured' # "feat." → "featured"
|
||
# Removed "ft." → "featured" (ambiguous: could be "feet" in measurements)
|
||
}
|
||
|
||
for pattern, replacement in abbreviation_map.items():
|
||
text = re.sub(pattern, replacement, text)
|
||
|
||
# --- IMPROVEMENT V4 ---
|
||
# The user correctly pointed out that replacing '$' with 's' was incorrect
|
||
# as it breaks searching for stylized names like A$AP Rocky.
|
||
# The new approach is to PRESERVE the '$' symbol during normalization.
|
||
|
||
# Replace common separators with spaces to preserve word boundaries.
|
||
text = re.sub(r'[._/]', ' ', text)
|
||
|
||
# Keep alphanumeric characters, spaces, hyphens, AND the '$' sign.
|
||
text = re.sub(r'[^a-z0-9\s$-]', '', text)
|
||
|
||
# Consolidate multiple spaces into one
|
||
text = re.sub(r'\s+', ' ', text).strip()
|
||
|
||
return text
|
||
|
||
def get_core_string(self, text: str) -> str:
|
||
"""Returns a 'core' version of a string with only letters and numbers for a strict comparison."""
|
||
if not text:
|
||
return ""
|
||
# Use normalize_string first to get abbreviation expansion, then strip to core
|
||
normalized = self.normalize_string(text)
|
||
return re.sub(r'[^a-z0-9]', '', normalized)
|
||
|
||
def clean_title(self, title: str) -> str:
|
||
"""Cleans title by removing common extra info using regex for fuzzy matching."""
|
||
cleaned = title
|
||
|
||
for pattern in self.title_patterns:
|
||
cleaned = re.sub(pattern, '', cleaned, flags=re.IGNORECASE).strip()
|
||
|
||
return self.normalize_string(cleaned)
|
||
|
||
def clean_artist(self, artist: str) -> str:
|
||
"""Cleans artist name by removing featured artists and other noise."""
|
||
cleaned = artist
|
||
|
||
for pattern in self.artist_patterns:
|
||
cleaned = re.sub(pattern, '', cleaned, flags=re.IGNORECASE).strip()
|
||
|
||
return self.normalize_string(cleaned)
|
||
|
||
def clean_album_name(self, album_name: str) -> str:
|
||
"""Clean album name by removing version info, deluxe editions, etc."""
|
||
if not album_name:
|
||
return ""
|
||
|
||
cleaned = album_name
|
||
|
||
# Common album suffixes to remove
|
||
album_patterns = [
|
||
# Add pattern to remove trailing info after a hyphen, common for remasters/editions.
|
||
r'\s-\s.*',
|
||
r'\s*\(deluxe\s*edition?\)',
|
||
r'\s*\(expanded\s*edition?\)',
|
||
r'\s*\(platinum\s*edition?\)', # Fix for "Fearless (Platinum Edition)"
|
||
r'\s*\(remastered?\)',
|
||
r'\s*\(remaster\)',
|
||
r'\s*\(anniversary\s*edition?\)',
|
||
r'\s*\(special\s*edition?\)',
|
||
r'\s*\(bonus\s*track\s*version\)',
|
||
r'\s*\(.*version\)', # Covers "Taylor's Version", "Radio Version", etc.
|
||
r'\s*\[deluxe\]',
|
||
r'\s*\[remastered?\]',
|
||
r'\s*\[.*version\]',
|
||
r'\s*-\s*deluxe',
|
||
r'\s*-\s*platinum\s*edition?', # Handle "Album - Platinum Edition"
|
||
r'\s*-\s*remastered?',
|
||
r'\s+platinum\s*edition?$', # Handle "Album Platinum Edition" at end
|
||
r'\s*\d{4}\s*remaster', # Year remaster
|
||
r'\s*\(\d{4}\s*remaster\)'
|
||
]
|
||
|
||
for pattern in album_patterns:
|
||
cleaned = re.sub(pattern, '', cleaned, flags=re.IGNORECASE).strip()
|
||
|
||
return self.normalize_string(cleaned)
|
||
|
||
def similarity_score(self, str1: str, str2: str) -> float:
|
||
"""Calculates similarity score between two strings with enhanced version handling."""
|
||
if not str1 or not str2:
|
||
return 0.0
|
||
|
||
# Standard similarity
|
||
standard_ratio = SequenceMatcher(None, str1, str2).ratio()
|
||
|
||
# Enhanced logic: Check if one string is a version of the other
|
||
# This handles cases like "Back & forth" vs "Back & forth original mix"
|
||
shorter, longer = (str1, str2) if len(str1) <= len(str2) else (str2, str1)
|
||
|
||
# If the shorter string is at the start of the longer string
|
||
if longer.startswith(shorter):
|
||
# Extract the extra content
|
||
extra_content = longer[len(shorter):].strip()
|
||
|
||
# Check if the extra content looks like version info
|
||
version_keywords = [
|
||
'original mix', 'radio mix', 'club mix', 'extended mix',
|
||
'slowed', 'reverb', 'sped up', 'acoustic', 'remix', 'remaster',
|
||
'live', 'demo', 'instrumental', 'clean', 'explicit',
|
||
'radio edit', 'extended', 'version'
|
||
]
|
||
|
||
# Normalize extra content for comparison
|
||
extra_normalized = extra_content.lower().strip(' -()[]')
|
||
|
||
# If the extra content matches version keywords, boost the similarity
|
||
for keyword in version_keywords:
|
||
if keyword in extra_normalized:
|
||
# High similarity but not perfect (to distinguish from exact matches)
|
||
return max(standard_ratio, 0.85)
|
||
|
||
return standard_ratio
|
||
|
||
def duration_similarity(self, duration1: int, duration2: int) -> float:
|
||
"""Calculates similarity score based on track duration (in ms)."""
|
||
if duration1 == 0 or duration2 == 0:
|
||
return 0.5 # Neutral score if a duration is missing
|
||
|
||
# Allow a 5-second tolerance (5000 ms)
|
||
if abs(duration1 - duration2) <= 5000:
|
||
return 1.0
|
||
|
||
diff_ratio = abs(duration1 - duration2) / max(duration1, duration2)
|
||
return max(0, 1.0 - diff_ratio * 5)
|
||
|
||
def calculate_match_confidence(self, spotify_track: SpotifyTrack, plex_track: PlexTrackInfo) -> Tuple[float, str]:
|
||
"""Calculates a confidence score using a prioritized model, starting with a strict 'core' title check."""
|
||
|
||
# --- Artist Scoring (calculated once) ---
|
||
spotify_artists_cleaned = [self.clean_artist(a) for a in spotify_track.artists if a]
|
||
plex_artist_normalized = self.normalize_string(plex_track.artist)
|
||
plex_artist_cleaned = self.clean_artist(plex_track.artist)
|
||
|
||
best_artist_score = 0.0
|
||
for spotify_artist in spotify_artists_cleaned:
|
||
if spotify_artist and spotify_artist in plex_artist_normalized:
|
||
best_artist_score = 1.0
|
||
break
|
||
score = self.similarity_score(spotify_artist, plex_artist_cleaned)
|
||
if score > best_artist_score:
|
||
best_artist_score = score
|
||
artist_score = best_artist_score
|
||
|
||
# --- Priority 1: Core Title Match (for exact matches like "Girls", "APT.", "LIL DEMON") ---
|
||
spotify_core_title = self.get_core_string(spotify_track.name)
|
||
plex_core_title = self.get_core_string(plex_track.title)
|
||
|
||
if spotify_core_title and spotify_core_title == plex_core_title:
|
||
# SAFETY CHECK: Only give high confidence if artist also matches reasonably well
|
||
# This prevents "Artist A - Girls" from matching "Artist Z - Girls" with high confidence
|
||
if artist_score >= 0.75: # Require decent artist match
|
||
# If the core titles are identical and artists match, we are highly confident
|
||
confidence = 0.90 + (artist_score * 0.09) # Max score of 0.99
|
||
return confidence, "core_title_match"
|
||
# If artist score is too low, fall through to standard weighted calculation
|
||
|
||
# --- Priority 2: Fuzzy Title Match (for variations, typos, etc.) ---
|
||
spotify_title_cleaned = self.clean_title(spotify_track.name)
|
||
plex_title_cleaned = self.clean_title(plex_track.title)
|
||
|
||
title_score = self.similarity_score(spotify_title_cleaned, plex_title_cleaned)
|
||
duration_score = self.duration_similarity(spotify_track.duration_ms, plex_track.duration if plex_track.duration else 0)
|
||
|
||
# Use a standard weighted calculation if the core titles didn't match
|
||
confidence = (title_score * 0.60) + (artist_score * 0.30) + (duration_score * 0.10)
|
||
match_type = "standard_match"
|
||
|
||
return confidence, match_type
|
||
|
||
def find_best_match(self, spotify_track: SpotifyTrack, plex_tracks: List[PlexTrackInfo]) -> MatchResult:
|
||
"""Finds the best Plex track match from a list of candidates."""
|
||
best_match = None
|
||
best_confidence = 0.0
|
||
best_match_type = "no_match"
|
||
|
||
if not plex_tracks:
|
||
return MatchResult(spotify_track, None, 0.0, "no_candidates")
|
||
|
||
for plex_track in plex_tracks:
|
||
confidence, match_type = self.calculate_match_confidence(spotify_track, plex_track)
|
||
|
||
if confidence > best_confidence:
|
||
best_confidence = confidence
|
||
best_match = plex_track
|
||
best_match_type = match_type
|
||
|
||
return MatchResult(
|
||
spotify_track=spotify_track,
|
||
plex_track=best_match,
|
||
confidence=best_confidence,
|
||
match_type=best_match_type
|
||
)
|
||
|
||
def detect_album_in_title(self, track_title: str, album_name: str = None) -> Tuple[str, bool]:
|
||
"""
|
||
Detect if album name appears in track title and return cleaned version.
|
||
Returns (cleaned_title, album_detected) tuple.
|
||
"""
|
||
if not track_title:
|
||
return "", False
|
||
|
||
original_title = track_title
|
||
title_lower = track_title.lower()
|
||
|
||
# Common patterns where album name appears in track titles
|
||
album_patterns = [
|
||
r'\s*-\s*(.+)$', # "Track - Album" (most common)
|
||
r'\s*\|\s*(.+)$', # "Track | Album"
|
||
r'\s*\(\s*(.+)\s*\)$' # "Track (Album)"
|
||
]
|
||
|
||
# If we have album name, check if it appears in the title
|
||
if album_name:
|
||
album_clean = album_name.lower().strip()
|
||
|
||
for pattern in album_patterns:
|
||
match = re.search(pattern, track_title)
|
||
if match:
|
||
potential_album = match.group(1).lower().strip()
|
||
|
||
# Check if the extracted part matches the album name with better fuzzy matching
|
||
similarity_threshold = 0.8
|
||
|
||
# Calculate similarity between potential album and actual album
|
||
if potential_album == album_clean:
|
||
similarity = 1.0 # Exact match
|
||
elif potential_album in album_clean or album_clean in potential_album:
|
||
# Substring match - calculate how much overlap
|
||
shorter = min(len(potential_album), len(album_clean))
|
||
longer = max(len(potential_album), len(album_clean))
|
||
similarity = shorter / longer if longer > 0 else 0.0
|
||
else:
|
||
# Use string similarity for fuzzy matching
|
||
similarity = self.similarity_score(potential_album, album_clean)
|
||
|
||
if similarity >= similarity_threshold:
|
||
# Remove the album part from the title
|
||
cleaned_title = re.sub(pattern, '', track_title).strip()
|
||
|
||
# SAFETY CHECK: Don't return empty or too-short titles
|
||
if not cleaned_title or len(cleaned_title.strip()) < 2:
|
||
logger.warning(f"Album removal would create empty title: '{original_title}' → '{cleaned_title}' - keeping original")
|
||
return track_title, False
|
||
|
||
# SAFETY CHECK: Don't remove if it would leave only articles or very short words
|
||
words = cleaned_title.split()
|
||
meaningful_words = [w for w in words if len(w) > 2 and w.lower() not in ['the', 'and', 'or', 'of', 'a', 'an']]
|
||
if not meaningful_words:
|
||
logger.warning(f"Album removal would leave only short words: '{original_title}' → '{cleaned_title}' - keeping original")
|
||
return track_title, False
|
||
|
||
logger.debug(f"Detected album in title: '{original_title}' → '{cleaned_title}' (removed: '{match.group(1)}', similarity: {similarity:.2f})")
|
||
return cleaned_title, True
|
||
|
||
# Fallback: detect common album-like suffixes even without album context
|
||
# Look for patterns that might be album names (usually after dash)
|
||
dash_pattern = r'\s*-\s*([A-Za-z][A-Za-z0-9\s&\-\']{3,30})$'
|
||
match = re.search(dash_pattern, track_title)
|
||
if match:
|
||
potential_album_part = match.group(1).strip()
|
||
|
||
# Heuristics: likely an album name if it:
|
||
# - Doesn't contain common track descriptors
|
||
# - Is reasonable length (4-30 chars)
|
||
# - Doesn't look like a feature/remix indicator
|
||
exclude_patterns = [
|
||
r'\b(remix|mix|edit|version|live|acoustic|instrumental|demo|feat|ft|featuring)\b'
|
||
]
|
||
|
||
is_likely_album = True
|
||
for exclude_pattern in exclude_patterns:
|
||
if re.search(exclude_pattern, potential_album_part.lower()):
|
||
is_likely_album = False
|
||
break
|
||
|
||
if is_likely_album and 4 <= len(potential_album_part) <= 30:
|
||
cleaned_title = re.sub(dash_pattern, '', track_title).strip()
|
||
print(f"🎵 Heuristic album detection: '{original_title}' → '{cleaned_title}' (removed: '{potential_album_part}')")
|
||
return cleaned_title, True
|
||
|
||
return track_title, False
|
||
|
||
def generate_download_queries(self, spotify_track: SpotifyTrack) -> List[str]:
|
||
"""
|
||
Generate multiple search query variations for better matching.
|
||
Returns queries in order of preference (cleaned titles first, then original).
|
||
"""
|
||
queries = []
|
||
|
||
if not spotify_track.artists:
|
||
# No artist info - just use track name variations
|
||
queries.append(self.clean_title(spotify_track.name))
|
||
return queries
|
||
|
||
artist = self.clean_artist(spotify_track.artists[0])
|
||
original_title = spotify_track.name
|
||
|
||
# Get album name if available - try multiple attribute names
|
||
album_name = None
|
||
for attr in ['album', 'album_name', 'album_title']:
|
||
album_name = getattr(spotify_track, attr, None)
|
||
if album_name:
|
||
break
|
||
|
||
# PRIORITY 1: Try removing potential album from title FIRST
|
||
cleaned_title, album_detected = self.detect_album_in_title(original_title, album_name)
|
||
if album_detected and cleaned_title != original_title:
|
||
cleaned_track = self.clean_title(cleaned_title)
|
||
if cleaned_track:
|
||
queries.append(f"{artist} {cleaned_track}".strip())
|
||
logger.debug(f"PRIORITY 1: Album-cleaned query: '{artist} {cleaned_track}'")
|
||
|
||
# PRIORITY 2: Try simplified versions, but preserve important version info
|
||
# Only remove content that's likely to be album names or noise, not version info
|
||
|
||
# Pattern 1: Intelligently handle content after " - "
|
||
# Only remove if it looks like album names, preserve version info like "slowed", "remix", etc.
|
||
dash_pattern = r'^([^-]+?)\s*-\s*(.+)$'
|
||
match = re.search(dash_pattern, original_title.strip())
|
||
if match:
|
||
title_part = match.group(1).strip()
|
||
dash_content = match.group(2).strip().lower()
|
||
|
||
# Define version keywords that should be preserved
|
||
preserve_keywords = [
|
||
'slowed', 'reverb', 'sped up', 'speed up', 'spedup', 'slowdown',
|
||
'remix', 'mix', 'edit', 'version', 'remaster', 'acoustic',
|
||
'live', 'demo', 'instrumental', 'radio', 'extended', 'club',
|
||
'original', 'clean', 'explicit', 'mashup', 'bootleg'
|
||
]
|
||
|
||
# Check if the dash content contains version keywords
|
||
should_preserve = any(keyword in dash_content for keyword in preserve_keywords)
|
||
|
||
if not should_preserve and title_part and len(title_part) >= 3:
|
||
# This looks like album content, safe to remove
|
||
dash_clean = self.clean_title(title_part)
|
||
if dash_clean and dash_clean not in [self.clean_title(q.split(' ', 1)[1]) for q in queries if ' ' in q]:
|
||
queries.append(f"{artist} {dash_clean}".strip())
|
||
logger.debug(f"PRIORITY 2: Dash-cleaned query (removed album): '{artist} {dash_clean}'")
|
||
elif should_preserve:
|
||
logger.debug(f"PRESERVED: Keeping dash content '{dash_content}' as it appears to be version info")
|
||
|
||
# Pattern 2: Only remove parentheses that contain noise (feat, explicit, etc), not version info
|
||
# Check if parentheses contain version-related keywords before removing
|
||
paren_pattern = r'^(.+?)\s*\(([^)]+)\)(.*)$'
|
||
paren_match = re.search(paren_pattern, original_title)
|
||
if paren_match:
|
||
before_paren = paren_match.group(1).strip()
|
||
paren_content = paren_match.group(2).strip().lower()
|
||
after_paren = paren_match.group(3).strip()
|
||
|
||
# Define what we consider "noise" vs "important version info"
|
||
noise_keywords = ['feat', 'ft', 'featuring', 'explicit', 'clean']
|
||
# Expanded version keywords to match the dash preserve keywords
|
||
version_keywords = [
|
||
'slowed', 'reverb', 'sped up', 'speed up', 'spedup', 'slowdown',
|
||
'remix', 'mix', 'edit', 'version', 'remaster', 'acoustic',
|
||
'live', 'demo', 'instrumental', 'radio', 'extended', 'club',
|
||
'original', 'mashup', 'bootleg'
|
||
]
|
||
|
||
# Only remove parentheses if they contain noise, not version info
|
||
is_noise = any(keyword in paren_content for keyword in noise_keywords)
|
||
is_version = any(keyword in paren_content for keyword in version_keywords)
|
||
|
||
if is_noise and not is_version and before_paren:
|
||
simple_title = (before_paren + ' ' + after_paren).strip()
|
||
if simple_title and len(simple_title) >= 3:
|
||
simple_clean = self.clean_title(simple_title)
|
||
if simple_clean and simple_clean not in [self.clean_title(q.split(' ', 1)[1]) for q in queries if ' ' in q]:
|
||
queries.append(f"{artist} {simple_clean}".strip())
|
||
logger.debug(f"PRIORITY 2: Noise-removed query: '{artist} {simple_clean}'")
|
||
elif is_version:
|
||
logger.debug(f"PRESERVED: Keeping parentheses content '({paren_content})' as it appears to be version info")
|
||
|
||
# PRIORITY 3: Original query (ONLY if no album was detected or if it's different)
|
||
original_track_clean = self.clean_title(original_title)
|
||
if not album_detected or not queries: # Only add original if no album detected or no other queries
|
||
if original_track_clean not in [q.split(' ', 1)[1] for q in queries if ' ' in q]:
|
||
queries.append(f"{artist} {original_track_clean}".strip())
|
||
logger.debug(f"PRIORITY 3: Original query: '{artist} {original_track_clean}'")
|
||
|
||
# Remove duplicates while preserving order
|
||
unique_queries = []
|
||
seen = set()
|
||
for query in queries:
|
||
if query.lower() not in seen:
|
||
unique_queries.append(query)
|
||
seen.add(query.lower())
|
||
|
||
return unique_queries
|
||
|
||
def generate_download_query(self, spotify_track: SpotifyTrack) -> str:
|
||
"""
|
||
Generate optimized search query for downloading tracks.
|
||
Returns the most specific query (backward compatibility).
|
||
"""
|
||
queries = self.generate_download_queries(spotify_track)
|
||
return queries[0] if queries else ""
|
||
|
||
|
||
def calculate_slskd_match_confidence(self, spotify_track: SpotifyTrack, slskd_track: TrackResult) -> float:
|
||
"""
|
||
Calculates a confidence score for a Soulseek track against a Spotify track.
|
||
This is the core of the new matching logic.
|
||
"""
|
||
# Normalize the Spotify track info once for efficiency
|
||
spotify_title_norm = self.normalize_string(spotify_track.name)
|
||
spotify_artists_norm = [self.normalize_string(a) for a in spotify_track.artists]
|
||
|
||
# The slskd filename is our primary source of truth, so normalize it
|
||
slskd_filename_norm = self.normalize_string(slskd_track.filename)
|
||
|
||
# 1. Title Score: How well does the Spotify title appear in the filename?
|
||
# We use the cleaned, core title for a strict check. This avoids matching remixes.
|
||
spotify_cleaned_title = self.clean_title(spotify_track.name)
|
||
title_score = 0.0
|
||
if spotify_cleaned_title in slskd_filename_norm:
|
||
title_score = 0.9 # High score for direct inclusion
|
||
# Bonus for being a standalone word/phrase, penalizing partial matches like 'in' in 'finland'
|
||
if re.search(r'\b' + re.escape(spotify_cleaned_title) + r'\b', slskd_filename_norm):
|
||
title_score = 1.0
|
||
|
||
# 2. Artist Score: How well do the Spotify artists appear in the filename?
|
||
artist_score = 0.0
|
||
for artist in spotify_artists_norm:
|
||
if artist in slskd_filename_norm:
|
||
artist_score = 1.0 # Perfect match if any artist is found
|
||
break
|
||
|
||
# 3. Duration Score: How similar are the track lengths?
|
||
# We give this a lower weight as slskd duration data can be unreliable.
|
||
duration_score = self.duration_similarity(spotify_track.duration_ms, slskd_track.duration if slskd_track.duration else 0)
|
||
|
||
# 4. Quality Bonus: Add a small bonus for higher quality formats
|
||
quality_bonus = 0.0
|
||
if slskd_track.quality:
|
||
if slskd_track.quality.lower() == 'flac':
|
||
quality_bonus = 0.07 # Reduced from 0.1 to prevent low-confidence FLAC beating high-confidence MP3
|
||
elif slskd_track.quality.lower() == 'mp3' and (slskd_track.bitrate or 0) >= 320:
|
||
quality_bonus = 0.05
|
||
|
||
# --- Final Weighted Score ---
|
||
# Title and Artist are the most important factors for an accurate match.
|
||
final_confidence = (title_score * 0.60) + (artist_score * 0.35) + (duration_score * 0.05)
|
||
|
||
# Add the quality bonus to the final score
|
||
final_confidence += quality_bonus
|
||
|
||
# Ensure the final score doesn't exceed 1.0
|
||
return min(final_confidence, 1.0)
|
||
|
||
|
||
def find_best_slskd_matches(self, spotify_track: SpotifyTrack, slskd_results: List[TrackResult]) -> List[TrackResult]:
|
||
"""
|
||
Scores and sorts a list of Soulseek results against a Spotify track.
|
||
Returns the list of candidates sorted from best to worst match.
|
||
"""
|
||
if not slskd_results:
|
||
return []
|
||
|
||
scored_results = []
|
||
for slskd_track in slskd_results:
|
||
confidence = self.calculate_slskd_match_confidence(spotify_track, slskd_track)
|
||
# We temporarily store the confidence score on the object itself for sorting
|
||
slskd_track.confidence = confidence
|
||
scored_results.append(slskd_track)
|
||
|
||
# Sort by confidence score (descending), and then by size as a tie-breaker
|
||
sorted_results = sorted(scored_results, key=lambda r: (r.confidence, r.size), reverse=True)
|
||
|
||
# Filter out very low-confidence results to avoid bad matches.
|
||
# A threshold of 0.6 means the title and artist had to have some reasonable similarity.
|
||
confident_results = [r for r in sorted_results if r.confidence > 0.6]
|
||
|
||
return confident_results
|
||
|
||
def detect_version_type(self, filename: str) -> Tuple[str, float]:
|
||
"""
|
||
Detect version type from filename and return (version_type, penalty).
|
||
Penalties are applied to prefer original versions over variants.
|
||
"""
|
||
if not filename:
|
||
return 'original', 0.0
|
||
|
||
filename_lower = filename.lower()
|
||
|
||
# Define version patterns and their penalties (higher penalty = lower priority)
|
||
version_patterns = {
|
||
'remix': {
|
||
'patterns': [r'\bremix\b', r'\brmx\b', r'\brework\b', r'\bedit\b(?!ion)'],
|
||
'penalty': 0.15 # -15% penalty for remixes
|
||
},
|
||
'live': {
|
||
'patterns': [r'\blive\b', r'\bconcert\b', r'\btour\b', r'\bperformance\b'],
|
||
'penalty': 0.20 # -20% penalty for live versions
|
||
},
|
||
'acoustic': {
|
||
'patterns': [r'\bacoustic\b', r'\bunplugged\b', r'\bstripped\b'],
|
||
'penalty': 0.12 # -12% penalty for acoustic
|
||
},
|
||
'instrumental': {
|
||
'patterns': [r'\binstrumental\b', r'\bkaraoke\b', r'\bminus one\b'],
|
||
'penalty': 0.25 # -25% penalty for instrumentals (most different from original)
|
||
},
|
||
'radio': {
|
||
'patterns': [r'\bradio\s*edit\b', r'\bradio\s*version\b', r'\bclean\s*edit\b'],
|
||
'penalty': 0.08 # -8% penalty for radio edits (minor difference)
|
||
},
|
||
'extended': {
|
||
'patterns': [r'\bextended\b', r'\bfull\s*version\b', r'\blong\s*version\b'],
|
||
'penalty': 0.05 # -5% penalty for extended (close to original)
|
||
},
|
||
'demo': {
|
||
'patterns': [r'\bdemo\b', r'\broughcut\b', r'\bunreleased\b'],
|
||
'penalty': 0.18 # -18% penalty for demos
|
||
},
|
||
'explicit': {
|
||
'patterns': [r'\bexplicit\b', r'\buncensored\b'],
|
||
'penalty': 0.02 # -2% minor penalty (might be preferred by some)
|
||
}
|
||
}
|
||
|
||
# Check each version type
|
||
for version_type, config in version_patterns.items():
|
||
for pattern in config['patterns']:
|
||
if re.search(pattern, filename_lower):
|
||
return version_type, config['penalty']
|
||
|
||
# No version indicators found - assume original
|
||
return 'original', 0.0
|
||
|
||
def calculate_slskd_match_confidence_enhanced(self, spotify_track: SpotifyTrack, slskd_track: TrackResult) -> Tuple[float, str]:
|
||
"""
|
||
Enhanced version of calculate_slskd_match_confidence with version-aware scoring.
|
||
Returns (confidence, version_type) tuple.
|
||
|
||
STRICT VERSION MATCHING:
|
||
- Live versions are ONLY accepted if Spotify track title contains "live" or "live version"
|
||
- Remixes are ONLY accepted if Spotify track title contains "remix" or "mix"
|
||
- Acoustic versions are ONLY accepted if Spotify track title contains "acoustic"
|
||
- etc.
|
||
"""
|
||
# Get base confidence using existing logic
|
||
base_confidence = self.calculate_slskd_match_confidence(spotify_track, slskd_track)
|
||
|
||
# Detect version type in Soulseek result
|
||
version_type, penalty = self.detect_version_type(slskd_track.filename)
|
||
|
||
# Check if Spotify track title contains version indicators
|
||
spotify_title_lower = spotify_track.name.lower()
|
||
|
||
# STRICT VERSION MATCHING: Reject mismatched versions
|
||
if version_type == 'live':
|
||
# Only accept live versions if Spotify title has live as a VERSION INDICATOR
|
||
# Patterns: (Live), - Live, [Live], Live at, Live from, Live in, Live Version
|
||
# NOT: words ending with 'live' like "Let Me Live" or starting like "Lively"
|
||
live_patterns = [
|
||
r'\(live\)', # (Live) or (Live at Wembley)
|
||
r'\[live\]', # [Live]
|
||
r'[-–—]\s*live\b', # - Live or – Live
|
||
r'\blive\s+at\b', # Live at
|
||
r'\blive\s+from\b', # Live from
|
||
r'\blive\s+in\b', # Live in
|
||
r'\blive\s+version\b', # Live Version
|
||
r'\blive\s+recording\b' # Live Recording
|
||
]
|
||
has_live_indicator = any(re.search(pattern, spotify_title_lower) for pattern in live_patterns)
|
||
|
||
if not has_live_indicator:
|
||
# Reject: Soulseek has live version but Spotify doesn't want it
|
||
return 0.0, 'rejected_version_mismatch'
|
||
|
||
elif version_type == 'remix':
|
||
# Only accept remixes if Spotify title has remix as a VERSION INDICATOR
|
||
# Patterns: (Remix), - Remix, [Remix], Remix, Mix
|
||
remix_patterns = [
|
||
r'\(.*?(remix|mix|rmx).*?\)', # (Remix) or (DJ Remix)
|
||
r'\[.*?(remix|mix|rmx).*?\]', # [Remix]
|
||
r'[-–—]\s*(remix|mix|rmx)\b', # - Remix
|
||
r'\b(remix|mix|rmx)\s*$', # Remix at end
|
||
]
|
||
has_remix_indicator = any(re.search(pattern, spotify_title_lower) for pattern in remix_patterns)
|
||
|
||
if not has_remix_indicator:
|
||
# Reject: Soulseek has remix but Spotify wants original
|
||
return 0.0, 'rejected_version_mismatch'
|
||
|
||
elif version_type == 'acoustic':
|
||
# Only accept acoustic if Spotify title has acoustic as a VERSION INDICATOR
|
||
acoustic_patterns = [
|
||
r'\(.*?acoustic.*?\)', # (Acoustic)
|
||
r'\[.*?acoustic.*?\]', # [Acoustic]
|
||
r'[-–—]\s*acoustic\b', # - Acoustic
|
||
r'\bacoustic\s+version\b', # Acoustic Version
|
||
]
|
||
has_acoustic_indicator = any(re.search(pattern, spotify_title_lower) for pattern in acoustic_patterns)
|
||
|
||
if not has_acoustic_indicator:
|
||
# Reject: Soulseek has acoustic but Spotify wants original
|
||
return 0.0, 'rejected_version_mismatch'
|
||
|
||
elif version_type == 'instrumental':
|
||
# Only accept instrumental if Spotify title has instrumental as a VERSION INDICATOR
|
||
instrumental_patterns = [
|
||
r'\(.*?instrumental.*?\)', # (Instrumental)
|
||
r'\[.*?instrumental.*?\]', # [Instrumental]
|
||
r'[-–—]\s*instrumental\b', # - Instrumental
|
||
r'\binstrumental\s+version\b', # Instrumental Version
|
||
]
|
||
has_instrumental_indicator = any(re.search(pattern, spotify_title_lower) for pattern in instrumental_patterns)
|
||
|
||
if not has_instrumental_indicator:
|
||
# Reject: Soulseek has instrumental but Spotify wants original
|
||
return 0.0, 'rejected_version_mismatch'
|
||
|
||
# Apply version penalty (for matching versions, slight penalty for quality differences)
|
||
if version_type != 'original':
|
||
adjusted_confidence = max(0.0, base_confidence - (penalty * 0.5)) # Reduced penalty since it's a match
|
||
# Store version info on the track object for UI display
|
||
slskd_track.version_type = version_type
|
||
slskd_track.version_penalty = penalty
|
||
else:
|
||
adjusted_confidence = base_confidence
|
||
slskd_track.version_type = 'original'
|
||
slskd_track.version_penalty = 0.0
|
||
|
||
return adjusted_confidence, version_type
|
||
|
||
def find_best_slskd_matches_enhanced(self, spotify_track: SpotifyTrack, slskd_results: List[TrackResult]) -> List[TrackResult]:
|
||
"""
|
||
Enhanced version of find_best_slskd_matches with version-aware scoring.
|
||
Returns candidates sorted by adjusted confidence (preferring originals).
|
||
"""
|
||
if not slskd_results:
|
||
return []
|
||
|
||
scored_results = []
|
||
for slskd_track in slskd_results:
|
||
# Use enhanced confidence calculation
|
||
confidence, version_type = self.calculate_slskd_match_confidence_enhanced(spotify_track, slskd_track)
|
||
|
||
# Store the adjusted confidence and version info
|
||
slskd_track.confidence = confidence
|
||
slskd_track.version_type = getattr(slskd_track, 'version_type', 'original')
|
||
scored_results.append(slskd_track)
|
||
|
||
# Sort by confidence score (descending), then by version preference, then by size
|
||
def sort_key(r):
|
||
# Primary: confidence score
|
||
# Secondary: prefer originals (original=0, others=penalty value for tie-breaking)
|
||
version_priority = 0.0 if r.version_type == 'original' else getattr(r, 'version_penalty', 0.1)
|
||
# Tertiary: file size
|
||
return (r.confidence, -version_priority, r.size)
|
||
|
||
sorted_results = sorted(scored_results, key=sort_key, reverse=True)
|
||
|
||
# Filter out very low-confidence results
|
||
# Lower the threshold to 0.45 to account for version penalties and album-in-title scenarios
|
||
confident_results = [r for r in sorted_results if r.confidence > 0.45]
|
||
|
||
# Debug logging for troubleshooting
|
||
if scored_results and not confident_results:
|
||
print(f"⚠️ DEBUG: Found {len(scored_results)} scored results but none met confidence threshold 0.45")
|
||
for i, result in enumerate(sorted_results[:3]): # Show top 3
|
||
print(f" {i+1}. {result.confidence:.3f} - {getattr(result, 'version_type', 'unknown')} - {result.filename[:60]}...")
|
||
elif confident_results:
|
||
print(f"✅ DEBUG: {len(confident_results)} results passed confidence threshold 0.45")
|
||
for i, result in enumerate(confident_results[:3]): # Show top 3
|
||
print(f" {i+1}. {result.confidence:.3f} - {getattr(result, 'version_type', 'unknown')} - {result.filename[:60]}...")
|
||
|
||
return confident_results
|
||
|
||
def calculate_album_confidence(self, spotify_album, plex_album_info: Dict[str, Any]) -> float:
|
||
"""Calculate confidence score for album matching"""
|
||
if not spotify_album or not plex_album_info:
|
||
return 0.0
|
||
|
||
score = 0.0
|
||
|
||
# 1. Album name similarity (40% weight)
|
||
spotify_album_clean = self.clean_album_name(spotify_album.name)
|
||
plex_album_clean = self.clean_album_name(plex_album_info['title'])
|
||
|
||
name_similarity = self.similarity_score(spotify_album_clean, plex_album_clean)
|
||
score += name_similarity * 0.4
|
||
|
||
# 2. Artist similarity (40% weight)
|
||
if spotify_album.artists and plex_album_info.get('artist'):
|
||
spotify_artist_clean = self.clean_artist(spotify_album.artists[0])
|
||
plex_artist_clean = self.clean_artist(plex_album_info['artist'])
|
||
|
||
artist_similarity = self.similarity_score(spotify_artist_clean, plex_artist_clean)
|
||
score += artist_similarity * 0.4
|
||
|
||
# 3. Track count similarity (10% weight)
|
||
spotify_track_count = getattr(spotify_album, 'total_tracks', 0)
|
||
plex_track_count = plex_album_info.get('track_count', 0)
|
||
|
||
if spotify_track_count > 0 and plex_track_count > 0:
|
||
# Calculate track count similarity (perfect match = 1.0, close matches get partial credit)
|
||
track_diff = abs(spotify_track_count - plex_track_count)
|
||
if track_diff == 0:
|
||
track_similarity = 1.0
|
||
elif track_diff <= 2: # Allow for slight differences (bonus tracks, etc.)
|
||
track_similarity = 0.8
|
||
elif track_diff <= 5:
|
||
track_similarity = 0.5
|
||
else:
|
||
track_similarity = 0.2
|
||
|
||
score += track_similarity * 0.1
|
||
|
||
# 4. Year similarity bonus (10% weight)
|
||
spotify_year = spotify_album.release_date[:4] if spotify_album.release_date else None
|
||
plex_year = str(plex_album_info.get('year', '')) if plex_album_info.get('year') else None
|
||
|
||
if spotify_year and plex_year:
|
||
if spotify_year == plex_year:
|
||
score += 0.1 # Perfect year match
|
||
elif abs(int(spotify_year) - int(plex_year)) <= 1:
|
||
score += 0.05 # Close year match (remaster, etc.)
|
||
|
||
return min(score, 1.0) # Cap at 1.0
|
||
|
||
def find_best_album_match(self, spotify_album, plex_albums: List[Dict[str, Any]]) -> Tuple[Optional[Dict[str, Any]], float]:
|
||
"""Find the best matching album from Plex candidates"""
|
||
if not plex_albums:
|
||
return None, 0.0
|
||
|
||
best_match = None
|
||
best_confidence = 0.0
|
||
|
||
for plex_album in plex_albums:
|
||
confidence = self.calculate_album_confidence(spotify_album, plex_album)
|
||
|
||
if confidence > best_confidence:
|
||
best_confidence = confidence
|
||
best_match = plex_album
|
||
|
||
# Only return matches above confidence threshold
|
||
if best_confidence >= 0.8: # High threshold for album matching
|
||
return best_match, best_confidence
|
||
else:
|
||
return None, best_confidence
|