docker-dosbox-novnc/scripts/setup_server.py

#!/usr/bin/env python3
"""HTML setup routine for DOSBox games.

Nothing about which games exist is hardcoded here: available games are
discovered by listing *.json manifests on the SMB share, one per game,
each describing how to install/uninstall/start/stop that game. Drop a new
<name>.json (and matching zip) on the share and it shows up here with no
image changes needed.

Manifest schema (all fields required):
{
  "name": "stuntcar",                 # must match the manifest's own filename
  "title": "Stuntcar Racer",
  "zip_path": "Games/dosbox/stuntcar.zip",   # path to the zip, relative to the share root -
                                              # games can live anywhere on the share, not just
                                              # next to their manifest
  "start_cmd": ["dosbox", "run.bat", "-conf", "dosbox-0.74-3.conf"]
}
Uninstall is implicit (remove GAMES_HOME/<name>); stop is implicit (terminate
the process tracked from the last start_cmd). Manifests themselves always live
in SMB_DIR (the catalog directory), even though the zip_path they point at may
not.
"""

import html
import json
import os
import re
import subprocess
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse

SMB_SERVER = "vlda-01"
SMB_SHARE = "software"
SMB_DIR = r"Games\dosbox"
GAMES_HOME = os.environ["GAMES_HOME"]
SETUP_PORT = int(os.environ.get("SETUP_PORT", "8080"))

state_lock = threading.RLock()
install_state = {}   # name -> "downloading" | "extracting" | "done" | "error: ..."
running_procs = {}   # name -> subprocess.Popen


def smb_run(*commands):
    result = subprocess.run(
        ["smbclient", "-N", f"//{SMB_SERVER}/{SMB_SHARE}", "-c", ";".join(commands)],
        capture_output=True, text=True, timeout=15,
    )
    return result.stdout


def discover_games():
    """Returns {name: manifest} for every *.json manifest found on the share."""
    listing = smb_run(f"ls {SMB_DIR}\\*.json")
    names = [m.group(1) for m in re.finditer(r"\s*(\S+)\.json\s+A\s+\d+", listing)]
    if not names:
        return {}

    tmpdir = "/tmp/manifests"
    os.makedirs(tmpdir, exist_ok=True)
    gets = [f"get {SMB_DIR}\\{name}.json {tmpdir}/{name}.json" for name in names]
    smb_run(*gets)

    games = {}
    for name in names:
        try:
            with open(f"{tmpdir}/{name}.json") as f:
                manifest = json.load(f)
            if manifest.get("name") == name:
                games[name] = manifest
        except (OSError, json.JSONDecodeError):
            continue
    return games


def is_installed(name):
    path = os.path.join(GAMES_HOME, name)
    return os.path.isdir(path) and os.listdir(path)


def is_running(name):
    with state_lock:
        proc = running_procs.get(name)
        if proc is None:
            return False
        if proc.poll() is None:
            return True
        del running_procs[name]
        return False


def install_game(name, manifest):
    with state_lock:
        if install_state.get(name) in ("downloading", "extracting"):
            return
        install_state[name] = "downloading"

    tmpdir = f"/tmp/setup-{name}"
    try:
        os.makedirs(tmpdir, exist_ok=True)
        zip_name = manifest["zip_path"].rsplit("/", 1)[-1]
        zip_path = os.path.join(tmpdir, zip_name)
        # smbget's -a (guest) can't be combined with -o (output file), so let it save
        # under its default name in tmpdir instead.
        subprocess.run(
            ["smbget", "-au", f"smb://{SMB_SERVER}/{SMB_SHARE}/{manifest['zip_path']}"],
            check=True, timeout=1800, cwd=tmpdir,
        )
        with state_lock:
            install_state[name] = "extracting"
        os.makedirs(GAMES_HOME, exist_ok=True)
        subprocess.run(["unzip", "-o", "-d", GAMES_HOME, zip_path], check=True, timeout=600)
        with state_lock:
            install_state[name] = "done"
    except Exception as exc:
        with state_lock:
            install_state[name] = f"error: {exc}"
    finally:
        subprocess.run(["rm", "-rf", tmpdir], check=False)


def uninstall_game(name):
    with state_lock:
        if is_running(name):
            return
        subprocess.run(["rm", "-rf", os.path.join(GAMES_HOME, name)], check=False)
        install_state.pop(name, None)


def start_game(name, manifest):
    with state_lock:
        if is_running(name):
            return
        proc = subprocess.Popen(manifest["start_cmd"], cwd=os.path.join(GAMES_HOME, name))
        running_procs[name] = proc


def stop_game(name):
    with state_lock:
        proc = running_procs.get(name)
        if proc is None:
            return
        proc.terminate()
    try:
        proc.wait(timeout=5)
    except subprocess.TimeoutExpired:
        proc.kill()
        proc.wait()
    with state_lock:
        running_procs.pop(name, None)


def render_page():
    games = discover_games()
    with state_lock:
        pending = dict(install_state)

    any_pending = any(v in ("downloading", "extracting") for v in pending.values())
    refresh_tag = '<meta http-equiv="refresh" content="3">' if any_pending else ""

    def button(action, name, label):
        return (
            f'<form method="post" action="/{action}">'
            f'<input type="hidden" name="name" value="{html.escape(name)}">'
            f'<button type="submit">{label}</button></form>'
        )

    rows = []
    for name, manifest in sorted(games.items()):
        title = html.escape(manifest.get("title", name))
        status = pending.get(name)
        if is_running(name):
            status_html = '<span class="status busy">Running</span>'
            actions = button("stop", name, "Stop")
        elif status in ("downloading", "extracting"):
            status_html = f'<span class="status busy">{status}...</span>'
            actions = ""
        elif status and status.startswith("error"):
            status_html = f'<span class="status err">{html.escape(status)}</span>'
            actions = button("install", name, "Retry")
        elif is_installed(name):
            status_html = '<span class="status ok">Installed</span>'
            actions = button("start", name, "Start") + button("uninstall", name, "Uninstall")
        else:
            status_html = "Not installed"
            actions = button("install", name, "Install")
        rows.append(f"<tr><td>{title}</td><td>{status_html}</td><td>{actions}</td></tr>")

    return f"""<!doctype html>
<html>
<head>
<title>dosbox-novnc game setup</title>
{refresh_tag}
<style>
body {{ font-family: sans-serif; margin: 2em; }}
table {{ border-collapse: collapse; width: 100%; max-width: 640px; }}
th, td {{ text-align: left; padding: 0.5em 1em; border-bottom: 1px solid #ccc; }}
form {{ margin: 0; display: inline; }}
button {{ padding: 0.3em 0.8em; margin-right: 0.3em; }}
.status.ok {{ color: #2a2; font-weight: bold; }}
.status.busy {{ color: #a70; font-weight: bold; }}
.status.err {{ color: #c22; }}
</style>
</head>
<body>
<h1>DOSBox games</h1>
<table>
<tr><th>Game</th><th>Status</th><th>Actions</th></tr>
{''.join(rows) if rows else '<tr><td colspan="3">No game manifests found on the share</td></tr>'}
</table>
</body>
</html>"""


class Handler(BaseHTTPRequestHandler):
    def log_message(self, fmt, *args):
        pass

    def _send_html(self, body, status=200):
        encoded = body.encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(encoded)))
        self.end_headers()
        self.wfile.write(encoded)

    def do_GET(self):
        if urlparse(self.path).path == "/":
            self._send_html(render_page())
        else:
            self._send_html("Not found", status=404)

    def do_POST(self):
        action = urlparse(self.path).path.lstrip("/")
        if action not in ("install", "uninstall", "start", "stop"):
            self._send_html("Not found", status=404)
            return

        length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(length).decode("utf-8")
        name = parse_qs(body).get("name", [""])[0]
        games = discover_games()

        if name in games:
            manifest = games[name]
            if action == "install":
                threading.Thread(target=install_game, args=(name, manifest), daemon=True).start()
            elif action == "uninstall":
                threading.Thread(target=uninstall_game, args=(name,), daemon=True).start()
            elif action == "start":
                threading.Thread(target=start_game, args=(name, manifest), daemon=True).start()
            elif action == "stop":
                threading.Thread(target=stop_game, args=(name,), daemon=True).start()

        self.send_response(303)
        self.send_header("Location", "/")
        self.end_headers()


if __name__ == "__main__":
    ThreadingHTTPServer(("0.0.0.0", SETUP_PORT), Handler).serve_forever()