Philip Guzman
115 lines
4.0 KiB
Python
115 lines
4.0 KiB
Python
import argparse
|
|
import json
|
|
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.health import analyze_health
|
|
from serato_doctor.models.library import Library
|
|
from serato_doctor.scanner import scan_filesystem
|
|
|
|
|
|
MAX_REQUEST_BYTES = 64 * 1024
|
|
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 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
|
|
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()
|