""" Versailles Airsoft - Application Web Principale (FastAPI) Plateforme de gestion et d'organisation de parties d'airsoft sur terrain privé à Versailles. """ import os import calendar from datetime import date, datetime, timedelta from typing import Optional, List from fastapi import ( FastAPI, Request, Depends, Form, HTTPException, status, Response ) 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 database import engine, Base, get_db, init_db from models import User, Availability, Game, VALID_SLOTS from auth import ( hash_password, verify_password, create_session_token, get_current_user, COOKIE_NAME, SESSION_MAX_AGE ) from discord import ( send_discord_game_notification, send_discord_game_cancellation_notification, format_french_date, MOIS_FR, JOURS_FR ) # Initialisation de la base de données SQLite init_db() # Application FastAPI app = FastAPI( title="Versailles Airsoft HQ", description="Plateforme tactique d'organisation de parties d'airsoft sur terrain privé", version="1.0.0" ) # 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 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 for a in avails: if a.user: user_names[a.user_id] = a.user.username 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 } return { SLOT_MATIN: { "uids": matin_uids, "usernames": [user_names[uid] for uid in matin_uids if uid in user_names], }, SLOT_APREM: { "uids": aprem_uids, "usernames": [user_names[uid] for uid in aprem_uids if uid in user_names], }, "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) .filter(Availability.date >= start_date, Availability.date <= end_date) .all() ) # Regroupement par date : date -> dict(user_id -> set(slots)) date_user_slots = {} for a in availabilities: 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 slots_summary.append({ "slot": slot, "count": count, "is_user_in": is_user_in }) 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) # ========================================== # AUTHENTIFICATION (LOGIN / REGISTER / LOGOUT) # ========================================== @app.get("/login", response_class=HTMLResponse) async def login_get( request: Request, redirect: 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 "" }) @app.post("/login", response_class=HTMLResponse) async def login_post( request: Request, username: str = Form(...), password: str = Form(...), redirect: Optional[str] = Form(None), db: Session = Depends(get_db) ): """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 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 "" }, status_code=status.HTTP_400_BAD_REQUEST) # Création du cookie de session token = create_session_token(user.id, user.username) response = RedirectResponse(url=redirect or "/", status_code=status.HTTP_303_SEE_OTHER) response.set_cookie( key=COOKIE_NAME, value=token, max_age=SESSION_MAX_AGE, httponly=True, samesite="lax", secure=False # Compatible HTTP local et HTTPS ) 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" }) @app.post("/register", response_class=HTMLResponse) async def register_post( request: Request, username: str = Form(...), password: str = Form(...), confirm_password: str = Form(...), db: Session = Depends(get_db) ): """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" }, 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" }, 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" }, 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" }, 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) response = RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER) response.set_cookie( key=COOKIE_NAME, value=token, max_age=SESSION_MAX_AGE, httponly=True, samesite="lax", secure=False ) 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) response.delete_cookie(COOKIE_NAME) 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 = date_cls.today() if 'date_cls' in locals() else datetime.now().date() 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.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) ): """ 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 = datetime.now().date() 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 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 } return templates.TemplateResponse(request=request, name="partials/day_modal.html", context=context) # ========================================== # 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)