import disnake from disnake.ext import commands import sqlite3 import aiohttp import io import os import logging import asyncio from PIL import Image, ImageDraw, ImageFont kuby_logger = logging.getLogger("KubyBot") class Welcome(commands.Cog): def __init__(self, bot): self.bot = bot # Base de données à la racine - Chemin absolu self.base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) self.db_path = os.path.join(self.base_dir, "config.db") self._init_db() def _init_db(self): conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS server_configs ( server_id INTEGER PRIMARY KEY, welcome_channel INTEGER, goodbye_channel INTEGER, welcome_dm_message TEXT, welcome_server_name TEXT, welcome_banner_background TEXT, welcome_channel_message TEXT ) ''') # Migration pour ajouter les colonnes si elles n'existent pas cursor.execute("PRAGMA table_info(server_configs)") columns = [col[1] for col in cursor.fetchall()] if 'welcome_server_name' not in columns: cursor.execute('ALTER TABLE server_configs ADD COLUMN welcome_server_name TEXT') if 'welcome_banner_background' not in columns: cursor.execute('ALTER TABLE server_configs ADD COLUMN welcome_banner_background TEXT') if 'welcome_channel_message' not in columns: cursor.execute('ALTER TABLE server_configs ADD COLUMN welcome_channel_message TEXT') # Migration: Conversion des chemins absolus en chemins relatifs pour les bannières cursor.execute("SELECT server_id, welcome_banner_background FROM server_configs WHERE welcome_banner_background IS NOT NULL") for server_id, path in cursor.fetchall(): if path and os.path.isabs(path): filename = os.path.basename(path) # Vérifier si c'est bien une bannière dans le dossier banners if "banner_" in filename: cursor.execute("UPDATE server_configs SET welcome_banner_background = ? WHERE server_id = ?", (filename, server_id)) conn.commit() conn.close() async def create_welcome_image(self, username, avatar_url, server_name=None, banner_background=None, custom_message=None): # Utiliser base_dir déjà calculé dans __init__ try: # Télécharger l'avatar async with aiohttp.ClientSession() as session: async with session.get(avatar_url) as resp: if resp.status != 200: kuby_logger.error(f"[Welcome] Impossible de télécharger l'avatar: {resp.status}") raise ValueError(f"Impossible de télécharger l'avatar (code: {resp.status})") avatar_bytes = await resp.read() # Vérifier et utiliser le fond personnalisé ou celui par défaut bg_path = None if banner_background: # Si c'est un chemin absolu (pour compatibilité temporaire) if os.path.isabs(banner_background) and os.path.exists(banner_background): bg_path = banner_background else: # Sinon on cherche dans le dossier banners/ potential_path = os.path.join(self.base_dir, "banners", banner_background) if os.path.exists(potential_path): bg_path = potential_path kuby_logger.debug(f"[Welcome] Fond personnalisé trouvé: {bg_path}") if not bg_path: default_bg = os.path.join(self.base_dir, 'background_template.png') if not os.path.exists(default_bg): kuby_logger.error(f"[Welcome] Fond par défaut introuvable: {default_bg}") raise FileNotFoundError(f"Fond par défaut introuvable: {default_bg}") bg_path = default_bg kuby_logger.debug(f"[Welcome] Utilisation du fond par défaut: {bg_path}") # Charger le fond background = Image.open(bg_path) draw = ImageDraw.Draw(background) # Charger les polices custom (comme sur EXEMPLE_ARRIVER.png) font_large_path = os.path.join(self.base_dir, 'custom_font_large.otf') font_small_path = os.path.join(self.base_dir, 'custom_font_small.otf') # Les 3 lignes de texte à afficher (format identique à l'exemple) line1_text = "BIENVENUE" line2_text = username.upper() line3_text = (server_name or "").upper() # Zone de texte disponible (à droite de l'avatar) text_area_left = 280 text_area_right = background.width - 20 text_area_width = text_area_right - text_area_left text_area_center_x = text_area_left + text_area_width // 2 # Fonction pour trouver la taille de police optimale def fit_font(font_path, text, max_size, max_width): size = max_size while size > 10: font = ImageFont.truetype(font_path, size) bbox = draw.textbbox((0, 0), text, font=font) w = bbox[2] - bbox[0] if w <= max_width: h = bbox[3] - bbox[1] return font, w, h size -= 2 font = ImageFont.truetype(font_path, 10) bbox = draw.textbbox((0, 0), text, font=font) return font, bbox[2] - bbox[0], bbox[3] - bbox[1] # Adapter chaque ligne à la zone disponible font1, w1, h1 = fit_font(font_large_path, line1_text, 65, text_area_width) font2, w2, h2 = fit_font(font_small_path, line2_text, 35, text_area_width) font3, w3, h3 = fit_font(font_large_path, line3_text, 65, text_area_width) # Espacement entre les lignes spacing_1_2 = 15 # entre BIENVENUE et username spacing_2_3 = 20 # entre username et nom du serveur total_height = h1 + spacing_1_2 + h2 + spacing_2_3 + h3 # Centrer verticalement le bloc de 3 lignes start_y = (background.height - total_height) // 2 # Dessiner chaque ligne centrée horizontalement dans la zone y = start_y draw.text((text_area_center_x - w1 // 2, y), line1_text, font=font1, fill=(255, 255, 255)) y += h1 + spacing_1_2 draw.text((text_area_center_x - w2 // 2, y), line2_text, font=font2, fill=(255, 255, 255)) y += h2 + spacing_2_3 draw.text((text_area_center_x - w3 // 2, y), line3_text, font=font3, fill=(255, 255, 255)) # Ajouter l'avatar avatar = Image.open(io.BytesIO(avatar_bytes)).convert("RGBA").resize((224, 224), Image.LANCZOS) mask = Image.new('L', avatar.size, 0) mask_draw = ImageDraw.Draw(mask) mask_draw.ellipse((12, 12, 212, 212), fill=255) background.paste(avatar, (40, 40), mask) # Retourner l'image img_byte_arr = io.BytesIO() background.save(img_byte_arr, format='PNG') img_byte_arr.seek(0) return img_byte_arr except Exception as e: kuby_logger.error(f"[Welcome] Erreur lors de la création de l'image: {e}", exc_info=True) return None @commands.Cog.listener() async def on_member_join(self, member): kuby_logger.info(f"📥 [Welcome] Événement on_member_join reçu pour {member.name} (ID: {member.id}) sur {member.guild.name}") welcome_channel = None self._init_db() try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute('SELECT welcome_channel, welcome_dm_message, welcome_server_name, welcome_banner_background, welcome_channel_message FROM server_configs WHERE server_id = ?', (member.guild.id,)) result = cursor.fetchone() conn.close() if not result: kuby_logger.debug(f"⚠️ [Welcome] Aucune configuration trouvée dans la base de données pour le serveur {member.guild.id}") return welcome_channel = result[0] welcome_dm_text = result[1] server_name = result[2] banner_background = result[3] welcome_channel_message = result[4] kuby_logger.debug(f"🔍 [Welcome] Config récupérée: DM={bool(welcome_dm_text)}, Channel={welcome_channel}") # 1. Gestion du DM if welcome_dm_text and not member.bot: kuby_logger.info(f"✉️ [Welcome] Tentative d'envoi de DM à {member.name}") try: template_values = { # Alias populaires "user": member.mention, "username": member.name, "member": member.mention, "member_name": member.name, "server": member.guild.name, "servername": member.guild.name, "guild": member.guild.name, "guild_name": member.guild.name, "member_count": member.guild.member_count, "count": member.guild.member_count, "members": member.guild.member_count, } class SafeDict(dict): def __missing__(self, key): return "{" + key + "}" formatted_message = welcome_dm_text.format_map(SafeDict(template_values)) # On combine le texte et le GIF dans un Embed élégant pour masquer le lien brut gif_url = "https://i.giphy.com/media/v1.Y2lkPTc5MGI3NjExNXV1NW02NTk0bGF2ZHcyMGFna2F6amRrbGpobGpsMHowOW5xa2E5eCZlcD12MV9pbnRlcm5hbF9naWZfYnlfaWQmY3Q9Zw/ZtFuW4rsUbfrRzPauJ/giphy.gif" embed = disnake.Embed( description=formatted_message, color=0x2b2d31 # Couleur sombre moderne et épurée ) embed.set_image(url=gif_url) await member.send(embed=embed) kuby_logger.info(f"✅ [Welcome] DM envoyé avec succès à {member.name} sous forme d'Embed") except disnake.Forbidden: kuby_logger.warning(f"🚫 [Welcome] Impossible d'envoyer le DM à {member.name} (DMs fermés)") except disnake.HTTPException as e: if e.code == 40003: # Rate limited (Opening DMs too fast) kuby_logger.warning(f"⏳ [Welcome] Limite de débit Discord atteinte pour les DMs : {member.name} n'a pas reçu son message.") elif e.code == 50007: # Cannot send messages to this user kuby_logger.warning(f"🚫 [Welcome] Impossible d'envoyer le DM à {member.name} (DMs fermés ou utilisateur non joignable)") else: kuby_logger.error(f"❌ [Welcome] Erreur HTTP lors de l'envoi du DM à {member.name}: {e}") except Exception as e: kuby_logger.error(f"❌ [Welcome] Erreur lors de l'envoi du DM à {member.name}: {e}", exc_info=True) else: kuby_logger.info(f"ℹ️ [Welcome] Aucun message de DM configuré pour le serveur {member.guild.id}") # 2. Gestion du salon de bienvenue if welcome_channel: channel = member.guild.get_channel(welcome_channel) if channel: try: # Créer l'image de bienvenue server_display_name = server_name if server_name else member.guild.name img = await self.create_welcome_image( member.name, member.avatar.url if member.avatar else member.default_avatar.url, server_name=server_display_name, banner_background=banner_background, custom_message=welcome_channel_message ) # Attendre un peu que le Cog d'invitations traite le join await asyncio.sleep(1) # Récupérer l'inviteur invites_cog = self.bot.get_cog("Invites") inviter_text = "" if invites_cog: try: inviter_id = invites_cog.get_inviter_info(member.guild.id, member.id) if inviter_id: inviter = member.guild.get_member(inviter_id) if not inviter: try: inviter = await self.bot.fetch_user(inviter_id) except (disnake.NotFound, disnake.HTTPException) as e: kuby_logger.warning(f"[Welcome] Impossible de récupérer l'inviteur {inviter_id}: {e}") inviter = None if inviter: total, current = invites_cog.get_invite_stats(member.guild.id, inviter_id) inviter_text = f"\n\n**Invité par :** {inviter.mention} ({current} invitations)" else: inviter_text = "\n\n**Invité par :** [Utilisateur introuvable]" except Exception as e: kuby_logger.error(f"[Welcome] Erreur lors de la récupération de l'inviteur: {e}", exc_info=True) inviter_text = "" # Envoyer le message avec le format exact requis if img: server_display_name = server_name if server_name else member.guild.name # IMPORTANT: transmettre aussi le nom réel du serveur à Pillow pour les placeholders {server} # (utile quand welcome_server_name n'est pas configuré) # (server_name est utilisé par create_welcome_image) embed = disnake.Embed( description=f"Bienvenue sur {server_display_name} {member.name} ! Grâce à toi, nous avons atteint les {member.guild.member_count} membres !{inviter_text}", color=0x9B59B6 ) embed.set_image(url="attachment://welcome.png") await channel.send(embed=embed, file=disnake.File(img, filename="welcome.png")) kuby_logger.info(f"[Welcome] Message de bienvenue avec image envoyé pour {member.name}") else: # Fallback : image non générée, envoyer un message texte kuby_logger.warning(f"[Welcome] Image non générée pour {member.name}, envoi d'un message texte") server_display_name = server_name if server_name else member.guild.name await channel.send(f"Bienvenue sur {server_display_name} {member.mention} ! Grâce à toi, nous avons atteint les {member.guild.member_count} membres !{inviter_text}") except Exception as e: kuby_logger.error(f"[Welcome] Erreur lors de l'envoi du message dans le salon: {e}", exc_info=True) # Fallback avec un message minimal try: await channel.send(f"{member.mention} Bienvenue dans le serveur !") kuby_logger.info(f"[Welcome] Message de fallback envoyé pour {member.name}") except Exception as e2: kuby_logger.error(f"[Welcome] Échec du message de fallback: {e2}", exc_info=True) else: kuby_logger.warning(f"[Welcome] Salon de bienvenue introuvable (ID: {welcome_channel}) pour {member.guild.name}") else: kuby_logger.debug(f"[Welcome] Aucun salon de bienvenue configuré pour {member.guild.name}") except Exception as e: kuby_logger.error(f"❌ [Welcome] Erreur critique dans on_member_join: {e}", exc_info=True) async def save_banner_attachment(self, attachment: disnake.Attachment, guild_id: int) -> str: """Sauvegarde l'attachment et retourne le chemin du fichier""" base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "banners")) os.makedirs(base_dir, exist_ok=True) # Vérifier que c'est une image if not attachment.content_type or not attachment.content_type.startswith('image/'): raise ValueError("Le fichier doit être une image.") # Vérifier le format PNG if not attachment.filename.lower().endswith('.png'): raise ValueError("Le fichier doit être au format PNG.") # Vérifier les dimensions banner_path = os.path.join(base_dir, f"banner_{guild_id}.png") async with aiohttp.ClientSession() as session: async with session.get(attachment.url) as resp: if resp.status == 200: # Vérifier les dimensions avant de sauvegarder img_data = await resp.read() from io import BytesIO with Image.open(BytesIO(img_data)) as img: width, height = img.size if width != 1000 or height != 300: raise ValueError(f"Les dimensions doivent être 1000x300 pixels (votre image: {width}x{height})") with open(banner_path, 'wb') as f: f.write(img_data) return os.path.basename(banner_path) @commands.slash_command(name="setwelcome", description="Définit le salon, les messages de bienvenue et le fond de la bannière") @commands.has_permissions(administrator=True) async def set_welcome(self, interaction: disnake.ApplicationCommandInteraction, channel: disnake.TextChannel, message_dm: str, message_salon: str, server_name: str, banner_file: disnake.Attachment = None): # ✅ DÉFÉRER LA RÉPONSE (le traitement peut prendre > 3s avec les images) await interaction.response.defer(ephemeral=True) banner_path = None error_message = None # Si un fichier est uploadé, le sauvegarder if banner_file: try: banner_path = await self.save_banner_attachment(banner_file, interaction.guild_id) except Exception as e: error_message = f"❌ Erreur avec le fichier: {e}" # Si pas d'erreur, sauvegarder en base de données if not error_message: try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute('''INSERT INTO server_configs (server_id, welcome_channel, welcome_dm_message, welcome_server_name, welcome_banner_background, welcome_channel_message) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(server_id) DO UPDATE SET welcome_channel=excluded.welcome_channel, welcome_dm_message=excluded.welcome_dm_message, welcome_server_name=excluded.welcome_server_name, welcome_banner_background=COALESCE(excluded.welcome_banner_background, welcome_banner_background), welcome_channel_message=excluded.welcome_channel_message''', (interaction.guild_id, channel.id, message_dm, server_name, banner_path, message_salon)) conn.commit() conn.close() bg_used = "background_template.png" if not banner_path else (banner_path.split('/')[-1]) error_message = f"✅ Configuration de bienvenue enregistrée !\nFond utilisé: {bg_used}" except Exception as e: error_message = f"❌ Erreur lors de la sauvegarde en base de données: {e}" # Envoyer la réponse finale (déferrée) await interaction.followup.send(error_message, ephemeral=True) @commands.slash_command(name="invitations", description="Affiche vos statistiques d'invitations") async def invitations_stats(self, interaction: disnake.ApplicationCommandInteraction, user: disnake.Member = None): user = user or interaction.author await self._send_invitation_stats(interaction, user) @commands.user_command(name="Stats Invitations") async def invitations_user(self, interaction: disnake.UserCommandInteraction, user: disnake.Member): await self._send_invitation_stats(interaction, user) async def _send_invitation_stats(self, interaction, user): invites_cog = self.bot.get_cog("Invites") if not invites_cog: return await interaction.response.send_message("❌ Le module d'invitations est désactivé.", ephemeral=True) total, current = invites_cog.get_invite_stats(interaction.guild.id, user.id) components = [ disnake.ui.Container( disnake.ui.Section( f"Statistiques d'invitations - {user.display_name}", accessory=disnake.ui.Thumbnail(user.display_avatar.url) ), disnake.ui.Separator(divider=True), disnake.ui.TextDisplay(f"➜ Total d'invitations : **{total}**"), disnake.ui.TextDisplay(f"➜ Membres présents : **{current}**") ) ] await interaction.response.send_message(components=components) def setup(bot): bot.add_cog(Welcome(bot))