tegwick
da3b7d66f0
Vendor whynot-design Layer 1 (tokens, CSS) and Layer 2 (<wn-*> components) via scripts/sync-whynot-design.sh with a pinned ref. Migrate the observatory shell to canonical web components, keep observatory-specific layout in styles.css, and add vendor integrity tests plus correct JS MIME types on the dev server.
81 lines
No EOL
2.8 KiB
Python
81 lines
No EOL
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
from .api import build_dashboard_payload
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
UI_DIR = ROOT / "ui"
|
|
|
|
UI_CONTENT_TYPES = {
|
|
".css": "text/css",
|
|
".js": "application/javascript",
|
|
".json": "application/json",
|
|
".html": "text/html; charset=utf-8",
|
|
}
|
|
|
|
|
|
class ObservatoryHandler(BaseHTTPRequestHandler):
|
|
data_dir: Path = ROOT / "data"
|
|
|
|
def _send(self, status: int, body: bytes, content_type: str) -> None:
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_GET(self) -> None:
|
|
parsed = urlparse(self.path)
|
|
if parsed.path == "/api/dashboard":
|
|
query = parse_qs(parsed.query)
|
|
period = query.get("period", [None])[0]
|
|
payload = build_dashboard_payload(self.data_dir, period)
|
|
self._send(200, json.dumps(payload).encode("utf-8"), "application/json")
|
|
return
|
|
|
|
if parsed.path == "/":
|
|
return self._serve_file(UI_DIR / "index.html", "text/html; charset=utf-8")
|
|
|
|
if parsed.path.startswith("/ui/"):
|
|
relative = parsed.path.removeprefix("/ui/")
|
|
target = UI_DIR / relative
|
|
if target.exists() and target.is_file():
|
|
content_type = UI_CONTENT_TYPES.get(target.suffix, "application/octet-stream")
|
|
if "charset" not in content_type:
|
|
content_type = f"{content_type}; charset=utf-8"
|
|
return self._serve_file(target, content_type)
|
|
|
|
self._send(404, b"Not found", "text/plain")
|
|
|
|
def _serve_file(self, path: Path, content_type: str) -> None:
|
|
self._send(200, path.read_bytes(), content_type)
|
|
|
|
def log_message(self, format: str, *args) -> None:
|
|
return
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description="Coulomb Economic Observatory UI server")
|
|
parser.add_argument("--host", default="127.0.0.1")
|
|
parser.add_argument("--port", type=int, default=8765)
|
|
parser.add_argument("--data-dir", type=Path, default=ROOT / "data")
|
|
args = parser.parse_args(argv)
|
|
|
|
ObservatoryHandler.data_dir = args.data_dir
|
|
server = ThreadingHTTPServer((args.host, args.port), ObservatoryHandler)
|
|
print(f"Economic Observatory UI: http://{args.host}:{args.port}/")
|
|
print(f"API: http://{args.host}:{args.port}/api/dashboard")
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
print("\nStopped.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main()) |