Philip Guzman
588 lines
22 KiB
Python
588 lines
22 KiB
Python
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()
|