""" Module d'intégration Discord Webhook pour Versailles Airsoft. Envoie des alertes tactiques au format Discord Embed lors de la confirmation d'une partie (8+ joueurs). """ import os import logging from datetime import date, datetime from typing import List, Optional import httpx from dotenv import load_dotenv load_dotenv() logger = logging.getLogger("airsoft.discord") DISCORD_WEBHOOK_URL = os.getenv("DISCORD_WEBHOOK_URL", "").strip() FIELD_NAME = os.getenv("FIELD_NAME", "Terrain Privé de Versailles") PLAYER_THRESHOLD = int(os.getenv("PLAYER_THRESHOLD", "8")) # Configuration Discord OAuth2 URL_GENEREE = os.getenv("URL_Generee") or os.getenv("URL_GENAREE") or os.getenv("DISCORD_AUTH_URL", "URL_Generee") DISCORD_CLIENT_ID = os.getenv("DISCORD_CLIENT_ID", "").strip() DISCORD_CLIENT_SECRET = os.getenv("DISCORD_CLIENT_SECRET", "").strip() DISCORD_REDIRECT_URI = os.getenv("DISCORD_REDIRECT_URI", "http://localhost:8000/auth/discord/callback").strip() # Traduction des mois et jours en français pour un affichage propre JOURS_FR = ["Lundi", "Mardi", "Mercredi", "Jeudi", "Vendredi", "Samedi", "Dimanche"] MOIS_FR = [ "", "Janvier", "Février", "Mars", "Avril", "Mai", "Juin", "Juillet", "Août", "Septembre", "Octobre", "Novembre", "Décembre" ] async def exchange_discord_code(code: str, redirect_uri: Optional[str] = None) -> Optional[dict]: """Échange le code d'autorisation OAuth2 contre un jeton d'accès Discord.""" client_id = os.getenv("DISCORD_CLIENT_ID", "").strip() or DISCORD_CLIENT_ID client_secret = os.getenv("DISCORD_CLIENT_SECRET", "").strip() or DISCORD_CLIENT_SECRET used_redirect_uri = redirect_uri or os.getenv("DISCORD_REDIRECT_URI", "").strip() or DISCORD_REDIRECT_URI or "http://localhost:8000/auth/discord/callback" if not client_id or not client_secret: logger.warning("[Discord OAuth] DISCORD_CLIENT_ID ou DISCORD_CLIENT_SECRET manquant pour l'échange automatique.") return None token_url = "https://discord.com/api/oauth2/token" data = { "grant_type": "authorization_code", "code": code, "redirect_uri": used_redirect_uri } headers = {"Content-Type": "application/x-www-form-urlencoded"} try: async with httpx.AsyncClient(timeout=10.0) as client: # 1. Tentative avec HTTP Basic Auth (Recommandé Discord RFC 6749) resp = await client.post( token_url, data=data, headers=headers, auth=(client_id, client_secret) ) if resp.status_code == 200: return resp.json() # 2. Fallback avec client_id & client_secret dans le corps de la requête data_with_creds = {**data, "client_id": client_id, "client_secret": client_secret} resp2 = await client.post(token_url, data=data_with_creds, headers=headers) if resp2.status_code == 200: return resp2.json() logger.warning( f"[Discord OAuth] Échec échange token ({resp.status_code}): {resp.text}. " "Vérifiez que DISCORD_CLIENT_SECRET dans .env est bien le Client Secret (onglet OAuth2) " "et non la Clé Publique (Public Key de l'onglet General Information)." ) return None except Exception as e: logger.error(f"[Discord OAuth] Erreur lors de l'échange de token: {e}") return None async def get_discord_user_profile(access_token: str) -> Optional[dict]: """Récupère le profil de l'utilisateur connecté via l'API Discord.""" user_url = "https://discord.com/api/users/@me" headers = {"Authorization": f"Bearer {access_token}"} try: async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.get(user_url, headers=headers) if resp.status_code == 200: return resp.json() else: logger.warning(f"[Discord OAuth] Échec récupération profil ({resp.status_code}): {resp.text}") return None except Exception as e: logger.error(f"[Discord OAuth] Erreur récupération profil: {e}") return None def format_french_date(d: date) -> str: """Formate une date en chaîne lisible en français.""" jour_nom = JOURS_FR[d.weekday()] mois_nom = MOIS_FR[d.month] return f"{jour_nom} {d.day} {mois_nom} {d.year}" async def send_discord_game_notification( target_date: date, slot: str, player_count: int, players: Optional[List[str]] = None, webhook_url: Optional[str] = None ) -> bool: """ Envoie un webhook Discord avec un Embed lorsqu'un créneau est confirmé (PLAYER_THRESHOLD+ joueurs). Retourne True si le message a été envoyé avec succès, False sinon. """ url = webhook_url or DISCORD_WEBHOOK_URL if not url: logger.info("[Discord] Aucun DISCORD_WEBHOOK_URL configuré. Notification ignorée.") return False date_str = format_french_date(target_date) # Construction de l'Embed Discord embed = { "title": f"⚜️ PARTIE D'AIRSOFT CONFIRMÉE ! ({PLAYER_THRESHOLD}+ JOUEURS)", "description": f"Le seuil de **{PLAYER_THRESHOLD} opérateurs** a été atteint pour ce créneau !", "color": 0xD4AF37, # Or de Versailles / Blason Royal "fields": [ { "name": "📅 Date de l'opération", "value": f"**{date_str}**", "inline": True }, { "name": "⏱️ Créneau Horaire", "value": f"**{slot}**", "inline": True }, { "name": "👥 Effectif", "value": f"**{player_count} opérateurs inscrits**", "inline": True }, { "name": "📍 Localisation", "value": f"**{FIELD_NAME}**", "inline": True } ], "footer": { "text": "Versailles Airsoft • Message Automatique", "icon_url": "https://img.icons8.com/color/96/shield.png" }, "timestamp": datetime.utcnow().isoformat() + "Z" } payload = { "username": "Versailles Airsoft HQ", "avatar_url": "https://img.icons8.com/color/96/military-badge.png", "content": f"🚨 Partie confirmée pour le **{date_str}** sur le créneau **{slot}** ({player_count} opérateurs) !", "embeds": [embed] } try: async with httpx.AsyncClient(timeout=10.0) as client: response = await client.post(url, json=payload) if response.status_code in (200, 204): logger.info(f"[Discord] Webhook envoyé avec succès pour {date_str} ({slot})") return True else: logger.warning(f"[Discord] Échec envoi webhook: code {response.status_code}, réponse: {response.text}") return False except Exception as e: logger.error(f"[Discord] Erreur lors de l'appel au webhook Discord: {e}") return False async def send_discord_game_cancellation_notification( target_date: date, slot: str, player_count: int, webhook_url: Optional[str] = None ) -> bool: """ Envoie un webhook Discord avec un Embed lorsqu'une partie confirmée repasse sous le seuil (< PLAYER_THRESHOLD joueurs). Retourne True si le message a été envoyé avec succès, False sinon. """ url = webhook_url or DISCORD_WEBHOOK_URL if not url: logger.info("[Discord] Aucun DISCORD_WEBHOOK_URL configuré. Notification d'annulation ignorée.") return False date_str = format_french_date(target_date) # Construction de l'Embed Discord d'annulation embed = { "title": "⚠️ PARTIE D'AIRSOFT ANNULÉE / EN ATTENTE", "description": f"Suite à des désistements, le créneau ne compte plus que **{player_count} opérateur(s)** (seuil requis : {PLAYER_THRESHOLD}). La partie est mise en attente.", "color": 0xE74C3C, # Rouge alerte "fields": [ { "name": "📅 Date de l'opération", "value": f"**{date_str}**", "inline": True }, { "name": "⏱️ Créneau Horaire", "value": f"**{slot}**", "inline": True }, { "name": "👥 Effectif actuel", "value": f"**{player_count} / {PLAYER_THRESHOLD} joueurs**", "inline": True }, { "name": "📍 Localisation", "value": f"**{FIELD_NAME}**", "inline": True } ], "footer": { "text": "Versailles Airsoft • Message Automatique", "icon_url": "https://img.icons8.com/color/96/shield.png" }, "timestamp": datetime.utcnow().isoformat() + "Z" } payload = { "username": "Versailles Airsoft HQ", "avatar_url": "https://img.icons8.com/color/96/military-badge.png", "content": f"⚠️ **Alerte : Partie annulée / en attente** pour le **{date_str}** sur le créneau **{slot}** (effectif insuffisant : {player_count}/{PLAYER_THRESHOLD} joueurs).", "embeds": [embed] } try: async with httpx.AsyncClient(timeout=10.0) as client: response = await client.post(url, json=payload) if response.status_code in (200, 204): logger.info(f"[Discord] Webhook d'annulation envoyé avec succès pour {date_str} ({slot})") return True else: logger.warning(f"[Discord] Échec envoi webhook annulation: code {response.status_code}, réponse: {response.text}") return False except Exception as e: logger.error(f"[Discord] Erreur lors de l'appel au webhook Discord (annulation): {e}") return False