196 lines
7.8 KiB
Python
196 lines
7.8 KiB
Python
|
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))
|