import disnake from disnake.ext import commands import sqlite3 import os import logging from typing import Dict, Optional kuby_logger = logging.getLogger("KubyBot") class Invites(commands.Cog): def __init__(self, bot): self.bot = bot self.db_path = os.path.join(os.path.dirname(__file__), "..", "config.db") self.invite_cache = {} # {guild_id: {code: uses}} self._init_db() def _init_db(self): conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS member_invites ( guild_id INTEGER, member_id INTEGER, inviter_id INTEGER, invite_code TEXT, PRIMARY KEY (guild_id, member_id) ) ''') cursor.execute(''' CREATE TABLE IF NOT EXISTS invite_configs ( guild_id INTEGER PRIMARY KEY, log_channel_id INTEGER ) ''') conn.commit() conn.close() async def update_invite_cache(self, guild): try: invites = await guild.invites() self.invite_cache[guild.id] = {invite.code: invite.uses for invite in invites} except disnake.Forbidden: pass @commands.Cog.listener() async def on_ready(self): for guild in self.bot.guilds: await self.update_invite_cache(guild) kuby_logger.info("✅ [Invites] Cache des invitations initialisé") # Debug: Lister toutes les commandes slash enregistrées slash_cmds = [cmd.name for cmd in self.bot.slash_commands] kuby_logger.info(f"🔍 [Debug] Commandes slash enregistrées : {slash_cmds}") @commands.Cog.listener() async def on_invite_create(self, invite): if invite.guild.id not in self.invite_cache: self.invite_cache[invite.guild.id] = {} self.invite_cache[invite.guild.id][invite.code] = invite.uses @commands.Cog.listener() async def on_invite_delete(self, invite): if invite.guild.id in self.invite_cache: self.invite_cache[invite.guild.id].pop(invite.code, None) @commands.Cog.listener() async def on_member_join(self, member): if member.bot: return guild = member.guild old_invites = self.invite_cache.get(guild.id, {}) try: new_invites = await guild.invites() except disnake.Forbidden: return used_invite = None for invite in new_invites: if invite.code in old_invites: if invite.uses > old_invites[invite.code]: used_invite = invite break elif invite.uses > 0: # C'est une nouvelle invitation qui a été utilisée used_invite = invite break # Mise à jour du cache self.invite_cache[guild.id] = {invite.code: invite.uses for invite in new_invites} if used_invite: inviter = used_invite.inviter if inviter: self.record_invite(guild.id, member.id, inviter.id, used_invite.code) kuby_logger.info(f"📥 [Invites] {member} invité par {inviter} ({used_invite.code})") # Modular join message conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute('SELECT log_channel_id FROM invite_configs WHERE guild_id = ?', (guild.id,)) result = cursor.fetchone() conn.close() if result: log_channel = guild.get_channel(result[0]) if log_channel: # Determine method method = "invitation standard" if guild.vanity_url_code and used_invite.code == guild.vanity_url_code: method = "lien personnalisé" elif used_invite.max_age == 0: method = "invitation permanente" elif used_invite.temporary: method = "invitation temporaire" elif used_invite.max_uses == 1: method = "invitation à usage unique" components = [ disnake.ui.Container( disnake.ui.Section( f"Nouveau membre via invitation", accessory=disnake.ui.Thumbnail(member.display_avatar.url) ), disnake.ui.Separator(divider=True), disnake.ui.TextDisplay(f"👤 {member.mention} vient de nous rejoindre !"), disnake.ui.TextDisplay(f"🤝 Grâce à {inviter.mention}"), disnake.ui.TextDisplay(f"🔗 Via **{method}**") ) ] await log_channel.send(components=components) def record_invite(self, guild_id, member_id, inviter_id, code): conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' INSERT INTO member_invites (guild_id, member_id, inviter_id, invite_code) VALUES (?, ?, ?, ?) ON CONFLICT(guild_id, member_id) DO UPDATE SET inviter_id=excluded.inviter_id, invite_code=excluded.invite_code ''', (guild_id, member_id, inviter_id, code)) conn.commit() conn.close() def get_inviter_info(self, guild_id, member_id): conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute('SELECT inviter_id FROM member_invites WHERE guild_id = ? AND member_id = ?', (guild_id, member_id)) result = cursor.fetchone() conn.close() return result[0] if result else None def get_invite_stats(self, guild_id, user_id): conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute('SELECT member_id FROM member_invites WHERE guild_id = ? AND inviter_id = ?', (guild_id, user_id)) invited_ids = [row[0] for row in cursor.fetchall()] conn.close() guild = self.bot.get_guild(guild_id) if not guild: return 0, 0 current_present = 0 for mid in invited_ids: if guild.get_member(mid): current_present += 1 return len(invited_ids), current_present @commands.slash_command(name="setinvitechannel", description="Définit le salon des logs d'invitations") @commands.has_permissions(administrator=True) async def set_invite_channel(self, interaction: disnake.ApplicationCommandInteraction, channel: disnake.TextChannel): conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' INSERT INTO invite_configs (guild_id, log_channel_id) VALUES (?, ?) ON CONFLICT(guild_id) DO UPDATE SET log_channel_id=excluded.log_channel_id ''', (interaction.guild_id, channel.id)) conn.commit() conn.close() components = [ disnake.ui.Container( disnake.ui.TextDisplay(f"✅ Salon des logs d'invitations défini sur {channel.mention}") ) ] await interaction.response.send_message(components=components, ephemeral=True) @commands.command(name="invitations_debug") async def invites_prefix(self, ctx, user: disnake.Member = None): user = user or ctx.author total, current = self.get_invite_stats(ctx.guild.id, user.id) await ctx.send(f"📊 **{user.display_name}** : {total} total, {current} présents.") def setup(bot): bot.add_cog(Invites(bot))