Add duplicate audio comparison controls

serato_doctor/web.py

@@ -1,12 +1,18 @@
 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 urlsplit
+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
@@ -28,6 +34,7 @@ from serato_doctor.repair import (
 
 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"),
@@ -41,6 +48,44 @@ 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:
@@ -159,6 +204,9 @@ def diagnostic_details(library: Library, limit: int = DETAIL_LIMIT) -> dict:
                 {
                     "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]
             ],
@@ -174,6 +222,9 @@ def diagnostic_details(library: Library, limit: int = DETAIL_LIMIT) -> dict:
                 {
                     "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]
             ],
@@ -298,7 +349,15 @@ def backup_history(serato: Path) -> dict:
 
 class SeratoDoctorHandler(BaseHTTPRequestHandler):
     def do_GET(self) -> None:
-        asset = STATIC_FILES.get(urlsplit(self.path).path)
+        request = urlsplit(self.path)
+        if request.path == "/api/audio":
+            try:
+                token = parse_qs(request.query)["token"][0]
+                self._audio_response(_verified_audio(token))
+            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
@@ -322,6 +381,7 @@ class SeratoDoctorHandler(BaseHTTPRequestHandler):
             "/api/duplicates/apply",
             "/api/backups/restore",
             "/api/backups",
+            "/api/reveal",
         }
         if self.path not in allowed:
             self._json_response(404, {"error": "Not found"})
@@ -338,6 +398,10 @@ class SeratoDoctorHandler(BaseHTTPRequestHandler):
                 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":
@@ -370,6 +434,38 @@ class SeratoDoctorHandler(BaseHTTPRequestHandler):
             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)