3 Commits

9 changed files with 1532 additions and 1592 deletions
+1
View File
@@ -32,6 +32,7 @@ htmlcov/
# IDE and editor files
.idea/
.vscode/
.codex/
*.swp
*.swo
+2 -6
View File
@@ -4,16 +4,12 @@
Core poker service code lives in `texas_holdem/`. Important modules include `engine.py` for Texas Hold'em rules, `service.py` for game management, `server.py` for the HTTP API, `agents.py` for local/HTTP agents, and `ai_client.py` / `human_client.py` for standalone agents. Prompt templates live in `texas_holdem/prompts/`.
Replay UI code lives in `texas_holdem_replay/`, with static browser assets under `texas_holdem_replay/static/`. Tests are in `tests/`, named by feature area such as `test_engine.py`, `test_service.py`, and `test_replay_server.py`. Design notes belong in `docs/`.
## Build, Test, and Development Commands
- `python -m unittest discover -v` runs the full test suite.
- `python -m compileall texas_holdem texas_holdem_replay tests` checks import and syntax validity.
- `python -m texas_holdem.server --host 127.0.0.1 --port 8000` starts the game service.
- `python -m texas_holdem.human_client --port 9001 --keep-history` starts an interactive human HTTP agent.
- `python -m texas_holdem.ai_client --port 9101 --api-key "$OPENAI_API_KEY" --model gpt-4o-mini` starts an OpenAI-compatible AI agent.
- `python -m texas_holdem_replay.server --port 8088` starts the replay viewer.
## Coding Style & Naming Conventions
@@ -21,11 +17,11 @@ Use Python 3.11+ standard-library APIs unless a dependency is intentionally adde
## Testing Guidelines
Use `unittest`. Add tests near the behavior changed: engine rules in `tests/test_engine.py`, HTTP/service behavior in `tests/test_service.py`, agent transport in `tests/test_agents.py`, and replay UI server helpers in `tests/test_replay_server.py`. New bug fixes should include a regression test. Avoid tests that require external network access or real LLM calls.
Use `unittest`. Add tests near the behavior changed: engine rules in `tests/test_engine.py`, HTTP/service behavior in `tests/test_service.py`, agent transport in `tests/test_agents.py`. New bug fixes should include a regression test. Avoid tests that require external network access or real LLM calls.
## Commit & Pull Request Guidelines
History uses short Conventional Commit-style subjects, for example `feat: add replay server and web client` and `fix: game service api block when a game is running`. Keep commits scoped to one behavior change. Pull requests should include a short summary, test commands run, linked issue or motivation, and screenshots only for replay UI or visible terminal-output changes.
History uses short Conventional Commit-style subjects, for example `feat: add server and agent client` and `fix: game service api block when a game is running`. Keep commits scoped to one behavior change. Pull requests should include a short summary, test commands run, linked issue or motivation, and screenshots only for visible terminal-output changes.
## Security & Configuration Tips
+19
View File
@@ -126,3 +126,22 @@ AI Agent 会在终端输出:
```bash
python -m unittest discover -v
```
## Web 回放与控制台
启动核心游戏服务后,可以单独启动 Web 回放服务:
```bash
python -m texas_holdem_replay.server --host 127.0.0.1 --port 8088
```
打开 `http://127.0.0.1:8088`。页面通过自身的代理接口访问核心服务,
避免浏览器跨域限制;它不会导入或耦合 `texas_holdem.engine` 内部代码。
页面支持:
- 拉取 `GET /games/{game_id}` 快照并按 `hands[].actions` 生成逐帧回放。
- 通过代理调用核心服务运行指定数量手牌。
- 可选覆盖下一批手牌的大小盲。
- 上传或粘贴静态 JSON 快照进行离线回放。
- 自动轮询正在运行的游戏,保留当前历史查看位置。
+7 -2
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
import unittest
from texas_holdem_replay.server import build_game_url
from texas_holdem_replay.server import build_core_url, build_game_url
class ReplayServerTests(unittest.TestCase):
@@ -22,7 +22,12 @@ class ReplayServerTests(unittest.TestCase):
with self.assertRaises(ValueError):
build_game_url({"url": ["file:///tmp/game.json"]})
def test_build_core_url_preserves_base_path(self) -> None:
self.assertEqual(
build_core_url("http://127.0.0.1:8000/api/", "/games/demo/hands/run"),
"http://127.0.0.1:8000/api/games/demo/hands/run",
)
if __name__ == "__main__":
unittest.main()
+1 -1
View File
@@ -1,2 +1,2 @@
"""Standalone web replay viewer for Texas Hold X game snapshots."""
"""Standalone web replay viewer for Texas Hold X."""
+149 -80
View File
@@ -8,26 +8,55 @@ 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
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({}))
STATIC_DIR = Path(__file__).with_name("static")
def build_game_url(params: dict[str, list[str]]) -> str:
"""Build a core-service game URL from query parameters.
def build_game_url(query: dict[str, list[str]]) -> str:
raw_url = _first(query, "url")
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(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_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")
base = _validate_http_url(base_url).rstrip("/")
safe_game_id = quote(game_id.strip("/"), safe="")
return f"{base}/games/{safe_game_id}"
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):
@@ -35,97 +64,152 @@ class ReplayRequestHandler(BaseHTTPRequestHandler):
def do_GET(self) -> None:
parsed = urlparse(self.path)
if parsed.path == "/api/health":
self._json({"ok": True, "service": "texas-holdem-replay"})
if parsed.path == "/health":
self._json({"ok": True})
return
if parsed.path == "/api/fetch-game":
self._handle_fetch_game(parsed.query)
self._handle_fetch_game(parse_qs(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 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, raw_query: str) -> None:
query = parse_qs(raw_query)
def _handle_fetch_game(self, params: dict[str, list[str]]) -> None:
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)
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 _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:
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 candidate.is_file():
candidate = STATIC_DIR / "index.html"
body = candidate.read_bytes()
content_type = mimetypes.guess_type(candidate.name)[0] or "application/octet-stream"
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._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 _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._cors_headers()
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Type", "application/json")
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 = 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)
print(f"Texas Hold X replay viewer listening on http://{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:
@@ -134,20 +218,5 @@ def main() -> None:
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()
File diff suppressed because it is too large Load Diff
+117 -105
View File
@@ -1,132 +1,144 @@
<!doctype html>
<html lang="zh-CN">
<html lang="en">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Texas Hold X Replay</title>
<link rel="stylesheet" href="/styles.css" />
<link rel="stylesheet" href="/styles.css">
</head>
<body>
<div class="app-shell">
<header class="topbar">
<div class="brand-lockup">
<div class="chip-mark" aria-hidden="true">TX</div>
<div class="brand-meta">
<h1>Texas Hold X Replay</h1>
<p id="subtitle">等待加载游戏数据</p>
</div>
<main class="shell">
<header class="topbar" aria-label="Application header">
<div>
<p class="eyebrow">Texas Hold X</p>
<h1>Game Replay Control</h1>
</div>
<div class="status-strip" aria-live="polite">
<span id="sourceBadge" class="badge badge-gold">No Data</span>
<span id="pollBadge" class="badge badge-blue">Auto Off</span>
<span id="connectionStatus" class="pill neutral">Idle</span>
<span id="gameStatus" class="pill">No game</span>
<span id="handCounter" class="metric">Hand 0</span>
</div>
</header>
<main class="layout-grid">
<!-- Stage zone: pure visualization (table, seats, animations).
Placed first in DOM so mobile/tablet layouts keep it on top. -->
<section class="stage-zone" aria-label="牌桌动画回放">
<div class="stage-head">
<div class="stage-head-left">
<span id="handBadge" class="badge badge-gold">Hand -</span>
<strong id="streetLabel">未加载</strong>
<section class="workspace">
<aside class="panel controls" aria-label="Game controls">
<div class="panel-header">
<h2>Source</h2>
</div>
<label class="field">
<span>Core service</span>
<input id="baseUrlInput" type="url" value="http://127.0.0.1:8000" spellcheck="false">
</label>
<label class="field">
<span>Game ID</span>
<input id="gameIdInput" type="text" value="demo" spellcheck="false">
</label>
<div class="button-row">
<button id="fetchButton" type="button" class="primary">Fetch</button>
<button id="togglePollButton" type="button">Auto</button>
</div>
<div class="divider"></div>
<div class="panel-header compact">
<h2>Run Hands</h2>
</div>
<div class="form-grid">
<label class="field">
<span>Count</span>
<input id="runCountInput" type="number" min="1" max="100" value="1">
</label>
<label class="field">
<span>Poll sec</span>
<input id="pollSecondsInput" type="number" min="1" max="60" value="3">
</label>
</div>
<div class="form-grid">
<label class="field">
<span>Small blind</span>
<input id="smallBlindInput" type="number" min="1" placeholder="keep">
</label>
<label class="field">
<span>Big blind</span>
<input id="bigBlindInput" type="number" min="1" placeholder="keep">
</label>
</div>
<label class="check-field">
<input id="untilOneLeftInput" type="checkbox">
<span>Run until one player remains</span>
</label>
<button id="runButton" type="button" class="wide">Run</button>
<div class="divider"></div>
<div class="panel-header compact">
<h2>Create Game</h2>
</div>
<textarea id="createGameInput" rows="10" spellcheck="false"></textarea>
<button id="createGameButton" type="button" class="wide">Create</button>
<div class="divider"></div>
<div class="panel-header compact">
<h2>Load JSON</h2>
</div>
<input id="fileInput" class="file-input" type="file" accept="application/json,.json">
<textarea id="jsonInput" rows="8" spellcheck="false" placeholder='Paste a GET /games/{id} response'></textarea>
<button id="loadJsonButton" type="button" class="wide">Load Snapshot</button>
</aside>
<section class="table-zone" aria-label="Poker table replay">
<div class="table-toolbar">
<div class="select-wrap">
<label for="handSelect">Hand</label>
<select id="handSelect"></select>
</div>
<div class="stage-head-right">
<span id="potLabel" class="badge badge-gold">Pot 0</span>
<div class="select-wrap">
<label for="speedInput">Pace</label>
<input id="speedInput" type="range" min="0.5" max="2" step="0.1" value="1">
</div>
</div>
<div id="table" class="poker-table">
<!-- felt-shell encapsulates the rounded green felt with overflow:hidden,
so player speech bubbles drawn in seat-layer can overflow freely
above and below the table without being clipped. -->
<div class="felt-shell" aria-hidden="true">
<div class="felt-rail"></div>
<div class="felt-surface">
<div class="felt-grid"></div>
<div class="felt-glow"></div>
<div class="felt-mark">TX</div>
<div class="felt-stage">
<div class="table-felt" id="tableFelt">
<div id="seatLayer" class="seat-layer"></div>
<div class="board-zone">
<div id="potDisplay" class="pot-display">Pot 0</div>
<div id="boardCards" class="cards board-cards"></div>
<div id="frameCaption" class="frame-caption">Load a game snapshot</div>
</div>
</div>
<div class="community-area">
<div id="boardCards" class="card-row board"></div>
<div id="tableMessage" class="table-message">上传 JSON 或从游戏服务获取快照</div>
</div>
<div id="seatLayer" class="seat-layer"></div>
</div>
<div class="transport" aria-label="Replay transport">
<button id="resetButton" type="button" title="Reset">|&lt;</button>
<button id="prevButton" type="button" title="Previous">&lt;</button>
<button id="playButton" type="button" class="primary">Play</button>
<button id="nextButton" type="button" title="Next">&gt;</button>
<span id="frameCounter" class="metric">0 / 0</span>
</div>
</section>
<!-- Interaction zone: data source + replay controls + summary. -->
<section class="control-panel" aria-label="数据与播放控制">
<div class="panel-section">
<h2>数据源</h2>
<label>
<span>游戏服务</span>
<input id="serverUrl" type="url" value="http://127.0.0.1:8000" placeholder="http://127.0.0.1:8000" />
</label>
<label>
<span>Game ID</span>
<input id="gameId" type="text" value="game1" placeholder="game1" />
</label>
<div class="button-row">
<button id="fetchBtn" class="primary-btn" type="button">获取</button>
<label class="file-btn">
上传 JSON
<input id="fileInput" type="file" accept="application/json,.json" />
</label>
</div>
<div class="auto-grid">
<label class="toggle-line">
<input id="autoPoll" type="checkbox" />
<span>自动获取</span>
</label>
<label>
<span>间隔秒</span>
<input id="pollSeconds" type="number" min="5" max="300" value="12" />
</label>
</div>
<aside class="panel log-panel" aria-label="Game details">
<div class="panel-header">
<h2>Table State</h2>
</div>
<div id="tableStats" class="stats-grid"></div>
<div class="panel-section">
<h2>回放</h2>
<label>
<span>手牌</span>
<select id="handSelect"></select>
</label>
<label>
<span>节奏</span>
<input id="pace" type="range" min="0.75" max="1.8" step="0.05" value="1" />
</label>
<div class="transport-row">
<button id="prevBtn" type="button" title="上一帧"></button>
<button id="playBtn" class="primary-btn" type="button" title="播放/暂停"></button>
<button id="nextBtn" type="button" title="下一帧"></button>
<button id="resetBtn" type="button" title="重置"></button>
</div>
<div class="progress-wrap">
<div id="progressBar"></div>
</div>
<div class="panel-header compact">
<h2>Players</h2>
</div>
<div id="playerList" class="player-list"></div>
<div class="panel-section dense">
<h2>牌局摘要</h2>
<dl class="stat-list">
<div><dt>状态</dt><dd id="gameStatus">-</dd></div>
<div><dt>玩家</dt><dd id="playerCount">-</dd></div>
<div><dt>总手数</dt><dd id="handCount">-</dd></div>
<div><dt>盲注</dt><dd id="blindLevel">-</dd></div>
</dl>
</div>
</section>
<aside class="event-panel" aria-label="事件日志">
<div class="panel-section">
<h2>事件</h2>
<ol id="eventLog" class="event-log"></ol>
<div class="panel-header compact">
<h2>Timeline</h2>
</div>
<ol id="eventLog" class="event-log"></ol>
</aside>
</main>
</div>
</section>
</main>
<script src="/app.js" type="module"></script>
<script src="/app.js"></script>
</body>
</html>
File diff suppressed because it is too large Load Diff