path: root/lib/project/src/producer/gui.py

"""Optional loopback-only web API. Importing the CLI never imports web dependencies.

There is no credential: every process on this machine can use the API, so the
server binds only to 127.0.0.1 and rejects cross-origin/rebinding requests.
Recordings arrive as browser uploads and are processed from a private
temporary work directory; outputs are written to an existing absolute
directory chosen per export.
"""
from contextlib import asynccontextmanager
from dataclasses import asdict
import math
from pathlib import Path
import re
import socket
import tomllib
import uuid

from .config import profile_label, resolve, toml
from .gui_jobs import Busy, Jobs, input_info
from .gui_schema import schema
from .pipeline import targets_for

BODY_LIMIT = 65536
UNSAFE_NAME = re.compile(r"[^\w .+()-]", re.ASCII)


class Boundary:
    """Reject cross-origin/rebinding requests and bound JSON bodies.

    Uploads bypass the bounded-body path: recordings stream straight to disk.
    """
    def __init__(self, app, authority):
        self.app, self.authority = app, authority

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            return await self.app(scope, receive, send)
        from starlette.responses import JSONResponse
        headers = scope.get("headers", [])
        def values(name):
            return [value.decode("latin1") for key, value in headers if key == name]
        status, detail = None, None
        if values(b"host") != [self.authority]:
            status, detail = 403, "Untrusted Host"
        elif values(b"origin") not in ([], [f"http://{self.authority}"]):
            status, detail = 403, "Untrusted Origin"
        elif any(value not in ("same-origin", "none") for value in values(b"sec-fetch-site")):
            status, detail = 403, "Cross-site request denied"
        async def secure_send(message):
            if message["type"] == "http.response.start":
                message.setdefault("headers", []).extend([
                    (b"cache-control", b"no-store"), (b"referrer-policy", b"no-referrer"),
                    (b"x-content-type-options", b"nosniff"), (b"x-frame-options", b"DENY"),
                    (b"content-security-policy", b"default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self' data:; media-src 'self' blob:; connect-src 'self'; object-src 'none'; base-uri 'none'; frame-ancestors 'none'; form-action 'none'"),
                ])
            await send(message)
        if status:
            return await JSONResponse({"detail": detail}, status_code=status)(scope, receive, secure_send)
        # Pre-read only the bounded JSON payloads. This also handles chunked bodies.
        if scope["path"].startswith("/api/") and scope["path"] != "/api/uploads":
            body = bytearray()
            while True:
                message = await receive()
                if message["type"] == "http.disconnect":
                    return
                body.extend(message.get("body", b""))
                if len(body) > BODY_LIMIT:
                    return await JSONResponse({"detail": "Request body exceeds 64 KiB"}, status_code=413)(scope, receive, secure_send)
                if not message.get("more_body", False):
                    break
            sent = False
            original_receive = receive
            async def bounded_receive():
                nonlocal sent
                if not sent:
                    sent = True
                    return {"type": "http.request", "body": bytes(body), "more_body": False}
                return await original_receive()
            receive = bounded_receive
        return await self.app(scope, receive, secure_send)


def object_body(value, allowed):
    if not isinstance(value, dict):
        raise ValueError("Expected a JSON object")
    if value.keys() - allowed:
        raise ValueError("Unknown request field(s): " + ", ".join(sorted(value.keys() - allowed)))
    return value


def output_directory(value, default):
    """An export needs an existing absolute directory on this machine."""
    if value in (None, ""):
        if default is None:
            raise ValueError("Choose an existing absolute output directory")
        default.mkdir(parents=True, exist_ok=True)
        return default
    if not isinstance(value, str) or "\x00" in value:
        raise ValueError("Choose an existing absolute output directory on this machine")
    path = Path(value).expanduser()
    if not path.is_absolute():
        raise ValueError("The output directory must be an absolute path on this machine")
    if not path.is_dir():
        raise ValueError(f"Output directory does not exist: {path}")
    return path.resolve()


def create_app(port, *, config=None, profile=None, overrides=None, output_dir=None):
    # These imports intentionally exist only behind the optional gui command.
    from fastapi import FastAPI, Request
    from fastapi.responses import FileResponse, JSONResponse, Response
    from fastapi.staticfiles import StaticFiles
    base = tomllib.loads(config.read_text()) if config else {}
    if profile is not None:
        base["profile"] = profile
    base.update(overrides or {})
    initial = asdict(resolve(overrides=base))

    def configured(values):
        if not isinstance(values, dict):
            raise ValueError("settings must be an object")
        return resolve(overrides=base | values)

    # Exports land in the launch directory by default — normally the same
    # directory as the loaded recording — unless an absolute directory is
    # given per export. Recording uploads are private temporary copies.
    default_output = Path(output_dir) if output_dir else Path.cwd()
    jobs = Jobs()

    @asynccontextmanager
    async def lifespan(app):
        try:
            yield
        finally:
            jobs.close()

    app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None, lifespan=lifespan)
    app.add_middleware(Boundary, authority=f"127.0.0.1:{port}")

    @app.exception_handler(ValueError)
    async def invalid(request, error):
        return JSONResponse({"detail": str(error)}, status_code=429 if isinstance(error, Busy) else 400)

    @app.exception_handler(OSError)
    async def inaccessible(request, error):
        return JSONResponse({"detail": str(error)}, status_code=400)

    @app.exception_handler(KeyError)
    async def missing(request, error):
        return JSONResponse({"detail": "Unknown or expired job/result, or missing required field"}, status_code=404)

    @app.get("/api/schema")
    def settings_schema():
        return schema(initial) | {"api_version": 1, "output_dir": str(default_output)}

    @app.post("/api/resolve")
    async def resolve_settings(request: Request):
        body = object_body(await request.json(), {"settings", "standalone"})
        values = body.get("settings", {})
        if not isinstance(values, dict):
            raise ValueError("settings must be an object")
        if body.get("standalone"):
            # Like a preset import: package defaults plus the profile's own
            # values, not the launch config, whose explicit values would
            # otherwise override most profile differences.
            return {"settings": asdict(resolve(overrides={"profile": values.get("profile", "natural")}))}
        return {"settings": asdict(configured(values))}

    @app.post("/api/presets/import")
    async def import_preset(request: Request):
        body = object_body(await request.json(), {"toml"})
        if not isinstance(body.get("toml"), str):
            raise ValueError("toml must be text")
        # Import is standalone: defaults/profile + TOML, not the launch config.
        return {"settings": asdict(resolve(overrides=tomllib.loads(body["toml"])))}

    @app.post("/api/presets/download")
    async def download_preset(request: Request):
        body = object_body(await request.json(), {"settings"})
        return Response(toml(configured(body.get("settings", {}))), media_type="application/toml",
                        headers={"Content-Disposition": 'attachment; filename="producer-preset.toml"'})

    @app.post("/api/uploads")
    async def upload(request: Request, name: str = "recording.wav"):
        """Stream a recording from the browser into the private work directory."""
        # Display name only: never used as a filesystem path component.
        safe_name = UNSAFE_NAME.sub("_", Path(name).name).strip(". ") or "recording.wav"
        if not safe_name.lower().endswith(".wav"):
            safe_name += ".wav"
        path = jobs.uploads_dir / f"{uuid.uuid4().hex}-{safe_name}"
        try:
            with path.open("wb") as sink:
                async for chunk in request.stream():
                    sink.write(chunk)
            try:
                info = input_info(path)
            except ValueError:
                path.unlink(missing_ok=True)
                raise
        except BaseException:
            path.unlink(missing_ok=True)
            raise
        upload_id = jobs.register_upload(path, safe_name)
        return {"upload_id": upload_id, "name": safe_name, **info}

    @app.get("/api/uploads/{upload_id}")
    def upload_info(upload_id: str):
        return jobs.upload_info(upload_id)

    @app.get("/api/jobs")
    def list_jobs():
        return {"jobs": jobs.list()}

    @app.get("/api/jobs/{job_id}")
    def get_job(job_id: str):
        return jobs.get(job_id)

    @app.post("/api/jobs/{job_id}/cancel")
    def cancel_job(job_id: str):
        return jobs.cancel(job_id)

    @app.post("/api/jobs", status_code=202)
    async def submit_job(request: Request):
        body = object_body(await request.json(),
                           {"kind", "upload_id", "input_id", "settings", "start", "duration",
                            "output_dir", "overwrite"})
        kind = body.get("kind")
        if kind not in ("scan", "preview", "export"):
            raise ValueError("kind must be scan, preview or export")
        info = jobs.upload_info(body.get("upload_id"))
        if body.get("input_id") != info["input_id"]:
            raise ValueError("Input changed; select and scan it again")
        snapshot = {"upload_id": body["upload_id"], "name": info["name"], "input_id": info["input_id"]}
        if kind != "scan":
            snapshot["settings"] = asdict(configured(body.get("settings", {})))
        if kind == "preview":
            start, duration = body.get("start"), body.get("duration")
            if (type(start) not in (int, float) or type(duration) not in (int, float)
                    or not math.isfinite(start) or not math.isfinite(duration)
                    or start < 0 or not 0 < duration <= 120 or start + duration > info["duration"]):
                raise ValueError("Preview must be within the recording: start >= 0, duration > 0 and <= 120 seconds")
            snapshot.update(start=start, duration=duration)
        if kind == "export":
            snapshot["output_dir"] = str(output_directory(body.get("output_dir"), default_output))
            overwrite = body.get("overwrite", False)
            if type(overwrite) is not bool:
                raise ValueError("overwrite must be a boolean")
            snapshot["overwrite"] = overwrite
        return jobs.submit(kind, snapshot)

    @app.post("/api/export/check")
    async def export_check(request: Request):
        """Preflight for the GUI export: exact target paths and whether any
        already exists, so the browser can ask before replacing audio."""
        body = object_body(await request.json(), {"upload_id", "input_id", "settings", "output_dir"})
        info = jobs.upload_info(body.get("upload_id"))
        if body.get("input_id") != info["input_id"]:
            raise ValueError("Input changed; select and scan it again")
        settings = configured(body.get("settings", {}))
        parent = output_directory(body.get("output_dir"), default_output)
        stem = Path(info["name"]).stem or "master"
        target = parent / f"{stem}.{profile_label(settings)}.wav"
        targets = targets_for(target, settings)
        return {"exists": any(target.exists() for target in targets),
                "paths": [str(path) for path in targets]}

    @app.get("/api/results/{result_id}/{kind}.wav")
    def audio(result_id: str, kind: str):
        return Response(jobs.media(result_id, kind), media_type="audio/wav")

    assets = Path(__file__).with_name("web")
    @app.get("/")
    def index():
        return FileResponse(assets / "index.html")
    app.mount("/assets", StaticFiles(directory=assets), name="assets")
    return app


def serve(port=8765, *, config=None, profile=None, overrides=None):
    if type(port) is not int or not 1 <= port <= 65535:
        raise ValueError("port must be between 1 and 65535")
    try:
        import uvicorn
        import fastapi  # noqa: F401 -- useful dependency diagnostic before binding
    except ImportError as error:
        raise RuntimeError("GUI dependencies are optional. Use './producer.sh gui' or install 'producer[gui]'.") from error
    app = create_app(port, config=config, profile=profile, overrides=overrides)
    # Pre-bind so a busy port fails before printing a misleading URL.
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        # Without SO_REUSEADDR a restart within a minute of a shutdown fails
        # with 'Address already in use': connections the server closed first
        # linger in TIME_WAIT while holding the local port. uvicorn sets this
        # flag only when it binds itself, not for pre-bound sockets.
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind(("127.0.0.1", port))
        sock.listen(128)
        print(f"Producer local GUI: http://127.0.0.1:{port}/", flush=True)
        server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port,
                                             access_log=False, proxy_headers=False))
        server.run(sockets=[sock])