src/chess_tutor/web/server.py

"""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()