import os import json import asyncio from datetime import datetime, timezone from typing import Dict, Any, List, Optional import aiohttp import disnake from disnake.ext import commands, tasks # Environment variables for GitLab access GITLAB_TOKEN = os.getenv("GITLAB_TOKEN") GITLAB_PROJECT_ID = os.getenv("GITLAB_PROJECT_ID") # numeric ID or URL-encoded path if not GITLAB_TOKEN or not GITLAB_PROJECT_ID: raise RuntimeError("GitLab token and project ID must be set in environment variables GITLAB_TOKEN and GITLAB_PROJECT_ID") # Files to persist issue tracking information ISSUE_TRACK_FILE = os.path.join(os.path.dirname(__file__), "gitlab_issues.json") LEGACY_REPORTS_FILE = os.path.join(os.path.dirname(__file__), "..", "data", "gitlab_reports.json") def load_issue_tracking() -> Dict[str, Any]: """Charge le tracking et fusionne automatiquement avec gitlab_reports.json (migration)""" tracking = {} # Charge le fichier principal if os.path.exists(ISSUE_TRACK_FILE): try: with open(ISSUE_TRACK_FILE, "r", encoding="utf-8") as f: tracking = json.load(f) except Exception as e: print(f"⚠️ Erreur chargement gitlab_issues.json: {e}") # Migration automatique depuis l'ancien système if os.path.exists(LEGACY_REPORTS_FILE): try: with open(LEGACY_REPORTS_FILE, "r", encoding="utf-8") as f: legacy_data = json.load(f) for issue_iid, user_id in legacy_data.items(): if issue_iid not in tracking: tracking[issue_iid] = { "requester_id": user_id, "last_note_id": 0, "type": "legacy_report" } print(f"🔄 Migration automatique: {len(legacy_data)} issues fusionnées depuis gitlab_reports.json") except Exception as e: print(f"⚠️ Erreur migration gitlab_reports.json: {e}") return tracking def save_issue_tracking(data: Dict[str, Any]) -> None: with open(ISSUE_TRACK_FILE, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) class GitLabIntegration(commands.Cog): """Cog providing slash commands to create GitLab issues (feature request / bug) and notify requesters when comments are added.""" def __init__(self, bot: commands.Bot): self.bot = bot self.issue_data: Dict[str, Any] = load_issue_tracking() # key: issue_iid (str) -> {"requester_id": int, "last_note_id": int} self.check_new_notes.start() def cog_unload(self): self.check_new_notes.cancel() async def _create_gitlab_issue(self, title: str, description: str, labels: List[str], requester_id: int = None) -> Optional[Dict[str, Any]]: """Crée une issue sur GitLab et la track automatiquement.""" url = f"https://gitlab.com/api/v4/projects/{GITLAB_PROJECT_ID}/issues" headers = { "PRIVATE-TOKEN": GITLAB_TOKEN, "Content-Type": "application/json", } payload = { "title": title, "description": description, "labels": ",".join(labels), } try: async with aiohttp.ClientSession() as session: async with session.post(url, headers=headers, json=payload) as resp: resp.raise_for_status() result = await resp.json() # Track automatiquement la nouvelle issue if requester_id: issue_iid = str(result["iid"]) self.issue_data[issue_iid] = { "requester_id": requester_id, "last_note_id": 0, "type": "discord-command" } save_issue_tracking(self.issue_data) return result except Exception as e: print(f"❌ Erreur création issue GitLab: {e}") return None async def _get_issue_notes(self, issue_iid: int) -> List[Dict[str, Any]]: """Fetch all notes (comments) for a given issue.""" url = f"https://gitlab.com/api/v4/projects/{GITLAB_PROJECT_ID}/issues/{issue_iid}/notes" headers = {"PRIVATE-TOKEN": GITLAB_TOKEN} async with aiohttp.ClientSession() as session: async with session.get(url, headers=headers) as resp: resp.raise_for_status() return await resp.json() async def _notify_requester(self, requester_id: int, issue_iid: int, note: Dict[str, Any]): """Send a private DM to the requester informing them of a new comment.""" user = await self.bot.fetch_user(requester_id) author = note.get("author", {}).get("name", "Someone") body = note.get("body", "") embed = disnake.Embed( title=f"Nouveau commentaire sur votre demande GitLab (#{issue_iid})", description=body, color=disnake.Color.blurple(), timestamp=datetime.fromisoformat(note["created_at"]).replace(tzinfo=timezone.utc), ) embed.set_footer(text=f"Commentaire de {author}") try: await user.send(embed=embed) except disnake.HTTPException: # The user may have DMs disabled; ignore silently. pass @tasks.loop(minutes=1) async def check_new_notes(self): """Vérifie TOUS les nouveaux commentaires (pas seulement le dernier) et notifie.""" for issue_iid_str, info in list(self.issue_data.items()): issue_iid = int(issue_iid_str) last_known_id = info.get("last_note_id", 0) requester_id = info.get("requester_id") try: notes = await self._get_issue_notes(issue_iid) if not notes: continue # ⚡ NOUVEAU: Notifie TOUS les commentaires manquants, pas seulement le dernier missing_notes = [n for n in notes if n["id"] > last_known_id] if missing_notes: for note in missing_notes: await self._notify_requester(requester_id, issue_iid, note) # Met à jour avec le dernier ID vu new_last_id = max(n["id"] for n in notes) self.issue_data[issue_iid_str]["last_note_id"] = new_last_id except Exception as e: print(f"⚠️ Erreur vérification issue #{issue_iid}: {e}") continue save_issue_tracking(self.issue_data) @commands.slash_command(name="recuperer_suivi_issues", description="Récupère le suivi des commentaires après un incident") async def recuperer_suivi(self, inter: disnake.ApplicationCommandInteraction): """Reconstruit gitlab_issues.json avec les derniers commentaires.""" await inter.response.defer(ephemeral=True) updated_count = 0 for issue_iid_str, info in list(self.issue_data.items()): issue_iid = int(issue_iid_str) try: notes = await self._get_issue_notes(issue_iid) if notes: latest_id = max(n["id"] for n in notes) if latest_id > info.get("last_note_id", 0): self.issue_data[issue_iid_str]["last_note_id"] = latest_id updated_count += 1 except Exception as e: print(f"⚠️ Erreur récupération issue #{issue_iid}: {e}") continue save_issue_tracking(self.issue_data) await inter.edit_original_message( content=f"✅ Suivi récupéré pour **{updated_count}/{len(self.issue_data)}** issues. " f"Fichier `{ISSUE_TRACK_FILE}` mis à jour." ) # ---------- Slash commands ---------- @commands.slash_command(name="suggerer_fonctionnalite", description="Suggérer une fonctionnalité pour le projet") async def suggere_fonctionnalite(self, inter: disnake.ApplicationCommandInteraction): await inter.response.send_modal(modal=FeatureModal(self)) @commands.slash_command(name="signaler_bug", description="Signaler un bug") async def signaler_bug(self, inter: disnake.ApplicationCommandInteraction): await inter.response.send_modal(modal=BugModal(self)) # ---------- Modals ---------- class FeatureModal(disnake.Modal): def __init__(self, cog: GitLabIntegration): components = [ disnake.ui.TextInput( label="Titre de la fonctionnalité", custom_id="title", style=disnake.TextInputStyle.short, max_length=100, required=True, ), disnake.ui.TextInput( label="Description détaillée", custom_id="description", style=disnake.TextInputStyle.paragraph, max_length=2000, required=True, ), ] super().__init__( title="Suggestion de fonctionnalité", custom_id="feature_modal", components=components, ) self.cog = cog async def callback(self, inter: disnake.ModalInteraction): title = inter.text_values["title"] description = inter.text_values["description"] await inter.response.defer(ephemeral=True) try: issue = await self.cog._create_gitlab_issue( title=title, description=description, labels=["feature", "discord-bot"], requester_id=inter.author.id # ⚡ NOUVEAU: Passe le requester_id pour tracking auto ) if issue: await inter.edit_original_message( content=f"✅ Fonctionnalité créée : [{title}](https://gitlab.com/{GITLAB_PROJECT_ID}/-/issues/{issue['iid']})" ) else: await inter.edit_original_message(content="❌ Erreur lors de la création de la fonctionnalité.") except Exception as e: await inter.edit_original_message(content=f"❌ Erreur lors de la création de la fonctionnalité : {e}") class BugModal(disnake.Modal): def __init__(self, cog: GitLabIntegration): components = [ disnake.ui.TextInput( label="Titre du bug", custom_id="title", style=disnake.TextInputStyle.short, max_length=100, required=True, ), disnake.ui.TextInput( label="Description et étapes pour reproduire", custom_id="description", style=disnake.TextInputStyle.paragraph, max_length=2000, required=True, ), ] super().__init__( title="Signalement de bug", custom_id="bug_modal", components=components, ) self.cog = cog async def callback(self, inter: disnake.ModalInteraction): title = inter.text_values["title"] description = inter.text_values["description"] await inter.response.defer(ephemeral=True) try: issue = await self.cog._create_gitlab_issue( title=title, description=description, labels=["bug", "discord-bot"], requester_id=inter.author.id # ⚡ NOUVEAU: Passe le requester_id pour tracking auto ) if issue: await inter.edit_original_message( content=f"✅ Bug signalé : [{title}](https://gitlab.com/{GITLAB_PROJECT_ID}/-/issues/{issue['iid']})" ) else: await inter.edit_original_message(content="❌ Erreur lors du signalement du bug.") except Exception as e: await inter.edit_original_message(content=f"❌ Erreur lors du signalement du bug : {e}") def setup(bot: commands.Bot): bot.add_cog(GitLabIntegration(bot))