Philip Guzman
399 lines
14 KiB
Python
399 lines
14 KiB
Python
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, Optional
|
|
from urllib.parse import 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.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,
|
|
list_backups,
|
|
plan_duplicate_repair,
|
|
restore_backup,
|
|
)
|
|
|
|
|
|
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"),
|
|
"/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 _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
|
|
|
|
|
|
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 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:
|
|
asset = STATIC_FILES.get(urlsplit(self.path).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/backups/restore",
|
|
"/api/backups",
|
|
}
|
|
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/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]}
|
|
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 _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()
|