""" Versailles Airsoft - Application Web Principale (FastAPI) Plateforme de gestion et d'organisation de parties d'airsoft sur terrain privé à Versailles. """ import time import os import calendar from contextlib import asynccontextmanager from datetime import date, datetime, timedelta dt_date = date import string from typing import Optional, List from fastapi import ( FastAPI, Request, Depends, Form, HTTPException, status, Response, WebSocket, WebSocketDisconnect ) from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from sqlalchemy.orm import Session from sqlalchemy import func from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded import asyncio from slowapi.middleware import SlowAPIMiddleware from database import engine, Base, SessionLocal, get_db, init_db from models import User, Availability, Game, VALID_SLOTS from scenarios_loader import ( get_all_scenarios, get_random_scenario, get_scenario_by_slug_or_id, ) from websocket_manager import tools_sync_manager, random_teams_sync_manager from auth import ( hash_password, verify_password, create_session_token, get_current_user, set_session_cookie, delete_session_cookie, generate_csrf_token, validate_csrf, COOKIE_NAME, SESSION_MAX_AGE, SESSION_DEFAULT_MAX_AGE, SESSION_REMEMBER_MAX_AGE ) from discord import ( send_discord_game_notification, send_discord_game_cancellation_notification, format_french_date, MOIS_FR, JOURS_FR, URL_GENEREE, exchange_discord_code, get_discord_user_profile ) @asynccontextmanager async def lifespan(app: FastAPI): """Initialise la base de données de l'application.""" init_db() yield # Initialisation synchrone de la base SQLite (compatibilité WSGI / Passenger) init_db() # Application FastAPI avec cycle de vie lifespan app = FastAPI( title="Versailles Airsoft HQ", description="Plateforme tactique d'organisation de parties d'airsoft sur terrain privé", version="1.0.0", lifespan=lifespan ) def get_client_ip(request: Request) -> str: """Récupère l'IP réelle du client en supportant les reverse-proxies (X-Forwarded-For).""" forwarded = request.headers.get("x-forwarded-for") if forwarded: return forwarded.split(",")[0].strip() if request.client and request.client.host: return request.client.host return "127.0.0.1" # Configuration du Rate Limiting (anti force-brute) limiter = Limiter(key_func=get_client_ip, default_limits=[]) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) app.add_middleware(SlowAPIMiddleware) # Middleware de sécurité HTTP (En-têtes de durcissement) @app.middleware("http") async def add_security_headers(request: Request, call_next): response = await call_next(request) response.headers["X-Frame-Options"] = "DENY" response.headers["X-Content-Type-Options"] = "nosniff" response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin" return response # Montage des fichiers statiques static_dir = os.path.join(os.path.dirname(__file__), "static") if os.path.exists(static_dir): app.mount("/static", StaticFiles(directory=static_dir), name="static") # Configuration des templates Jinja2 templates_dir = os.path.join(os.path.dirname(__file__), "templates") templates = Jinja2Templates(directory=templates_dir) PLAYER_THRESHOLD = int(os.getenv("PLAYER_THRESHOLD", "8")) templates.env.globals["player_threshold"] = PLAYER_THRESHOLD templates.env.globals["URL_Generee"] = URL_GENEREE templates.env.globals["discord_auth_url"] = URL_GENEREE templates.env.globals["csrf_token"] = generate_csrf_token SLOT_MATIN = "Matin (09h-13h)" SLOT_APREM = "Après-midi (13h-17h)" # ========================================== # FONCTIONS UTILITAIRES & LOGIQUE CALENDRIER # ========================================== def calculate_effective_players_for_date(db: Session, target_date: date): """ Calcule les listes d'opérateurs pour chaque créneau (Matin et Après-midi) d'une date donnée. """ avails = ( db.query(Availability) .join(User, Availability.user_id == User.id) .filter(Availability.date == target_date) .all() ) user_slots = {} # user_id -> set(slots) user_names = {} # user_id -> username user_avatars = {} # user_id -> avatar_url for a in avails: if a.user: user_names[a.user_id] = a.user.username user_avatars[a.user_id] = a.user.avatar_url if a.user_id not in user_slots: user_slots[a.user_id] = set() user_slots[a.user_id].add(a.slot) # Utilisateurs effectifs pour chaque créneau matin_uids = { uid for uid, slots in user_slots.items() if SLOT_MATIN in slots } aprem_uids = { uid for uid, slots in user_slots.items() if SLOT_APREM in slots } matin_players = [user_names[uid] for uid in matin_uids if uid in user_names] aprem_players = [user_names[uid] for uid in aprem_uids if uid in user_names] matin_players_info = [ {"id": uid, "username": user_names[uid], "avatar_url": user_avatars.get(uid)} for uid in matin_uids if uid in user_names ] aprem_players_info = [ {"id": uid, "username": user_names[uid], "avatar_url": user_avatars.get(uid)} for uid in aprem_uids if uid in user_names ] return { SLOT_MATIN: { "uids": matin_uids, "usernames": sorted(matin_players, key=lambda x: x.lower()), "players_info": sorted(matin_players_info, key=lambda x: x["username"].lower()), }, SLOT_APREM: { "uids": aprem_uids, "usernames": sorted(aprem_players, key=lambda x: x.lower()), "players_info": sorted(aprem_players_info, key=lambda x: x["username"].lower()), }, "user_direct_slots": user_slots, "user_names": user_names } async def sync_and_trigger_games_for_date(db: Session, target_date: date) -> dict: """ Met à jour la table Game et déclenche les webhooks Discord : - Envoi d'une alerte Discord dès que 8+ joueurs sont atteints (confirmation). - Envoi d'une alerte Discord d'annulation si des désistements font passer l'effectif sous 8 joueurs. - Ré-envoi de confirmation si le seuil de 8 joueurs est de nouveau atteint par la suite. """ effective_data = calculate_effective_players_for_date(db, target_date) notifications_triggered = [] notifications_cancelled = [] for slot in VALID_SLOTS: players = effective_data[slot]["usernames"] count = len(players) game = db.query(Game).filter(Game.date == target_date, Game.slot == slot).first() if count >= PLAYER_THRESHOLD: if not game: game = Game( date=target_date, slot=slot, player_count=count, notified_discord=False ) db.add(game) db.flush() else: game.player_count = count # Déclenchement Discord (Confirmation) if not game.notified_discord: webhook_sent = await send_discord_game_notification( target_date=target_date, slot=slot, player_count=count, players=players ) if webhook_sent: game.notified_discord = True notifications_triggered.append(slot) db.commit() else: if game: game.player_count = count # Si la partie avait été confirmée et notifiée sur Discord, on envoie l'annulation if game.notified_discord: webhook_sent = await send_discord_game_cancellation_notification( target_date=target_date, slot=slot, player_count=count ) # On réinitialise notified_discord à False pour permettre une ré-alerte si de nouveaux joueurs reviennent game.notified_discord = False if webhook_sent: notifications_cancelled.append(slot) db.commit() return { "effective_data": effective_data, "notifications_triggered": notifications_triggered, "notifications_cancelled": notifications_cancelled } def get_month_calendar_data(year: int, month: int, db: Session, current_user: Optional[User] = None): """ Calcule la structure du calendrier pour le mois et l'année demandés, incluant le décompte des joueurs par créneau (avec cumul journée complète). """ cal = calendar.Calendar(firstweekday=0) # 0 = Lundi month_days = cal.monthdatescalendar(year, month) start_date = month_days[0][0] end_date = month_days[-1][-1] # Récupération de toutes les disponibilités pour la plage affichée availabilities = ( db.query(Availability) .join(User, Availability.user_id == User.id) .filter(Availability.date >= start_date, Availability.date <= end_date) .all() ) # Regroupement par date : date -> dict(user_id -> set(slots)) date_user_slots = {} user_names_map = {} for a in availabilities: if a.user: user_names_map[a.user_id] = a.user.username if a.date not in date_user_slots: date_user_slots[a.date] = {} if a.user_id not in date_user_slots[a.date]: date_user_slots[a.date][a.user_id] = set() date_user_slots[a.date][a.user_id].add(a.slot) # Récupération des parties confirmées confirmed_games = ( db.query(Game) .filter(Game.date >= start_date, Game.date <= end_date, Game.player_count >= PLAYER_THRESHOLD) .all() ) confirmed_game_keys = {(g.date, g.slot) for g in confirmed_games} today = date.today() calendar_days = [] for week in month_days: for d in week: is_current_month = (d.month == month) is_today = (d == today) user_slots_for_day = date_user_slots.get(d, {}) # Calcul des uids inscrits pour chaque créneau matin_uids = { uid for uid, slots in user_slots_for_day.items() if SLOT_MATIN in slots } aprem_uids = { uid for uid, slots in user_slots_for_day.items() if SLOT_APREM in slots } slot_counts = { SLOT_MATIN: len(matin_uids), SLOT_APREM: len(aprem_uids) } slot_uids_map = { SLOT_MATIN: matin_uids, SLOT_APREM: aprem_uids } slots_summary = [] day_has_confirmed = False day_user_registered = False for slot in VALID_SLOTS: count = slot_counts[slot] uids = slot_uids_map[slot] is_user_in = bool(current_user and current_user.id in uids) if (d, slot) in confirmed_game_keys or count >= PLAYER_THRESHOLD: day_has_confirmed = True if is_user_in: day_user_registered = True slot_players = sorted([user_names_map[uid] for uid in uids if uid in user_names_map], key=lambda x: x.lower()) slots_summary.append({ "slot": slot, "count": count, "is_user_in": is_user_in, "players": slot_players }) calendar_days.append({ "date": d, "date_str": d.strftime("%Y-%m-%d"), "day_num": d.day, "is_current_month": is_current_month, "is_today": is_today, "is_past": (d < today), "has_confirmed_game": day_has_confirmed, "is_user_registered": day_user_registered, "slots_summary": slots_summary }) # Calcul des mois précédent et suivant pour la navigation if month == 1: prev_year, prev_month = year - 1, 12 else: prev_year, prev_month = year, month - 1 if month == 12: next_year, next_month = year + 1, 1 else: next_year, next_month = year, month + 1 return { "calendar_days": calendar_days, "current_month_name": MOIS_FR[month], "current_year": year, "current_month": month, "prev_year": prev_year, "prev_month": prev_month, "next_year": next_year, "next_month": next_month, } def get_confirmed_games_count(db: Session) -> int: """Retourne le nombre total de parties confirmées futures (>= 8 joueurs et date >= aujourd'hui).""" today = date.today() return db.query(Game).filter(Game.player_count >= PLAYER_THRESHOLD, Game.date >= today).count() # ========================================== # ROUTES PRINCIPALES (PAGES HTML) # ========================================== @app.get("/", response_class=HTMLResponse) async def index_page( request: Request, year: Optional[int] = None, month: Optional[int] = None, db: Session = Depends(get_db) ): """ Page principale : Calendrier tactique des disponibilités et opérations. """ current_user = get_current_user(request, db) today = date.today() target_year = year or today.year target_month = month or today.month # Sécurisation des valeurs de mois if target_month < 1 or target_month > 12: target_month = today.month target_year = today.year cal_data = get_month_calendar_data(target_year, target_month, db, current_user) total_confirmed = get_confirmed_games_count(db) total_availabilities = db.query(Availability).count() context = { "request": request, "current_user": current_user, "active_page": "calendar", "confirmed_games_count": total_confirmed, "total_confirmed_games": total_confirmed, "total_availabilities": total_availabilities, **cal_data } return templates.TemplateResponse(request=request, name="index.html", context=context) @app.get("/parties", response_class=HTMLResponse) async def games_page(request: Request, db: Session = Depends(get_db)): """ Page des parties validées futures uniquement (celles ayant atteint 8 joueurs et plus à partir d'aujourd'hui). """ current_user = get_current_user(request, db) today = date.today() confirmed_games = ( db.query(Game) .filter(Game.player_count >= PLAYER_THRESHOLD, Game.date >= today) .order_by(Game.date.asc(), Game.slot.asc()) .all() ) games_data = [] for g in confirmed_games: effective_data = calculate_effective_players_for_date(db, g.date) slot_info = effective_data.get(g.slot, {"usernames": [], "players_info": []}) players = slot_info["usernames"] players_info = slot_info.get("players_info", []) is_user_registered = bool(current_user and current_user.username in players) games_data.append({ "game": g, "players": players, "players_info": players_info, "is_user_registered": is_user_registered, "french_date": format_french_date(g.date) }) context = { "request": request, "current_user": current_user, "active_page": "games", "confirmed_games_count": len(confirmed_games), "games": games_data } return templates.TemplateResponse(request=request, name="games.html", context=context) @app.get("/regles", response_class=HTMLResponse) async def rules_page(request: Request, db: Session = Depends(get_db)): """ Page de la charte de sécurité, limites FPS et règles du terrain. """ current_user = get_current_user(request, db) total_confirmed = get_confirmed_games_count(db) context = { "request": request, "current_user": current_user, "active_page": "rules", "confirmed_games_count": total_confirmed } return templates.TemplateResponse(request=request, name="rules.html", context=context) # ========================================== # BANQUE & GESTION CRUD DES SCÉNARIOS TACTIQUES # ========================================== @app.get("/scenarios", response_class=HTMLResponse) async def scenarios_page(request: Request, db: Session = Depends(get_db)): """ Banque de scénarios tactiques chargés depuis le catalogue JSON. """ current_user = get_current_user(request, db) total_confirmed = get_confirmed_games_count(db) scenarios = get_all_scenarios() context = { "request": request, "current_user": current_user, "active_page": "scenarios", "confirmed_games_count": total_confirmed, "scenarios": scenarios, "scenarios_data": scenarios } return templates.TemplateResponse(request=request, name="scenarios.html", context=context) @app.get("/scenarios/random") async def random_scenario_redirect( request: Request, app_required: Optional[bool] = None, db: Session = Depends(get_db) ): """ Redirection instantanée vers un scénario aléatoire, avec filtre optionnel sur l'usage d'application. """ chosen = get_random_scenario(app_required=app_required) if not chosen: raise HTTPException(status_code=404, detail="Aucun scénario disponible.") return RedirectResponse(url=f"/scenarios/{chosen['slug']}", status_code=status.HTTP_303_SEE_OTHER) @app.get("/scenarios/{id_or_slug}", response_class=HTMLResponse) async def scenario_detail_page( id_or_slug: str, request: Request, db: Session = Depends(get_db) ): """ Fiche détaillée d'un scénario tactique avec cartographie Leaflet interactive calée sur Satory. Accessible par slug (ex: /scenarios/guerre-de-gangs) ou par identifiant ID (ex: /scenarios/1). """ current_user = get_current_user(request, db) total_confirmed = get_confirmed_games_count(db) scenario = get_scenario_by_slug_or_id(id_or_slug) if not scenario: raise HTTPException(status_code=404, detail="Scénario tactique non trouvé.") # Scénarios pour la navigation rapide all_scenarios = get_all_scenarios() context = { "request": request, "current_user": current_user, "active_page": "scenarios", "confirmed_games_count": total_confirmed, "scenario": scenario, "all_scenarios": all_scenarios, } return templates.TemplateResponse(request=request, name="scenario_detail.html", context=context) @app.get("/outils", response_class=HTMLResponse) async def tools_page( request: Request, room: Optional[str] = None, db: Session = Depends(get_db) ): """ Page des outils tactiques de terrain autonomes (Bombe factice, Minuteur, Compteur, Capture de base et tirage des équipes). Permet la synchronisation multi-téléphones en direct sans historique ni base de données. """ current_user = get_current_user(request, db) total_confirmed = get_confirmed_games_count(db) context = { "request": request, "current_user": current_user, "active_page": "tools", "confirmed_games_count": total_confirmed, "initial_room": room.upper().strip() if room else "" } return templates.TemplateResponse(request=request, name="tools.html", context=context) # ========================================== # SYNCHRONISATION MULTI-TÉLÉPHONES DES OUTILS TACTIQUES # ========================================== @app.post("/api/tools/create-room") async def api_create_tools_room(): """Génère un salon éphémère de synchronisation pour les outils.""" room_code = tools_sync_manager.generate_room_code() tools_sync_manager.get_or_create_room(room_code) return {"room_code": room_code} @app.websocket("/ws/tools/{room_code}") async def websocket_tools_endpoint(websocket: WebSocket, room_code: str): """ Canal WebSocket ultra-rapide pour synchroniser en temps réel les minuteurs et les compteurs de points/respawns entre plusieurs téléphones connectés. """ clean_code = room_code.upper().strip() client_id = websocket.query_params.get("client_id", "default")[:80] await tools_sync_manager.connect(clean_code, websocket, client_id) room = tools_sync_manager.get_or_create_room(clean_code) base = room.get_ecapture_base(client_id) # Diffusion immédiate de l'état actuel et mise à jour du compteur d'appareils await tools_sync_manager.broadcast_room_state(clean_code) try: while True: data = await websocket.receive_json() msg_type = data.get("type") if msg_type == "timer_action": action = data.get("action") if action == "start": if not room.timer_running and room.get_current_timer_seconds() > 0: room.timer_running = True room.timer_last_start_time = time.time() elif action == "pause": if room.timer_running: room.timer_seconds_left = room.get_current_timer_seconds() room.timer_running = False room.timer_last_start_time = None elif action == "reset": room.timer_running = False room.timer_last_start_time = None room.timer_seconds_left = room.timer_initial_seconds elif action == "set_duration": dur_seconds = int(data.get("seconds", 1200)) room.timer_initial_seconds = dur_seconds room.timer_seconds_left = dur_seconds room.timer_running = False room.timer_last_start_time = None await tools_sync_manager.broadcast_room_state(clean_code) elif msg_type == "counter_action": target = data.get("target", "single") # "single", "alpha", "bravo", "charlie", "delta" delta = int(data.get("delta", 0)) is_reset = data.get("reset", False) mode = data.get("mode") label = data.get("label") teams_count = data.get("teams_count") if mode in ["single", "teams"]: room.counter_mode = mode if label: room.counter_label = label if teams_count: try: tc = int(teams_count) if tc in [2, 3, 4]: room.counter_teams_count = tc except (ValueError, TypeError): pass if is_reset: if target == "alpha": room.counter_alpha = 0 elif target == "bravo": room.counter_bravo = 0 elif target == "charlie": room.counter_charlie = 0 elif target == "delta": room.counter_delta = 0 elif target == "all_teams": room.counter_alpha = 0 room.counter_bravo = 0 room.counter_charlie = 0 room.counter_delta = 0 else: room.counter_single = 0 else: if target == "alpha": room.counter_alpha = max(0, room.counter_alpha + delta) elif target == "bravo": room.counter_bravo = max(0, room.counter_bravo + delta) elif target == "charlie": room.counter_charlie = max(0, room.counter_charlie + delta) elif target == "delta": room.counter_delta = max(0, room.counter_delta + delta) else: room.counter_single = max(0, room.counter_single + delta) await tools_sync_manager.broadcast_room_state(clean_code) elif msg_type == "ecapture_action": action = data.get("action") if action == "capture": team = data.get("team") if team in ["alpha", "bravo", "charlie", "delta"]: base["owner"] = team base["running"] = True base["holding_team"] = None tools_sync_manager.ensure_room_loop(clean_code) elif action == "start_hold": team = data.get("team") if team in ["alpha", "bravo", "charlie", "delta"]: base["holding_team"] = team elif action == "cancel_hold": base["holding_team"] = None elif action == "pause": base["running"] = False elif action == "resume": if base["owner"]: base["running"] = True tools_sync_manager.ensure_room_loop(clean_code) elif action == "neutralize": base["owner"] = None base["running"] = False base["holding_team"] = None base["control_seconds"] = 0 elif action == "reset": base.update(owner=None, running=False, holding_team=None, control_seconds=0) elif action == "set_duration": try: duration = int(data.get("duration", 10)) if duration in [5, 10, 30]: room.ecapture_hold_duration = duration except (ValueError, TypeError): pass elif action == "set_teams_count": try: teams_count = int(data.get("teams_count", 2)) if teams_count in [2, 3, 4]: room.ecapture_teams_count = teams_count except (ValueError, TypeError): pass if not any(item["running"] and item["owner"] for item in room.ecapture_bases.values()): tools_sync_manager.stop_room_loop(clean_code) await tools_sync_manager.broadcast_room_state(clean_code) elif msg_type == "switch_tab": tab = data.get("tab") if tab in ["bomb", "timer", "counter", "ecapture", "team_draw"]: room.active_tab = tab await tools_sync_manager.broadcast_room_state(clean_code) except WebSocketDisconnect: tools_sync_manager.disconnect(clean_code, websocket) await tools_sync_manager.broadcast_room_state(clean_code) except Exception: tools_sync_manager.disconnect(clean_code, websocket) # ============================================================================== # OUTIL DE RÉPARTITION ALÉATOIRE DES ÉQUIPES (SALONS TEMPS RÉEL) # ============================================================================== @app.post("/api/tools/team-draw/create-room") async def api_create_team_draw_room(): """Crée un salon temporaire pour le tirage équilibré des équipes.""" room_code = random_teams_sync_manager.generate_room_code() random_teams_sync_manager.get_or_create_room(room_code) return {"room_code": room_code} @app.websocket("/ws/tools/team-draw/{room_code}") async def websocket_team_draw_endpoint(websocket: WebSocket, room_code: str): """Synchronise les participants et l'attribution secrète des équipes.""" clean_code = room_code.upper().strip() await websocket.accept() room = random_teams_sync_manager.get_or_create_room(clean_code) current_player_id: Optional[str] = None try: while True: data = await websocket.receive_json() msg_type = data.get("type") if msg_type == "join": player_id = str(data.get("player_id", "")).strip() or f"P_{int(time.time() * 1000) % 100000}" player_name = str(data.get("name", "")).strip() or "Opérateur" current_player_id = player_id random_teams_sync_manager.active_connections.setdefault(clean_code, {})[websocket] = player_id room.get_or_create_player(player_id, player_name) await random_teams_sync_manager.broadcast_room_state(clean_code) elif msg_type == "set_teams_count": player = room.players.get(current_player_id) if current_player_id else None if player and player.is_host: try: room.set_teams_count(int(data.get("teams_count", 2))) except (TypeError, ValueError): pass await random_teams_sync_manager.broadcast_room_state(clean_code) elif msg_type == "start_game": player = room.players.get(current_player_id) if current_player_id else None if player and player.is_host and room.status == "lobby" and room.players: room.start_game() await random_teams_sync_manager.broadcast_room_state(clean_code) if room.status == "countdown": async def run_countdown(initial_round: int, duration: int): for _ in range(duration): await asyncio.sleep(1) if room.status != "countdown" or room.round_number != initial_round: return remaining = max(0, int(round(room.countdown_end_time - time.time()))) if room.countdown_end_time else 0 if remaining <= 0: room.trigger_reveal() await random_teams_sync_manager.broadcast_room_state(clean_code) if room.status == "started": return asyncio.create_task(run_countdown(room.round_number, 5)) elif msg_type == "reset_round": player = room.players.get(current_player_id) if current_player_id else None if player and player.is_host: room.reset_round() await random_teams_sync_manager.broadcast_room_state(clean_code) elif msg_type == "claim_host": if current_player_id and current_player_id in room.players: room.host_player_id = current_player_id for player_id, player in room.players.items(): player.is_host = player_id == current_player_id await random_teams_sync_manager.broadcast_room_state(clean_code) elif msg_type == "kick_player": player = room.players.get(current_player_id) if current_player_id else None target_id = data.get("player_id") if player and player.is_host and target_id in room.players: del room.players[target_id] await random_teams_sync_manager.broadcast_room_state(clean_code) elif msg_type == "ping": await websocket.send_json({"type": "pong"}) except WebSocketDisconnect: random_teams_sync_manager.disconnect(clean_code, websocket) await random_teams_sync_manager.broadcast_room_state(clean_code) except Exception: random_teams_sync_manager.disconnect(clean_code, websocket) # ========================================== # AUTHENTIFICATION (LOGIN / REGISTER / LOGOUT / DISCORD OAUTH) # ========================================== def get_discord_oauth_url(request: Request) -> str: """Génère l'URL d'autorisation Discord selon l'hôte actuel (localhost vs production).""" raw_url = os.getenv("URL_Generee", "").strip() or os.getenv("URL_GENAREE", "").strip() client_id = os.getenv("DISCORD_CLIENT_ID", "").strip() host = request.headers.get("host", "localhost:8000") scheme = "https" if ("mmi-velizy.fr" in host or request.url.scheme == "https") else "http" callback_url = f"{scheme}://{host}/auth/discord/callback" if client_id: import urllib.parse encoded_redirect = urllib.parse.quote(callback_url, safe="") return f"https://discord.com/oauth2/authorize?client_id={client_id}&response_type=code&redirect_uri={encoded_redirect}&scope=identify" return raw_url or "/login?error=Configuration+Discord+manquante" @app.get("/login", response_class=HTMLResponse) async def login_get( request: Request, redirect: Optional[str] = None, error: Optional[str] = None, db: Session = Depends(get_db) ): """Affiche le formulaire de connexion.""" current_user = get_current_user(request, db) if current_user: return RedirectResponse(url=redirect or "/", status_code=status.HTTP_303_SEE_OTHER) return templates.TemplateResponse(request=request, name="login.html", context={ "request": request, "current_user": None, "active_page": "login", "redirect_url": redirect or "", "discord_auth_url": "/auth/discord/login", "error": error }) @app.get("/auth/discord/login") @app.get("/login/discord") async def discord_login(request: Request): """Redirige l'utilisateur vers l'URL d'autorisation Discord OAuth2.""" target_url = get_discord_oauth_url(request) return RedirectResponse(url=target_url, status_code=status.HTTP_303_SEE_OTHER) @app.get("/auth/discord/callback") @app.get("/callback") @app.get("/auth/callback") @app.get("/discord/callback") @app.get("/login/callback") @app.get("/api/auth/callback/discord") async def discord_callback( request: Request, code: Optional[str] = None, error: Optional[str] = None, error_description: Optional[str] = None, db: Session = Depends(get_db) ): """ Point de retour (callback) OAuth2 Discord : Échange le code contre un profil utilisateur et connecte automatiquement l'opérateur. """ if error or not code: err_msg = error_description or error or "Connexion Discord annulée ou échouée." return RedirectResponse(url=f"/login?error={err_msg}", status_code=status.HTTP_303_SEE_OTHER) # Détection de l'URI de redirection utilisée lors de l'appel host = request.headers.get("host", "localhost:8000") scheme = "https" if ("mmi-velizy.fr" in host or request.url.scheme == "https") else "http" current_redirect_uri = f"{scheme}://{host}{request.url.path}" token_data = await exchange_discord_code(code, redirect_uri=current_redirect_uri) if not token_data and os.getenv("DISCORD_REDIRECT_URI"): token_data = await exchange_discord_code(code, redirect_uri=os.getenv("DISCORD_REDIRECT_URI")) if not token_data or "access_token" not in token_data: return RedirectResponse( url="/login?error=Impossible+de+valider+l'authentification+Discord.", status_code=status.HTTP_303_SEE_OTHER ) profile = await get_discord_user_profile(token_data["access_token"]) if not profile or "id" not in profile: return RedirectResponse( url="/login?error=Impossible+de+récupérer+votre+profil+Discord.", status_code=status.HTTP_303_SEE_OTHER ) discord_id = str(profile["id"]) discord_username = profile.get("global_name") or profile.get("username") or f"Discord_{discord_id[:6]}" avatar_hash = profile.get("avatar") avatar_url = f"https://cdn.discordapp.com/avatars/{discord_id}/{avatar_hash}.png" if avatar_hash else None # Recherche si l'utilisateur existe déjà par son Discord ID user = db.query(User).filter(User.discord_id == discord_id).first() if not user: # Recherche par nom d'utilisateur existant user = db.query(User).filter(func.lower(User.username) == discord_username.lower()).first() if user: # Association du compte existant avec Discord user.discord_id = discord_id if avatar_url: user.avatar_url = avatar_url db.commit() db.refresh(user) else: # Création d'un nouvel utilisateur lié à Discord user = User( username=discord_username, discord_id=discord_id, avatar_url=avatar_url, hashed_password=None ) db.add(user) db.commit() db.refresh(user) else: # Mise à jour des informations d'avatar / nom si besoin if avatar_url and user.avatar_url != avatar_url: user.avatar_url = avatar_url db.commit() db.refresh(user) # Création du cookie de session token = create_session_token(user.id, user.username, remember_me=True) response = RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER) set_session_cookie(response, token, request, remember_me=True) return response @app.post("/login", response_class=HTMLResponse) @limiter.limit("5/minute") async def login_post( request: Request, username: str = Form(...), password: str = Form(...), remember_me: Optional[str] = Form(None), redirect: Optional[str] = Form(None), db: Session = Depends(get_db), _csrf: bool = Depends(validate_csrf) ): """Traite la connexion et émet le cookie de session sécurisé.""" username_clean = username.strip() user = db.query(User).filter(func.lower(User.username) == username_clean.lower()).first() if not user or not user.hashed_password or not verify_password(password, user.hashed_password): return templates.TemplateResponse(request=request, name="login.html", context={ "request": request, "current_user": None, "error": "Identifiants invalides. Vérifiez votre pseudo et mot de passe.", "active_page": "login", "redirect_url": redirect or "", "discord_auth_url": URL_GENEREE }, status_code=status.HTTP_400_BAD_REQUEST) # Création du cookie de session sécurisé is_remember = remember_me in ("true", "1", "on", True) token = create_session_token(user.id, user.username, remember_me=is_remember) response = RedirectResponse(url=redirect or "/", status_code=status.HTTP_303_SEE_OTHER) set_session_cookie(response, token, request, remember_me=is_remember) return response @app.get("/register", response_class=HTMLResponse) async def register_get(request: Request, db: Session = Depends(get_db)): """Affiche le formulaire d'inscription.""" current_user = get_current_user(request, db) if current_user: return RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER) return templates.TemplateResponse(request=request, name="register.html", context={ "request": request, "current_user": None, "active_page": "register", "discord_auth_url": URL_GENEREE }) @app.post("/register", response_class=HTMLResponse) @limiter.limit("5/minute") async def register_post( request: Request, username: str = Form(...), password: str = Form(...), confirm_password: str = Form(...), db: Session = Depends(get_db), _csrf: bool = Depends(validate_csrf) ): """Crée un nouvel utilisateur et connecte automatiquement.""" username_clean = username.strip() if len(username_clean) < 3 or len(username_clean) > 30: return templates.TemplateResponse(request=request, name="register.html", context={ "request": request, "current_user": None, "error": "Le pseudo doit contenir entre 3 et 30 caractères.", "active_page": "register", "discord_auth_url": URL_GENEREE }, status_code=status.HTTP_400_BAD_REQUEST) if len(password) < 6: return templates.TemplateResponse(request=request, name="register.html", context={ "request": request, "current_user": None, "error": "Le mot de passe doit contenir au moins 6 caractères.", "active_page": "register", "discord_auth_url": URL_GENEREE }, status_code=status.HTTP_400_BAD_REQUEST) if password != confirm_password: return templates.TemplateResponse(request=request, name="register.html", context={ "request": request, "current_user": None, "error": "Les mots de passe ne correspondent pas.", "active_page": "register", "discord_auth_url": URL_GENEREE }, status_code=status.HTTP_400_BAD_REQUEST) existing = db.query(User).filter(func.lower(User.username) == username_clean.lower()).first() if existing: return templates.TemplateResponse(request=request, name="register.html", context={ "request": request, "current_user": None, "error": f"Le pseudo '{username_clean}' est déjà utilisé par un autre opérateur.", "active_page": "register", "discord_auth_url": URL_GENEREE }, status_code=status.HTTP_400_BAD_REQUEST) # Hachage et création de l'utilisateur hashed_pwd = hash_password(password) new_user = User(username=username_clean, hashed_password=hashed_pwd) db.add(new_user) db.commit() db.refresh(new_user) # Connexion automatique token = create_session_token(new_user.id, new_user.username, remember_me=True) response = RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER) set_session_cookie(response, token, request, remember_me=True) return response @app.get("/logout") async def logout(): """Déconnecte l'utilisateur en supprimant le cookie.""" response = RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER) delete_session_cookie(response) return response # ========================================== # ENDPOINTS API & HTMX DYNAMIQUES # ========================================== @app.get("/api/day-details", response_class=HTMLResponse) async def get_day_details( request: Request, date: str, db: Session = Depends(get_db) ): """ Renvoie le fragment HTML des créneaux pour une date donnée (chargé dans la modale Alpine.js). """ current_user = get_current_user(request, db) try: target_date = datetime.strptime(date, "%Y-%m-%d").date() except ValueError: raise HTTPException(status_code=400, detail="Format de date invalide (YYYY-MM-DD)") today = dt_date.today() is_past = (target_date < today) effective_data = calculate_effective_players_for_date(db, target_date) user_direct_slots = effective_data["user_direct_slots"].get(current_user.id, set()) if current_user else set() any_confirmed = False slots_data = [] for slot in VALID_SLOTS: slot_info = effective_data[slot] players = slot_info["usernames"] players_info = slot_info.get("players_info", []) count = len(players) if count >= PLAYER_THRESHOLD: any_confirmed = True slots_data.append({ "slot": slot, "count": count, "players": players, "players_info": players_info, "is_user_registered": (slot in user_direct_slots) }) context = { "request": request, "current_user": current_user, "target_date": target_date, "date_str": date, "french_date_str": format_french_date(target_date), "slots_data": slots_data, "any_confirmed": any_confirmed, "is_past": is_past } return templates.TemplateResponse(request=request, name="partials/day_modal.html", context=context) @app.get("/api/day-cell", response_class=HTMLResponse) async def get_day_cell( request: Request, date: str, db: Session = Depends(get_db) ): """ Renvoie le fragment HTML d'une case unique du calendrier (rafraîchissement ciblé par HTMX). """ current_user = get_current_user(request, db) try: target_date = datetime.strptime(date, "%Y-%m-%d").date() except ValueError: raise HTTPException(status_code=400, detail="Format de date invalide (YYYY-MM-DD)") today = dt_date.today() effective_data = calculate_effective_players_for_date(db, target_date) user_direct_slots = effective_data["user_direct_slots"].get(current_user.id, set()) if current_user else set() matin_count = len(effective_data[SLOT_MATIN]["usernames"]) aprem_count = len(effective_data[SLOT_APREM]["usernames"]) confirmed_games = ( db.query(Game) .filter(Game.date == target_date, Game.player_count >= PLAYER_THRESHOLD) .all() ) has_confirmed = (len(confirmed_games) > 0) or (matin_count >= PLAYER_THRESHOLD) or (aprem_count >= PLAYER_THRESHOLD) is_user_in_matin = bool(current_user and SLOT_MATIN in user_direct_slots) is_user_in_aprem = bool(current_user and SLOT_APREM in user_direct_slots) is_user_registered = is_user_in_matin or is_user_in_aprem slots_summary = [ { "slot": SLOT_MATIN, "count": matin_count, "is_user_in": is_user_in_matin }, { "slot": SLOT_APREM, "count": aprem_count, "is_user_in": is_user_in_aprem } ] day_info = { "date": target_date, "date_str": date, "day_num": target_date.day, "is_current_month": True, "is_today": (target_date == today), "is_past": (target_date < today), "has_confirmed_game": has_confirmed, "is_user_registered": is_user_registered, "slots_summary": slots_summary } return templates.TemplateResponse(request=request, name="partials/day_cell.html", context={ "request": request, "day_info": day_info }) @app.post("/api/availability/toggle", response_class=HTMLResponse) async def toggle_availability( request: Request, date: str = Form(...), slot: str = Form(...), loop_index: Optional[int] = Form(1), db: Session = Depends(get_db), _csrf: bool = Depends(validate_csrf) ): """ Bascule (Ajoute ou Retire) la disponibilité de l'utilisateur connecté pour une date et un créneau. Interdit toute modification sur une date passée. """ current_user = get_current_user(request, db) if not current_user: raise HTTPException(status_code=401, detail="Connexion requise") if slot not in VALID_SLOTS: raise HTTPException(status_code=400, detail="Créneau invalide") try: target_date = datetime.strptime(date, "%Y-%m-%d").date() except ValueError: raise HTTPException(status_code=400, detail="Format de date invalide") today = dt_date.today() if target_date < today: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Impossible de s'inscrire ou de se désister pour une date passée." ) # Vérifier si l'utilisateur est déjà inscrit existing_avail = db.query(Availability).filter( Availability.user_id == current_user.id, Availability.date == target_date, Availability.slot == slot ).first() if existing_avail: # Retrait de la disponibilité db.delete(existing_avail) db.commit() else: # Ajout de la disponibilité new_avail = Availability( user_id=current_user.id, date=target_date, slot=slot ) db.add(new_avail) db.commit() # Synchronisation des parties créées pour cette date et envoi Discord si >= 8 sync_result = await sync_and_trigger_games_for_date(db, target_date) effective_data = sync_result["effective_data"] user_direct_slots = effective_data["user_direct_slots"].get(current_user.id, set()) any_confirmed = False slots_data = [] for s in VALID_SLOTS: s_info = effective_data[s] s_players = s_info["usernames"] s_players_info = s_info.get("players_info", []) s_count = len(s_players) if s_count >= PLAYER_THRESHOLD: any_confirmed = True slots_data.append({ "slot": s, "count": s_count, "players": s_players, "players_info": s_players_info, "is_user_registered": (s in user_direct_slots) }) # Rendu du partial day_modal.html mis à jour pour HTMX avec trigger de rafraîchissement de la cellule context = { "request": request, "current_user": current_user, "target_date": target_date, "date_str": date, "french_date_str": format_french_date(target_date), "slots_data": slots_data, "any_confirmed": any_confirmed, "is_past": False } response = templates.TemplateResponse(request=request, name="partials/day_modal.html", context=context) response.headers["HX-Trigger"] = f"refreshDayCell_{date}" return response # ========================================== # GESTION PROPRE DES ERREURS # ========================================== @app.exception_handler(404) async def not_found_handler(request: Request, exc: HTTPException): """Page d'erreur 404 personnalisée.""" return HTMLResponse( content=""" 404 - QG Non Trouvé
404

COORDONNÉES INTROUVABLES

Le secteur demandé n'existe pas ou a été déplacé.

Retour au QG
""", status_code=404 ) if __name__ == "__main__": import uvicorn uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)