from __future__ import annotations import argparse import json import mimetypes from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any from urllib.error import HTTPError, URLError from urllib.parse import parse_qs, quote, urlencode, urlparse from urllib.request import ProxyHandler, Request, build_opener STATIC_DIR = Path(__file__).resolve().parent / "static" DEFAULT_CORE_BASE_URL = "http://127.0.0.1:8000" NO_PROXY_OPENER = build_opener(ProxyHandler({})) def build_game_url(params: dict[str, list[str]]) -> str: """Build a core-service game URL from query parameters. Supported forms: - ``url=https://host/games/demo`` for callers that already have a full URL. - ``base_url=http://host&game_id=demo`` for the common local service case. """ raw_url = _first(params, "url") if raw_url: return _validate_http_url(raw_url) base_url = _first(params, "base_url") or DEFAULT_CORE_BASE_URL game_id = _first(params, "game_id") if not game_id: raise ValueError("game_id is required when url is not provided") parsed = urlparse(_validate_http_url(base_url.rstrip("/"))) base_path = parsed.path.rstrip("/") game_path = f"{base_path}/games/{quote(game_id, safe='')}" return parsed._replace(path=game_path, query="", fragment="").geturl() def build_core_url(base_url: str, path: str) -> str: parsed = urlparse(_validate_http_url(base_url.rstrip("/"))) base_path = parsed.path.rstrip("/") target_path = f"{base_path}/{path.lstrip('/')}" return parsed._replace(path=target_path, query="", fragment="").geturl() def _first(params: dict[str, list[str]], key: str) -> str | None: values = params.get(key) if not values: return None return values[0] def _validate_http_url(value: str) -> str: parsed = urlparse(value) if parsed.scheme not in {"http", "https"} or not parsed.netloc: raise ValueError("url must be an absolute http(s) URL") return value class ReplayRequestHandler(BaseHTTPRequestHandler): server_version = "TexasHoldemReplay/0.1" def do_GET(self) -> None: parsed = urlparse(self.path) if parsed.path == "/health": self._json({"ok": True}) return if parsed.path == "/api/fetch-game": self._handle_fetch_game(parse_qs(parsed.query)) return self._serve_static(parsed.path) def do_POST(self) -> None: parsed = urlparse(self.path) if parsed.path == "/api/create-game": self._proxy_json_request("POST", "/games") return if parsed.path == "/api/run-hands": try: payload = self._read_json() base_url = str(payload.pop("base_url", DEFAULT_CORE_BASE_URL)) game_id = str(payload.pop("game_id", "")).strip() if not game_id: raise ValueError("game_id is required") path = f"/games/{quote(game_id, safe='')}/hands/run" self._proxy_json_request("POST", path, base_url=base_url, payload=payload) except ValueError as exc: self._json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) return self._json({"error": "not found"}, HTTPStatus.NOT_FOUND) def log_message(self, format: str, *args: Any) -> None: return def _handle_fetch_game(self, params: dict[str, list[str]]) -> None: try: target_url = build_game_url(params) payload, status = self._request_json("GET", target_url) self._json(payload, HTTPStatus(status)) except ValueError as exc: self._json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) except RuntimeError as exc: self._json({"error": str(exc)}, HTTPStatus.BAD_GATEWAY) def _proxy_json_request( self, method: str, path: str, base_url: str | None = None, payload: dict[str, Any] | None = None, ) -> None: try: if payload is None: payload = self._read_json() base_url = str(payload.pop("base_url", DEFAULT_CORE_BASE_URL)) target_url = build_core_url(base_url or DEFAULT_CORE_BASE_URL, path) response_payload, status = self._request_json(method, target_url, payload) self._json(response_payload, HTTPStatus(status)) except ValueError as exc: self._json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) except RuntimeError as exc: self._json({"error": str(exc)}, HTTPStatus.BAD_GATEWAY) def _request_json( self, method: str, url: str, payload: dict[str, Any] | None = None, ) -> tuple[dict[str, Any], int]: data = None headers = {"Accept": "application/json"} if payload is not None: data = json.dumps(payload).encode("utf-8") headers["Content-Type"] = "application/json" request = Request(url, data=data, headers=headers, method=method) try: with NO_PROXY_OPENER.open(request, timeout=20) as response: raw = response.read().decode("utf-8") status = response.status except HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") status = exc.code except (OSError, URLError) as exc: raise RuntimeError(f"core service request failed: {url}") from exc try: parsed = json.loads(raw) if raw else {} except json.JSONDecodeError as exc: raise RuntimeError(f"core service returned invalid JSON: {url}") from exc if not isinstance(parsed, dict): raise RuntimeError("core service response must be a JSON object") return parsed, status def _serve_static(self, path: str) -> None: relative = "index.html" if path in {"", "/"} else path.lstrip("/") if relative.startswith("static/"): relative = relative[len("static/") :] if "/" in relative: safe_parts = [part for part in relative.split("/") if part not in {"", ".", ".."}] relative = "/".join(safe_parts) target = STATIC_DIR / relative try: target.relative_to(STATIC_DIR) except ValueError: self._json({"error": "not found"}, HTTPStatus.NOT_FOUND) return if not target.is_file(): self._json({"error": "not found"}, HTTPStatus.NOT_FOUND) return content_type = mimetypes.guess_type(str(target))[0] or "application/octet-stream" body = target.read_bytes() self.send_response(HTTPStatus.OK) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _read_json(self) -> dict[str, Any]: length = int(self.headers.get("Content-Length", "0")) if length <= 0: return {} try: payload = json.loads(self.rfile.read(length).decode("utf-8")) except json.JSONDecodeError as exc: raise ValueError("request body must be valid JSON") from exc if not isinstance(payload, dict): raise ValueError("request body must be a JSON object") return payload def _json(self, payload: dict[str, Any], status: HTTPStatus = HTTPStatus.OK) -> None: body = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def create_server(host: str, port: int) -> ThreadingHTTPServer: return ThreadingHTTPServer((host, port), ReplayRequestHandler) def main() -> None: parser = argparse.ArgumentParser(description="Run the Texas Hold X web replay viewer.") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", default=8088, type=int) args = parser.parse_args() server = create_server(args.host, args.port) query = urlencode({"base_url": DEFAULT_CORE_BASE_URL}) print(f"Texas Hold X replay viewer listening on http://{args.host}:{args.port}/?{query}") try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() if __name__ == "__main__": main()