nomad_mcp/app/services/gitea_client.py

import os
import logging
import requests
from typing import Dict, Any, List, Optional, Tuple
from urllib.parse import urlparse
from fastapi import HTTPException

# Configure logging
logger = logging.getLogger(__name__)

class GiteaClient:
    """Client for interacting with Gitea API."""
    
    def __init__(self):
        """Initialize Gitea client with configuration from environment variables."""
        self.api_base_url = os.getenv("GITEA_API_URL", "").rstrip("/")
        self.token = os.getenv("GITEA_API_TOKEN")
        self.username = os.getenv("GITEA_USERNAME")
        self.verify_ssl = os.getenv("GITEA_VERIFY_SSL", "true").lower() == "true"
        
        if not self.api_base_url:
            logger.warning("GITEA_API_URL is not configured. Gitea integration will not work.")
        
        if not self.token and (self.username and os.getenv("GITEA_PASSWORD")):
            self.token = self._get_token_from_credentials()
    
    def _get_token_from_credentials(self) -> Optional[str]:
        """Get a token using username and password if provided."""
        try:
            response = requests.post(
                f"{self.api_base_url}/users/{self.username}/tokens",
                auth=(self.username, os.getenv("GITEA_PASSWORD", "")),
                json={
                    "name": "nomad-mcp-service",
                    "scopes": ["repo", "read:org"]
                },
                verify=self.verify_ssl
            )
            
            if response.status_code == 201:
                return response.json().get("sha1")
            else:
                logger.error(f"Failed to get Gitea token: {response.text}")
                return None
        except Exception as e:
            logger.error(f"Failed to get Gitea token: {str(e)}")
            return None
    
    def _get_headers(self) -> Dict[str, str]:
        """Get request headers with authentication."""
        headers = {
            "Content-Type": "application/json",
            "Accept": "application/json"
        }
        
        if self.token:
            headers["Authorization"] = f"token {self.token}"
        
        return headers
    
    def parse_repo_url(self, repo_url: str) -> Tuple[str, str]:
        """
        Parse a Gitea repository URL to extract owner and repo name.
        
        Examples:
        - http://gitea.internal.example.com/username/repo-name -> (username, repo-name)
        - https://gitea.example.com/org/project -> (org, project)
        """
        try:
            # Parse the URL
            parsed_url = urlparse(repo_url)
            
            # Get the path and remove leading/trailing slashes
            path = parsed_url.path.strip("/")
            
            # Split the path
            parts = path.split("/")
            
            if len(parts) < 2:
                raise ValueError(f"Invalid repository URL: {repo_url}")
            
            # Extract owner and repo
            owner = parts[0]
            repo = parts[1]
            
            return owner, repo
        except Exception as e:
            logger.error(f"Failed to parse repository URL: {repo_url}, error: {str(e)}")
            raise ValueError(f"Invalid repository URL: {repo_url}")
    
    def check_repository_exists(self, repo_url: str) -> bool:
        """Check if a repository exists in Gitea."""
        if not self.api_base_url:
            # No Gitea integration configured, assume repository exists
            return True
        
        try:
            owner, repo = self.parse_repo_url(repo_url)
            
            response = requests.get(
                f"{self.api_base_url}/repos/{owner}/{repo}",
                headers=self._get_headers(),
                verify=self.verify_ssl
            )
            
            return response.status_code == 200
        except Exception as e:
            logger.error(f"Failed to check repository: {repo_url}, error: {str(e)}")
            return False
    
    def get_repository_info(self, repo_url: str) -> Optional[Dict[str, Any]]:
        """Get repository information from Gitea."""
        if not self.api_base_url:
            # No Gitea integration configured
            return None
        
        try:
            owner, repo = self.parse_repo_url(repo_url)
            
            response = requests.get(
                f"{self.api_base_url}/repos/{owner}/{repo}",
                headers=self._get_headers(),
                verify=self.verify_ssl
            )
            
            if response.status_code == 200:
                return response.json()
            else:
                logger.error(f"Failed to get repository info: {response.text}")
                return None
        except Exception as e:
            logger.error(f"Failed to get repository info: {repo_url}, error: {str(e)}")
            return None
    
    def list_repositories(self, limit: int = 100) -> List[Dict[str, Any]]:
        """List available repositories from Gitea."""
        if not self.api_base_url:
            # No Gitea integration configured
            return []
        
        try:
            response = requests.get(
                f"{self.api_base_url}/user/repos",
                headers=self._get_headers(),
                params={"limit": limit},
                verify=self.verify_ssl
            )
            
            if response.status_code == 200:
                return response.json()
            else:
                logger.error(f"Failed to list repositories: {response.text}")
                return []
        except Exception as e:
            logger.error(f"Failed to list repositories: {str(e)}")
            return []
    
    def get_repository_branches(self, repo_url: str) -> List[Dict[str, Any]]:
        """Get branches for a repository."""
        if not self.api_base_url:
            # No Gitea integration configured
            return []
        
        try:
            owner, repo = self.parse_repo_url(repo_url)
            
            response = requests.get(
                f"{self.api_base_url}/repos/{owner}/{repo}/branches",
                headers=self._get_headers(),
                verify=self.verify_ssl
            )
            
            if response.status_code == 200:
                return response.json()
            else:
                logger.error(f"Failed to get repository branches: {response.text}")
                return []
        except Exception as e:
            logger.error(f"Failed to get repository branches: {repo_url}, error: {str(e)}")
            return []