Kuby/utils/gitlab_client.py

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.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"

    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 ""
        }
        
        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()