87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
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)))
|
|
)
|