from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from fastapi.responses import HTMLResponse, JSONResponse import json, time, random, sqlite3, os from datetime import datetime # ── CONFIG ──────────────────────────────────────────────────────────────────── CONFIG_PATH = os.environ.get("CONFIG_PATH", "config.json") QUESTIONS_PATH = os.environ.get("QUESTIONS_PATH", "questions.json") DB_PATH = os.environ.get("DB_PATH", "/data/si-quiz.db") with open(CONFIG_PATH) as f: CONFIG = json.load(f) APP_NAME = CONFIG.get("app_name", "SI Quiz") APP_SUB = CONFIG.get("app_subtitle", "Space Invaders Quiz") ORG_NAME = CONFIG.get("org_name", "") HOST_PASSWORD = CONFIG.get("host_password", "changeme") DEPARTMENTS = CONFIG.get("departments", ["Engineering", "Operations", "IT", "Other"]) TIME_LIMIT = int(CONFIG.get("time_limit_seconds", 30)) Q_SELECTION = CONFIG.get("question_selection", {"mode": "all"}) with open(QUESTIONS_PATH) as f: ALL_QUESTIONS = json.load(f) def pick_questions() -> list[int]: """ Returns a list of question indices for a session. config.json modes: "mode": "all" → use every question, shuffled "mode": "sample" → pick N random questions (requires "count") "mode": "grouped" → pick N from each tagged group (requires "groups") Each group: {"tag": "general", "pick": 4} Questions without a matching tag are ignored for that group. """ mode = Q_SELECTION.get("mode", "all") if mode == "all": indices = list(range(len(ALL_QUESTIONS))) random.shuffle(indices) return indices if mode == "sample": count = int(Q_SELECTION.get("count", len(ALL_QUESTIONS))) indices = list(range(len(ALL_QUESTIONS))) return random.sample(indices, min(count, len(indices))) if mode == "grouped": selected = [] for group in Q_SELECTION.get("groups", []): tag = group.get("tag") pick = int(group.get("pick", 1)) pool = [i for i, q in enumerate(ALL_QUESTIONS) if q.get("group") == tag] selected += random.sample(pool, min(pick, len(pool))) random.shuffle(selected) return selected # fallback indices = list(range(len(ALL_QUESTIONS))) random.shuffle(indices) return indices # ── DATABASE ────────────────────────────────────────────────────────────────── def get_db(): os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def init_db(): with get_db() as conn: conn.executescript(""" CREATE TABLE IF NOT EXISTS sessions ( id INTEGER PRIMARY KEY AUTOINCREMENT, started_at TEXT NOT NULL, ended_at TEXT, total_players INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS scores ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id INTEGER NOT NULL REFERENCES sessions(id), player TEXT NOT NULL, score INTEGER NOT NULL DEFAULT 0, rank INTEGER, dept TEXT NOT NULL DEFAULT '', finished_at TEXT NOT NULL ); """) try: conn.execute("ALTER TABLE scores ADD COLUMN dept TEXT NOT NULL DEFAULT ''") except Exception: pass init_db() def save_session(players: list) -> int: now = datetime.utcnow().isoformat() with get_db() as conn: cur = conn.execute( "INSERT INTO sessions (started_at, ended_at, total_players) VALUES (?,?,?)", (now, now, len(players)) ) session_id = cur.lastrowid ranked = sorted(players, key=lambda p: p["score"], reverse=True) for rank, p in enumerate(ranked, 1): conn.execute( "INSERT INTO scores (session_id, player, score, rank, dept, finished_at) VALUES (?,?,?,?,?,?)", (session_id, p.get("display", p["name"]), p["score"], rank, p.get("dept", ""), now) ) return session_id # ── APP ─────────────────────────────────────────────────────────────────────── app = FastAPI() app.mount("/static", StaticFiles(directory="static"), name="static") templates = Jinja2Templates(directory="templates") def base_ctx(request: Request) -> dict: return { "request": request, "app_name": APP_NAME, "app_sub": APP_SUB, "org_name": ORG_NAME, "host_password": HOST_PASSWORD, "departments": DEPARTMENTS, } # ── GAME STATE ──────────────────────────────────────────────────────────────── class GameState: def __init__(self): self.reset() def reset(self): self.phase = "lobby" self.players: dict = {} self.current_q = -1 self.q_start_time = 0 self.answers_this_round = [] self.question_order = pick_questions() game = GameState() # ── CONNECTION MANAGER ──────────────────────────────────────────────────────── class ConnectionManager: def __init__(self): self.players: dict = {} self.hosts: list = [] async def connect_player(self, ws_id: str, ws: WebSocket): await ws.accept() self.players[ws_id] = ws async def connect_host(self, ws: WebSocket): await ws.accept() self.hosts.append(ws) def disconnect(self, ws_id: str): self.players.pop(ws_id, None) game.players.pop(ws_id, None) def disconnect_host(self, ws: WebSocket): if ws in self.hosts: self.hosts.remove(ws) async def send_to(self, ws_id: str, data: dict): ws = self.players.get(ws_id) if ws: try: await ws.send_json(data) except Exception: pass async def broadcast_players(self, data: dict): dead = [] for ws_id, ws in list(self.players.items()): try: await ws.send_json(data) except Exception: dead.append(ws_id) for d in dead: self.disconnect(d) async def broadcast_hosts(self, data: dict): dead = [] for ws in list(self.hosts): try: await ws.send_json(data) except Exception: dead.append(ws) for d in dead: self.disconnect_host(d) async def broadcast_all(self, data: dict): await self.broadcast_players(data) await self.broadcast_hosts(data) mgr = ConnectionManager() def get_leaderboard(): ranked = sorted(game.players.values(), key=lambda p: p["score"], reverse=True) return [{"name": p.get("display", p["name"]), "score": p["score"], "dept": p.get("dept", "")} for p in ranked] # ── ROUTES ──────────────────────────────────────────────────────────────────── @app.get("/", response_class=HTMLResponse) async def player_page(request: Request): return templates.TemplateResponse("player.html", base_ctx(request)) @app.get("/host", response_class=HTMLResponse) async def host_page(request: Request): return templates.TemplateResponse("host.html", base_ctx(request)) @app.get("/scores", response_class=HTMLResponse) async def scores_page(request: Request): return templates.TemplateResponse("scores.html", base_ctx(request)) @app.delete("/api/scores/clear") async def clear_scores(): with get_db() as conn: conn.execute("DELETE FROM scores") conn.execute("DELETE FROM sessions") return {"ok": True} @app.get("/api/scores") async def api_scores(): with get_db() as conn: sessions = conn.execute( "SELECT * FROM sessions ORDER BY started_at DESC LIMIT 50" ).fetchall() result = [] for s in sessions: players = conn.execute( "SELECT player, score, rank, dept FROM scores WHERE session_id=? ORDER BY rank", (s["id"],) ).fetchall() result.append({ "id": s["id"], "started_at": s["started_at"], "ended_at": s["ended_at"], "total_players": s["total_players"], "players": [{"player": p["player"], "score": p["score"], "rank": p["rank"], "dept": p["dept"] if "dept" in p.keys() else ""} for p in players] }) return JSONResponse(result) # ── PLAYER WEBSOCKET ────────────────────────────────────────────────────────── @app.websocket("/ws/player/{ws_id}") async def player_ws(ws: WebSocket, ws_id: str): await mgr.connect_player(ws_id, ws) try: await mgr.send_to(ws_id, {"type": "phase", "phase": game.phase}) async for raw in ws.iter_text(): msg = json.loads(raw) if msg["type"] == "join": name = msg["name"].strip()[:40] dept = msg.get("dept", "").strip()[:40] display = msg.get("display", name).strip()[:40] if not name: continue game.players[ws_id] = {"name": name, "score": 0, "answered": False, "dept": dept, "display": display} await mgr.send_to(ws_id, {"type": "joined", "name": name, "display": display, "dept": dept}) await mgr.broadcast_hosts({ "type": "lobby_update", "players": [{"name": p["name"], "display": p.get("display", p["name"]), "dept": p.get("dept", "")} for p in game.players.values()], "count": len(game.players) }) elif msg["type"] == "answer": if game.phase != "question": continue player = game.players.get(ws_id) if not player or player["answered"]: continue elapsed = time.time() - game.q_start_time if elapsed > TIME_LIMIT: continue player["answered"] = True chosen = msg["choice"] q_idx = game.question_order[game.current_q] correct = ALL_QUESTIONS[q_idx]["correct"] is_correct = chosen == correct points = 0 if is_correct: speed_bonus = max(0, (TIME_LIMIT - elapsed) / TIME_LIMIT) points = int(200 + 800 * speed_bonus) player["score"] += points game.answers_this_round.append({ "name": player["name"], "correct": is_correct, "elapsed": round(elapsed, 2) }) await mgr.send_to(ws_id, { "type": "answer_result", "correct": is_correct, "points": points, "total": player["score"], "correct_idx": correct }) answered = sum(1 for p in game.players.values() if p["answered"]) await mgr.broadcast_hosts({ "type": "answer_update", "answered": answered, "total": len(game.players), "details": game.answers_this_round }) except WebSocketDisconnect: mgr.disconnect(ws_id) await mgr.broadcast_hosts({ "type": "lobby_update", "players": [{"name": p["name"], "display": p.get("display", p["name"]), "dept": p.get("dept", "")} for p in game.players.values()], "count": len(game.players) }) # ── HOST WEBSOCKET ──────────────────────────────────────────────────────────── @app.websocket("/ws/host") async def host_ws(ws: WebSocket): await mgr.connect_host(ws) try: await ws.send_json({ "type": "lobby_update", "players": [{"name": p["name"], "display": p.get("display", p["name"]), "dept": p.get("dept", "")} for p in game.players.values()], "count": len(game.players) }) async for raw in ws.iter_text(): msg = json.loads(raw) if msg["type"] == "start_game": existing_players = dict(game.players) game.reset() game.players = existing_players for p in game.players.values(): p["score"] = 0 p["answered"] = False await mgr.broadcast_all({"type": "phase", "phase": "lobby"}) elif msg["type"] == "next_question": game.current_q += 1 if game.current_q >= len(game.question_order): lb = get_leaderboard() save_session(list(game.players.values())) game.phase = "podium" await mgr.broadcast_all({"type": "podium", "leaderboard": lb}) continue game.phase = "question" game.q_start_time = time.time() game.answers_this_round = [] for p in game.players.values(): p["answered"] = False q_idx = game.question_order[game.current_q] q = ALL_QUESTIONS[q_idx] await mgr.broadcast_all({ "type": "question", "number": game.current_q + 1, "total": len(game.question_order), "q": q["q"], "options": q["options"], "time_limit": TIME_LIMIT, "correct_idx_hint": q["correct"] }) elif msg["type"] == "show_results": game.phase = "results" q_idx = game.question_order[game.current_q] q = ALL_QUESTIONS[q_idx] await mgr.broadcast_all({ "type": "results", "correct_idx": q["correct"], "fun_fact": q["fun_fact"], "leaderboard": get_leaderboard() }) elif msg["type"] == "clear_lobby": await mgr.broadcast_players({"type": "kicked"}) game.reset() await mgr.broadcast_hosts({"type": "lobby_update", "players": [], "count": 0}) elif msg["type"] == "reset": game.reset() await mgr.broadcast_all({"type": "phase", "phase": "lobby"}) except WebSocketDisconnect: mgr.disconnect_host(ws)