#!/usr/bin/env python3
"""HTML setup routine: lets the user browse games available on the SMB
share and install them into GAMES_HOME on demand, instead of the image
auto-downloading everything at container startup."""
import html
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.Lock()
install_state = {} # name -> "downloading" | "extracting" | "done" | "error: ..."
def list_remote_games():
result = subprocess.run(
["smbclient", "-N", f"//{SMB_SERVER}/{SMB_SHARE}", "-c", f"ls {SMB_DIR}\\*.zip"],
capture_output=True, text=True, timeout=15,
)
games = []
for line in result.stdout.splitlines():
m = re.match(r"\s*(\S+)\.zip\s+A\s+(\d+)", line)
if m:
games.append((m.group(1), int(m.group(2))))
return sorted(games)
def list_installed_games():
if not os.path.isdir(GAMES_HOME):
return set()
return {
name for name in os.listdir(GAMES_HOME)
if os.path.isdir(os.path.join(GAMES_HOME, name)) and os.listdir(os.path.join(GAMES_HOME, name))
}
def install_game(name):
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_path = os.path.join(tmpdir, f"{name}.zip")
# smbget's -a (guest) can't be combined with -o (output file), so let it save
# under the default name in tmpdir instead, same as the original build-time fetch.
subprocess.run(
["smbget", "-au", f"smb://{SMB_SERVER}/{SMB_SHARE}/Games/dosbox/{name}.zip"],
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 render_page():
remote = list_remote_games()
installed = list_installed_games()
with state_lock:
pending = dict(install_state)
any_pending = any(v in ("downloading", "extracting") for v in pending.values())
refresh_tag = '' if any_pending else ""
rows = []
for name, size in remote:
status = pending.get(name)
if name in installed and status not in ("downloading", "extracting"):
action = 'Installed'
elif status in ("downloading", "extracting"):
action = f'{status}...'
elif status and status.startswith("error"):
action = (
f'{html.escape(status)}'
f''
)
else:
action = (
f''
)
rows.append(
f"| {html.escape(name)} | {size / 1_000_000:.1f} MB | {action} |
"
)
return f"""
dosbox-novnc game setup
{refresh_tag}
DOSBox games
| Game | Size | Status / Action |
|---|
{''.join(rows) if rows else '| No games found on the share |
'}
"""
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):
if urlparse(self.path).path != "/install":
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]
valid_names = {n for n, _ in list_remote_games()}
if name in valid_names:
threading.Thread(target=install_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()