carlitosbond/route-rush
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Update main.py

Browse Source
This commit is contained in:
carlitosbond
2026-05-13 15:50:51 +00:00
parent c7f1b5ebe8
commit 767956406f
1 changed files with 223 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+215 -230
View File
@@ -1,12 +1,11 @@
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse, JSONResponse from fastapi.responses import HTMLResponse, JSONResponse
import json, time, random, sqlite3, os import json, time, random, sqlite3, os
from datetime import datetime from datetime import datetime
DB_PATH = os.environ.get("DB_PATH", "/data/rr.db") DB_PATH = os.environ.get("DB_PATH", "/data/quiz.db")
app = FastAPI()
templates = Jinja2Templates(directory="templates")
# ── DATABASE ────────────────────────────────────────────────────────────────── # ── DATABASE ──────────────────────────────────────────────────────────────────
def get_db(): def get_db():
@@ -22,108 +21,105 @@ def init_db():
id INTEGER PRIMARY KEY AUTOINCREMENT, id INTEGER PRIMARY KEY AUTOINCREMENT,
started_at TEXT NOT NULL, started_at TEXT NOT NULL,
ended_at TEXT, ended_at TEXT,
rounds INTEGER DEFAULT 0 total_players INTEGER DEFAULT 0
); );
CREATE TABLE IF NOT EXISTS scores ( CREATE TABLE IF NOT EXISTS scores (
id INTEGER PRIMARY KEY AUTOINCREMENT, id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id INTEGER REFERENCES sessions(id), session_id INTEGER NOT NULL REFERENCES sessions(id),
player TEXT NOT NULL, player TEXT NOT NULL,
score INTEGER DEFAULT 0, score INTEGER NOT NULL DEFAULT 0,
rank INTEGER, rank INTEGER,
dept TEXT NOT NULL DEFAULT '', dept TEXT NOT NULL DEFAULT '',
finished_at TEXT NOT NULL finished_at TEXT NOT NULL
); );
""") """)
# safe migration if dept column missing from older DB # Safe migration: add dept column if it doesn't exist yet
try: try:
conn.execute("ALTER TABLE scores ADD COLUMN dept TEXT NOT NULL DEFAULT ''") conn.execute("ALTER TABLE scores ADD COLUMN dept TEXT NOT NULL DEFAULT ''")
except Exception: except Exception:
pass pass # column already exists
init_db() init_db()
# ── GRID ────────────────────────────────────────────────────────────────────── def save_session(players: list) -> int:
GRID_SIZE = 4 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
MODIFIER_TYPES = [ # ── APP ───────────────────────────────────────────────────────────────────────
"normal", "normal", "normal", "normal", "normal", "normal", app = FastAPI()
"slow", "slow", "boost", "blocked" app.mount("/static", StaticFiles(directory="static"), name="static")
] templates = Jinja2Templates(directory="templates")
def generate_grid(): # ── QUESTIONS ─────────────────────────────────────────────────────────────────
nodes = [] QUESTIONS = [
for row in range(GRID_SIZE): {
for col in range(GRID_SIZE): "q": "¿Cuál es el enfoque principal del equipo de Network Intelligence?",
nodes.append({"id": row * GRID_SIZE + col, "row": row, "col": col}) "options": [
"Vender licencias de software a otras empresas",
corners = [0, 3, 12, 15] "Construir herramientas propias que empoderan a los equipos operacionales",
start = random.choice(corners) "Administrar el presupuesto de TI de LCPR",
opposite = {0: 15, 3: 12, 12: 3, 15: 0} "Instalar equipos físicos en planta externa"
end = opposite[start] ],
"correct": 1,
for attempt in range(20): "fun_fact": "¡Exacto! Construimos para quienes operan la red día a día — no solo dashboards bonitos."
edges = [] },
for row in range(GRID_SIZE): {
for col in range(GRID_SIZE): "q": "¿Qué problema resolvió el equipo al reemplazar Power BI con frontend propio?",
nid = row * GRID_SIZE + col "options": [
if col + 1 < GRID_SIZE: "La pantalla se veía más bonita",
edges.append({"a": nid, "b": nid + 1, "modifier": random.choice(MODIFIER_TYPES)}) "Eliminamos la dependencia de licencias costosas y ganamos control total",
if row + 1 < GRID_SIZE: "Fue más fácil de instalar en los servidores",
edges.append({"a": nid, "b": nid + GRID_SIZE, "modifier": random.choice(MODIFIER_TYPES)}) "Los colores del dashboard coincidían con los de Liberty"
],
# Validate: start node must have at least one non-blocked neighbor "correct": 1,
start_edges = [e for e in edges if e["a"] == start or e["b"] == start] "fun_fact": "$0 en licencias de visualización. El código es nuestro y lo extendemos como queremos."
if any(e["modifier"] != "blocked" for e in start_edges): },
break {
# If all attempts failed (extremely unlikely), force one start edge open "q": "¿Qué es Pathfinder?",
else: "options": [
for e in edges: "Una aplicación para rastrear camiones de campo",
if e["a"] == start or e["b"] == start: "Un sistema de GPS para la red de fibra",
e["modifier"] = "normal" "Una plataforma web que centraliza herramientas operacionales de red en un solo lugar",
break "Un reporte mensual de disponibilidad de red"
],
return {"nodes": nodes, "edges": edges, "start": start, "end": end} "correct": 2,
"fun_fact": "Pathfinder es el 'panel de control' propio de LCPR — construido exactamente para nuestros flujos de trabajo."
def score_path(path: list, edges: list, elapsed: float, time_limit: float, end_node: int) -> dict: },
if not path or path[-1] != end_node: {
return {"points": 0, "reason": "no_path"} "q": "¿Qué logró el proyecto de Firmware Upgrade Automation?",
"options": [
edge_map = {} "Actualizar manualmente 10 modems por semana",
for e in edges: "Contratar más técnicos para visitas a domicilio",
edge_map[(e["a"], e["b"])] = e["modifier"] "Automatizar la actualización masiva de modems sin intervención humana",
edge_map[(e["b"], e["a"])] = e["modifier"] "Comprar equipos nuevos para reemplazar los modems viejos"
],
hop_cost = 0 "correct": 2,
valid = True "fun_fact": "400+ modems actualizados en batches automáticos. Antes era trabajo manual uno por uno."
for i in range(len(path) - 1): },
a, b = path[i], path[i + 1] {
mod = edge_map.get((a, b)) "q": "¿Hacia dónde se dirige el equipo en su visión AIOps?",
if mod is None or mod == "blocked": "options": [
valid = False "Depender más de herramientas comerciales como ServiceNow",
break "Detectar y predecir fallas antes de que el cliente las reporte",
elif mod == "slow": "Reducir el equipo de ingeniería a la mitad",
hop_cost += 2 "Migrar toda la red a la nube pública"
elif mod == "boost": ],
hop_cost += 0.5 "correct": 1,
else: "fun_fact": "La meta: que el técnico salga a campo con el diagnóstico ya hecho — no a diagnosticar."
hop_cost += 1
if not valid:
return {"points": 0, "reason": "invalid_path", "hops": len(path) - 1}
base = 1000
speed_bonus = int(500 * max(0, (time_limit - elapsed) / time_limit))
efficiency_penalty = int(max(0, hop_cost - (GRID_SIZE - 1)) * 50)
points = max(0, base + speed_bonus - efficiency_penalty)
return {
"points": points,
"reason": "ok",
"hops": len(path) - 1,
"hop_cost": hop_cost,
"speed_bonus": speed_bonus,
"efficiency_penalty": efficiency_penalty,
} }
]
# ── GAME STATE ──────────────────────────────────────────────────────────────── # ── GAME STATE ────────────────────────────────────────────────────────────────
class GameState: class GameState:
@@ -132,58 +128,56 @@ class GameState:
def reset(self): def reset(self):
self.phase = "lobby" self.phase = "lobby"
self.players = {} self.players: dict = {}
self.current_round = 0 self.current_q = -1
self.total_rounds = 3 self.q_start_time = 0
self.grid = None self.answers_this_round = []
self.round_start = 0 self.question_order = list(range(len(QUESTIONS)))
self.time_limit = 30 random.shuffle(self.question_order)
self.submissions = {}
self.session_id = None
game = GameState() game = GameState()
# ── CONNECTION MANAGER ──────────────────────────────────────────────────────── # ── CONNECTION MANAGER ────────────────────────────────────────────────────────
class ConnMgr: class ConnectionManager:
def __init__(self): def __init__(self):
self.players = {} self.players: dict = {}
self.hosts = [] self.hosts: list = []
async def connect_player(self, pid, ws): async def connect_player(self, ws_id: str, ws: WebSocket):
await ws.accept() await ws.accept()
self.players[pid] = ws self.players[ws_id] = ws
async def connect_host(self, ws): async def connect_host(self, ws: WebSocket):
await ws.accept() await ws.accept()
self.hosts.append(ws) self.hosts.append(ws)
def disconnect(self, pid): def disconnect(self, ws_id: str):
self.players.pop(pid, None) self.players.pop(ws_id, None)
game.players.pop(pid, None) game.players.pop(ws_id, None)
def disconnect_host(self, ws): def disconnect_host(self, ws: WebSocket):
if ws in self.hosts: if ws in self.hosts:
self.hosts.remove(ws) self.hosts.remove(ws)
async def send(self, pid, data): async def send_to(self, ws_id: str, data: dict):
ws = self.players.get(pid) ws = self.players.get(ws_id)
if ws: if ws:
try: try:
await ws.send_json(data) await ws.send_json(data)
except Exception: except Exception:
pass pass
async def broadcast_players(self, data): async def broadcast_players(self, data: dict):
dead = [] dead = []
for pid, ws in list(self.players.items()): for ws_id, ws in list(self.players.items()):
try: try:
await ws.send_json(data) await ws.send_json(data)
except Exception: except Exception:
dead.append(pid) dead.append(ws_id)
for d in dead: for d in dead:
self.disconnect(d) self.disconnect(d)
async def broadcast_hosts(self, data): async def broadcast_hosts(self, data: dict):
dead = [] dead = []
for ws in list(self.hosts): for ws in list(self.hosts):
try: try:
@@ -193,37 +187,15 @@ class ConnMgr:
for d in dead: for d in dead:
self.disconnect_host(d) self.disconnect_host(d)
async def broadcast_all(self, data): async def broadcast_all(self, data: dict):
await self.broadcast_players(data) await self.broadcast_players(data)
await self.broadcast_hosts(data) await self.broadcast_hosts(data)
mgr = ConnMgr() mgr = ConnectionManager()
def leaderboard(): def get_leaderboard():
ranked = sorted(game.players.values(), key=lambda p: p["score"], reverse=True) ranked = sorted(game.players.values(), key=lambda p: p["score"], reverse=True)
return [{"name": p.get("display", p["name"]), "score": p["score"]} for p in ranked] return [{"name": p.get("display", p["name"]), "score": p["score"], "dept": p.get("dept","")} for p in ranked]
def save_session():
now = datetime.utcnow().isoformat()
with get_db() as conn:
if not game.session_id:
cur = conn.execute(
"INSERT INTO sessions (started_at, ended_at, rounds) VALUES (?,?,?)",
(now, now, game.current_round)
)
game.session_id = cur.lastrowid
else:
conn.execute(
"UPDATE sessions SET ended_at=?, rounds=? WHERE id=?",
(now, game.current_round, game.session_id)
)
ranked = sorted(game.players.values(), key=lambda p: p["score"], reverse=True)
conn.execute("DELETE FROM scores WHERE session_id=?", (game.session_id,))
for rank, p in enumerate(ranked, 1):
conn.execute(
"INSERT INTO scores (session_id, player, score, rank, dept, finished_at) VALUES (?,?,?,?,?,?)",
(game.session_id, p.get("display", p["name"]), p["score"], rank, p.get("dept", ""), now)
)
# ── ROUTES ──────────────────────────────────────────────────────────────────── # ── ROUTES ────────────────────────────────────────────────────────────────────
@app.get("/", response_class=HTMLResponse) @app.get("/", response_class=HTMLResponse)
@@ -238,11 +210,18 @@ async def host_page(request: Request):
async def scores_page(request: Request): async def scores_page(request: Request):
return templates.TemplateResponse("scores.html", {"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") @app.get("/api/scores")
async def api_scores(): async def api_scores():
with get_db() as conn: with get_db() as conn:
sessions = conn.execute( sessions = conn.execute(
"SELECT * FROM sessions ORDER BY started_at DESC LIMIT 20" "SELECT * FROM sessions ORDER BY started_at DESC LIMIT 50"
).fetchall() ).fetchall()
result = [] result = []
for s in sessions: for s in sessions:
@@ -253,94 +232,84 @@ async def api_scores():
result.append({ result.append({
"id": s["id"], "id": s["id"],
"started_at": s["started_at"], "started_at": s["started_at"],
"rounds": s["rounds"], "ended_at": s["ended_at"],
"players": [ "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]
"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) return JSONResponse(result)
@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}
# ── PLAYER WEBSOCKET ────────────────────────────────────────────────────────── # ── PLAYER WEBSOCKET ──────────────────────────────────────────────────────────
@app.websocket("/ws/player/{pid}") @app.websocket("/ws/player/{ws_id}")
async def player_ws(ws: WebSocket, pid: str): async def player_ws(ws: WebSocket, ws_id: str):
await mgr.connect_player(pid, ws) await mgr.connect_player(ws_id, ws)
try: try:
await mgr.send(pid, {"type": "phase", "phase": game.phase}) await mgr.send_to(ws_id, {"type": "phase", "phase": game.phase})
async for raw in ws.iter_text(): async for raw in ws.iter_text():
msg = json.loads(raw) msg = json.loads(raw)
if msg["type"] == "join": if msg["type"] == "join":
name = msg["name"].strip()[:30] name = msg["name"].strip()[:40]
dept = msg.get("dept", "").strip()[:40] dept = msg.get("dept", "").strip()[:40]
display = msg.get("display", name).strip()[:30] display = msg.get("display", name).strip()[:40]
if not name or len(game.players) >= 4: if not name:
await mgr.send(pid, {"type": "error", "msg": "Game full or invalid name"})
continue continue
game.players[pid] = { game.players[ws_id] = {"name": name, "score": 0, "answered": False, "dept": dept, "display": display}
"name": name, await mgr.send_to(ws_id, {"type": "joined", "name": name, "display": display, "dept": dept})
"display": display,
"dept": dept,
"score": 0,
"submitted": False,
}
await mgr.send(pid, {"type": "joined", "name": name, "display": display, "dept": dept})
await mgr.broadcast_hosts({ await mgr.broadcast_hosts({
"type": "lobby_update", "type": "lobby_update",
"players": [p.get("display", p["name"]) for p in game.players.values()], "players": [{"name": p["name"], "display": p.get("display", p["name"]), "dept": p.get("dept","")} for p in game.players.values()],
"count": len(game.players), "count": len(game.players)
}) })
elif msg["type"] == "submit_path": elif msg["type"] == "answer":
if game.phase != "playing": if game.phase != "question":
continue continue
player = game.players.get(pid) player = game.players.get(ws_id)
if not player or player["submitted"]: if not player or player["answered"]:
continue continue
elapsed = time.time() - game.round_start elapsed = time.time() - game.q_start_time
path = msg.get("path", []) TIME_LIMIT = 15
result = score_path(path, game.grid["edges"], elapsed, game.time_limit, game.grid["end"]) if elapsed > TIME_LIMIT:
player["score"] += result["points"] continue
player["submitted"] = True
game.submissions[pid] = { player["answered"] = True
"name": player.get("display", player["name"]), chosen = msg["choice"]
"path": path, q_idx = game.question_order[game.current_q]
"points": result["points"], correct = QUESTIONS[q_idx]["correct"]
"elapsed": round(elapsed, 2), is_correct = chosen == correct
"reason": result["reason"],
} points = 0
await mgr.send(pid, { if is_correct:
"type": "path_result", speed_bonus = max(0, (TIME_LIMIT - elapsed) / TIME_LIMIT)
"points": result["points"], points = int(200 + 800 * speed_bonus)
"total": player["score"], player["score"] += points
"reason": result["reason"],
"elapsed": round(elapsed, 2), 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({ await mgr.broadcast_hosts({
"type": "submission_update", "type": "answer_update",
"submitted": sum(1 for p in game.players.values() if p["submitted"]), "answered": answered,
"total": len(game.players), "total": len(game.players),
"submissions": list(game.submissions.values()), "details": game.answers_this_round
}) })
except WebSocketDisconnect: except WebSocketDisconnect:
mgr.disconnect(pid) mgr.disconnect(ws_id)
await mgr.broadcast_hosts({ await mgr.broadcast_hosts({
"type": "lobby_update", "type": "lobby_update",
"players": [p.get("display", p["name"]) for p in game.players.values()], "players": [{"name": p["name"], "display": p.get("display", p["name"]), "dept": p.get("dept","")} for p in game.players.values()],
"count": len(game.players), "count": len(game.players)
}) })
# ── HOST WEBSOCKET ──────────────────────────────────────────────────────────── # ── HOST WEBSOCKET ────────────────────────────────────────────────────────────
@@ -350,50 +319,66 @@ async def host_ws(ws: WebSocket):
try: try:
await ws.send_json({ await ws.send_json({
"type": "lobby_update", "type": "lobby_update",
"players": [p.get("display", p["name"]) for p in game.players.values()], "players": [{"name": p["name"], "display": p.get("display", p["name"]), "dept": p.get("dept","")} for p in game.players.values()],
"count": len(game.players), "count": len(game.players)
}) })
async for raw in ws.iter_text(): async for raw in ws.iter_text():
msg = json.loads(raw) msg = json.loads(raw)
if msg["type"] == "start_round": if msg["type"] == "start_game":
# accept config params from host on first round existing_players = dict(game.players) # preserve joined players
if "total_rounds" in msg: game.reset()
game.total_rounds = int(msg["total_rounds"]) game.players = existing_players # restore them
if "time_limit" in msg: # Reset scores and state for a fresh game
game.time_limit = int(msg["time_limit"])
game.grid = generate_grid()
game.phase = "playing"
game.round_start = time.time()
game.current_round += 1
game.submissions = {}
for p in game.players.values(): for p in game.players.values():
p["submitted"] = False 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({ await mgr.broadcast_all({
"type": "round_start", "type": "question",
"round": game.current_round, "number": game.current_q + 1,
"total_rounds": game.total_rounds, "total": len(QUESTIONS),
"grid": game.grid, "q": q["q"],
"time_limit": game.time_limit, "options": q["options"],
"time_limit": 15,
"correct_idx_hint": -1 # not revealed until show_results
}) })
elif msg["type"] == "end_round": elif msg["type"] == "show_results":
game.phase = "results" game.phase = "results"
lb = leaderboard() q_idx = game.question_order[game.current_q]
save_session() q = QUESTIONS[q_idx]
await mgr.broadcast_all({ await mgr.broadcast_all({
"type": "round_results", "type": "results",
"round": game.current_round, "correct_idx": q["correct"],
"grid": game.grid, "fun_fact": q["fun_fact"],
"submissions": list(game.submissions.values()), "leaderboard": get_leaderboard()
"leaderboard": lb,
"is_final": game.current_round >= game.total_rounds,
}) })
elif msg["type"] == "reset": elif msg["type"] == "reset":
# Kick all players back to join screen, then wipe state
await mgr.broadcast_players({"type": "kicked"})
game.reset() game.reset()
await mgr.broadcast_all({"type": "phase", "phase": "lobby"}) await mgr.broadcast_hosts({"type": "lobby_update", "players": [], "count": 0})
except WebSocketDisconnect: except WebSocketDisconnect:
mgr.disconnect_host(ws) mgr.disconnect_host(ws)