Ajout d'un système de signalisation de bugs automatiser et manuel

utils/gitlab_client.py

@@ -0,0 +1,110 @@
+import aiohttp
+import os
+import logging
+from datetime import datetime, timedelta
+from typing import Optional, List
+
+# 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
+
+class GitLabClient:
+    def __init__(self):
+        self.token = os.getenv("GITLAB_TOKEN")
+        self.project_id = os.getenv("GITLAB_PROJECT_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"
+
+    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
+
+        headers = {
+            "PRIVATE-TOKEN": self.token
+        }
+        
+        data = {
+            "title": title,
+            "description": description,
+            "labels": ",".join(labels) if labels else ""
+        }
+
+        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 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()