112 lines
4.1 KiB
Python
112 lines
4.1 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] == "games":
|
|
self._json(MANAGER.get_game(path[1]).to_dict())
|
|
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 == ["games"]:
|
|
game = MANAGER.create_game(self._read_json())
|
|
self._json(game.to_dict(), HTTPStatus.CREATED)
|
|
return
|
|
if len(path) == 3 and path[0] == "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))
|
|
summaries = MANAGER.run_hands(path[1], count, until_one_left)
|
|
self._json({"hands": summaries, "game": MANAGER.get_game(path[1]).to_dict()})
|
|
return
|
|
if len(path) == 4 and path[0] == "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))
|
|
summaries = MANAGER.run_hands(path[1], count, until_one_left)
|
|
self._json({"hands": summaries, "game": MANAGER.get_game(path[1]).to_dict()})
|
|
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
|
|
|
|
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()
|