from pathlib import Path from typing import Iterable, Tuple from serato_doctor.models.crate import Crate from serato_doctor.models.reference import TrackReference def read_crate_text(crate_path: Path) -> str: return crate_path.read_bytes().decode("utf-16-le", errors="ignore").replace("\x00", "") def clean_path(raw: str) -> str: # Common Serato decode artifacts where final extension char gets merged. raw = raw.replace(".mp漳", ".mp3") raw = raw.replace(".MP漳", ".MP3") raw = raw.replace(".m4愠", ".m4a") raw = raw.replace(".M4愠", ".M4A") raw = raw.replace(".wa瘠", ".wav") raw = raw.replace(".WA瘠", ".WAV") raw = raw.replace(".ai映", ".aif") raw = raw.replace(".AI映", ".AIF") return raw.strip() DEFAULT_PATH_MARKERS = ("Users/", "Volumes/") def path_markers(reference_roots: Iterable[Path]) -> Tuple[str, ...]: configured = tuple( root.expanduser().as_posix().strip("/") + "/" for root in reference_roots ) return configured or DEFAULT_PATH_MARKERS def load_crate(crate_path: Path, reference_roots: Iterable[Path] = ()) -> Crate: text = read_crate_text(crate_path) refs = [] # Serato record markers seen after paths in UTF-16-LE decoded crate data. stop_markers = ["牴k", "otrk", "ptrk", "tvcn", "ovct"] for marker in path_markers(reference_roots): for part in text.split(marker)[1:]: candidate = marker + part stops = [ candidate.find(stop) for stop in stop_markers if candidate.find(stop) != -1 ] if not stops: continue raw_path = "/" + candidate[: min(stops)] raw_path = clean_path(raw_path) path = Path(raw_path) refs.append( TrackReference( source=crate_path, path=path, filename=path.name, ) ) return Crate(path=crate_path, references=tuple(refs)) def parse_crate( crate_path: Path, reference_roots: Iterable[Path] = () ) -> list[TrackReference]: """Parse references from one crate, preserving the prototype API.""" return list(load_crate(crate_path, reference_roots).references) def parse_crates( root: Path, reference_roots: Iterable[Path] = () ) -> list[TrackReference]: refs = [] reference_roots = tuple(reference_roots) for crate in root.rglob("*.crate"): refs.extend(parse_crate(crate, reference_roots)) return refs