145 lines
5.4 KiB
Python
145 lines
5.4 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from http import HTTPStatus
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from typing import Any
|
|
from urllib.parse import urlparse
|
|
|
|
from texas_holdem.engine import GameComplete
|
|
from texas_holdem.service import GameManager
|
|
|
|
MANAGER = GameManager()
|
|
|
|
|
|
class PokerRequestHandler(BaseHTTPRequestHandler):
|
|
server_version = "TexasHoldemService/0.1"
|
|
|
|
def do_GET(self) -> None:
|
|
path = self._path_parts()
|
|
try:
|
|
if path == ["health"]:
|
|
self._json({"ok": True})
|
|
return
|
|
if path == ["games"]:
|
|
self._json({"games": MANAGER.list_games()})
|
|
return
|
|
if len(path) == 2 and path[0] in {"game", "games"}:
|
|
self._json(MANAGER.get_game_state(path[1]))
|
|
return
|
|
self._json({"error": "not found"}, HTTPStatus.NOT_FOUND)
|
|
except KeyError as exc:
|
|
self._json({"error": str(exc)}, HTTPStatus.NOT_FOUND)
|
|
|
|
def do_POST(self) -> None:
|
|
path = self._path_parts()
|
|
try:
|
|
if path in (["game"], ["games"]):
|
|
game = MANAGER.create_game(self._read_json())
|
|
self._json(game.snapshot_completed(), HTTPStatus.CREATED)
|
|
return
|
|
if len(path) == 3 and path[0] in {"game", "games"} and path[2] == "hands":
|
|
body = self._read_json()
|
|
count = int(body.get("count", 1))
|
|
until_one_left = bool(body.get("until_one_left", False))
|
|
small_blind, big_blind = self._extract_blinds(body)
|
|
summaries = MANAGER.run_hands(
|
|
path[1],
|
|
count,
|
|
until_one_left,
|
|
small_blind=small_blind,
|
|
big_blind=big_blind,
|
|
)
|
|
self._json({"hands": summaries, "game": MANAGER.get_game_state(path[1])})
|
|
return
|
|
if len(path) == 4 and path[0] in {"game", "games"} and path[2] == "hands" and path[3] == "run":
|
|
body = self._read_json()
|
|
count = int(body.get("count", 1))
|
|
until_one_left = bool(body.get("until_one_left", False))
|
|
small_blind, big_blind = self._extract_blinds(body)
|
|
summaries = MANAGER.run_hands(
|
|
path[1],
|
|
count,
|
|
until_one_left,
|
|
small_blind=small_blind,
|
|
big_blind=big_blind,
|
|
)
|
|
self._json({"hands": summaries, "game": MANAGER.get_game_state(path[1])})
|
|
return
|
|
self._json({"error": "not found"}, HTTPStatus.NOT_FOUND)
|
|
except KeyError as exc:
|
|
self._json({"error": str(exc)}, HTTPStatus.NOT_FOUND)
|
|
except (GameComplete, ValueError) as exc:
|
|
self._json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
|
|
|
|
def log_message(self, format: str, *args: Any) -> None:
|
|
return
|
|
|
|
def _path_parts(self) -> list[str]:
|
|
parsed = urlparse(self.path)
|
|
return [part for part in parsed.path.split("/") if part]
|
|
|
|
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
|
|
|
|
@staticmethod
|
|
def _extract_blinds(body: dict[str, Any]) -> tuple[int | None, int | None]:
|
|
"""Parse optional blind overrides from a /hands POST body.
|
|
|
|
Callers may omit both keys (keep current level), or supply both to
|
|
raise the blinds for the upcoming batch. Providing only one is
|
|
treated as a client error and surfaced via ``ValueError`` so the
|
|
handler can reply with 400.
|
|
"""
|
|
raw_small = body.get("small_blind")
|
|
raw_big = body.get("big_blind")
|
|
if raw_small is None and raw_big is None:
|
|
return None, None
|
|
if raw_small is None or raw_big is None:
|
|
raise ValueError(
|
|
"small_blind and big_blind must be provided together"
|
|
)
|
|
return int(raw_small), int(raw_big)
|
|
|
|
def _json(self, payload: dict[str, Any], status: HTTPStatus = HTTPStatus.OK) -> None:
|
|
body = json.dumps(payload, ensure_ascii=True).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), PokerRequestHandler)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Run the Texas Hold'em multi-agent service.")
|
|
parser.add_argument("--host", default="127.0.0.1")
|
|
parser.add_argument("--port", default=8000, type=int)
|
|
args = parser.parse_args()
|
|
|
|
server = create_server(args.host, args.port)
|
|
print(f"Texas Hold'em service listening on http://{args.host}:{args.port}")
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
server.server_close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|