301 lines
16 KiB
Python
301 lines
16 KiB
Python
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
|
||
self.db_path = os.path.join(os.path.dirname(__file__), "..", "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):
|
||
current_dir = os.path.dirname(__file__)
|
||
base_dir = os.path.abspath(os.path.join(current_dir, ".."))
|
||
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.get(avatar_url) as resp:
|
||
avatar_bytes = await resp.read()
|
||
|
||
# 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(base_dir, "banners", banner_background)
|
||
if os.path.exists(potential_path):
|
||
bg_path = potential_path
|
||
|
||
if bg_path:
|
||
background = Image.open(bg_path)
|
||
else:
|
||
background = Image.open(os.path.join(base_dir, 'background_template.png'))
|
||
|
||
draw = ImageDraw.Draw(background)
|
||
|
||
font_large = ImageFont.truetype(os.path.join(base_dir, 'custom_font_large.otf'), 70)
|
||
font_small = ImageFont.truetype(os.path.join(base_dir, 'custom_font_small.otf'), 50)
|
||
|
||
draw.text((325, 70), "Bienvenue", font=font_large, fill=(255,255,255))
|
||
draw.text((325, 135), f"{username}", font=font_small, fill=(255,255,255))
|
||
|
||
# Afficher le nom du serveur uniquement si configuré
|
||
if server_name:
|
||
font_server = ImageFont.truetype(os.path.join(base_dir, 'custom_font_large.otf'), 55)
|
||
draw.text((325, 195), server_name, font=font_server, fill=(255,255,255))
|
||
|
||
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)
|
||
|
||
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:
|
||
logging.error(f"Erreur image: {e}")
|
||
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}")
|
||
|
||
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.warning(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:
|
||
formatted_message = welcome_dm_text.format(
|
||
username=member.name,
|
||
servername=member.guild.name,
|
||
member_count=member.guild.member_count
|
||
)
|
||
# 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}")
|
||
except Exception as e:
|
||
kuby_logger.error(f"❌ [Welcome] Erreur critique dans on_member_join: {e}", exc_info=True)
|
||
|
||
# 2. Gestion du salon de bienvenue
|
||
if welcome_channel:
|
||
channel = member.guild.get_channel(welcome_channel)
|
||
if channel:
|
||
img = await self.create_welcome_image(
|
||
member.name,
|
||
member.avatar.url if member.avatar else member.default_avatar.url,
|
||
server_name=server_name,
|
||
banner_background=banner_background
|
||
)
|
||
if img:
|
||
# Attendre un peu que le Cog d'invitations traite le join
|
||
await asyncio.sleep(1)
|
||
invites_cog = self.bot.get_cog("Invites")
|
||
inviter_text = ""
|
||
if invites_cog:
|
||
inviter_id = invites_cog.get_inviter_info(member.guild.id, member.id)
|
||
if inviter_id:
|
||
inviter = member.guild.get_member(inviter_id) or await self.bot.fetch_user(inviter_id)
|
||
total, current = invites_cog.get_invite_stats(member.guild.id, inviter_id)
|
||
inviter_text = f"\n\n**Invité par :** {inviter.mention} ({current} invitations)"
|
||
|
||
# Utiliser le message personnalisé ou un message par défaut
|
||
channel_description = (welcome_channel_message.format(
|
||
username=member.name,
|
||
servername=member.guild.name,
|
||
member_count=member.guild.member_count
|
||
) if welcome_channel_message else "Bienvenue dans le serveur !\nSi vous cherchez des RolePlay de qualité,\nVous êtes au bon endroit. <a:sip:1316891821858619452>") + inviter_text
|
||
|
||
components = [
|
||
disnake.ui.Container(
|
||
disnake.ui.TextDisplay(f"Merci d'accueillir {member} 🤝"),
|
||
disnake.ui.Separator(divider=True),
|
||
disnake.ui.TextDisplay(channel_description),
|
||
disnake.ui.MediaGallery(disnake.MediaGalleryItem(url="attachment://welcome.png"))
|
||
)
|
||
]
|
||
await channel.send(content=member.mention, components=components, file=disnake.File(img, filename="welcome.png"))
|
||
|
||
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):
|
||
banner_path = 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:
|
||
await interaction.response.send_message(f"❌ Erreur avec le fichier: {e}", ephemeral=True)
|
||
return
|
||
|
||
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])
|
||
await interaction.response.send_message(f"✅ Configuration de bienvenida enregistrée !\nFond utilisé: {bg_used}", 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))
|
||
|