Kuby/utils/forgejo_client.py

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