texas_hold_x/texas_holdem_replay/server.py

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 quote, unquote, urlparse, parse_qs
from urllib.request import Request, urlopen


STATIC_DIR = Path(__file__).with_name("static")


def build_game_url(query: dict[str, list[str]]) -> str:
    raw_url = _first(query, "url")
    if raw_url:
        return _validate_http_url(raw_url)

    base_url = _first(query, "base_url")
    game_id = _first(query, "game_id")
    if not base_url or not game_id:
        raise ValueError("provide either url or both base_url and game_id")

    base = _validate_http_url(base_url).rstrip("/")
    safe_game_id = quote(game_id.strip("/"), safe="")
    return f"{base}/games/{safe_game_id}"


class ReplayRequestHandler(BaseHTTPRequestHandler):
    server_version = "TexasHoldemReplay/0.1"

    def do_GET(self) -> None:
        parsed = urlparse(self.path)
        if parsed.path == "/api/health":
            self._json({"ok": True, "service": "texas-holdem-replay"})
            return
        if parsed.path == "/api/fetch-game":
            self._handle_fetch_game(parsed.query)
            return
        self._serve_static(parsed.path)

    def do_OPTIONS(self) -> None:
        self.send_response(HTTPStatus.NO_CONTENT)
        self._cors_headers()
        self.end_headers()

    def log_message(self, format: str, *args: Any) -> None:
        return

    def _handle_fetch_game(self, raw_query: str) -> None:
        query = parse_qs(raw_query)
        try:
            target = build_game_url(query)
            timeout = float(_first(query, "timeout") or 8)
            request = Request(target, headers={"Accept": "application/json"})
            with urlopen(request, timeout=max(1, min(timeout, 30))) as response:
                payload = response.read()
                content_type = response.headers.get("Content-Type", "")
            try:
                data = json.loads(payload.decode("utf-8"))
            except json.JSONDecodeError as exc:
                raise ValueError("target did not return valid JSON") from exc
            if not isinstance(data, dict):
                raise ValueError("target JSON must be an object")
            self._json({"source": target, "content_type": content_type, "game": data})
        except HTTPError as exc:
            body = exc.read().decode("utf-8", errors="replace")[:600]
            self._json(
                {"error": f"upstream returned HTTP {exc.code}", "detail": body},
                HTTPStatus.BAD_GATEWAY,
            )
        except (URLError, TimeoutError) as exc:
            self._json({"error": "failed to reach upstream", "detail": str(exc)}, HTTPStatus.BAD_GATEWAY)
        except ValueError as exc:
            self._json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)

    def _serve_static(self, raw_path: str) -> None:
        relative = unquote(raw_path.lstrip("/")) or "index.html"
        if relative.endswith("/"):
            relative += "index.html"
        candidate = (STATIC_DIR / relative).resolve()
        root = STATIC_DIR.resolve()
        if root not in candidate.parents and candidate != root:
            self._json({"error": "not found"}, HTTPStatus.NOT_FOUND)
            return
        if not candidate.is_file():
            candidate = STATIC_DIR / "index.html"
        body = candidate.read_bytes()
        content_type = mimetypes.guess_type(candidate.name)[0] or "application/octet-stream"
        self.send_response(HTTPStatus.OK)
        self._cors_headers()
        self.send_header("Content-Type", content_type)
        self.send_header("Cache-Control", "no-store" if candidate.name == "index.html" else "public, max-age=60")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    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._cors_headers()
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _cors_headers(self) -> None:
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")


def create_server(host: str, port: int) -> ThreadingHTTPServer:
    return ThreadingHTTPServer((host, port), ReplayRequestHandler)


def main() -> None:
    parser = argparse.ArgumentParser(description="Run the standalone Texas Hold X replay web 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)
    print(f"Texas Hold X replay viewer listening on http://{args.host}:{args.port}")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()


def _first(query: dict[str, list[str]], key: str) -> str | None:
    values = query.get(key)
    if not values:
        return None
    value = values[0].strip()
    return value or None


def _validate_http_url(value: str) -> str:
    parsed = urlparse(value.strip())
    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
        raise ValueError("url must be an absolute http(s) URL")
    return value.strip()


if __name__ == "__main__":
    main()