OBS-Overlays/gauntlet/server.py

#!/usr/bin/env python3
"""Tiny local server for the Gauntlet OBS overlay.

Serves overlay.html/editor.html and adds local control endpoints:
POST /api/gauntlet -> validates/writes gauntlet.json
POST /api/advance  -> mark active game cleared, activate next queued game
POST /api/reset    -> first game active, remaining games queued
"""
from __future__ import annotations

import argparse
import json
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any

VALID_STATUSES = {"active", "queued", "cleared", "paused", "failed"}
MAX_GAMES = 10


def clean_text(value: Any, max_len: int = 120) -> str:
    return str(value or "").strip()[:max_len]


def normalize_game(game: Any, fallback_status: str = "queued") -> dict[str, str] | None:
    if not isinstance(game, dict):
        return None
    title = clean_text(game.get("title"), 80)
    subtitle = clean_text(game.get("subtitle"), 120)
    icon = clean_text(game.get("icon"), 8)
    status = clean_text(game.get("status"), 20).lower()
    if status not in VALID_STATUSES:
        status = fallback_status
    if not title:
        return None
    return {"title": title, "subtitle": subtitle, "icon": icon or "◆", "status": status}


def normalize_gauntlet(data: Any) -> dict[str, list[dict[str, str]]]:
    games = data.get("games") if isinstance(data, dict) else data
    if not isinstance(games, list):
        raise ValueError("Expected JSON object with a games array, or a games array.")

    normalized: list[dict[str, str]] = []
    for item in games[:MAX_GAMES]:
        game = normalize_game(item)
        if game:
            normalized.append(game)

    active_seen = False
    for game in normalized:
        if game["status"] == "active":
            if active_seen:
                game["status"] = "queued"
            else:
                active_seen = True

    if normalized and not active_seen:
        first_not_cleared = next((g for g in normalized if g["status"] != "cleared"), normalized[0])
        first_not_cleared["status"] = "active"

    return {"games": normalized}


def load_gauntlet(directory: str) -> dict[str, list[dict[str, str]]]:
    path = Path(directory).resolve() / "gauntlet.json"
    if not path.exists():
        return {"games": []}
    return normalize_gauntlet(json.loads(path.read_text(encoding="utf-8")))


def save_gauntlet(directory: str, data: dict[str, list[dict[str, str]]]) -> None:
    path = Path(directory).resolve() / "gauntlet.json"
    path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")


def advance_gauntlet(data: dict[str, list[dict[str, str]]]) -> dict[str, list[dict[str, str]]]:
    games = data.get("games", [])
    active_index = next((i for i, g in enumerate(games) if g.get("status") == "active"), None)
    if active_index is None:
        if games:
            games[0]["status"] = "active"
        return {"games": games}

    games[active_index]["status"] = "cleared"
    for i in range(active_index + 1, len(games)):
        if games[i].get("status") != "cleared":
            games[i]["status"] = "active"
            break
    return {"games": games}


def reset_gauntlet(data: dict[str, list[dict[str, str]]]) -> dict[str, list[dict[str, str]]]:
    games = data.get("games", [])
    for index, game in enumerate(games):
        game["status"] = "active" if index == 0 else "queued"
    return {"games": games}


class GauntletHandler(SimpleHTTPRequestHandler):
    server_version = "GauntletOverlay/1.0"

    def do_POST(self) -> None:  # noqa: N802
        route = self.path.split("?", 1)[0]
        try:
            if route == "/api/gauntlet":
                length = int(self.headers.get("Content-Length", "0"))
                raw = self.rfile.read(length).decode("utf-8")
                data = normalize_gauntlet(json.loads(raw))
                save_gauntlet(self.directory, data)
                self.send_json({"ok": True, "message": "gauntlet.json saved", **data})
                return
            if route == "/api/advance":
                data = advance_gauntlet(load_gauntlet(self.directory))
                save_gauntlet(self.directory, data)
                self.send_json({"ok": True, "message": "advanced to next game", **data})
                return
            if route == "/api/reset":
                data = reset_gauntlet(load_gauntlet(self.directory))
                save_gauntlet(self.directory, data)
                self.send_json({"ok": True, "message": "gauntlet reset", **data})
                return
        except Exception as exc:
            self.send_json({"ok": False, "error": str(exc)}, status=400)
            return
        self.send_json({"ok": False, "error": "Not found"}, status=404)

    def send_json(self, payload: dict[str, Any], status: int = 200) -> None:
        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)


def main() -> None:
    parser = argparse.ArgumentParser(description="Run the Gauntlet OBS overlay server.")
    parser.add_argument("--host", default="127.0.0.1", help="Host to bind. Default: 127.0.0.1")
    parser.add_argument("--port", type=int, default=8770, help="Port to bind. Default: 8770")
    parser.add_argument("--directory", default=str(Path(__file__).resolve().parent), help="Overlay folder to serve.")
    args = parser.parse_args()
    handler = lambda *handler_args, **kwargs: GauntletHandler(*handler_args, directory=args.directory, **kwargs)  # noqa: E731
    with ThreadingHTTPServer((args.host, args.port), handler) as httpd:
        print(f"Gauntlet Overlay server running at http://{args.host}:{args.port}/")
        print(f"Editor:  http://{args.host}:{args.port}/editor.html")
        print(f"Overlay: http://{args.host}:{args.port}/overlay.html")
        print("Controls: run advance-game.bat to clear the current game and activate the next one.")
        print("Keep this window open while streaming. Press Ctrl+C to stop.")
        httpd.serve_forever()


if __name__ == "__main__":
    main()