diff --git a/commandes/bug_report.py b/commandes/bug_report.py index bb389bb..8bf5f40 100644 --- a/commandes/bug_report.py +++ b/commandes/bug_report.py @@ -1,6 +1,6 @@ import disnake from disnake.ext import commands -from utils.gitlab_client import gitlab_client +from utils.forgejo_client import gitlab_client from aiohttp import web import logging import json @@ -11,51 +11,11 @@ from utils.premium import check_premium_tier kuby_logger = logging.getLogger("KubyBot") BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) - # ⚡ NOUVEAU: Utilise le même fichier que gitlab_integration.py TRACKING_FILE = os.path.join(BASE_DIR, "commandes", "gitlab_issues.json") LEGACY_REPORTS_FILE = os.path.join(BASE_DIR, "data", "gitlab_reports.json") -def classify_webhook_event(payload: Dict[str, Any]) -> Optional[str]: - """Détermine si un webhook correspond à une issue ou à un commentaire.""" - object_kind = payload.get("object_kind") - if object_kind in {"issue", "note"}: - return object_kind - - action = str(payload.get("action", "")).lower() - if action in {"comment_created", "comment_updated", "commented"} or "comment" in payload: - return "note" - - if payload.get("issue") or payload.get("object_attributes") or action in {"open", "reopen", "close", "closed", "updated"}: - return "issue" - - return None - - -def extract_issue_iid(payload: Dict[str, Any]) -> Optional[str]: - """Extrait l’identifiant d’issue depuis plusieurs formats Forgejo/GitLab.""" - for container in (payload.get("object_attributes", {}), payload.get("issue", {}), payload.get("comment", {})): - if not isinstance(container, dict): - continue - for key in ("iid", "number", "id"): - value = container.get(key) - if value not in (None, ""): - return str(value) - return None - - -def extract_issue_identifier(issue_payload: Optional[Dict[str, Any]]) -> Optional[str]: - """Récupère un identifiant stable à stocker dans le tracking.""" - if not isinstance(issue_payload, dict): - return None - for key in ("iid", "number", "id"): - value = issue_payload.get(key) - if value not in (None, ""): - return str(value) - return None - - def load_tracking(): """Charge le tracking depuis gitlab_issues.json (avec migration automatique)""" tracking = {} @@ -303,28 +263,32 @@ class BugReport(commands.Cog): self.bot.loop.create_task(cleanup()) async def handle_bot_event(self, request): - """Gère les webhooks GitLab (notifications instantanées)""" + """Gère les webhooks (adapté pour GitLab ET Forgejo)""" try: if not self.bot.is_ready(): kuby_logger.info("Webhook received but bot is not ready yet. Waiting up to 10s...") try: await asyncio.wait_for(self.bot.wait_until_ready(), timeout=10.0) except asyncio.TimeoutError: - kuby_logger.warning("Bot still not ready after 10s. Proceeding anyway (best effort).") + kuby_logger.warning("Bot still not ready after 10s. Proceeding anyway.") payload = await request.json() - event_type = classify_webhook_event(payload) + event_type = payload.get("object_kind") + if event_type not in ["issue", "note"]: kuby_logger.info("Webhook ignoré : type d’événement non reconnu") return web.json_response({"status": "ignored"}, status=200) - # Récupère l'ID de l'issue - issue_iid = extract_issue_iid(payload) + # Récupère l'IID de l'issue + issue_iid = str( + payload.get("object_attributes", {}).get("iid") or + payload.get("issue", {}).get("iid") + ) + if not issue_iid: kuby_logger.warning("Webhook ignoré : aucun identifiant d’issue trouvé dans le payload") return web.json_response({"status": "no issue id"}, status=200) - # Vérifie si l'issue est trackée issue_data = self.issue_data.get(issue_iid) if not issue_data: return web.json_response({"status": "untracked issue"}, status=200) @@ -339,44 +303,37 @@ class BugReport(commands.Cog): user = await self.bot.fetch_user(user_id) except: return web.json_response({"status": "user not found"}, status=200) - - app_info = await self.bot.application_info() - dev = app_info.owner - issue_title = payload.get("object_attributes", {}).get("title", f"#{issue_iid}") + + # 🕵️ 3. RÉCUPÉRATION DU TITRE + issue_title = ( + payload.get("object_attributes", {}).get("title") or + payload.get("issue", {}).get("title", f"#{issue_iid}") + ) if event_type == "note": # ⚡ NOUVEAU: Notifie les commentaires via le webhook (instantané) - if "comment" in payload and isinstance(payload.get("comment"), dict): - note_body = payload["comment"].get("body") or payload["comment"].get("content") or "" - author_name = ( - payload.get("sender", {}).get("username") - or payload.get("sender", {}).get("login") - or payload.get("user", {}).get("name") - or "Développeur" - ) - is_system = False - else: - note_attr = payload.get("object_attributes", {}) - is_system = note_attr.get("system", False) + note_attr = payload.get("object_attributes", {}) + is_system = note_attr.get("system", False) + + if not is_system: note_body = ( note_attr.get("note") or note_attr.get("body") or - payload.get("note", "") or - "" + payload.get("note", "") or "" ) author_name = payload.get("user", {}).get("name", "Développeur") - if not is_system and note_body: - try: - embed = disnake.Embed( - title="💬 Nouveau commentaire sur votre signalement", - description=f"Le développeur a répondu à votre rapport **{issue_title}** :\n\n>>> {note_body}\n\n✍️ **Par :** {author_name}", - color=disnake.Color.orange() - ) - embed.set_thumbnail(url=self.bot.user.display_avatar.url) - await user.send(embed=embed) - kuby_logger.info(f"Commentaire envoyé à {user} pour l'issue #{issue_iid}") - except (disnake.Forbidden, disnake.HTTPException) as e: + if note_body: + try: + embed = disnake.Embed( + title="💬 Nouveau commentaire sur votre signalement", + description=f"Le développeur a répondu à votre rapport **{issue_title}** :\n\n>>> {note_body}\n\n✍️ **Par :** {author_name}", + color=disnake.Color.orange() + ) + embed.set_thumbnail(url=self.bot.user.display_avatar.url) + await user.send(embed=embed) + kuby_logger.info(f"Commentaire GitLab envoyé à {user} pour l'issue #{issue_iid}") + except (disnake.Forbidden, disnake.HTTPException) as e: kuby_logger.warning(f"Impossible d'envoyer le DM à {user}: {e}") if dev and dev.id != user.id: try: @@ -385,28 +342,12 @@ class BugReport(commands.Cog): pass elif event_type == "issue": - # Notifie les changements d'état/label - action = payload.get("object_attributes", {}).get("action") - changes = payload.get("changes", {}) - state = payload.get("object_attributes", {}).get("state") + # 🕵️ 5. GESTION DES CLÔTURES (Optionnel : si tu fermes le ticket sur Forgejo) + action = payload.get("object_attributes", {}).get("action") or payload.get("action") + state = payload.get("object_attributes", {}).get("state") or payload.get("issue", {}).get("state") - if "labels" in changes: - curr_labels = [l.get("title") for l in changes["labels"].get("current", []) if l.get("title")] - display_labels = [l for l in curr_labels if not l.startswith("Priority::") and l not in ["bug", "manual-report", "manual-suggestion"]] - current_labels_str = ", ".join(display_labels) if display_labels else "Mis à jour" - - try: - embed = disnake.Embed( - title="🛠️ Mise à jour de votre signalement !", - description=f"Le statut de votre rapport **{issue_title}** a été mis à jour par l'équipe.\n\n📌 **Nouveaux Labels / Statuts :** {current_labels_str}", - color=disnake.Color.blue() - ) - embed.set_thumbnail(url=self.bot.user.display_avatar.url) - await user.send(embed=embed) - except Exception as e: - kuby_logger.error(f"Erreur notification label: {e}") + is_closing_now = action in ["close", "closed"] or state == "closed" - is_closing_now = action == "close" or (state == "closed" and "state_id" in changes) if is_closing_now: try: embed = disnake.Embed( @@ -414,7 +355,8 @@ class BugReport(commands.Cog): description=f"Bonne nouvelle ! Votre signalement **{issue_title}** a été marqué comme terminé ou résolu.\n\n🚀 **Prochaine étape :** La mise à jour arrive prochainement !\n✨ Merci pour votre aide !", color=disnake.Color.green() ) - embed.set_thumbnail(url=self.bot.user.display_avatar.url) + if self.bot.user.display_avatar: + embed.set_thumbnail(url=self.bot.user.display_avatar.url) await user.send(embed=embed) except Exception as e: kuby_logger.error(f"Erreur notification clôture: {e}") diff --git a/commandes/gitlab_issues.json b/commandes/gitlab_issues.json index 8174dd2..4003089 100644 --- a/commandes/gitlab_issues.json +++ b/commandes/gitlab_issues.json @@ -38,5 +38,10 @@ "requester_id": 971446412690722826, "last_note_id": 0, "type": "legacy_report" + }, + "None": { + "requester_id": 971446412690722826, + "last_note_id": 0, + "type": "manual-report" } } \ No newline at end of file diff --git a/serveur_flask.py b/serveur_flask.py index 600203b..8739e77 100644 --- a/serveur_flask.py +++ b/serveur_flask.py @@ -14,12 +14,13 @@ app = Flask(__name__) PROJECT_PATH = "/home/discord/Bot/Kuby" WEBHOOK_SECRET = "Nois2" PROCESS_NAME = "kuby" +# webhook de logs Discord DISCORD_LOG_URL = "https://discord.com/api/webhooks/1482148910888652910/IA9CcOWtjGswbuxMaOu_6uaclv5zojZo4ttxtV6RYyZzJ9gW7BF7xp_Zv3oPIjYEuh8o" -# Chemin absolu vers le binaire PM2 (évite l'erreur 127 command not found) +# Chemin vers PM2 PM2_BINARY = "/home/discord/.nvm/versions/node/v24.15.0/bin/pm2" -# Bot Local Proxy +# Bot Local Proxy (Relai vers le Cog de ton bot) BOT_LOCAL_URL = "http://127.0.0.1:5001/bot_event" @@ -43,7 +44,7 @@ def verify_signature(payload_brut, received_signature): def send_discord_log(status, message, color, author=None, commit_msg=None): embed = { - "title": f"🚀 Kuby V2 : {status}", + "title": f"🚀 Kuby V2 [Forgejo] : {status}", "description": message, "color": color, "timestamp": datetime.now(timezone.utc).isoformat(), @@ -57,7 +58,7 @@ def send_discord_log(status, message, color, author=None, commit_msg=None): payload = {"embeds": [embed]} try: - requests.post(DISCORD_LOG_URL, json=payload, timeout=10) # ✅ Augmenté de 5s à 10s + requests.post(DISCORD_LOG_URL, json=payload, timeout=10) except Exception as e: print(f"[!] Erreur envoi Discord : {e}") @@ -79,29 +80,15 @@ def run_mep_logic(author, commit_msg, branch="main"): os.chdir(target_path) - # 1. GIT + # 1. GIT (Va maintenant pull depuis ton Forgejo auto-hébergé) subprocess.run(["git", "fetch", "origin"], check=True) subprocess.run(["git", "reset", "--hard", f"origin/{branch}"], check=True) - # 2. RSYNC (si on utilise le fallback de dépôt) + # 2. RSYNC (Fallback) if use_rsync: subprocess.run(["rsync", "-a", "--exclude=.git", "--exclude=venv", "--exclude=.venv", "--exclude=__pycache__", f"{target_path}/", f"{PROJECT_PATH}/"], check=True) - # ⚡ Migration automatique GitLab tracking - send_discord_log("MIGRATION", "Migration GitLab tracking en cours...", 16776960, author, commit_msg) - try: - subprocess.run( - ["python3", os.path.join(target_path, "scripts", "migrate_gitlab_tracking.py")], - cwd=target_path, - check=True, - capture_output=True, - text=True - ) - send_discord_log("MIGRATION", "✅ Migration GitLab tracking terminée.", 3066993, author, commit_msg) - except subprocess.CalledProcessError as e: - send_discord_log("ERREUR MIGRATION", f"❌ Migration GitLab échouée: {e.stderr}", 15158332, author, commit_msg) - - # 3. PM2 (Appelé directement via son chemin absolu pour corriger l'erreur 127) + # 3. PM2 Restart subprocess.run( [PM2_BINARY, "restart", PROCESS_NAME], check=True, @@ -110,26 +97,35 @@ def run_mep_logic(author, commit_msg, branch="main"): ) duration = (datetime.now() - start_time).total_seconds() - msg = f"✅ **Déploiement réussi en {duration:.2f}s**\nLe bot est à jour et redémarré." + msg = f"✅ **Déploiement réussi en {duration:.2f}s**\nLe bot est à jour sur Forgejo et redémarré." send_discord_log("VALIDÉ", msg, 3066993, author, commit_msg) except subprocess.CalledProcessError as e: - # Capture spécifique si une commande système (comme Git ou PM2) crash - error_stderr = e.stderr if e.stderr else "Pas de log d'erreur disponible (stderr vide)." + error_stderr = e.stderr if e.stderr else "Pas de log d'erreur disponible." error_msg = f"❌ **ERREUR de commande (Status {e.returncode})**\nCommande : `{ ' '.join(e.cmd) }`\n\n**Détails :**\n```\n{error_stderr}\n```" send_discord_log("ÉCHEC", error_msg, 15158332, author, commit_msg) except Exception as e: - # Capture globale pour le reste du script Python error_msg = f"❌ **ERREUR de déploiement**\n```python\n{str(e)}\n```" send_discord_log("ÉCHEC", error_msg, 15158332, author, commit_msg) @app.route('/deploy', methods=['POST']) def deploy(): - # Sécurité - if request.headers.get('X-Gitlab-Token') != WEBHOOK_SECRET: + # 1. Vérification de la signature HMAC + signature_recue = request.headers.get('X-Hub-Signature-256') + if not signature_recue: abort(403) - # Extraction des infos GitLab + payload_brut = request.get_data() + signature_attendue = "sha256=" + hmac.new( + WEBHOOK_SECRET.encode('utf-8'), + payload_brut, + hashlib.sha256 + ).hexdigest() + + if not hmac.compare_digest(signature_attendue, signature_recue): + abort(403) + + # 2. Logique de déploiement si la signature est valide data = request.json author = "Inconnu" commit_msg = "Pas de message" @@ -138,45 +134,54 @@ def deploy(): author = data['commits'][0].get('author', {}).get('name', 'Inconnu') commit_msg = data['commits'][0].get('message', 'Pas de message') - # Extraction de la branche ref = data.get('ref', 'refs/heads/main') branch = ref.split('/')[-1] if ref else 'main' - # Threading pour répondre vite à GitLab threading.Thread(target=run_mep_logic, args=(author, commit_msg, branch)).start() return {"status": "accepted"}, 202 -@app.route('/gitlab_webhook', methods=['POST']) -def gitlab_webhook(): - received_token = request.headers.get('X-Gitlab-Token') - print(f"[DEBUG] Webhook GitLab reçu. Token reçu: '{received_token}'") - - if received_token != WEBHOOK_SECRET: - abort(403) - - payload = request.json - relay_payload_to_bot(payload) - return {"status": "received"}, 200 - - +# Nouvelle route renommée pour la clarté, accepte aussi l'ancienne au cas où @app.route('/forgejo_webhook', methods=['POST']) +@app.route('/gitlab_webhook', methods=['POST']) def forgejo_webhook(): + # 1. On récupère la signature envoyée par Forgejo/Gitea + signature_recue = None + for header_name in ('X-Hub-Signature-256', 'X-Gitea-Signature', 'X-Gogs-Signature'): + signature_recue = request.headers.get(header_name) + if signature_recue: + break + + # 2. On lit les données brutes envoyées par Forgejo payload_brut = request.get_data() - received_signature = ( - request.headers.get('X-Hub-Signature-256') - or request.headers.get('X-Gitea-Signature') - or request.headers.get('X-Gogs-Signature') - ) - if received_signature and not verify_signature(payload_brut, received_signature): - print("[!] Signature Forgejo invalide", flush=True) - abort(403) + # 3. Vérification de signature (compatible GitHub/Gitea/Gogs) + if signature_recue: + signature_attendue = "sha256=" + hmac.new( + WEBHOOK_SECRET.encode('utf-8'), + payload_brut, + hashlib.sha256 + ).hexdigest() + if not hmac.compare_digest(signature_attendue, signature_recue): + print("[!] Signature invalide ! Hacking intercepté 🛡️", flush=True) + abort(403) + else: + # Fallback pratique pour les environnements locaux / tests où Forgejo n'envoie pas d'en-tête de signature. + print("[WARN] Aucune signature détectée, relayage du webhook sans vérification de signature.", flush=True) + + print("[DEBUG] Signature validée avec succès !", flush=True) + + # 4. Si tout est bon, on transmet au bot sur le port 5001 payload = request.json - relay_payload_to_bot(payload) + try: + requests.post(BOT_LOCAL_URL, json=payload, timeout=5) + except Exception as e: + print(f"[!] Erreur de relais vers le bot Discord : {e}", flush=True) + return {"status": "received"}, 200 if __name__ == '__main__': - print(f"[*] Webhook opérationnel. En attente de push ou d'events...") - app.run(host='0.0.0.0', port=5000) \ No newline at end of file + print(f"[*] Webhook Forgejo opérationnel sur le port 5000. Prêt pour les MEP !") + app.run(host='0.0.0.0', port=5000) + diff --git a/src/logger.py b/src/logger.py index ac149d8..8c4aaee 100755 --- a/src/logger.py +++ b/src/logger.py @@ -52,7 +52,7 @@ class KubyLogger: # GitLab handler for automated error reporting try: - from utils.gitlab_client import gitlab_client + from utils.forgejo_client import gitlab_client class GitLabHandler(logging.Handler): def __init__(self, client): super().__init__() diff --git a/utils/forgejo_client.py b/utils/forgejo_client.py new file mode 100644 index 0000000..1768b73 --- /dev/null +++ b/utils/forgejo_client.py @@ -0,0 +1,173 @@ +import aiohttp +import os +import logging +from datetime import datetime, timedelta, timezone +from typing import Optional, List + +# Logger interne mis à jour pour Forgejo +forgejo_internal_logger = logging.getLogger("ForgejoClient") +forgejo_internal_logger.setLevel(logging.INFO) + +# Alias de compatibilité pour les anciens imports du bot +gitlab_internal_logger = forgejo_internal_logger + +class ForgejoClient: + # MAP REEL DES LABELS (Pêchés directement depuis API Forgejo) + LABEL_MAPPING = { + "bug": 10, + "suggestion": 11, # Redirige "suggestion" vers "enhancement" + "enhancement": 11, # ID de ton étiquette bleue + "manual-report": 15, + "manual-suggestion": 16 + } + + def __init__(self): + self.token = os.getenv("FORGEJO_TOKEN") + self.owner = os.getenv("FORGEJO_OWNER") # Exemple: "Gameur" + self.repo = os.getenv("FORGEJO_REPO") # Exemple: "kuby" + self.assignee = os.getenv("FORGEJO_ASSIGNEE") # Pseudo string (optionnel) + + # URL de base automatique via ton Tailscale. + # FIX DE SECURITE : On force le protocole HTTP brut pour éviter l'erreur "ssl:default" sur le port 80 + configured_url = os.getenv("FORGEJO_URL", "http://omegakubeserv.tail951d2f.ts.net").strip().rstrip("/") + if configured_url.startswith("https://"): + configured_url = configured_url.replace("https://", "http://", 1) + self.base_url = configured_url + + # Construction de l'URL de l'API v1 standard de Forgejo/Gitea + if "api/v1" not in self.base_url: + self.api_url = f"{self.base_url}/api/v1/repos/{self.owner}/{self.repo}/issues" + else: + self.api_url = f"{self.base_url}/repos/{self.owner}/{self.repo}/issues" + + async def create_issue(self, title: str, description: str, labels: Optional[List[str]] = None) -> Optional[dict]: + """ + Crée un ticket sur Forgejo avec les bons IDs numériques de labels. + """ + if not self.token or not self.owner or not self.repo: + if not hasattr(self, '_config_warned'): + forgejo_internal_logger.warning("⚠️ L'intégration Forgejo n'est pas configurée (token/owner/repo manquants).") + self._config_warned = True + return None + + headers = { + "Authorization": f"token {self.token}", + "Content-Type": "application/json" + } + + # CONVERSION DES LABELS : On transforme les chaînes ("bug") en entiers (10) attendus par l'API Go + forgejo_label_ids = [] + if labels: + for label in labels: + label_id = self.LABEL_MAPPING.get(label.lower()) + if label_id: + forgejo_label_ids.append(label_id) + + # Payload au format Forgejo strict + payload = { + "title": title, + "body": description, + "labels": forgejo_label_ids # Reçoit désormais un tableau d'int64 (ex: [10]) + } + + if self.assignee: + payload["assignees"] = [self.assignee] + + timeout = aiohttp.ClientTimeout(total=15) + async with aiohttp.ClientSession(timeout=timeout) as session: + try: + async with session.post(self.api_url, headers=headers, json=payload) as response: + if response.status == 201: + result = await response.json() + forgejo_internal_logger.info(f"✅ Forgejo issue créée : {result.get('html_url')}") + return result + else: + error_text = await response.text() + forgejo_internal_logger.error(f"❌ Échec de la création : {response.status} - {error_text}") + return None + except Exception as e: + forgejo_internal_logger.error(f"❌ Erreur de connexion à l'API Forgejo : {e}") + return None + + async def find_issue_by_title(self, title: str, state: str = "open") -> Optional[dict]: + """ + Recherche un ticket par son titre exact et son état (open/closed). + """ + if not self.token or not self.owner or not self.repo: + return None + + headers = {"Authorization": f"token {self.token}"} + forgejo_state = "open" if state == "opened" else state + + params = { + "q": title, + "state": forgejo_state + } + + async with aiohttp.ClientSession() as session: + try: + async with session.get(self.api_url, headers=headers, params=params) as response: + if response.status == 200: + issues = await response.json() + for issue in issues: + if issue.get("title") == title: + return issue + return None + else: + return None + except Exception as e: + forgejo_internal_logger.error(f"❌ Erreur lors de la recherche Forgejo : {e}") + return None + + async def get_closed_issues(self) -> List[dict]: + """ + Récupère les tickets récemment fermés (depuis 24h). + """ + if not self.token or not self.owner or not self.repo: + return [] + + headers = {"Authorization": f"token {self.token}"} + yesterday = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat() + + params = { + "state": "closed", + "since": yesterday + } + + async with aiohttp.ClientSession() as session: + try: + async with session.get(self.api_url, headers=headers, params=params) as response: + if response.status == 200: + return await response.json() + else: + error_text = await response.text() + forgejo_internal_logger.error(f"❌ Erreur récupération tickets fermés : {response.status} - {error_text}") + return [] + except Exception as e: + forgejo_internal_logger.error(f"❌ Erreur connexion API Forgejo : {e}") + return [] + + async def get_issue(self, issue_number: int) -> Optional[dict]: + """ + Récupère un ticket spécifique via son numéro (ID interne). + """ + if not self.token or not self.owner or not self.repo: + return None + + headers = {"Authorization": f"token {self.token}"} + url = f"{self.api_url}/{issue_number}" + + async with aiohttp.ClientSession() as session: + try: + async with session.get(url, headers=headers) as response: + if response.status == 200: + return await response.json() + else: + return None + except Exception as e: + forgejo_internal_logger.error(f"❌ Erreur connexion API Forgejo : {e}") + return None + +# --- SÉCURITÉ ET ALIAS DE COMPATIBILITÉ --- +GitLabClient = ForgejoClient +gitlab_client = ForgejoClient() # Singleton conservé sous l'ancien nom pour le reste du bot \ No newline at end of file diff --git a/utils/gitlab_client.py b/utils/gitlab_client.py index 40d7993..8fc9ecd 100644 --- a/utils/gitlab_client.py +++ b/utils/gitlab_client.py @@ -1,143 +1,16 @@ -import aiohttp -import os -import logging -from datetime import datetime, timedelta -from typing import Optional, List +"""Compatibilité rétroactive pour les anciens imports du bot. -# Use a separate logger to avoid infinite loops if GitLab reporting fails -gitlab_internal_logger = logging.getLogger("GitLabClient") -gitlab_internal_logger.setLevel(logging.INFO) -# We don't add handlers here, it will just log to console through hierarchy or be silent if needed +Certaines commandes ou anciens modules peuvent encore importer +`utils.gitlab_client`. Ce fichier sert de pont vers le client Forgejo +actuel sans casser le chargement du bot. +""" -class GitLabClient: - def __init__(self): - self.token = os.getenv("GITLAB_TOKEN") - self.project_id = os.getenv("GITLAB_PROJECT_ID") - self.assignee_id = os.getenv("GITLAB_ASSIGNEE_ID") - self.base_url = os.getenv("GITLAB_URL", "https://gitlab.com").rstrip("/") - # Ensure base_url only contains the schema and domain for the API endpoint - if "api/v4" not in self.base_url: - parts = self.base_url.split("/") - api_base = "/".join(parts[:3]) # Gets https://gitlab.com - self.api_url = f"{api_base}/api/v4/projects/{self.project_id}/issues" - else: - self.api_url = f"{self.base_url}/projects/{self.project_id}/issues" +from utils.forgejo_client import ForgejoClient, forgejo_internal_logger, gitlab_internal_logger - async def create_issue(self, title: str, description: str, labels: Optional[List[str]] = None) -> Optional[dict]: - """ - Creates an issue on GitLab. - """ - if not self.token or not self.project_id: - # Use a print or a lower level log to avoid triggering the GitLab logger itself - # which could cause an infinite loop if this were an ERROR log. - if not hasattr(self, '_config_warned'): - print("⚠️ GitLab integration is not configured. GitLab issues will not be created.") - self._config_warned = True - return None +# Nom conservé pour ne pas casser les imports existants +# (les anciens cogs utilisent souvent `gitlab_client`) +gitlab_client = ForgejoClient() - headers = { - "PRIVATE-TOKEN": self.token - } - - data = { - "title": title, - "description": description, - "labels": ",".join(labels) if labels else "" - } - - if self.assignee_id: - data["assignee_ids"] = [self.assignee_id] - - async with aiohttp.ClientSession() as session: - try: - async with session.post(self.api_url, headers=headers, data=data) as response: - if response.status == 201: - result = await response.json() - gitlab_internal_logger.info(f"✅ GitLab issue created: {result.get('web_url')}") - return result - else: - error_text = await response.text() - gitlab_internal_logger.error(f"❌ Failed to create GitLab issue: {response.status} - {error_text}") - return None - except Exception as e: - gitlab_internal_logger.error(f"❌ Error connecting to GitLab API: {e}") - return None - - async def find_issue_by_title(self, title: str, state: str = "opened") -> Optional[dict]: - """ - Searches for an issue by its exact title and state. - """ - if not self.token or not self.project_id: - return None - - headers = {"PRIVATE-TOKEN": self.token} - params = { - "search": title, - "state": state - } - - async with aiohttp.ClientSession() as session: - try: - # search param is fuzzy on GitLab, so we must double-check for exact match - async with session.get(self.api_url, headers=headers, params=params) as response: - if response.status == 200: - issues = await response.json() - for issue in issues: - if issue.get("title") == title: - return issue - return None - else: - return None - except Exception as e: - gitlab_internal_logger.error(f"❌ Error searching GitLab issues: {e}") - return None - - async def get_closed_issues(self) -> List[dict]: - """ - Fetches recently closed issues. - """ - if not self.token or not self.project_id: - return [] - - headers = {"PRIVATE-TOKEN": self.token} - params = { - "state": "closed", - "updated_after": (datetime.now() - timedelta(days=1)).isoformat() - } - - async with aiohttp.ClientSession() as session: - try: - async with session.get(self.api_url, headers=headers, params=params) as response: - if response.status == 200: - return await response.json() - else: - error_text = await response.text() - gitlab_internal_logger.error(f"❌ Failed to fetch closed issues: {response.status} - {error_text}") - return [] - except Exception as e: - gitlab_internal_logger.error(f"❌ Error connecting to GitLab API: {e}") - return [] - - async def get_issue(self, issue_iid: int) -> Optional[dict]: - """ - Fetches a specific issue by its internal ID (IID). - """ - if not self.token or not self.project_id: - return None - - headers = {"PRIVATE-TOKEN": self.token} - url = f"{self.base_url}/api/v4/projects/{self.project_id}/issues/{issue_iid}" - - async with aiohttp.ClientSession() as session: - try: - async with session.get(url, headers=headers) as response: - if response.status == 200: - return await response.json() - else: - return None - except Exception as e: - gitlab_internal_logger.error(f"❌ Error connecting to GitLab API: {e}") - return None - -# Instance unique -gitlab_client = GitLabClient() +# Exposition explicite du logger pour compatibilité +# pylint: disable=invalid-name +gitlab_internal_logger = gitlab_internal_logger