import disnake from disnake.ext import commands import json import os from collections import deque from datetime import datetime, timedelta # Chemins des fichiers SETTINGS_FILE = "data/security_settings.json" WHITELIST_FILE = "data/whitelist.json" # Cache TTL en secondes (5 secondes pour avoir une réactivité acceptable) CACHE_TTL = 5 # Triggers pour détection de liens LINK_TRIGGERS = ("disnake.gg/", "http://", "https://", "www.") class AntiSpam(commands.Cog): def __init__(self, bot): self.bot = bot # Optimisation: deque avec maxlen pour O(1) au lieu de O(n) list filtering self.message_counts = {} self.spam_warnings = {} self.link_warnings = {} # Cache avec timestamps pour éviter la lecture fichier à chaque message self._settings_cache = {} # {guild_id: (timestamp, settings_dict)} self._whitelist_cache = {} # {guild_id: (timestamp, whitelist_set)} def _get_cached_settings(self, guild_id): """Récupère les paramètres avec cache (TTL de 5 secondes)""" now = disnake.utils.utcnow().timestamp() cached = self._settings_cache.get(guild_id) if cached: cached_time, settings = cached if now - cached_time < CACHE_TTL: return settings # Cache expiré ou absent - lire le fichier try: if os.path.exists(SETTINGS_FILE): with open(SETTINGS_FILE, "r", encoding='utf-8') as f: all_settings = json.load(f) settings = all_settings.get(str(guild_id), {}) else: settings = {} except: settings = {} self._settings_cache[guild_id] = (now, settings) return settings def _get_cached_whitelist(self, guild_id): """Récupère la whitelist avec cache (TTL de 5 secondes)""" now = disnake.utils.utcnow().timestamp() cached = self._whitelist_cache.get(guild_id) if cached: cached_time, whitelist_set = cached if now - cached_time < CACHE_TTL: return whitelist_set # Cache expiré ou absent - lire le fichier whitelist_set = set() try: if os.path.exists(WHITELIST_FILE): with open(WHITELIST_FILE, "r", encoding='utf-8') as f: data = json.load(f) server_id_str = str(guild_id) whitelist_ids = data.get(server_id_str, []) whitelist_set = set(str(uid) for uid in whitelist_ids) except: pass self._whitelist_cache[guild_id] = (now, whitelist_set) return whitelist_set def is_whitelisted(self, guild_id, user_id): """Vérifie l'immunité (Propriétaire ou Whitelist JSON avec cache)""" guild = self.bot.get_guild(guild_id) if guild and user_id == guild.owner_id: return True # Utilisation du cache whitelist whitelist = self._get_cached_whitelist(guild_id) return str(user_id) in whitelist @commands.Cog.listener() async def on_message(self, message): # 1. Filtres de base (early return pour performance) if message.author.bot or not message.guild: return guild_id = message.guild.id user_id = message.author.id # 2. VÉRIFICATION WHITELIST (Immunité totale) if self.is_whitelisted(guild_id, user_id): return # --- LECTURE CACHÉ (dynamique mais rapide) --- settings = self._get_cached_settings(guild_id) # --- TEST ANTI-LIENS --- # Bloque tout le monde sauf whitelist ET manage_messages anti_link_enabled = settings.get("anti_link", False) if settings else False has_perm = message.author.guild_permissions.manage_messages if anti_link_enabled and not has_perm: content_lower = message.content.lower() if any(trigger in content_lower for trigger in LINK_TRIGGERS): await self.handle_violation(message, "link") return # Sortir si pas de settings if not settings: return # --- TEST ANTI-SPAM (deque pour O(1)) --- if settings.get("anti_spam") is True: now = disnake.utils.utcnow().timestamp() # Initialiser ou récupérer le deque pour cet utilisateur if user_id not in self.message_counts: # maxlen=10 suffit pour la fenêtre de 5 secondes avec seuil typique self.message_counts[user_id] = deque(maxlen=10) user_deque = self.message_counts[user_id] user_deque.append(now) # Nettoyer les timestamps expirés (plus vieux que 5 secondes) # Le deque supprime automatiquement les anciens éléments quand il dépasse maxlen # Mais on doit aussi filtrer par temps pour être précis while user_deque and now - user_deque[0] >= 5: user_deque.popleft() threshold = settings.get("spam_threshold", 5) if len(user_deque) > threshold: await self.handle_violation(message, "spam") async def handle_violation(self, message, type_violation): user_id = message.author.id if type_violation == "link": try: await message.delete() except: pass self.link_warnings[user_id] = self.link_warnings.get(user_id, 0) + 1 current_warns = self.link_warnings[user_id] reason = "Envoi de liens interdits" else: try: await message.channel.purge(limit=5, check=lambda m: m.author == message.author) except: pass self.spam_warnings[user_id] = self.spam_warnings.get(user_id, 0) + 1 current_warns = self.spam_warnings[user_id] reason = "Spam excessif" # Réinitialiser le deque après spam détecté self.message_counts[user_id] = deque(maxlen=10) if current_warns <= 3: embed = disnake.Embed( title="🛡️ Sécurité Kuby", description=f"{message.author.mention}, {reason}.\n**Avertissement : {current_warns}/3**", color=0xFFA500 ) await message.channel.send(embed=embed, delete_after=5) else: await self.apply_mute(message, reason) if type_violation == "link": self.link_warnings[user_id] = 0 else: self.spam_warnings[user_id] = 0 async def apply_mute(self, message, reason): try: until = disnake.utils.utcnow() + timedelta(minutes=5) await message.author.timeout(until, reason=f"Kuby Security: {reason} (3 Warns)") embed = disnake.Embed( title="⛔ Exclusion Temporaire", description=f"**Membre :** {message.author.mention}\n**Durée :** 5 minutes\n**Raison :** {reason} répétitif.", color=0xFF0000 ) await message.channel.send(embed=embed) except Exception as e: print(f"Erreur lors de la sanction : {e}") def setup(bot): bot.add_cog(AntiSpam(bot))