Update README.md

app/services/gitea_client.py

@@ -0,0 +1,180 @@
+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 []