Commit Helper MCP

""" Manages repository targeting and multi-repository operations. This service handles repository path resolution, validation, and manages repository-specific service instances. """ import logging import os from pathlib import Path from typing import Dict, Optional, Tuple, Any from .commitizen_core import CommitzenCore from .gitpython_core import GitPythonCore from ..config import get_settings, RepositoryConfig from ..errors import ( handle_errors, RepositoryError, ConfigurationError, ServiceError, create_success_response, ) logger = logging.getLogger(__name__) class RepositoryManager: """Manages repository targeting and multi-repository operations.""" def __init__(self): """Initialize repository manager with empty cache.""" self.settings = get_settings() self._repository_cache: Dict[ str, Tuple[CommitzenCore, Optional[GitPythonCore]] ] = {} self._repository_configs: Dict[str, RepositoryConfig] = {} logger.info("Initialized RepositoryManager") @handle_errors(log_errors=True) def resolve_repository_path(self, repo_path: Optional[str] = None) -> str: """ Resolve repository path with priority order. Priority: 1. Explicit repo_path parameter 2. Settings default_repo_path (from environment or config) 3. Current working directory (default behavior) Args: repo_path: Optional explicit repository path Returns: Resolved repository path as string """ # Priority 1: Explicit parameter if repo_path: resolved_path = Path(repo_path).resolve() logger.debug(f"Using explicit repo_path: {resolved_path}") return str(resolved_path) # Priority 2: Settings (which includes environment variable) if self.settings.default_repo_path: resolved_path = Path(self.settings.default_repo_path).resolve() logger.debug(f"Using default_repo_path from settings: {resolved_path}") return str(resolved_path) # Priority 3: Current working directory resolved_path = Path.cwd() logger.debug(f"Using current working directory: {resolved_path}") return str(resolved_path) @handle_errors(log_errors=True) def validate_repository_access(self, repo_path: str) -> bool: """ Validate repository exists and is accessible. Args: repo_path: Repository path to validate Returns: True if repository is valid and accessible """ try: path = Path(repo_path) # Check if path exists if not path.exists(): logger.warning(f"Repository path does not exist: {repo_path}") return False # Check if it's a directory if not path.is_dir(): logger.warning(f"Repository path is not a directory: {repo_path}") return False # Check if we have read access if not os.access(repo_path, os.R_OK): logger.warning(f"No read access to repository: {repo_path}") return False # Check if it's a git repository git_dir = path / ".git" if not git_dir.exists(): logger.info(f"Not a git repository (no .git directory): {repo_path}") # This is okay - we can still use Commitizen without git return True return True except Exception as e: logger.error(f"Error validating repository access: {e}") return False @handle_errors(log_errors=True) def get_repository_services( self, repo_path: Optional[str] = None ) -> Tuple[CommitzenCore, Optional[GitPythonCore]]: """ Get or create services for specific repository. Uses caching to avoid recreating services for the same repository. Args: repo_path: Optional repository path Returns: Tuple of (CommitzenCore, GitPythonCore or None) """ # Resolve repository path resolved_path = self.resolve_repository_path(repo_path) # Check cache if resolved_path in self._repository_cache: logger.debug(f"Using cached services for: {resolved_path}") return self._repository_cache[resolved_path] # Validate repository access if not self.validate_repository_access(resolved_path): raise RepositoryError( f"Invalid or inaccessible repository: {resolved_path}", repo_path=resolved_path, ) # Create new services logger.info(f"Creating new services for repository: {resolved_path}") # Create CommitzenCore (always available) try: commitizen_core = CommitzenCore(repo_path=resolved_path) except Exception as e: raise ConfigurationError( f"Failed to initialize CommitzenCore: {e}", config_file=resolved_path, cause=e, ) # Try to create GitPythonCore gitpython_core = None try: gitpython_core = GitPythonCore(repo_path=resolved_path) logger.info(f"GitPython service created for: {resolved_path}") except RepositoryError as e: # This is expected for non-git repositories logger.info( f"Not a git repository, GitPython service disabled: {resolved_path}" ) except Exception as e: logger.warning(f"Failed to create GitPython service: {e}") # Cache the services self._repository_cache[resolved_path] = (commitizen_core, gitpython_core) return commitizen_core, gitpython_core @handle_errors(log_errors=True) def manage_environment_variables(self) -> Dict[str, str]: """ Handle COMMITIZEN_REPO_PATH and other environment variables. Returns: Dict of relevant environment variables and their values """ env_vars = {} # Check for COMMITIZEN_REPO_PATH if "COMMITIZEN_REPO_PATH" in os.environ: env_vars["COMMITIZEN_REPO_PATH"] = os.environ["COMMITIZEN_REPO_PATH"] logger.info( f"COMMITIZEN_REPO_PATH set to: {env_vars['COMMITIZEN_REPO_PATH']}" ) # Check for other relevant environment variables git_vars = ["GIT_DIR", "GIT_WORK_TREE", "GIT_AUTHOR_NAME", "GIT_AUTHOR_EMAIL"] for var in git_vars: if var in os.environ: env_vars[var] = os.environ[var] return env_vars @handle_errors(log_errors=True) def clear_repository_cache(self, repo_path: Optional[str] = None) -> None: """ Clear cached repository services. Args: repo_path: Optional specific repository to clear, or None to clear all """ if repo_path: resolved_path = self.resolve_repository_path(repo_path) if resolved_path in self._repository_cache: del self._repository_cache[resolved_path] logger.info(f"Cleared cache for repository: {resolved_path}") else: self._repository_cache.clear() logger.info("Cleared all repository caches") @handle_errors(log_errors=True) def get_repository_config( self, repo_path: Optional[str] = None ) -> RepositoryConfig: """ Get repository configuration, using cache if available. Args: repo_path: Optional repository path Returns: RepositoryConfig instance """ resolved_path = self.resolve_repository_path(repo_path) if resolved_path not in self._repository_configs: try: self._repository_configs[resolved_path] = RepositoryConfig( resolved_path ) except Exception as e: raise ConfigurationError( f"Failed to load repository configuration: {e}", config_file=resolved_path, cause=e, ) return self._repository_configs[resolved_path] @handle_errors(log_errors=True) def get_cached_repositories(self) -> Dict[str, Dict[str, Any]]: """ Get information about all cached repositories. Returns: Dict with repository paths and their service status """ repositories = {} for repo_path, ( commitizen_core, gitpython_core, ) in self._repository_cache.items(): repositories[repo_path] = { "has_commitizen": commitizen_core is not None, "has_git": gitpython_core is not None, "commitizen_plugin": commitizen_core.committer.__class__.__name__ if commitizen_core else None, "git_enabled": gitpython_core is not None, } return repositories @handle_errors(log_errors=True) def switch_repository(self, new_repo_path: str) -> Dict[str, Any]: """ Switch to a different repository and return its status. Args: new_repo_path: Path to the new repository Returns: Dict with repository switch status and information """ # Validate the new repository if not self.validate_repository_access(new_repo_path): raise RepositoryError( f"Invalid or inaccessible repository: {new_repo_path}", repo_path=new_repo_path, ) # Get services for the new repository commitizen_core, gitpython_core = self.get_repository_services(new_repo_path) # Get repository information repo_info = { "repository_path": new_repo_path, "has_commitizen": True, "has_git": gitpython_core is not None, "commitizen_plugin": commitizen_core.committer.__class__.__name__, "git_enabled": gitpython_core is not None, } # Add git status if available if gitpython_core: try: status = gitpython_core.get_repository_status() repo_info["git_status"] = { "current_branch": status.get("current_branch"), "staged_files_count": status.get("staged_files_count", 0), "unstaged_files_count": status.get("unstaged_files_count", 0), "untracked_files_count": status.get("untracked_files_count", 0), } except Exception as e: logger.warning(f"Could not get git status: {e}") repo_info["git_status_error"] = str(e) return create_success_response(repo_info)