""" Cog Tâches - Système de gestion des tâches modulaire pour Kuby Utilise disnake >= 2.11 avec Components V2 """ import disnake import re from disnake.ext import commands, tasks from disnake import ui from datetime import datetime from zoneinfo import ZoneInfo from typing import Optional, Dict, List, Any from utils.gestion_taches import ( ajouter_tache, lister_taches_utilisateur, lister_toutes_taches, get_tache, update_tache_status, update_tache_priorite, supprimer_tache, taches_proches, get_taches_settings, set_taches_settings, PRIORITES, STATUTS ) from src.logger import kuby_logger # ============================================================================ # MODALS # ============================================================================ class NouvelleTacheModal(ui.Modal): """Modal pour créer une nouvelle tâche.""" def __init__(self, guild_id: int): self.guild_id = guild_id # Create components list - Modals can only contain TextInput components components = [ ui.TextInput( label="Titre", custom_id="tache_titre", placeholder="Ex: Corriger le bug du login", max_length=100, required=True ), ui.TextInput( label="Description", custom_id="tache_description", style=disnake.TextInputStyle.paragraph, placeholder="Décrivez la tâche en détail...", max_length=1000, required=True ), ui.TextInput( label="ID de l'utilisateur assigné (ou mention)", custom_id="tache_assigne", placeholder="Ex: 123456789 ou @utilisateur", max_length=100, required=True ), ui.TextInput( label="Date d'échéance (JJ/MM/AAAA)", custom_id="tache_echeance", placeholder="31/12/2026", required=True ), ui.TextInput( label="Priorité", custom_id="tache_priorite", placeholder=f"Une de: {', '.join(PRIORITES)}", max_length=20, required=False ) ] super().__init__(title="📌 Nouvelle Tâche", custom_id="nouvelle_tache_modal", components=components) async def callback(self, inter: disnake.ModalInteraction) -> None: """Callback quand le modal est soumis.""" try: titre = inter.text_values["tache_titre"] description = inter.text_values["tache_description"] assigne_input = inter.text_values["tache_assigne"] echeance_str = inter.text_values["tache_echeance"] priorite = inter.text_values.get("tache_priorite", "Moyenne") # Parse assigne_id from input (could be ID or mention) mention_match = re.match(r'<@(\d+)>', assigne_input) if mention_match: assigne_id = int(mention_match.group(1)) else: try: assigne_id = int(assigne_input) except ValueError: await inter.response.send_message( "❌ ID utilisateur invalide. Veuillez entrer un ID numérique ou une mention valide.", ephemeral=True ) return # Validate priority if priorite and priorite not in PRIORITES: priorite = "Moyenne" try: echeance_date = datetime.strptime(echeance_str, "%d/%m/%Y") echeance = echeance_date.strftime("%Y-%m-%d") except ValueError: return await inter.response.send_message( "❌ Format de date invalide. Utilisez **JJ/MM/AAAA**.", ephemeral=True ) tache_id = await ajouter_tache( titre=titre, description=description, assigne_id=assigne_id, priorite=priorite, echeance=echeance, cree_par_id=inter.author.id, serveur_id=self.guild_id ) await inter.response.send_message( f"✅ **Tâche #{tache_id} créée avec succès !**\n" f"→ Assignée à : <@{assigne_id}>\n" f"→ Échéance : {echeance_str}", ephemeral=True ) try: assigne = inter.bot.get_user(assigne_id) or await inter.bot.fetch_user(assigne_id) if assigne: await assigne.send( f"📌 **Nouvelle tâche assignée** par {inter.author.mention} :\n\n" f"**#{tache_id} - {titre}**\n" f"📅 Échéance : {echeance_str}\n" f"⚠️ Priorité : {priorite}" ) except Exception: pass except Exception as e: kuby_logger.error(f"Erreur création tâche: {e}") await inter.response.send_message( "❌ Une erreur est survenue. Veuillez réessayer.", ephemeral=True ) class ConfirmDeleteModal(ui.Modal): """Modal de confirmation pour la suppression.""" def __init__(self, tache_id: str, tache_titre: str): self.tache_id = tache_id self.tache_titre = tache_titre components = [ ui.TextInput( label=f'Pour confirmer, tapez : "{tache_titre}"', custom_id="confirm_text", placeholder=f"Tapez exactement : {tache_titre}", required=True ) ] super().__init__(title=f"⚠️ Supprimer la tâche #{tache_id}", custom_id=f"confirm_delete_{tache_id}", components=components) async def callback(self, inter: disnake.ModalInteraction) -> None: confirmation = inter.text_values["confirm_text"] if confirmation != self.tache_titre: return await inter.response.send_message( "❌ Confirmation incorrecte. La tâche n'a pas été supprimée.", ephemeral=True ) success = await supprimer_tache(self.tache_id) if success: await inter.response.send_message( f"🗑️ **Tâche #{self.tache_id} supprimée avec succès.**", ephemeral=True ) else: await inter.response.send_message( "❌ Impossible de supprimer la tâche.", ephemeral=True ) # ============================================================================ # COMPONENTS V2 HELPER FUNCTIONS # ============================================================================ def _get_statut_emoji(statut: str) -> str: return {"À faire": "📌", "En cours": "⏳", "Terminée": "✅", "Annulée": "❌"}.get(statut, "📌") def _get_priorite_emoji(priorite: str) -> str: return {"Critique": "🔴", "Haute": "🟠", "Moyenne": "🟡", "Basse": "🟢", "Faible": "⚪"}.get(priorite, "🟡") def get_main_menu_components(guild_id: int) -> list: """Menu principal des tâches""" return [ ui.Container( ui.TextDisplay("📝 **Gestion des Tâches**\nUtilisez les boutons ci-dessous pour interagir."), ui.Section("➕ Nouvelle Tâche", accessory=ui.Button(label="Créer", style=disnake.ButtonStyle.green, custom_id=f"taches_nouvelle:{guild_id}")), ui.Section("📋 Mes Tâches", accessory=ui.Button(label="Mes tâches", style=disnake.ButtonStyle.blurple, custom_id=f"taches_mes:{guild_id}")), ui.Section("🌐 Toutes les Tâches", accessory=ui.Button(label="Toutes", style=disnake.ButtonStyle.blurple, custom_id=f"taches_toutes:{guild_id}")), ui.Section("🔄 Rafraîchir", accessory=ui.Button(label="Rafraîchir", style=disnake.ButtonStyle.gray, custom_id=f"taches_rafraichir:{guild_id}")), ) ] def get_no_permission_components(guild_id: int) -> list: """Message de permission refusée""" return [ ui.Container( ui.TextDisplay("❌ **Permission refusée**\nSeuls les administrateurs peuvent utiliser cette fonctionnalité."), ui.Section("⬅️ Retour", accessory=ui.Button(label="Retour", style=disnake.ButtonStyle.gray, custom_id=f"taches_back:{guild_id}")), ) ] def get_module_disabled_components(guild_id: int) -> list: """Message de module désactivé""" return [ ui.Container( ui.TextDisplay("❌ **Module désactivé**\nLe système de tâches est désactivé sur ce serveur."), ui.Section("⬅️ Retour", accessory=ui.Button(label="Retour", style=disnake.ButtonStyle.gray, custom_id=f"taches_back:{guild_id}")), ) ] def get_no_taches_components(user_id: int, guild_id: int, is_all: bool = False) -> list: """Message lorsqu'il n'y a pas de tâches""" message = "📭 **Aucune tâche assignée**\nVous n'avez aucune tâche en cours sur ce serveur." if is_all: message = "📭 **Aucune tâche**\nAucune tâche trouvée sur ce serveur." return [ ui.Container( ui.TextDisplay(message), ui.Section("⬅️ Retour au menu", accessory=ui.Button(label="Retour", style=disnake.ButtonStyle.gray, custom_id=f"taches_back_to_menu:{user_id}:{guild_id}")), ) ] def get_taches_pagination_components(user_id: int, guild_id: int, taches_list: List[tuple], page: int, is_all: bool) -> list: """Pagination des tâches - divisé en plusieurs Containers pour respecter la limite de 5 éléments""" start = page * 4 end = start + 4 page_taches = taches_list[start:end] total_pages = max(1, (len(taches_list) - 1) // 4 + 1) containers = [] # Premier Container: Titre + tâches (max 4) container1_sections = [ ui.TextDisplay(f"📝 **{'Toutes les tâches du serveur' if is_all else 'Vos tâches'}** :") ] if page_taches: for tid, tache in page_taches: container1_sections.append(ui.Section( f"#{tid} - {tache['titre'][:30]}", accessory=ui.Button( label="Voir", style=disnake.ButtonStyle.blurple, custom_id=f"tache_detail:{tid}:{user_id}:{guild_id}:{page}:{1 if is_all else 0}" ) )) else: container1_sections.append(ui.TextDisplay("Aucune tâche à afficher.")) containers.append(ui.Container(*container1_sections)) # Deuxième Container: Navigation prev_disabled = page == 0 next_disabled = end >= len(taches_list) container2_sections = [] if prev_disabled and next_disabled: container2_sections.append(ui.Section( "⬅️ Retour au menu", accessory=ui.Button( label="Retour", style=disnake.ButtonStyle.gray, custom_id=f"taches_back_to_menu:{user_id}:{guild_id}" ) )) else: if not prev_disabled: container2_sections.append(ui.Section( "⬅️ Précédent", accessory=ui.Button( label="Précédent", style=disnake.ButtonStyle.secondary, custom_id=f"taches_prev:{user_id}:{guild_id}:{page}:{1 if is_all else 0}", disabled=prev_disabled ) )) container2_sections.append(ui.TextDisplay(f"Page {page + 1}/{total_pages}")) if not next_disabled: container2_sections.append(ui.Section( "Suivant ➡️", accessory=ui.Button( label="Suivant", style=disnake.ButtonStyle.secondary, custom_id=f"taches_next:{user_id}:{guild_id}:{page}:{1 if is_all else 0}", disabled=next_disabled ) )) if len(container2_sections) >= 5: containers.append(ui.Container(*container2_sections)) container2_sections = [] container2_sections.append(ui.Section( "⬅️ Retour au menu", accessory=ui.Button( label="Retour", style=disnake.ButtonStyle.gray, custom_id=f"taches_back_to_menu:{user_id}:{guild_id}" ) )) containers.append(ui.Container(*container2_sections)) return containers def get_tache_detail_components(tache: Dict[str, Any], tache_id: str, user_id: int, guild_id: int, can_edit: bool) -> list: """Détails d'une tâche""" statut_emoji = _get_statut_emoji(tache.get("statut", "À faire")) priorite_emoji = _get_priorite_emoji(tache.get("priorite", "Moyenne")) tache_text = ( f"{statut_emoji} **Tâche #{tache_id}**\n\n" f"**Titre** : {tache.get('titre', 'N/A')}\n" f"**Description** : {tache.get('description', 'N/A')}\n" f"**Assigné à** : <@{tache.get('assigne', '0')}>\n" f"**Créé par** : <@{tache.get('cree_par', '0')}>\n" f"**Priorité** : {priorite_emoji} {tache.get('priorite', 'Moyenne')}\n" f"**Statut** : {tache.get('statut', 'À faire')}\n" f"**Échéance** : {tache.get('echeance', 'N/A')}\n" f"**Créée le** : {tache.get('date_creation', 'N/A')}" ) sections = [ui.TextDisplay(tache_text)] if can_edit: sections.append(ui.Section("✅ Marquer terminée", accessory=ui.Button(label="Terminée", style=disnake.ButtonStyle.green, custom_id=f"tache_done:{tache_id}:{guild_id}"))) sections.append(ui.Section("⚡ Changer priorité", accessory=ui.Button(label="Priorité", style=disnake.ButtonStyle.blurple, custom_id=f"tache_prio:{tache_id}:{guild_id}"))) sections.append(ui.Section("🗑️ Supprimer", accessory=ui.Button(label="Supprimer", style=disnake.ButtonStyle.red, custom_id=f"tache_delete:{tache_id}:{guild_id}"))) sections.append(ui.Section("⬅️ Retour", accessory=ui.Button(label="Retour", style=disnake.ButtonStyle.gray, custom_id=f"tache_back:{guild_id}"))) return [ui.Container(*sections)] def get_tache_done_components(tache: Dict[str, Any], user_id: int, guild_id: int) -> list: """Confirmation que la tâche est terminée (avec titre barré)""" priorite_emoji = _get_priorite_emoji(tache.get("priorite", "Moyenne")) text = ( f"✅ **Tâche terminée !**\n\n" f"📌 **Titre** : ~~{tache.get('titre', 'Inconnu')}~~\n" f"📅 Échéance : {tache.get('echeance', 'N/A')}\n" f"⚠️ Priorité : {tache.get('priorite', 'Moyenne')}\n" f"📝 Statut : **Terminée**" ) return [ ui.Container( ui.TextDisplay(text), # user_id est passé dans le custom_id pour que le handler puisse vérifier l'autorisation ui.Section("⬅️ Retour", accessory=ui.Button(label="Retour", style=disnake.ButtonStyle.gray, custom_id=f"taches_back_to_menu:{user_id}:{guild_id}")), ) ] def get_change_priority_components(tache_id: str, current_prio: str, user_id: int, guild_id: int) -> list: """Sélection de la priorité - divisé en plusieurs Containers pour respecter la limite de 5 éléments""" containers = [] container1_sections = [ ui.TextDisplay("📝 Sélectionnez la nouvelle priorité :") ] for prio in PRIORITES: style = disnake.ButtonStyle.green if prio == current_prio else disnake.ButtonStyle.secondary emoji = _get_priorite_emoji(prio) container1_sections.append(ui.Section( f"{emoji} {prio}", accessory=ui.Button(label=prio, style=style, custom_id=f"prio_select:{tache_id}:{prio}:{user_id}:{guild_id}") )) if len(container1_sections) >= 5: containers.append(ui.Container(*container1_sections)) container1_sections = [] if container1_sections: containers.append(ui.Container(*container1_sections)) containers.append(ui.Container( ui.Section( "❌ Annuler", accessory=ui.Button(label="Annuler", style=disnake.ButtonStyle.red, custom_id=f"prio_cancel:{tache_id}:{user_id}:{guild_id}") ) )) return containers # ============================================================================ # COG PRINCIPAL # ============================================================================ class TachesCog(commands.Cog): """Cog principal pour la gestion des tâches.""" def __init__(self, bot): self.bot = bot async def cog_load(self): """Démarre la tâche de rappel une fois le cog chargé.""" self.rappels_loop.start() def cog_unload(self): self.rappels_loop.cancel() # ========================================================================== # HANDLER MANUEL DES INTERACTIONS DE BOUTONS # ========================================================================== @commands.Cog.listener("on_message_interaction") async def on_taches_interaction(self, interaction: disnake.MessageInteraction): """Gère toutes les interactions de boutons pour le système de tâches.""" custom_id = interaction.data.custom_id if not custom_id or not (custom_id.startswith("taches_") or custom_id.startswith("tache_") or custom_id.startswith("prio_")): return kuby_logger.debug(f"Interaction tâches reçue : {custom_id} par {interaction.user}") try: parts = custom_id.split(":") action = parts[0] # Menu principal if action == "taches_nouvelle": guild_id = int(parts[1]) if not interaction.author.guild_permissions.administrator: await interaction.response.edit_message( components=get_no_permission_components(guild_id)) return modal = NouvelleTacheModal(guild_id) await interaction.response.send_modal(modal) elif action == "taches_mes": guild_id = int(parts[1]) user_id = interaction.author.id settings = await get_taches_settings(guild_id) if not settings.get("taches_enabled", True): await interaction.response.edit_message( components=get_module_disabled_components(guild_id)) return taches = await lister_taches_utilisateur(user_id, guild_id) if not taches: await interaction.response.edit_message( components=get_no_taches_components(user_id, guild_id, is_all=False)) return taches_list = list(taches.items()) await interaction.response.edit_message( components=get_taches_pagination_components(user_id, guild_id, taches_list, 0, False)) elif action == "taches_toutes": guild_id = int(parts[1]) user_id = interaction.author.id if not interaction.author.guild_permissions.administrator: await interaction.response.edit_message( components=get_no_permission_components(guild_id)) return settings = await get_taches_settings(guild_id) if not settings.get("taches_enabled", True): await interaction.response.edit_message( components=get_module_disabled_components(guild_id)) return taches = await lister_toutes_taches(guild_id) if not taches: await interaction.response.edit_message( components=get_no_taches_components(user_id, guild_id, is_all=True)) return taches_list = list(taches.items()) await interaction.response.edit_message( components=get_taches_pagination_components(user_id, guild_id, taches_list, 0, True)) elif action == "taches_rafraichir": guild_id = int(parts[1]) await interaction.response.edit_message( components=get_main_menu_components(guild_id)) # Navigation elif action == "taches_prev": user_id = int(parts[1]) guild_id = int(parts[2]) current_page = int(parts[3]) is_all = bool(int(parts[4])) if interaction.author.id != user_id: return if current_page > 0: new_page = current_page - 1 taches = await lister_toutes_taches(guild_id) if is_all else await lister_taches_utilisateur(user_id, guild_id) taches_list = list(taches.items()) if taches else [] await interaction.response.edit_message( components=get_taches_pagination_components(user_id, guild_id, taches_list, new_page, is_all)) elif action == "taches_next": user_id = int(parts[1]) guild_id = int(parts[2]) current_page = int(parts[3]) is_all = bool(int(parts[4])) if interaction.author.id != user_id: return taches = await lister_toutes_taches(guild_id) if is_all else await lister_taches_utilisateur(user_id, guild_id) taches_list = list(taches.items()) if taches else [] if (current_page + 1) * 4 < len(taches_list): new_page = current_page + 1 await interaction.response.edit_message( components=get_taches_pagination_components(user_id, guild_id, taches_list, new_page, is_all)) elif action == "taches_back_to_menu": if interaction.author.id != int(parts[1]): return guild_id = int(parts[2]) await interaction.response.edit_message( components=get_main_menu_components(guild_id)) # Détails tâche elif action == "tache_detail": tache_id = parts[1] user_id = int(parts[2]) guild_id = int(parts[3]) page = int(parts[4]) is_all = bool(int(parts[5])) if interaction.author.id != user_id: await interaction.response.send_message("❌ Non autorisé.", ephemeral=True) return tache = await get_tache(tache_id) if not tache: await interaction.response.edit_message( components=get_taches_pagination_components(user_id, guild_id, [], page, is_all)) return if isinstance(tache, dict) and len(tache) == 1: tache = next(iter(tache.values())) can_edit = tache.get("assigne") == user_id or tache.get("cree_par") == user_id await interaction.response.edit_message( components=get_tache_detail_components(tache, tache_id, user_id, guild_id, can_edit)) # Actions tâche elif action == "tache_done": tache_id = parts[1] guild_id = int(parts[2]) tache = await get_tache(tache_id) if not tache: await interaction.response.send_message("❌ Tâche introuvable.", ephemeral=True) return if isinstance(tache, dict) and len(tache) == 1: tache = next(iter(tache.values())) if interaction.author.id != tache.get("assigne") and interaction.author.id != tache.get("cree_par"): await interaction.response.send_message("❌ Non autorisé.", ephemeral=True) return success = await update_tache_status(tache_id, "Terminée") if success: tache["statut"] = "Terminée" await interaction.response.edit_message( components=get_tache_done_components(tache, interaction.author.id, guild_id)) else: await interaction.response.send_message("❌ Erreur.", ephemeral=True) elif action == "tache_prio": tache_id = parts[1] guild_id = int(parts[2]) tache = await get_tache(tache_id) if not tache: await interaction.response.send_message("❌ Tâche introuvable.", ephemeral=True) return if isinstance(tache, dict) and len(tache) == 1: tache = next(iter(tache.values())) if interaction.author.id != tache.get("assigne") and interaction.author.id != tache.get("cree_par"): await interaction.response.send_message("❌ Non autorisé.", ephemeral=True) return current_prio = tache.get("priorite", "Moyenne") await interaction.response.edit_message( components=get_change_priority_components(tache_id, current_prio, interaction.author.id, guild_id)) elif action == "tache_delete": tache_id = parts[1] guild_id = int(parts[2]) tache = await get_tache(tache_id) if not tache: await interaction.response.send_message("❌ Tâche introuvable.", ephemeral=True) return if isinstance(tache, dict) and len(tache) == 1: tache = next(iter(tache.values())) if interaction.author.id != tache.get("assigne") and interaction.author.id != tache.get("cree_par"): await interaction.response.send_message("❌ Non autorisé.", ephemeral=True) return tache_titre = tache.get("titre", "Tâche") modal = ConfirmDeleteModal(tache_id, tache_titre) await interaction.response.send_modal(modal) # Retour elif action == "tache_back": guild_id = int(parts[1]) await interaction.response.edit_message( components=get_main_menu_components(guild_id)) elif action == "taches_back": guild_id = int(parts[1]) await interaction.response.edit_message( components=get_main_menu_components(guild_id)) # Changement de priorité elif action == "prio_select": tache_id = parts[1] new_prio = parts[2] user_id = int(parts[3]) guild_id = int(parts[4]) if interaction.author.id != user_id: await interaction.response.send_message("❌ Non autorisé.", ephemeral=True) return success = await update_tache_priorite(tache_id, new_prio) if success: await interaction.response.edit_message( components=get_main_menu_components(guild_id)) await interaction.followup.send(f"✅ Priorité mise à jour : **{new_prio}**", ephemeral=True) else: await interaction.response.send_message("❌ Erreur.", ephemeral=True) elif action == "prio_cancel": guild_id = int(parts[3]) await interaction.response.edit_message( components=get_main_menu_components(guild_id)) except Exception as e: kuby_logger.error(f"Erreur dans on_taches_interaction: {e}", exc_info=True) try: await interaction.response.send_message("❌ Une erreur est survenue.", ephemeral=True) except: pass # ========================================================================== # COMMANDE SLASH # ========================================================================== @commands.slash_command(name="taches", description="Ouvrir le menu de gestion des tâches") async def taches_command(self, inter: disnake.ApplicationCommandInteraction): """Commande principale pour accéder au menu des tâches.""" settings = await get_taches_settings(inter.guild.id) if not settings.get("taches_enabled", True): return await inter.response.send_message( "❌ Le système de tâches est désactivé sur ce serveur.", ephemeral=True ) await inter.response.send_message( components=get_main_menu_components(inter.guild.id) ) # ========================================================================== # BOUCLE DE RAPPELS # ========================================================================== @tasks.loop(minutes=15) async def rappels_loop(self): """Vérifie les tâches proches de leur échéance tous les 15 minutes.""" for guild in self.bot.guilds: try: settings = await get_taches_settings(guild.id) if not settings.get("taches_enabled", True): continue taches = await taches_proches(guild.id, heures_avant=1) for tid, tache in taches.items(): try: user = await self.bot.fetch_user(tache["assigne"]) if user: statut_emoji = _get_statut_emoji(tache.get("statut", "À faire")) priorite_emoji = _get_priorite_emoji(tache.get("priorite", "Moyenne")) await user.send( content=( f"⏰ **Rappel : Tâche #{tid}**\n\n" f"{statut_emoji} **{tache['titre']}** {priorite_emoji}\n" f"📅 **Échéance dans 1 heure**\n" f"👤 Assigné par : <@{tache['cree_par']}>" ) ) except Exception as e: kuby_logger.error(f"Erreur rappel tâche {tid}: {e}") except Exception as e: kuby_logger.error(f"Erreur dans rappels_loop pour {guild.id}: {e}") @rappels_loop.before_loop async def avant_rappels(self): await self.bot.wait_until_ready() # ============================================================================ # SETUP # ============================================================================ def setup(bot): """Setup du cog Tâches.""" bot.add_cog(TachesCog(bot))