Kuby/commandes/modules_security/antispam.py

import disnake
from disnake.ext import commands
import json
import os
from collections import deque
from datetime import datetime, timedelta

# Chemins ABSOLUS
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.path.join(BASE_DIR, "..", "..", "data")
SETTINGS_FILE = os.path.join(DATA_DIR, "security_settings.json")
WHITELIST_FILE = os.path.join(DATA_DIR, "whitelist.json")

# Créer le dossier data/ s'il n'existe pas
os.makedirs(DATA_DIR, exist_ok=True)

# 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))