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)

class ForgejoClient:
    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")  # Nom d'utilisateur (optionnel)
        
        # URL de base automatique via ton Tailscale
        self.base_url = os.getenv("FORGEJO_URL", "https://omegakubeserv.tail951d2f.ts.net").rstrip("/")
        
        # 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.
        """
        if not self.token or not self.owner or not self.repo:
            if not hasattr(self, '_config_warned'):
                print("⚠️ L'intégration Forgejo n'est pas configurée.")
                self._config_warned = True
            return None

        headers = {
            "Authorization": f"token {self.token}",
            "Content-Type": "application/json"
        }
        
        # Payload au format Forgejo (JSON)
        payload = {
            "title": title,
            "body": description,  # 'description' devient 'body'
            "labels": labels if labels else []  # Devient une vraie liste []
        }
        
        if self.assignee:
            payload["assignees"] = [self.assignee] # Liste de pseudos strings

        async with aiohttp.ClientSession() as session:
            try:
                # Forgejo attend du JSON strict (json=payload)
                async with session.post(self.api_url, headers=headers, json=payload) as response:
                    if response.status == 201:
                        result = await response.json()
                        # Forgejo renvoie l'URL dans 'html_url' au lieu de 'web_url'
                        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}"}
        # GitLab utilise 'opened', Forgejo utilise 'open'
        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}"}
        # Forgejo utilise 'since' au format ISO pour filtrer par date
        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

# Instance unique
gitlab_client = ForgejoClient()  # Nom conservé pour éviter de réécrire les imports du bot