import argparse import json 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 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.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 MAX_REQUEST_BYTES = 64 * 1024 DETAIL_LIMIT = 50 STATIC_FILES = { "/": ("index.html", "text/html; charset=utf-8"), "/app.css": ("app.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 _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], } 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], } 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 class SeratoDoctorHandler(BaseHTTPRequestHandler): def do_GET(self) -> None: asset = STATIC_FILES.get(self.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("Content-Length", str(len(content))) self.end_headers() self.wfile.write(content) def do_POST(self) -> None: if self.path != "/api/analyze": 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") roots = [Path(value) for value in payload.get("reference_roots", [])] result = analyze_paths( Path(payload["serato"]), Path(payload["music"]), roots ) except (KeyError, TypeError, json.JSONDecodeError, ValueError) as error: self._json_response(400, {"error": str(error)}) return self._json_response(200, result) 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()