"""Serve the local GeoAI Workbench console from the repository root."""
|
|
from __future__ import annotations
|
|
import argparse
|
import os
|
from pathlib import PurePosixPath
|
from http import HTTPStatus
|
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
|
from pathlib import Path
|
from urllib.parse import unquote, urlsplit
|
|
|
DEFAULT_HOST = "127.0.0.1"
|
DEFAULT_PORT = 6173
|
ALLOWED_PATH_PREFIXES = (
|
"apps/workbench-console",
|
"shared/outputs",
|
"shared/data/raw/01-object-detection",
|
)
|
|
|
class WorkbenchConsoleHandler(SimpleHTTPRequestHandler):
|
"""Read-only static handler rooted at the workbench repository."""
|
|
def do_GET(self) -> None: # noqa: N802 - inherited standard-library method name
|
if urlsplit(self.path).path == "/":
|
self.send_response(HTTPStatus.FOUND)
|
self.send_header("Location", "/apps/workbench-console/")
|
self.end_headers()
|
return
|
super().do_GET()
|
|
def translate_path(self, path: str) -> str:
|
"""Expose only the static UI and artifacts required by the local console."""
|
decoded_path = unquote(urlsplit(path).path).lstrip("/")
|
requested = PurePosixPath(decoded_path)
|
is_allowed = any(
|
decoded_path == prefix or decoded_path.startswith(f"{prefix}/")
|
for prefix in ALLOWED_PATH_PREFIXES
|
)
|
if ".." in requested.parts or not is_allowed:
|
return os.fspath(Path(self.directory) / ".console-forbidden")
|
if decoded_path == "apps/workbench-console" or decoded_path.startswith("apps/workbench-console/"):
|
console_relative = requested.parts[2:]
|
return os.fspath(Path(self.directory) / "apps" / "workbench-console" / "dist" / Path(*console_relative))
|
return os.fspath(Path(self.directory).joinpath(*requested.parts))
|
|
def end_headers(self) -> None:
|
self.send_header("Cache-Control", "no-store")
|
self.send_header("X-Content-Type-Options", "nosniff")
|
super().end_headers()
|
|
|
def parse_args() -> argparse.Namespace:
|
root = Path(__file__).resolve().parents[1]
|
parser = argparse.ArgumentParser(description="Serve the local GeoAI Workbench console.")
|
parser.add_argument("--host", default=DEFAULT_HOST, help="Bind address. Defaults to loopback only.")
|
parser.add_argument("--port", type=int, default=DEFAULT_PORT, help="TCP port in the 6000-6999 range.")
|
parser.add_argument("--root", type=Path, default=root, help="Workbench repository root to serve.")
|
return parser.parse_args()
|
|
|
def main() -> int:
|
args = parse_args()
|
if not 6000 <= args.port <= 6999:
|
raise SystemExit("Port must be in the 6000-6999 range.")
|
if args.host not in {"127.0.0.1", "localhost", "::1"}:
|
raise SystemExit("This console is local-only. Use 127.0.0.1, localhost, or ::1.")
|
root = args.root.resolve()
|
app_dir = root / "apps" / "workbench-console"
|
if not (root / "shared").is_dir() or not (app_dir / "dist" / "index.html").is_file():
|
raise SystemExit(f"Not a GeoAI Workbench root: {root}")
|
|
handler = lambda *handler_args, **handler_kwargs: WorkbenchConsoleHandler( # noqa: E731
|
*handler_args, directory=os.fspath(root), **handler_kwargs
|
)
|
server = ThreadingHTTPServer((args.host, args.port), handler)
|
print(f"GeoAI Workbench console: http://{args.host}:{args.port}")
|
print(f"Serving built console and read-only artifacts from: {root}")
|
try:
|
server.serve_forever()
|
except KeyboardInterrupt:
|
print("\nConsole stopped.")
|
finally:
|
server.server_close()
|
return 0
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
