"""Local HTTP server for the Chess Tutor web UI.""" from __future__ import annotations import json import mimetypes import os import traceback from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any from urllib.parse import urlparse from chess_tutor.env import load_project_dotenv from chess_tutor.inference.pipeline import run_inference from chess_tutor.inference.sites import DEFAULT_INFERENCE_GAMES_LIMIT from chess_tutor.playstyle_coaching.actionable_config import ActionableConfig from chess_tutor.playstyle_coaching.catalog import load_playstyle_catalog from chess_tutor.web.payload import build_web_payload from chess_tutor.web.sample_payload import build_sample_web_payload from chess_tutor.web.sections import ui_layout_config from chess_tutor.web.version import version_info STATIC_DIR = Path(__file__).resolve().parent / "static" API_ONLY = False MAX_ANALYZE_GAMES = 200 def _parse_analyze_body(body: dict[str, Any]) -> dict[str, Any]: username = str(body.get("username", "")).strip() if not username: raise ValueError("username is required") site = str(body.get("site", "lichess")).strip().lower() if site not in ("lichess", "chess.com", "chesscom"): raise ValueError("site must be lichess or chess.com") if site == "chesscom": site = "chess.com" raw_games = body.get("games", DEFAULT_INFERENCE_GAMES_LIMIT) try: games_limit = int(raw_games) except (TypeError, ValueError): raise ValueError("games must be an integer") from None if games_limit < 1 or games_limit > MAX_ANALYZE_GAMES: raise ValueError(f"games must be between 1 and {MAX_ANALYZE_GAMES}") lichess_token = body.get("lichess_token") or os.environ.get("LICHESS_TOKEN") if lichess_token is not None and not isinstance(lichess_token, str): raise ValueError("lichess_token must be a string") lichess_study_id = body.get("lichess_study_id") if lichess_study_id is not None and not isinstance(lichess_study_id, str): raise ValueError("lichess_study_id must be a string") return { "username": username, "site": site, "games_limit": games_limit, "time_control": str(body.get("time_control", "600+0")), "policy": str(body.get("policy", "teacher")), "corpus_id": str(body.get("corpus_id", "standard_600")), "lichess_token": lichess_token, "lichess_study_id": lichess_study_id, } def _json_response(handler: BaseHTTPRequestHandler, status: int, payload: Any) -> None: body = json.dumps(payload, ensure_ascii=False).encode("utf-8") handler.send_response(status) handler.send_header("Content-Type", "application/json; charset=utf-8") handler.send_header("Content-Length", str(len(body))) handler.end_headers() handler.wfile.write(body) def _read_json_body(handler: BaseHTTPRequestHandler) -> dict[str, Any]: length = int(handler.headers.get("Content-Length", "0")) if length <= 0: return {} raw = handler.rfile.read(length) data = json.loads(raw.decode("utf-8")) if not isinstance(data, dict): raise ValueError("JSON body must be an object") return data def run_inference_for_web( *, username: str, site: str, games_limit: int = DEFAULT_INFERENCE_GAMES_LIMIT, time_control: str = "600+0", policy: str = "teacher", corpus_id: str = "standard_600", lichess_token: str | None = None, lichess_study_id: str | None = None, ) -> dict[str, Any]: catalog = load_playstyle_catalog() cfg = ActionableConfig.from_catalog(catalog) result = run_inference( username, site, games_limit=games_limit, time_control=time_control, coaching_policy=policy, corpus_id=corpus_id, actionable_config=cfg, lichess_token=lichess_token, lichess_study_id=lichess_study_id, show_progress=False, ) return build_web_payload(result, corpus_id=corpus_id) class ChessTutorHandler(BaseHTTPRequestHandler): server_version = "ChessTutorWeb/0.1" def log_message(self, format: str, *args: Any) -> None: print(f"[web] {self.address_string()} {format % args}") def do_GET(self) -> None: path = urlparse(self.path).path if path == "/api/version": return _json_response(self, HTTPStatus.OK, version_info()) if path == "/api/ui-layout": return _json_response(self, HTTPStatus.OK, ui_layout_config()) if path == "/api/sample-report": try: payload = build_sample_web_payload() except RuntimeError as exc: _json_response( self, HTTPStatus.SERVICE_UNAVAILABLE, {"error": str(exc)}, ) return return _json_response(self, HTTPStatus.OK, payload) if API_ONLY: self.send_error(HTTPStatus.NOT_FOUND) return if path in ("/", "/index.html"): return self._serve_file(STATIC_DIR / "index.html", "text/html; charset=utf-8") if path in ("/puzzle-demo", "/puzzle-demo.html"): return self._serve_file(STATIC_DIR / "puzzle-demo.html", "text/html; charset=utf-8") if path.startswith("/assets/"): rel = path.removeprefix("/assets/") target = (STATIC_DIR / "assets" / rel).resolve() assets_root = (STATIC_DIR / "assets").resolve() if not str(target).startswith(str(assets_root)): self.send_error(HTTPStatus.FORBIDDEN) return if not target.is_file(): self.send_error(HTTPStatus.NOT_FOUND) return mime, _ = mimetypes.guess_type(str(target)) return self._serve_file(target, mime or "application/octet-stream") if path.startswith("/static/"): rel = path.removeprefix("/static/") target = (STATIC_DIR / rel).resolve() if not str(target).startswith(str(STATIC_DIR.resolve())): self.send_error(HTTPStatus.FORBIDDEN) return if not target.is_file(): self.send_error(HTTPStatus.NOT_FOUND) return mime, _ = mimetypes.guess_type(str(target)) return self._serve_file(target, mime or "application/octet-stream") self.send_error(HTTPStatus.NOT_FOUND) def do_POST(self) -> None: path = urlparse(self.path).path if path != "/api/analyze": self.send_error(HTTPStatus.NOT_FOUND) return try: body = _read_json_body(self) params = _parse_analyze_body(body) payload = run_inference_for_web(**params) _json_response(self, HTTPStatus.OK, payload) except json.JSONDecodeError: _json_response( self, HTTPStatus.BAD_REQUEST, {"error": "Invalid JSON body"}, ) except ValueError as exc: _json_response( self, HTTPStatus.BAD_REQUEST, {"error": str(exc)}, ) except RuntimeError as exc: _json_response( self, HTTPStatus.BAD_REQUEST, {"error": str(exc)}, ) except Exception as exc: traceback.print_exc() _json_response( self, HTTPStatus.INTERNAL_SERVER_ERROR, {"error": str(exc)}, ) def _serve_file(self, path: Path, content_type: str) -> None: data = path.read_bytes() self.send_response(HTTPStatus.OK) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def serve(host: str = "127.0.0.1", port: int = 8765, *, api_only: bool = False) -> None: global API_ONLY API_ONLY = api_only load_project_dotenv() server = ThreadingHTTPServer((host, port), ChessTutorHandler) if api_only: url = f"http://{host}:{port}/api/" print(f"Chess Tutor API at {url}") print("Frontend: use the Vite dev server (e.g. npm run web:dev) on port 5173.") print("Python sources under src/chess_tutor/ auto-reload in that mode.") else: url = f"http://{host}:{port}/" print(f"Chess Tutor web UI at {url}") if not (STATIC_DIR / "index.html").is_file(): print( "Warning: static build missing. " "Run: cd src/chess_tutor/web/frontend && npm run build" ) print("Press Ctrl+C to stop.") try: server.serve_forever() except KeyboardInterrupt: print("\nStopping.") finally: server.server_close()