carlitosbondandClaude Sonnet 4.6
3539b1f75f
- Updated Dockerfile to Python 3.11, selective COPY for cleaner image - Expanded .gitignore with Python/OS patterns - Added config.json, questions.json, env_example, and static assets - Updated templates (host, player, scores) with Space Invaders UI - Rewrote README for SI Quiz branding and setup docs Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
393 lines
16 KiB
Python
393 lines
16 KiB
Python
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) |