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 # Configuration des variables d'environnement pour Forgejo FORGEJO_TOKEN = os.getenv("FORGEJO_TOKEN") FORGEJO_OWNER = os.getenv("FORGEJO_OWNER") # Exemple: "Gameur" ou "Omega_Kube" FORGEJO_REPO = os.getenv("FORGEJO_REPO") # Exemple: "kuby" BASE_URL = "https://omegakubeserv.tail951d2f.ts.net/api/v1" if not FORGEJO_TOKEN or not FORGEJO_OWNER or not FORGEJO_REPO: raise RuntimeError("Forgejo variables (FORGEJO_TOKEN, FORGEJO_OWNER, FORGEJO_REPO) must be set.") # Fichiers de persistance mis à jour pour Forgejo ISSUE_TRACK_FILE = os.path.join(os.path.dirname(__file__), "forgejo_issues.json") def load_issue_tracking() -> Dict[str, Any]: tracking = {} 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 forgejo_issues.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 ForgejoIntegration(commands.Cog): """Cog fournissant des commandes slash pour créer des tickets sur Forgejo et notifier les utilisateurs.""" def __init__(self, bot: commands.Bot): self.bot = bot self.issue_data: Dict[str, Any] = load_issue_tracking() self.check_new_comments.start() def cog_unload(self): self.check_new_comments.cancel() async def _create_forgejo_issue(self, title: str, description: str, labels: List[str], requester_id: int = None) -> Optional[Dict[str, Any]]: """Crée un ticket sur Forgejo et le track automatiquement.""" url = f"{BASE_URL}/repos/{FORGEJO_OWNER}/{FORGEJO_REPO}/issues" headers = { "Authorization": f"token {FORGEJO_TOKEN}", "Content-Type": "application/json", } payload = { "title": title, "body": description, # GitLab utilisait 'description', Forgejo utilise 'body' "labels": labels, # Forgejo accepte directement une liste ['bug', 'bot'] } 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() if requester_id: issue_number = str(result["number"]) # 'iid' devient 'number' self.issue_data[issue_number] = { "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 Forgejo: {e}") return None async def _get_issue_comments(self, issue_number: int) -> List[Dict[str, Any]]: """Récupère tous les commentaires d'un ticket spécifique.""" url = f"{BASE_URL}/repos/{FORGEJO_OWNER}/{FORGEJO_REPO}/issues/{issue_number}/comments" headers = {"Authorization": f"token {FORGEJO_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_number: int, comment: Dict[str, Any]): """Envoie un DM privé à l'utilisateur lors d'un nouveau commentaire.""" user = await self.bot.fetch_user(requester_id) author = comment.get("user", {}).get("username", "Quelqu'un") # 'author.name' devient 'user.username' body = comment.get("body", "") embed = disnake.Embed( title=f"Nouveau commentaire sur votre demande Forgejo (#{issue_number})", description=body, color=disnake.Color.brand_green(), timestamp=datetime.fromisoformat(comment["created_at"].replace("Z", "+00:00")), ) embed.set_footer(text=f"Commentaire de {author}") try: await user.send(embed=embed) except disnake.HTTPException: pass @tasks.loop(minutes=1) async def check_new_comments(self): """Boucle de vérification des commentaires Forgejo.""" for issue_num_str, info in list(self.issue_data.items()): issue_number = int(issue_num_str) last_known_id = info.get("last_note_id", 0) requester_id = info.get("requester_id") try: comments = await self._get_issue_comments(issue_number) if not comments: continue missing_comments = [c for c in comments if c["id"] > last_known_id] if missing_comments: for comment in missing_comments: await self._notify_requester(requester_id, issue_number, comment) new_last_id = max(c["id"] for c in comments) self.issue_data[issue_num_str]["last_note_id"] = new_last_id except Exception as e: print(f"⚠️ Erreur vérification issue #{issue_number}: {e}") continue save_issue_tracking(self.issue_data) @commands.slash_command(name="recuperer_suivi_issues", description="Récupère le suivi des commentaires Forgejo après un incident") async def recuperer_suivi(self, inter: disnake.ApplicationCommandInteraction): await inter.response.defer(ephemeral=True) updated_count = 0 for issue_num_str, info in list(self.issue_data.items()): issue_number = int(issue_num_str) try: comments = await self._get_issue_comments(issue_number) if comments: latest_id = max(c["id"] for c in comments) if latest_id > info.get("last_note_id", 0): self.issue_data[issue_num_str]["last_note_id"] = latest_id updated_count += 1 except Exception as e: print(f"⚠️ Erreur récupération issue #{issue_number}: {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." ) @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 sur Kuby") async def signaler_bug(self, inter: disnake.ApplicationCommandInteraction): await inter.response.send_modal(modal=BugModal(self)) # ---------- Modals Acteurs ---------- class FeatureModal(disnake.Modal): def __init__(self, cog: ForgejoIntegration): components = [ disnake.ui.TextInput(label="Titre de la fonctionnalité", custom_id="title", 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) issue = await self.cog._create_forgejo_issue( title=title, description=description, labels=["feature", "discord-bot"], requester_id=inter.author.id ) if issue: url = f"https://omegakubeserv.tail951d2f.ts.net/{FORGEJO_OWNER}/{FORGEJO_REPO}/issues/{issue['number']}" await inter.edit_original_message(content=f"✅ Fonctionnalité créée : [{title}]({url})") else: await inter.edit_original_message(content="❌ Erreur lors de la création sur Forgejo.") class BugModal(disnake.Modal): def __init__(self, cog: ForgejoIntegration): components = [ disnake.ui.TextInput(label="Titre du bug", custom_id="title", max_length=100, required=True), disnake.ui.TextInput(label="Description et étapes", style=disnake.TextInputStyle.paragraph, custom_id="description", 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) issue = await self.cog._create_forgejo_issue( title=title, description=description, labels=["bug", "discord-bot"], requester_id=inter.author.id ) if issue: url = f"https://omegakubeserv.tail951d2f.ts.net/{FORGEJO_OWNER}/{FORGEJO_REPO}/issues/{issue['number']}" await inter.edit_original_message(content=f"✅ Bug signalé : [{title}]({url})") else: await inter.edit_original_message(content="❌ Erreur lors du signalement sur Forgejo.") def setup(bot: commands.Bot): bot.add_cog(ForgejoIntegration(bot))