| 21 | |
| 22 | |
| 23 | def build_handler(scan_root: Path): |
| 24 | class DashboardHandler(BaseHTTPRequestHandler): |
| 25 | server_version = "LedgerForgeBenchmarkDashboard/1.0" |
| 26 | |
| 27 | def do_GET(self) -> None: # noqa: N802 |
| 28 | parsed = urlparse(self.path) |
| 29 | path = parsed.path |
| 30 | |
| 31 | if path in {"/", ""}: |
| 32 | self.redirect("/dashboard/") |
| 33 | return |
| 34 | if path == "/api/runs": |
| 35 | self.send_json({"runs": discover_runs(scan_root)}) |
| 36 | return |
| 37 | if path.startswith("/api/runs/"): |
| 38 | run_id = unquote(path.removeprefix("/api/runs/")) |
| 39 | self.handle_run(run_id) |
| 40 | return |
| 41 | if path == "/dashboard" or path == "/dashboard/": |
| 42 | self.serve_file(DASHBOARD_DIR / "index.html") |
| 43 | return |
| 44 | if path.startswith("/dashboard/"): |
| 45 | relative = Path(path.removeprefix("/dashboard/")) |
| 46 | target = (DASHBOARD_DIR / relative).resolve() |
| 47 | if DASHBOARD_DIR not in target.parents and target != DASHBOARD_DIR: |
| 48 | self.send_error(HTTPStatus.NOT_FOUND, "Not found") |
| 49 | return |
| 50 | self.serve_file(target) |
| 51 | return |
| 52 | |
| 53 | self.send_error(HTTPStatus.NOT_FOUND, "Not found") |
| 54 | |
| 55 | def log_message(self, format: str, *args) -> None: # noqa: A003 |
| 56 | return |
| 57 | |
| 58 | def handle_run(self, run_id: str) -> None: |
| 59 | for run in discover_runs(scan_root): |
| 60 | if run.get("id") == run_id: |
| 61 | self.send_json(run) |
| 62 | return |
| 63 | self.send_error(HTTPStatus.NOT_FOUND, "Run not found") |
| 64 | |
| 65 | def send_json(self, payload: object) -> None: |
| 66 | body = json.dumps(payload).encode("utf-8") |
| 67 | etag = payload_etag(payload) |
| 68 | if self.headers.get("If-None-Match") == etag: |
| 69 | self.send_response(HTTPStatus.NOT_MODIFIED) |
| 70 | self.send_header("ETag", etag) |
| 71 | self.end_headers() |
| 72 | return |
| 73 | self.send_response(HTTPStatus.OK) |
| 74 | self.send_header("Content-Type", "application/json; charset=utf-8") |
| 75 | self.send_header("Content-Length", str(len(body))) |
| 76 | self.send_header("Cache-Control", "no-store") |
| 77 | self.send_header("ETag", etag) |
| 78 | self.end_headers() |
| 79 | try: |
| 80 | self.wfile.write(body) |