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") # Configure SSL verification with certificate bundle ssl_cert_file = os.getenv("SSL_CERT_FILE") requests_ca_bundle = os.getenv("REQUESTS_CA_BUNDLE") # Use certificate bundle if available, otherwise fall back to boolean verification if ssl_cert_file and os.path.exists(ssl_cert_file): self.verify_ssl = ssl_cert_file elif requests_ca_bundle and os.path.exists(requests_ca_bundle): self.verify_ssl = requests_ca_bundle else: # Check for project-local certificate bundle project_ca_bundle = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "certs", "mei_sheng_ca_bundle.pem") if os.path.exists(project_ca_bundle): self.verify_ssl = project_ca_bundle else: 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 []