host.controller_web_editor.serve

HTTP server for the controller web editor.

stdlib http.server is sufficient: this is a localhost dev tool, not a production service. Three concerns are wired together:

  • Serve files out of static/ (the built SPA, or a placeholder page).
  • Expose a small JSON API under /api/ for config I/O.
  • Keep all filesystem access through paths.safe_resolve.

Run with python -m host.controller_web_editor.

  1"""HTTP server for the controller web editor.
  2
  3stdlib ``http.server`` is sufficient: this is a localhost dev tool, not a
  4production service.  Three concerns are wired together:
  5
  6* Serve files out of ``static/`` (the built SPA, or a placeholder page).
  7* Expose a small JSON API under ``/api/`` for config I/O.
  8* Keep all filesystem access through :mod:`paths.safe_resolve`.
  9
 10Run with ``python -m host.controller_web_editor``.
 11"""
 12
 13from __future__ import annotations
 14
 15import argparse
 16import json
 17import logging
 18from http import HTTPStatus
 19from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 20from pathlib import Path
 21from urllib.parse import parse_qs, urlsplit
 22
 23from .build import maybe_build_spa
 24from .paths import PathNotAllowedError, list_yaml_configs, safe_resolve
 25from .routes.config import load_as_json, save_from_json
 26from .routes.export import ExportRequestError, export_from_yaml
 27from .routes.hitboxes import (
 28    LayoutNotAllowedError,
 29    list_layouts,
 30    load_layout,
 31    save_layout,
 32)
 33from .routes.prefs import load_prefs, save_prefs
 34
 35
 36log = logging.getLogger("controller_web_editor")
 37
 38STATIC_DIR = Path(__file__).resolve().parent / "static"
 39
 40# Shown in the browser when ``static/index.html`` is missing -- e.g. a
 41# bare checkout without Node and without the committed bundle.  Plain
 42# string so we don't need a template file on disk.
 43_MISSING_BUILD_HTML = """<!doctype html>
 44<html><head><meta charset="utf-8"><title>Controller Editor: build missing</title>
 45<style>
 46body { font-family: system-ui, sans-serif; max-width: 40rem; margin: 4rem auto;
 47       padding: 0 1rem; color: #222; }
 48h1 { font-size: 1.4rem; }
 49code, pre { background: #f3f3f3; padding: 0.1rem 0.4rem; border-radius: 4px;
 50            font-family: Consolas, "SF Mono", monospace; }
 51pre { padding: 0.6rem 0.8rem; overflow-x: auto; }
 52.tip { color: #555; font-size: 0.9em; margin-top: 1.5rem; }
 53</style></head><body>
 54<h1>The SPA isn't built yet</h1>
 55<p>The server is running but <code>static/index.html</code> doesn't exist,
 56so there's nothing to serve at <code>/</code>.</p>
 57<p>Build the frontend with:</p>
 58<pre>cd host/controller_web_editor/web
 59npm install
 60npm run build</pre>
 61<p>Then reload this page.  After that, the server will rebuild
 62automatically whenever you edit files under <code>web/src/</code>.</p>
 63<p class="tip">If you don't have Node installed, grab it from
 64<a href="https://nodejs.org">nodejs.org</a> (LTS is fine).  Or pull a
 65fresh checkout -- the compiled bundle is checked in.</p>
 66</body></html>
 67"""
 68
 69# MIME types we explicitly recognise.  Anything else falls back to octet-stream.
 70_MIME = {
 71    ".html": "text/html; charset=utf-8",
 72    ".js": "application/javascript; charset=utf-8",
 73    ".mjs": "application/javascript; charset=utf-8",
 74    ".css": "text/css; charset=utf-8",
 75    ".json": "application/json; charset=utf-8",
 76    ".svg": "image/svg+xml",
 77    ".png": "image/png",
 78    ".ico": "image/x-icon",
 79    ".map": "application/json; charset=utf-8",
 80}
 81
 82
 83class Handler(BaseHTTPRequestHandler):
 84    # Quieter access log; we'll route through logging instead of stderr.
 85    def log_message(self, fmt, *args):  # noqa: N802 (stdlib name)
 86        log.info("%s - %s", self.address_string(), fmt % args)
 87
 88    # --- response helpers ---
 89
 90    def _send_json(self, status: int, payload: object) -> None:
 91        body = json.dumps(payload).encode("utf-8")
 92        self.send_response(status)
 93        self.send_header("Content-Type", "application/json; charset=utf-8")
 94        self.send_header("Content-Length", str(len(body)))
 95        self.end_headers()
 96        self.wfile.write(body)
 97
 98    def _send_error_json(self, status: int, message: str) -> None:
 99        self._send_json(status, {"error": message})
100
101    def _send_file(self, path: Path) -> None:
102        try:
103            data = path.read_bytes()
104        except OSError as exc:
105            self._send_error_json(HTTPStatus.NOT_FOUND, str(exc))
106            return
107        self.send_response(HTTPStatus.OK)
108        self.send_header(
109            "Content-Type", _MIME.get(path.suffix.lower(), "application/octet-stream"))
110        self.send_header("Content-Length", str(len(data)))
111        self.send_header("Cache-Control", "no-store")
112        self.end_headers()
113        self.wfile.write(data)
114
115    # --- request entry points ---
116
117    def do_GET(self):  # noqa: N802 (stdlib name)
118        parts = urlsplit(self.path)
119        query = parse_qs(parts.query)
120
121        if parts.path == "/api/configs":
122            self._send_json(HTTPStatus.OK, {"configs": list_yaml_configs()})
123            return
124
125        if parts.path == "/api/config":
126            path_param = (query.get("path") or [""])[0]
127            try:
128                resolved = safe_resolve(path_param)
129            except PathNotAllowedError as exc:
130                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
131                return
132            if not resolved.is_file():
133                self._send_error_json(
134                    HTTPStatus.NOT_FOUND, f"no such file: {path_param}")
135                return
136            try:
137                self._send_json(HTTPStatus.OK, load_as_json(resolved))
138            except Exception as exc:
139                log.exception("load failed for %s", resolved)
140                self._send_error_json(
141                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
142            return
143
144        if parts.path == "/api/hitboxes":
145            layout = (query.get("layout") or ["xbox"])[0]
146            try:
147                self._send_json(HTTPStatus.OK, load_layout(layout))
148            except LayoutNotAllowedError as exc:
149                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
150            except FileNotFoundError as exc:
151                self._send_error_json(HTTPStatus.NOT_FOUND, str(exc))
152            except Exception as exc:
153                log.exception("hitboxes load failed for %s", layout)
154                self._send_error_json(
155                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
156            return
157
158        if parts.path == "/api/hitboxes/layouts":
159            self._send_json(HTTPStatus.OK, {"layouts": list_layouts()})
160            return
161
162        if parts.path == "/api/prefs":
163            self._send_json(HTTPStatus.OK, load_prefs())
164            return
165
166        if parts.path == "/api/export":
167            path_param = (query.get("path") or [""])[0]
168            orientation = (query.get("orientation") or ["landscape"])[0]
169            fmt = (query.get("format") or ["pdf"])[0]
170            hide_raw = (query.get("hide_unassigned") or ["0"])[0].lower()
171            hide_unassigned = hide_raw in ("1", "true", "yes", "on")
172            try:
173                resolved = safe_resolve(path_param)
174            except PathNotAllowedError as exc:
175                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
176                return
177            if not resolved.is_file():
178                self._send_error_json(
179                    HTTPStatus.NOT_FOUND, f"no such file: {path_param}")
180                return
181            try:
182                data, content_type, filename = export_from_yaml(
183                    resolved, orientation, fmt, hide_unassigned)
184            except ExportRequestError as exc:
185                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
186                return
187            except ValueError as exc:
188                # No controllers, multi-page PNG, etc.
189                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
190                return
191            except Exception as exc:
192                log.exception("export failed for %s", resolved)
193                self._send_error_json(
194                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
195                return
196            self.send_response(HTTPStatus.OK)
197            self.send_header("Content-Type", content_type)
198            self.send_header("Content-Length", str(len(data)))
199            self.send_header(
200                "Content-Disposition", f'attachment; filename="{filename}"')
201            self.send_header("Cache-Control", "no-store")
202            self.end_headers()
203            self.wfile.write(data)
204            return
205
206        # Static files.  Default route serves index.html.
207        rel = parts.path.lstrip("/") or "index.html"
208        target = (STATIC_DIR / rel).resolve()
209        try:
210            target.relative_to(STATIC_DIR.resolve())
211        except ValueError:
212            self._send_error_json(HTTPStatus.FORBIDDEN, "outside static root")
213            return
214        if not target.is_file():
215            # SPA fallback — anything unknown returns index.html so the
216            # frontend router can decide.
217            target = STATIC_DIR / "index.html"
218            if not target.is_file():
219                # Friendly explainer when nobody has built the bundle.
220                body = _MISSING_BUILD_HTML.encode("utf-8")
221                self.send_response(HTTPStatus.OK)
222                self.send_header("Content-Type", "text/html; charset=utf-8")
223                self.send_header("Content-Length", str(len(body)))
224                self.send_header("Cache-Control", "no-store")
225                self.end_headers()
226                self.wfile.write(body)
227                return
228        self._send_file(target)
229
230    def do_POST(self):  # noqa: N802 (stdlib name)
231        parts = urlsplit(self.path)
232        query = parse_qs(parts.query)
233
234        if parts.path == "/api/config":
235            path_param = (query.get("path") or [""])[0]
236            try:
237                resolved = safe_resolve(path_param)
238            except PathNotAllowedError as exc:
239                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
240                return
241            length = int(self.headers.get("Content-Length", "0"))
242            try:
243                payload = json.loads(self.rfile.read(length).decode("utf-8"))
244            except json.JSONDecodeError as exc:
245                self._send_error_json(
246                    HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}")
247                return
248            try:
249                save_from_json(resolved, payload)
250            except Exception as exc:
251                log.exception("save failed for %s", resolved)
252                self._send_error_json(
253                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
254                return
255            self._send_json(HTTPStatus.OK, {"saved": path_param})
256            return
257
258        if parts.path == "/api/hitboxes":
259            layout = (query.get("layout") or ["xbox"])[0]
260            length = int(self.headers.get("Content-Length", "0"))
261            try:
262                payload = json.loads(self.rfile.read(length).decode("utf-8"))
263            except json.JSONDecodeError as exc:
264                self._send_error_json(
265                    HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}")
266                return
267            try:
268                save_layout(layout, payload)
269            except LayoutNotAllowedError as exc:
270                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
271                return
272            except ValueError as exc:
273                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
274                return
275            except Exception as exc:
276                log.exception("hitboxes save failed for %s", layout)
277                self._send_error_json(
278                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
279                return
280            self._send_json(HTTPStatus.OK, {"saved": layout})
281            return
282
283        if parts.path == "/api/prefs":
284            length = int(self.headers.get("Content-Length", "0"))
285            try:
286                payload = json.loads(self.rfile.read(length).decode("utf-8"))
287            except json.JSONDecodeError as exc:
288                self._send_error_json(
289                    HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}")
290                return
291            if not isinstance(payload, dict):
292                self._send_error_json(
293                    HTTPStatus.BAD_REQUEST, "payload must be a JSON object")
294                return
295            try:
296                updated = save_prefs(payload)
297            except ValueError as exc:
298                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
299                return
300            except Exception as exc:
301                log.exception("prefs save failed")
302                self._send_error_json(
303                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
304                return
305            self._send_json(HTTPStatus.OK, updated)
306            return
307
308        self._send_error_json(HTTPStatus.NOT_FOUND, parts.path)
309
310
311def run(host: str = "127.0.0.1", port: int = 8071) -> None:
312    logging.basicConfig(
313        level=logging.INFO,
314        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
315    )
316    # Auto-build the SPA if web/src/ is newer than static/.  Best-effort;
317    # skipped silently when Node isn't installed or
318    # CONTROLLER_WEB_EDITOR_SKIP_BUILD is set.
319    status = maybe_build_spa()
320    log.info("spa build: %s", status)
321    server = ThreadingHTTPServer((host, port), Handler)
322    log.info("serving on http://%s:%d (static=%s)", host, port, STATIC_DIR)
323    try:
324        server.serve_forever()
325    except KeyboardInterrupt:
326        log.info("shutting down")
327    finally:
328        server.server_close()
329
330
331def main(argv: list[str] | None = None) -> int:
332    parser = argparse.ArgumentParser(description=__doc__)
333    parser.add_argument("--host", default="127.0.0.1")
334    parser.add_argument("--port", type=int, default=8071)
335    args = parser.parse_args(argv)
336    run(args.host, args.port)
337    return 0
log = <Logger controller_web_editor (DEBUG)>
STATIC_DIR = PosixPath('/home/runner/work/Robot-2026/Robot-2026/host/controller_web_editor/static')
class Handler(http.server.BaseHTTPRequestHandler):
 84class Handler(BaseHTTPRequestHandler):
 85    # Quieter access log; we'll route through logging instead of stderr.
 86    def log_message(self, fmt, *args):  # noqa: N802 (stdlib name)
 87        log.info("%s - %s", self.address_string(), fmt % args)
 88
 89    # --- response helpers ---
 90
 91    def _send_json(self, status: int, payload: object) -> None:
 92        body = json.dumps(payload).encode("utf-8")
 93        self.send_response(status)
 94        self.send_header("Content-Type", "application/json; charset=utf-8")
 95        self.send_header("Content-Length", str(len(body)))
 96        self.end_headers()
 97        self.wfile.write(body)
 98
 99    def _send_error_json(self, status: int, message: str) -> None:
100        self._send_json(status, {"error": message})
101
102    def _send_file(self, path: Path) -> None:
103        try:
104            data = path.read_bytes()
105        except OSError as exc:
106            self._send_error_json(HTTPStatus.NOT_FOUND, str(exc))
107            return
108        self.send_response(HTTPStatus.OK)
109        self.send_header(
110            "Content-Type", _MIME.get(path.suffix.lower(), "application/octet-stream"))
111        self.send_header("Content-Length", str(len(data)))
112        self.send_header("Cache-Control", "no-store")
113        self.end_headers()
114        self.wfile.write(data)
115
116    # --- request entry points ---
117
118    def do_GET(self):  # noqa: N802 (stdlib name)
119        parts = urlsplit(self.path)
120        query = parse_qs(parts.query)
121
122        if parts.path == "/api/configs":
123            self._send_json(HTTPStatus.OK, {"configs": list_yaml_configs()})
124            return
125
126        if parts.path == "/api/config":
127            path_param = (query.get("path") or [""])[0]
128            try:
129                resolved = safe_resolve(path_param)
130            except PathNotAllowedError as exc:
131                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
132                return
133            if not resolved.is_file():
134                self._send_error_json(
135                    HTTPStatus.NOT_FOUND, f"no such file: {path_param}")
136                return
137            try:
138                self._send_json(HTTPStatus.OK, load_as_json(resolved))
139            except Exception as exc:
140                log.exception("load failed for %s", resolved)
141                self._send_error_json(
142                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
143            return
144
145        if parts.path == "/api/hitboxes":
146            layout = (query.get("layout") or ["xbox"])[0]
147            try:
148                self._send_json(HTTPStatus.OK, load_layout(layout))
149            except LayoutNotAllowedError as exc:
150                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
151            except FileNotFoundError as exc:
152                self._send_error_json(HTTPStatus.NOT_FOUND, str(exc))
153            except Exception as exc:
154                log.exception("hitboxes load failed for %s", layout)
155                self._send_error_json(
156                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
157            return
158
159        if parts.path == "/api/hitboxes/layouts":
160            self._send_json(HTTPStatus.OK, {"layouts": list_layouts()})
161            return
162
163        if parts.path == "/api/prefs":
164            self._send_json(HTTPStatus.OK, load_prefs())
165            return
166
167        if parts.path == "/api/export":
168            path_param = (query.get("path") or [""])[0]
169            orientation = (query.get("orientation") or ["landscape"])[0]
170            fmt = (query.get("format") or ["pdf"])[0]
171            hide_raw = (query.get("hide_unassigned") or ["0"])[0].lower()
172            hide_unassigned = hide_raw in ("1", "true", "yes", "on")
173            try:
174                resolved = safe_resolve(path_param)
175            except PathNotAllowedError as exc:
176                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
177                return
178            if not resolved.is_file():
179                self._send_error_json(
180                    HTTPStatus.NOT_FOUND, f"no such file: {path_param}")
181                return
182            try:
183                data, content_type, filename = export_from_yaml(
184                    resolved, orientation, fmt, hide_unassigned)
185            except ExportRequestError as exc:
186                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
187                return
188            except ValueError as exc:
189                # No controllers, multi-page PNG, etc.
190                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
191                return
192            except Exception as exc:
193                log.exception("export failed for %s", resolved)
194                self._send_error_json(
195                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
196                return
197            self.send_response(HTTPStatus.OK)
198            self.send_header("Content-Type", content_type)
199            self.send_header("Content-Length", str(len(data)))
200            self.send_header(
201                "Content-Disposition", f'attachment; filename="{filename}"')
202            self.send_header("Cache-Control", "no-store")
203            self.end_headers()
204            self.wfile.write(data)
205            return
206
207        # Static files.  Default route serves index.html.
208        rel = parts.path.lstrip("/") or "index.html"
209        target = (STATIC_DIR / rel).resolve()
210        try:
211            target.relative_to(STATIC_DIR.resolve())
212        except ValueError:
213            self._send_error_json(HTTPStatus.FORBIDDEN, "outside static root")
214            return
215        if not target.is_file():
216            # SPA fallback — anything unknown returns index.html so the
217            # frontend router can decide.
218            target = STATIC_DIR / "index.html"
219            if not target.is_file():
220                # Friendly explainer when nobody has built the bundle.
221                body = _MISSING_BUILD_HTML.encode("utf-8")
222                self.send_response(HTTPStatus.OK)
223                self.send_header("Content-Type", "text/html; charset=utf-8")
224                self.send_header("Content-Length", str(len(body)))
225                self.send_header("Cache-Control", "no-store")
226                self.end_headers()
227                self.wfile.write(body)
228                return
229        self._send_file(target)
230
231    def do_POST(self):  # noqa: N802 (stdlib name)
232        parts = urlsplit(self.path)
233        query = parse_qs(parts.query)
234
235        if parts.path == "/api/config":
236            path_param = (query.get("path") or [""])[0]
237            try:
238                resolved = safe_resolve(path_param)
239            except PathNotAllowedError as exc:
240                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
241                return
242            length = int(self.headers.get("Content-Length", "0"))
243            try:
244                payload = json.loads(self.rfile.read(length).decode("utf-8"))
245            except json.JSONDecodeError as exc:
246                self._send_error_json(
247                    HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}")
248                return
249            try:
250                save_from_json(resolved, payload)
251            except Exception as exc:
252                log.exception("save failed for %s", resolved)
253                self._send_error_json(
254                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
255                return
256            self._send_json(HTTPStatus.OK, {"saved": path_param})
257            return
258
259        if parts.path == "/api/hitboxes":
260            layout = (query.get("layout") or ["xbox"])[0]
261            length = int(self.headers.get("Content-Length", "0"))
262            try:
263                payload = json.loads(self.rfile.read(length).decode("utf-8"))
264            except json.JSONDecodeError as exc:
265                self._send_error_json(
266                    HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}")
267                return
268            try:
269                save_layout(layout, payload)
270            except LayoutNotAllowedError as exc:
271                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
272                return
273            except ValueError as exc:
274                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
275                return
276            except Exception as exc:
277                log.exception("hitboxes save failed for %s", layout)
278                self._send_error_json(
279                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
280                return
281            self._send_json(HTTPStatus.OK, {"saved": layout})
282            return
283
284        if parts.path == "/api/prefs":
285            length = int(self.headers.get("Content-Length", "0"))
286            try:
287                payload = json.loads(self.rfile.read(length).decode("utf-8"))
288            except json.JSONDecodeError as exc:
289                self._send_error_json(
290                    HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}")
291                return
292            if not isinstance(payload, dict):
293                self._send_error_json(
294                    HTTPStatus.BAD_REQUEST, "payload must be a JSON object")
295                return
296            try:
297                updated = save_prefs(payload)
298            except ValueError as exc:
299                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
300                return
301            except Exception as exc:
302                log.exception("prefs save failed")
303                self._send_error_json(
304                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
305                return
306            self._send_json(HTTPStatus.OK, updated)
307            return
308
309        self._send_error_json(HTTPStatus.NOT_FOUND, parts.path)

HTTP request handler base class.

The following explanation of HTTP serves to guide you through the code as well as to expose any misunderstandings I may have about HTTP (so you don't need to read the code to figure out I'm wrong :-).

HTTP (HyperText Transfer Protocol) is an extensible protocol on top of a reliable stream transport (e.g. TCP/IP). The protocol recognizes three parts to a request:

  1. One line identifying the request type and path
  2. An optional set of RFC-822-style headers
  3. An optional data part

The headers and data are separated by a blank line.

The first line of the request has the form

where is a (case-sensitive) keyword such as GET or POST, is a string containing path information for the request, and should be the string "HTTP/1.0" or "HTTP/1.1". is encoded using the URL encoding scheme (using %xx to signify the ASCII character with hex code xx).

The specification specifies that lines are separated by CRLF but for compatibility with the widest range of clients recommends servers also handle LF. Similarly, whitespace in the request line is treated sensibly (allowing multiple spaces between components and allowing trailing whitespace).

Similarly, for output, lines ought to be separated by CRLF pairs but most clients grok LF characters just fine.

If the first line of the request has the form

(i.e. is left out) then this is assumed to be an HTTP 0.9 request; this form has no optional headers and data part and the reply consists of just the data.

The reply form of the HTTP 1.x protocol again has three parts:

  1. One line giving the response code
  2. An optional set of RFC-822-style headers
  3. The data

Again, the headers and data are separated by a blank line.

The response code line has the form

where is the protocol version ("HTTP/1.0" or "HTTP/1.1"), is a 3-digit response code indicating success or failure of the request, and is an optional human-readable string explaining what the response code means.

This server parses the request and the headers, and then calls a function specific to the request type (). Specifically, a request SPAM will be handled by a method do_SPAM(). If no such method exists the server sends an error response to the client. If it exists, it is called with no arguments:

do_SPAM()

Note that the request name is case sensitive (i.e. SPAM and spam are different requests).

The various request details are stored in instance variables:

  • client_address is the client IP address in the form (host, port);

  • command, path and version are the broken-down request line;

  • headers is an instance of email.message.Message (or a derived class) containing the header information;

  • rfile is a file object open for reading positioned at the start of the optional input data part;

  • wfile is a file object open for writing.

IT IS IMPORTANT TO ADHERE TO THE PROTOCOL FOR WRITING!

The first thing to be written must be the response line. Then follow 0 or more header lines, then a blank line, and then the actual data (if any). The meaning of the header lines depends on the command executed by the server; in most cases, when data is returned, there should be at least one header line of the form

Content-type: /

where and should be registered MIME types, e.g. "text/html" or "text/plain".

def log_message(self, fmt, *args):
86    def log_message(self, fmt, *args):  # noqa: N802 (stdlib name)
87        log.info("%s - %s", self.address_string(), fmt % args)

Log an arbitrary message.

This is used by all other logging functions. Override it if you have specific logging wishes.

The first argument, FORMAT, is a format string for the message to be logged. If the format string contains any % escapes requiring parameters, they should be specified as subsequent arguments (it's just like printf!).

The client ip and current date/time are prefixed to every message.

Unicode control characters are replaced with escaped hex before writing the output to stderr.

def do_GET(self):
118    def do_GET(self):  # noqa: N802 (stdlib name)
119        parts = urlsplit(self.path)
120        query = parse_qs(parts.query)
121
122        if parts.path == "/api/configs":
123            self._send_json(HTTPStatus.OK, {"configs": list_yaml_configs()})
124            return
125
126        if parts.path == "/api/config":
127            path_param = (query.get("path") or [""])[0]
128            try:
129                resolved = safe_resolve(path_param)
130            except PathNotAllowedError as exc:
131                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
132                return
133            if not resolved.is_file():
134                self._send_error_json(
135                    HTTPStatus.NOT_FOUND, f"no such file: {path_param}")
136                return
137            try:
138                self._send_json(HTTPStatus.OK, load_as_json(resolved))
139            except Exception as exc:
140                log.exception("load failed for %s", resolved)
141                self._send_error_json(
142                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
143            return
144
145        if parts.path == "/api/hitboxes":
146            layout = (query.get("layout") or ["xbox"])[0]
147            try:
148                self._send_json(HTTPStatus.OK, load_layout(layout))
149            except LayoutNotAllowedError as exc:
150                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
151            except FileNotFoundError as exc:
152                self._send_error_json(HTTPStatus.NOT_FOUND, str(exc))
153            except Exception as exc:
154                log.exception("hitboxes load failed for %s", layout)
155                self._send_error_json(
156                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
157            return
158
159        if parts.path == "/api/hitboxes/layouts":
160            self._send_json(HTTPStatus.OK, {"layouts": list_layouts()})
161            return
162
163        if parts.path == "/api/prefs":
164            self._send_json(HTTPStatus.OK, load_prefs())
165            return
166
167        if parts.path == "/api/export":
168            path_param = (query.get("path") or [""])[0]
169            orientation = (query.get("orientation") or ["landscape"])[0]
170            fmt = (query.get("format") or ["pdf"])[0]
171            hide_raw = (query.get("hide_unassigned") or ["0"])[0].lower()
172            hide_unassigned = hide_raw in ("1", "true", "yes", "on")
173            try:
174                resolved = safe_resolve(path_param)
175            except PathNotAllowedError as exc:
176                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
177                return
178            if not resolved.is_file():
179                self._send_error_json(
180                    HTTPStatus.NOT_FOUND, f"no such file: {path_param}")
181                return
182            try:
183                data, content_type, filename = export_from_yaml(
184                    resolved, orientation, fmt, hide_unassigned)
185            except ExportRequestError as exc:
186                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
187                return
188            except ValueError as exc:
189                # No controllers, multi-page PNG, etc.
190                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
191                return
192            except Exception as exc:
193                log.exception("export failed for %s", resolved)
194                self._send_error_json(
195                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
196                return
197            self.send_response(HTTPStatus.OK)
198            self.send_header("Content-Type", content_type)
199            self.send_header("Content-Length", str(len(data)))
200            self.send_header(
201                "Content-Disposition", f'attachment; filename="{filename}"')
202            self.send_header("Cache-Control", "no-store")
203            self.end_headers()
204            self.wfile.write(data)
205            return
206
207        # Static files.  Default route serves index.html.
208        rel = parts.path.lstrip("/") or "index.html"
209        target = (STATIC_DIR / rel).resolve()
210        try:
211            target.relative_to(STATIC_DIR.resolve())
212        except ValueError:
213            self._send_error_json(HTTPStatus.FORBIDDEN, "outside static root")
214            return
215        if not target.is_file():
216            # SPA fallback — anything unknown returns index.html so the
217            # frontend router can decide.
218            target = STATIC_DIR / "index.html"
219            if not target.is_file():
220                # Friendly explainer when nobody has built the bundle.
221                body = _MISSING_BUILD_HTML.encode("utf-8")
222                self.send_response(HTTPStatus.OK)
223                self.send_header("Content-Type", "text/html; charset=utf-8")
224                self.send_header("Content-Length", str(len(body)))
225                self.send_header("Cache-Control", "no-store")
226                self.end_headers()
227                self.wfile.write(body)
228                return
229        self._send_file(target)
def do_POST(self):
231    def do_POST(self):  # noqa: N802 (stdlib name)
232        parts = urlsplit(self.path)
233        query = parse_qs(parts.query)
234
235        if parts.path == "/api/config":
236            path_param = (query.get("path") or [""])[0]
237            try:
238                resolved = safe_resolve(path_param)
239            except PathNotAllowedError as exc:
240                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
241                return
242            length = int(self.headers.get("Content-Length", "0"))
243            try:
244                payload = json.loads(self.rfile.read(length).decode("utf-8"))
245            except json.JSONDecodeError as exc:
246                self._send_error_json(
247                    HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}")
248                return
249            try:
250                save_from_json(resolved, payload)
251            except Exception as exc:
252                log.exception("save failed for %s", resolved)
253                self._send_error_json(
254                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
255                return
256            self._send_json(HTTPStatus.OK, {"saved": path_param})
257            return
258
259        if parts.path == "/api/hitboxes":
260            layout = (query.get("layout") or ["xbox"])[0]
261            length = int(self.headers.get("Content-Length", "0"))
262            try:
263                payload = json.loads(self.rfile.read(length).decode("utf-8"))
264            except json.JSONDecodeError as exc:
265                self._send_error_json(
266                    HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}")
267                return
268            try:
269                save_layout(layout, payload)
270            except LayoutNotAllowedError as exc:
271                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
272                return
273            except ValueError as exc:
274                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
275                return
276            except Exception as exc:
277                log.exception("hitboxes save failed for %s", layout)
278                self._send_error_json(
279                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
280                return
281            self._send_json(HTTPStatus.OK, {"saved": layout})
282            return
283
284        if parts.path == "/api/prefs":
285            length = int(self.headers.get("Content-Length", "0"))
286            try:
287                payload = json.loads(self.rfile.read(length).decode("utf-8"))
288            except json.JSONDecodeError as exc:
289                self._send_error_json(
290                    HTTPStatus.BAD_REQUEST, f"invalid JSON: {exc}")
291                return
292            if not isinstance(payload, dict):
293                self._send_error_json(
294                    HTTPStatus.BAD_REQUEST, "payload must be a JSON object")
295                return
296            try:
297                updated = save_prefs(payload)
298            except ValueError as exc:
299                self._send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
300                return
301            except Exception as exc:
302                log.exception("prefs save failed")
303                self._send_error_json(
304                    HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
305                return
306            self._send_json(HTTPStatus.OK, updated)
307            return
308
309        self._send_error_json(HTTPStatus.NOT_FOUND, parts.path)
def run(host: str = '127.0.0.1', port: int = 8071) -> None:
312def run(host: str = "127.0.0.1", port: int = 8071) -> None:
313    logging.basicConfig(
314        level=logging.INFO,
315        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
316    )
317    # Auto-build the SPA if web/src/ is newer than static/.  Best-effort;
318    # skipped silently when Node isn't installed or
319    # CONTROLLER_WEB_EDITOR_SKIP_BUILD is set.
320    status = maybe_build_spa()
321    log.info("spa build: %s", status)
322    server = ThreadingHTTPServer((host, port), Handler)
323    log.info("serving on http://%s:%d (static=%s)", host, port, STATIC_DIR)
324    try:
325        server.serve_forever()
326    except KeyboardInterrupt:
327        log.info("shutting down")
328    finally:
329        server.server_close()
def main(argv: list[str] | None = None) -> int:
332def main(argv: list[str] | None = None) -> int:
333    parser = argparse.ArgumentParser(description=__doc__)
334    parser.add_argument("--host", default="127.0.0.1")
335    parser.add_argument("--port", type=int, default=8071)
336    args = parser.parse_args(argv)
337    run(args.host, args.port)
338    return 0