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, asyncio from datetime import datetime DB_PATH = os.environ.get("DB_PATH", "/data/quiz.db") # ── 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 ); """) # Safe migration: add dept column if it doesn't exist yet try: conn.execute("ALTER TABLE scores ADD COLUMN dept TEXT NOT NULL DEFAULT ''") except Exception: pass # column already exists 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") # ── QUESTIONS ───────────────────────────────────────────────────────────────── QUESTIONS = [ { "q": "¿Cuál es el enfoque principal del equipo de Network Intelligence?", "options": [ "Vender licencias de software a otras empresas", "Construir herramientas propias que empoderan a los equipos operacionales", "Administrar el presupuesto de TI de LCPR", "Instalar equipos físicos en planta externa" ], "correct": 1, "fun_fact": "¡Exacto! Construimos para quienes operan la red día a día — no solo dashboards bonitos." }, { "q": "¿Qué problema resolvió el equipo al reemplazar Power BI con frontend propio?", "options": [ "La pantalla se veía más bonita", "Eliminamos la dependencia de licencias costosas y ganamos control total", "Fue más fácil de instalar en los servidores", "Los colores del dashboard coincidían con los de Liberty" ], "correct": 1, "fun_fact": "$0 en licencias de visualización. El código es nuestro y lo extendemos como queremos." }, { "q": "¿Qué es Pathfinder?", "options": [ "Una aplicación para rastrear camiones de campo", "Un sistema de GPS para la red de fibra", "Una plataforma web que centraliza herramientas operacionales de red en un solo lugar", "Un reporte mensual de disponibilidad de red" ], "correct": 2, "fun_fact": "Pathfinder es el 'panel de control' propio de LCPR — construido exactamente para nuestros flujos de trabajo." }, { "q": "¿Qué logró el proyecto de Firmware Upgrade Automation?", "options": [ "Actualizar manualmente 10 modems por semana", "Contratar más técnicos para visitas a domicilio", "Automatizar la actualización masiva de modems sin intervención humana", "Comprar equipos nuevos para reemplazar los modems viejos" ], "correct": 2, "fun_fact": "400+ modems actualizados en batches automáticos. Antes era trabajo manual uno por uno." }, { "q": "¿Hacia dónde se dirige el equipo en su visión AIOps?", "options": [ "Depender más de herramientas comerciales como ServiceNow", "Detectar y predecir fallas antes de que el cliente las reporte", "Reducir el equipo de ingeniería a la mitad", "Migrar toda la red a la nube pública" ], "correct": 1, "fun_fact": "La meta: que el técnico salga a campo con el diagnóstico ya hecho — no a diagnosticar." } ] # ── 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 = list(range(len(QUESTIONS))) random.shuffle(self.question_order) game = GameState() # ── CONNECTION MANAGER ──────────────────────────────────────────────────────── class ConnectionManager: def __init__(self): self.players: dict = {} self.hosts: list = [] self._reset_task: asyncio.Task = None 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) # Cancel pending auto-reset if host reconnected in time if self._reset_task and not self._reset_task.done(): self._reset_task.cancel() self._reset_task = None 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) # If no hosts remain and game is not in lobby, schedule auto-reset if not self.hosts and game.phase != "lobby": self._reset_task = asyncio.create_task(self._auto_reset(delay=300)) async def _auto_reset(self, delay: int): """Wait `delay` seconds, then reset the game if no host has reconnected.""" await asyncio.sleep(delay) if not self.hosts: game.reset() await self.broadcast_players({"type": "phase", "phase": "lobby"}) 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", {"request": request}) @app.get("/host", response_class=HTMLResponse) async def host_page(request: Request): return templates.TemplateResponse("host.html", {"request": request}) @app.get("/scores", response_class=HTMLResponse) async def scores_page(request: Request): return templates.TemplateResponse("scores.html", {"request": 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 TIME_LIMIT = 20 if elapsed > TIME_LIMIT: continue player["answered"] = True chosen = msg["choice"] q_idx = game.question_order[game.current_q] correct = 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) # preserve joined players game.reset() game.players = existing_players # restore them # Reset scores and state for a fresh game 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(QUESTIONS): lb = get_leaderboard() save_session(list(game.players.values())) # ← persist to SQLite 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 = QUESTIONS[q_idx] await mgr.broadcast_all({ "type": "question", "number": game.current_q + 1, "total": len(QUESTIONS), "q": q["q"], "options": q["options"], "time_limit": 20, "correct_idx_hint": -1 # not revealed until show_results }) elif msg["type"] == "show_results": game.phase = "results" q_idx = game.question_order[game.current_q] q = QUESTIONS[q_idx] await mgr.broadcast_all({ "type": "results", "correct_idx": q["correct"], "fun_fact": q["fun_fact"], "leaderboard": get_leaderboard() }) elif msg["type"] == "reset": game.reset() await mgr.broadcast_all({"type": "phase", "phase": "lobby"}) except WebSocketDisconnect: mgr.disconnect_host(ws)