import re import unicodedata from collections import defaultdict from pathlib import Path from typing import DefaultDict, Iterable, List, Tuple from serato_doctor.models.match import MatchEvidence, TrackMatch from serato_doctor.models.reference import TrackReference from serato_doctor.models.track import DiskTrack def normalize(value: str) -> str: return unicodedata.normalize("NFKC", value).casefold() def cloud_conflict_name(filename: str) -> str: """Remove a trailing numeric cloud-conflict suffix from a filename stem.""" path = Path(filename) stem = re.sub(r" \d+$", "", path.stem) return normalize(stem + path.suffix) def score_candidate(reference: TrackReference, track: DiskTrack) -> TrackMatch: reference_name = reference.filename track_name = track.filename if reference_name == track_name: filename_points = 60 filename_reason = "Filename is identical" elif normalize(reference_name) == normalize(track_name): filename_points = 55 filename_reason = "Filename matches after case and Unicode normalization" elif cloud_conflict_name(reference_name) == cloud_conflict_name(track_name): filename_points = 50 filename_reason = "Filename matches after removing a numeric conflict suffix" else: filename_points = 0 filename_reason = "Filename does not match" same_extension = normalize(reference.path.suffix) == normalize(track.suffix) same_parent = normalize(reference.path.parent.name) == normalize( track.path.parent.name ) evidence = ( MatchEvidence( "filename", filename_points > 0, filename_points, 60, filename_reason, ), MatchEvidence( "extension", same_extension, 10 if same_extension else 0, 10, "File extension matches" if same_extension else "File extension differs", ), MatchEvidence( "parent_folder", same_parent, 20 if same_parent else 0, 20, "Parent folder matches" if same_parent else "Parent folder differs", ), ) return TrackMatch(reference, track, evidence) class MatchingEngine: """Find and rank filename-related disk candidates without modifying files.""" def __init__(self, tracks: Iterable[DiskTrack]): self._by_conflict_name: DefaultDict[str, List[DiskTrack]] = defaultdict(list) for track in tracks: self._by_conflict_name[cloud_conflict_name(track.filename)].append(track) def candidates_for(self, reference: TrackReference) -> Tuple[TrackMatch, ...]: candidates = self._by_conflict_name.get( cloud_conflict_name(reference.filename), [] ) matches = [score_candidate(reference, track) for track in candidates] return tuple( sorted(matches, key=lambda match: (-match.score, str(match.track.path))) )