import argparse import base64 import binascii import hmac import json import mimetypes import secrets import subprocess from collections import Counter from dataclasses import asdict from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from importlib import resources from pathlib import Path from typing import Iterable, Optional from urllib.parse import parse_qs, quote, urlsplit from serato_doctor.crate_parser import load_library_crates from serato_doctor.database_parser import parse_database from serato_doctor.duplicates import find_duplicate_groups from serato_doctor.health import analyze_health from serato_doctor.hashing import compare_audio_files from serato_doctor.matching import MatchingEngine, normalize from serato_doctor.models.crate import CrateKind from serato_doctor.models.duplicate import DuplicateKind from serato_doctor.models.library import Library from serato_doctor.scanner import scan_filesystem from serato_doctor.repair import ( BACKUP_FOLDER, apply_duplicate_repair, apply_duplicate_repair_batch, list_backups, plan_duplicate_repair, restore_backup, ) MAX_REQUEST_BYTES = 64 * 1024 DETAIL_LIMIT = 50 FILE_TOKEN_SECRET = secrets.token_bytes(32) STATIC_FILES = { "/": ("index.html", "text/html; charset=utf-8"), "/app.css": ("app.css", "text/css; charset=utf-8"), "/recovery.css": ("recovery.css", "text/css; charset=utf-8"), "/layout-fixes.css": ("layout-fixes.css", "text/css; charset=utf-8"), "/app.js": ("app.js", "text/javascript; charset=utf-8"), } def _display_path(path: Path) -> str: return str(path) def _file_token(path: Path) -> str: encoded = base64.urlsafe_b64encode(str(path.resolve()).encode()).decode() signature = hmac.digest(FILE_TOKEN_SECRET, encoded.encode(), "sha256").hex() return f"{encoded}.{signature}" def _verified_audio(token: str) -> Path: try: encoded, signature = token.rsplit(".", 1) expected = hmac.digest( FILE_TOKEN_SECRET, encoded.encode(), "sha256" ).hex() if not hmac.compare_digest(signature, expected): raise ValueError path = Path(base64.urlsafe_b64decode(encoded.encode()).decode()) except (binascii.Error, ValueError, UnicodeDecodeError): raise ValueError("Invalid or expired file preview") if not path.is_file() or path.suffix.casefold() not in { ".mp3", ".m4a", ".wav", ".aif", ".aiff", ".flac", }: raise ValueError("Audio file is no longer available") return path def _file_preview(path: Path) -> dict: token = _file_token(path) return { "path": _display_path(path), "audio_url": f"/api/audio?token={quote(token)}", "reveal_token": token, } def _first_reason(match) -> str: for item in match.evidence: if item.matched: return item.explanation return "Filename-related candidate" def diagnostic_details(library: Library, limit: int = DETAIL_LIMIT) -> dict: dynamic_sources = { crate.path for crate in library.crates if crate.kind is CrateKind.SMART } results = tuple( result for result in library.reconcile_by_filename() if result.reference.source not in dynamic_sources ) missing = [result for result in results if not result.exists_by_filename] matcher = MatchingEngine(library.tracks) suggested_matches = [] for result in missing: candidates = matcher.candidates_for(result.reference) if not candidates: continue best = candidates[0] suggested_matches.append( { "filename": result.reference.filename, "crate": _display_path(result.reference.source), "saved_path": _display_path(result.reference.path), "candidate": _display_path(best.track.path), "score": f"{best.score_percent}%", "reason": _first_reason(best), } ) duplicate_groups = find_duplicate_groups(library.tracks) exact_duplicates = [ group for group in duplicate_groups if group.kind is DuplicateKind.EXACT_NAME ] cloud_conflicts = [ group for group in duplicate_groups if group.kind is DuplicateKind.CLOUD_CONFLICT ] database_tracks = library.database.tracks if library.database else () missing_database_tracks = [ track for track in database_tracks if not track.path.exists() ] database_filename_counts = Counter( normalize(track.filename) for track in missing_database_tracks ) referenced_names = {reference.filename for reference in library.references} unused_tracks = [ track for track in library.tracks if track.filename not in referenced_names ] return { "database_missing_tracks": { "title": "Missing tracks in Serato", "summary": ( "These are Serato database entries whose saved file location " "does not currently exist on disk." ), "total": len(missing_database_tracks), "items": [ { "filename": track.filename, "saved_path": _display_path(track.path), "artist": track.artist or "Unknown artist", "title": track.title or track.filename, "repeated_filename": database_filename_counts[ normalize(track.filename) ] > 1, } for track in missing_database_tracks[:limit] ], }, "old_crate_references": { "title": "Old crate references", "summary": ( "These are regular crate appearances whose exact filename was " "not found in the selected music folder." ), "total": len(missing), "items": [ { "filename": result.reference.filename, "crate": _display_path(result.reference.source), "saved_path": _display_path(result.reference.path), } for result in missing[:limit] ], }, "suggested_matches": { "title": "Suggested matches", "summary": ( "These are read-only guesses where Serato Doctor found a " "filename-related candidate on disk." ), "total": len(suggested_matches), "items": suggested_matches[:limit], }, "duplicate_filenames": { "title": "Duplicate filenames", "summary": ( "These groups contain different files with the same cleaned-up " "filename. Review before making any decisions." ), "total": len(exact_duplicates), "items": [ { "filename": group.display_name, "files": [_display_path(track.path) for track in group.tracks], "file_previews": [ _file_preview(track.path) for track in group.tracks ], } for group in exact_duplicates[:limit] ], }, "cloud_conflicts": { "title": "Possible cloud conflicts", "summary": ( "These filename families look like cloud sync conflict copies, " "such as a duplicate ending in a number." ), "total": len(cloud_conflicts), "items": [ { "filename": group.display_name, "files": [_display_path(track.path) for track in group.tracks], "file_previews": [ _file_preview(track.path) for track in group.tracks ], } for group in cloud_conflicts[:limit] ], }, "broken_symlinks": { "title": "Broken shortcuts", "summary": ( "These symbolic links point somewhere that no longer resolves." ), "total": len(library.broken_symlinks), "items": [ { "path": _display_path(link.path), "target": _display_path(link.target) if link.target else "Unknown", } for link in library.broken_symlinks[:limit] ], }, "unused_tracks": { "title": "Unused tracks", "summary": ( "These scanned files were not referenced by any loaded crate. " "That does not mean they should be deleted." ), "total": len(unused_tracks), "items": [ {"filename": track.filename, "path": _display_path(track.path)} for track in unused_tracks[:limit] ], }, } def analyze_paths( serato: Path, music: Path, reference_roots: Iterable[Path] = () ) -> dict: serato = serato.expanduser() music = music.expanduser() reference_roots = tuple(root.expanduser() for root in reference_roots) if not serato.is_dir(): raise ValueError(f"Serato folder does not exist: {serato}") if not music.is_dir(): raise ValueError(f"Music folder does not exist: {music}") crates = load_library_crates(serato, reference_roots) filesystem = scan_filesystem(music) database_path = serato / "database V2" database = parse_database(database_path) if database_path.is_file() else None library = Library.from_crates( crates, filesystem.tracks, filesystem.broken_symlinks, database, ) report = analyze_health(library) result = asdict(report) result["score_basis"] = report.score_basis result["details"] = diagnostic_details(library) return result def duplicate_repair( serato: Path, music: Path, keeper: Path, group_files: Iterable[Path], backup_limit: Optional[int], apply: bool = False, ) -> dict: """Validate a current duplicate group and preview or apply consolidation.""" serato = serato.expanduser().resolve() music = music.expanduser().resolve() keeper = keeper.expanduser().resolve() requested = tuple(path.expanduser().resolve() for path in group_files) if not serato.is_dir() or not music.is_dir(): raise ValueError("Analyze the library again before repairing duplicates") tracks = scan_filesystem(music).tracks groups = find_duplicate_groups(tracks) valid_groups = [ {track.path.resolve() for track in group.tracks} for group in groups ] if set(requested) not in valid_groups: raise ValueError("This duplicate group changed; analyze the library again") plan = plan_duplicate_repair(keeper, requested, serato) result = { "keeper": str(plan.keeper), "replaced": [str(path) for path in plan.replaced], "metadata_backups": len(plan.metadata_files), "strategy": "shortcut", "database_v2_modified": False, } if apply: receipt = apply_duplicate_repair(plan, serato, backup_limit) result.update({"applied": True, "backup": str(receipt.backup)}) else: result["applied"] = False return result def duplicate_repair_batch( serato: Path, music: Path, choices: Iterable[dict], backup_limit: Optional[int], apply: bool = False, ) -> dict: """Validate and preview or apply several keeper choices together.""" serato = serato.expanduser().resolve() music = music.expanduser().resolve() if not serato.is_dir() or not music.is_dir(): raise ValueError("Analyze the library again before repairing duplicates") groups = find_duplicate_groups(scan_filesystem(music).tracks) valid_groups = [ {track.path.resolve() for track in group.tracks} for group in groups ] plans = [] selected_groups = set() for choice in choices: requested = tuple(Path(value).expanduser().resolve() for value in choice["group_files"]) group_key = frozenset(requested) if set(requested) not in valid_groups: raise ValueError("A duplicate group changed; analyze the library again") if group_key in selected_groups: raise ValueError("A duplicate group was selected more than once") selected_groups.add(group_key) plans.append( plan_duplicate_repair(Path(choice["keeper"]), requested, serato) ) if not plans: raise ValueError("Choose at least one duplicate group") replaced = [str(path) for plan in plans for path in plan.replaced] comparisons = [ compare_audio_files((plan.keeper,) + plan.replaced) for plan in plans ] result = { "choice_count": len(plans), "replaced": replaced, "decisions": [ { "keeper": str(plan.keeper), "replaced": [str(path) for path in plan.replaced], "hash_status": comparison["status"], } for plan, comparison in zip(plans, comparisons) ], "metadata_backups": len(plans[0].metadata_files), "strategy": "shortcut", "database_v2_modified": False, } if apply: receipt = apply_duplicate_repair_batch(plans, serato, backup_limit) result.update({"applied": True, "backup": str(receipt.backup)}) else: result["applied"] = False return result def backup_history(serato: Path) -> dict: serato = serato.expanduser().resolve() if not serato.is_dir(): raise ValueError(f"Serato folder does not exist: {serato}") backups = list_backups(serato) return { "total": len(backups), "size_bytes": sum(backup.size_bytes for backup in backups), "items": [ { "path": str(backup.path), "created_at": backup.created_at, "keeper": str(backup.keeper), "replaced": [str(path) for path in backup.replaced], "size_bytes": backup.size_bytes, "status": backup.status, "restored_at": backup.restored_at, } for backup in backups ], } class SeratoDoctorHandler(BaseHTTPRequestHandler): def do_GET(self) -> None: request = urlsplit(self.path) if request.path == "/api/audio": try: token = parse_qs(request.query)["token"][0] self._audio_response(_verified_audio(token)) except (BrokenPipeError, ConnectionResetError): return except (KeyError, IndexError, OSError, ValueError) as error: self._json_response(404, {"error": str(error)}) return asset = STATIC_FILES.get(request.path) if asset is None: self._json_response(404, {"error": "Not found"}) return filename, content_type = asset content = ( resources.files("serato_doctor.webui") .joinpath(filename) .read_bytes() ) self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(content))) self.end_headers() self.wfile.write(content) def do_POST(self) -> None: allowed = { "/api/analyze", "/api/duplicates/preview", "/api/duplicates/apply", "/api/duplicates/batch/preview", "/api/duplicates/batch/apply", "/api/duplicates/hash", "/api/backups/restore", "/api/backups", "/api/reveal", } if self.path not in allowed: self._json_response(404, {"error": "Not found"}) return try: length = int(self.headers.get("Content-Length", "0")) if length <= 0 or length > MAX_REQUEST_BYTES: raise ValueError("Invalid request size") payload = json.loads(self.rfile.read(length)) if not isinstance(payload, dict): raise ValueError("Request body must be a JSON object") if self.path == "/api/analyze": roots = [Path(value) for value in payload.get("reference_roots", [])] result = analyze_paths( Path(payload["serato"]), Path(payload["music"]), roots ) elif self.path == "/api/reveal": path = _verified_audio(payload["token"]) subprocess.run(["open", "-R", str(path)], check=True) result = {"revealed": str(path)} elif self.path == "/api/backups": result = backup_history(Path(payload["serato"])) elif self.path == "/api/backups/restore": serato = Path(payload["serato"]).expanduser().resolve() backup = Path(payload["backup"]).expanduser().resolve() backup_root = (serato / BACKUP_FOLDER).resolve() if backup.parent != backup_root: raise ValueError("That backup does not belong to this library") restored = restore_backup(backup) result = {"restored": [str(path) for path in restored]} elif self.path == "/api/duplicates/hash": tokens = payload["tokens"] if not isinstance(tokens, list) or not 2 <= len(tokens) <= 10: raise ValueError("Compare between 2 and 10 audio files") comparison = compare_audio_files( _verified_audio(token) for token in tokens ) result = { "status": comparison["status"], "files": [ { "path": item["path"], "size": item["size"], "fingerprint": item["sha256"][:12], } for item in comparison["files"] ], } elif self.path.startswith("/api/duplicates/batch/"): raw_limit = payload.get("backup_limit", 10) backup_limit = None if raw_limit is None else int(raw_limit) result = duplicate_repair_batch( Path(payload["serato"]), Path(payload["music"]), payload["choices"], backup_limit, apply=self.path.endswith("/apply"), ) else: raw_limit = payload.get("backup_limit", 10) backup_limit = None if raw_limit is None else int(raw_limit) result = duplicate_repair( Path(payload["serato"]), Path(payload["music"]), Path(payload["keeper"]), (Path(value) for value in payload["group_files"]), backup_limit, apply=self.path.endswith("/apply"), ) except ( KeyError, TypeError, OSError, json.JSONDecodeError, ValueError, ) as error: self._json_response(400, {"error": str(error)}) return self._json_response(200, result) def _audio_response(self, path: Path) -> None: size = path.stat().st_size start, end = 0, size - 1 status = 200 range_header = self.headers.get("Range") if range_header and range_header.startswith("bytes="): raw_start, _, raw_end = range_header[6:].partition("-") start = int(raw_start or 0) end = min(int(raw_end) if raw_end else end, end) if start < 0 or start > end: raise ValueError("Invalid audio range") status = 206 length = end - start + 1 self.send_response(status) self.send_header( "Content-Type", mimetypes.guess_type(path.name)[0] or "audio/mpeg" ) self.send_header("Accept-Ranges", "bytes") self.send_header("Content-Length", str(length)) if status == 206: self.send_header("Content-Range", f"bytes {start}-{end}/{size}") self.end_headers() with path.open("rb") as audio: audio.seek(start) remaining = length while remaining: chunk = audio.read(min(64 * 1024, remaining)) if not chunk: break self.wfile.write(chunk) remaining -= len(chunk) def _json_response(self, status: int, payload: dict) -> None: content = json.dumps(payload).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(content))) self.end_headers() self.wfile.write(content) def log_message(self, format: str, *args: object) -> None: return def main() -> None: parser = argparse.ArgumentParser(prog="serato-doctor-web") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8765) args = parser.parse_args() server = ThreadingHTTPServer((args.host, args.port), SeratoDoctorHandler) print(f"Serato Doctor web interface: http://{args.host}:{args.port}") print("Press Ctrl+C to stop.") try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close()