# main.py : point d'entree FastAPI (routes + websocket + lancement game loop) import asyncio import logging import os from contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from game_loop import GameLoop from lobby import Lobby, LobbyManager from stats import get_leaderboard, init_db, is_discord_taken, save_result from websocket import ConnectionManager LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper() logging.basicConfig( level=getattr(logging, LOG_LEVEL, logging.INFO), format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) logger = logging.getLogger(__name__) @asynccontextmanager async def lifespan(app: FastAPI): await init_db() logger.info("Base de données initialisée") yield app = FastAPI(title="SOULGATE", lifespan=lifespan) app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) manager = ConnectionManager() lobby_manager = LobbyManager() active_games: dict[str, tuple[GameLoop, asyncio.Task]] = {} logger.info("Serveur SOULGATE démarré") # --- routes leaderboard --- class SubmitPayload(BaseModel): team_name: str score: int time_seconds: int waves_completed: int victory: bool lobby_code: str = "" submitter_id: str = "" p1_username: str = "" p1_class: str = "" p1_discord: str = "" p2_username: str = "" p2_class: str = "" p2_discord: str = "" p3_username: str = "" p3_class: str = "" p3_discord: str = "" # lobbies dont le score a deja ete soumis : empeche les doublons _submitted_lobbies: set[str] = set() @app.get("/leaderboard") async def route_leaderboard() -> list[dict]: return await get_leaderboard() @app.get("/leaderboard/check-discord/{tag}") async def route_check_discord(tag: str) -> dict: taken = await is_discord_taken(tag) return {"taken": taken} @app.post("/leaderboard/submit") async def route_submit(payload: SubmitPayload) -> dict: if not payload.team_name.strip(): raise HTTPException(400, "Le nom d'équipe est obligatoire.") # une defaite n'a rien a faire dans le leaderboard if not payload.victory: raise HTTPException(403, "Seules les victoires peuvent être enregistrées.") # seul le host peut soumettre, et 1 seule fois par lobby code = payload.lobby_code.upper().strip() submitter = payload.submitter_id.strip() if not code or not submitter: raise HTTPException(400, "Identifiants de session manquants.") if code in _submitted_lobbies: raise HTTPException(409, "Le score de cette partie a déjà été enregistré.") lobby = lobby_manager.lobbies.get(code) if not lobby: raise HTTPException(404, "Lobby introuvable ou déjà nettoyé.") player = lobby.get_player(submitter) if not player or not player.is_host: raise HTTPException(403, "Seul le chef d'équipe peut enregistrer le score.") # pas de doublon de discord for tag in (payload.p1_discord, payload.p2_discord, payload.p3_discord): if tag and await is_discord_taken(tag): raise HTTPException(409, f"Le Discord '{tag}' est déjà dans le leaderboard.") rank = await save_result(payload.model_dump(exclude={"lobby_code", "submitter_id"})) _submitted_lobbies.add(code) return {"rank": rank} @app.get("/health") async def health() -> dict: return {"status": "ok"} @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket) -> None: conn_id = await manager.connect(websocket) try: while True: try: data = await websocket.receive_json() except WebSocketDisconnect: raise except Exception: try: await manager.send(conn_id, {"type": "error", "code": "invalid_json", "message": "Message JSON invalide."}) except Exception: pass continue await _handle_message(conn_id, data.get("type", ""), data) except WebSocketDisconnect: messages = lobby_manager.disconnect(conn_id) await manager.dispatch(messages) manager.disconnect(conn_id) async def _handle_message(conn_id: str, msg_type: str, data: dict) -> None: username = manager.get_username(conn_id) if msg_type == "set_username": uname = str(data.get("username", "")).strip() if not uname: await manager.send(conn_id, {"type": "error", "code": "invalid_username", "message": "Le pseudo ne peut pas être vide."}) return manager.set_username(conn_id, uname) await manager.send(conn_id, {"type": "username_set", "username": uname, "my_id": conn_id}) elif msg_type in ("create_lobby", "join_lobby"): if not username: await manager.send(conn_id, {"type": "error", "code": "no_username", "message": "Définis ton pseudo d'abord (set_username)."}) return if msg_type == "create_lobby": messages = lobby_manager.create_lobby(conn_id, username) else: messages = lobby_manager.join_lobby(conn_id, username, data.get("code", "")) await manager.dispatch(messages) elif msg_type == "leave_lobby": await manager.dispatch(lobby_manager.leave_lobby(conn_id)) elif msg_type == "select_class": await manager.dispatch(lobby_manager.select_class(conn_id, data.get("class", ""))) elif msg_type == "ready": await manager.dispatch(lobby_manager.set_ready(conn_id)) elif msg_type == "start_game": messages = lobby_manager.start_game(conn_id) await manager.dispatch(messages) if messages and messages[0][1].get("type") == "game_starting": code = lobby_manager.conn_to_lobby.get(conn_id) if code and code in lobby_manager.lobbies: await _launch_game(lobby_manager.lobbies[code]) elif msg_type == "input": code = lobby_manager.conn_to_lobby.get(conn_id) if code and code in active_games: game, _ = active_games[code] try: dx = float(data.get("dx", 0)) dy = float(data.get("dy", 0)) except (ValueError, TypeError): return game.handle_input(conn_id, dx, dy) elif msg_type == "attack": code = lobby_manager.conn_to_lobby.get(conn_id) if code and code in active_games: game, _ = active_games[code] try: tx = float(data.get("tx", 0)) ty = float(data.get("ty", 0)) except (ValueError, TypeError): return game.handle_attack(conn_id, tx, ty) elif msg_type == "ability": code = lobby_manager.conn_to_lobby.get(conn_id) if code and code in active_games: game, _ = active_games[code] try: ability_id = int(data.get("id", 0)) tx = float(data.get("tx", 0)) ty = float(data.get("ty", 0)) except (ValueError, TypeError): return if ability_id in (1, 2, 3): game.handle_ability(conn_id, ability_id, tx, ty) elif msg_type == "player_upgrade": code = lobby_manager.conn_to_lobby.get(conn_id) if code and code in active_games: game, _ = active_games[code] game.handle_upgrade(conn_id, data.get("upgrade_id", "")) elif msg_type == "displacement": code = lobby_manager.conn_to_lobby.get(conn_id) if code and code in active_games: game, _ = active_games[code] try: tx = float(data.get("tx", 0)) ty = float(data.get("ty", 0)) except (ValueError, TypeError): return game.handle_displacement(conn_id, tx, ty) else: logger.warning("Message inconnu de %s : %s", conn_id, msg_type) await manager.send(conn_id, {"type": "error", "code": "unknown_message", "message": f"Type de message inconnu : {msg_type}"}) async def _launch_game(lobby: Lobby) -> None: async def broadcast(msg: dict) -> None: for conn_id in lobby.get_conn_ids(): await manager.send(conn_id, msg) game = GameLoop(lobby, broadcast) task = asyncio.create_task(game.run()) active_games[lobby.code] = (game, task) logger.info("Game loop lancée — lobby %s", lobby.code)