Commit Helper MCP

""" Commitizen Service Facade service that coordinates the new service architecture while maintaining backward compatibility with existing code. """ import logging from typing import Dict, List, Any, Optional from .services import ( CommitzenCore, GitPythonCore, CommitOrchestrator, RepositoryManager, ValidationService, ) from .errors import ( handle_errors, RepositoryError, ConfigurationError, ServiceError, create_success_response, ) logger = logging.getLogger(__name__) class CommitzenService: """ Facade service that coordinates the new service architecture. Maintains backward compatibility while delegating to specialized services. """ @handle_errors(log_errors=True) def __init__(self, repo_path: Optional[str] = None): """ Initialize the facade with new service architecture. Args: repo_path: Optional path to git repository. If not provided, will check COMMITIZEN_REPO_PATH environment variable, then fall back to current working directory. """ try: # Initialize repository manager self.repository_manager = RepositoryManager() self.repo_path = self.repository_manager.resolve_repository_path(repo_path) # Get core services for the repository self.commitizen_core, self.git_core = ( self.repository_manager.get_repository_services(repo_path) ) # Initialize orchestrator if git is available self.orchestrator = ( CommitOrchestrator(self.commitizen_core, self.git_core) if self.git_core else None ) # Maintain compatibility properties self.git_enabled = self.git_core is not None self.git_implementation = "GitPython" if self.git_core else None self.git_service = self.git_core # Backward compatibility alias # Maintain compatibility with old properties self.config = self.commitizen_core.config self.committer = self.commitizen_core.committer self.adapter = self.commitizen_core.adapter logger.info(f"Initialized CommitzenService facade for: {self.repo_path}") logger.info(f"Git operations enabled: {self.git_enabled}") except Exception as e: logger.error(f"Failed to initialize CommitzenService: {e}") raise ServiceError( f"Service initialization failed: {e}", service_name="CommitzenService", cause=e, ) # Delegate methods to appropriate services while maintaining existing interface @handle_errors(log_errors=True) def refresh_config(self) -> None: """Reload configuration if it changes.""" self.commitizen_core.refresh_config() # Update local references for compatibility self.config = self.commitizen_core.config self.committer = self.commitizen_core.committer self.adapter = self.commitizen_core.adapter @handle_errors(log_errors=True) def get_questions(self) -> List[Dict[str, Any]]: """Get interactive questions from current plugin.""" return self.commitizen_core.get_questions() @handle_errors(log_errors=True) def generate_message(self, answers: Dict[str, Any]) -> str: """Generate commit message from answers dictionary.""" return self.commitizen_core.generate_message(answers) @handle_errors(log_errors=True) def validate_message(self, message: str) -> bool: """Validate commit message format.""" return self.commitizen_core.validate_message(message) @handle_errors(log_errors=True) def get_example(self) -> str: """Get example commit message.""" return self.commitizen_core.get_example() @handle_errors(log_errors=True) def get_schema(self) -> Dict[str, Any]: """Get commit message schema.""" return self.commitizen_core.get_schema() @handle_errors(log_errors=True) def get_info(self) -> Dict[str, Any]: """Get comprehensive plugin information.""" return self.commitizen_core.get_info() @handle_errors(log_errors=True) def get_commit_types(self) -> List[Dict[str, str]]: """Extract available commit types from questions.""" return self.commitizen_core.get_commit_types() @handle_errors(log_errors=True) def preview_commit_operation( self, message: str, sign_off: bool = True, **kwargs ) -> Dict[str, Any]: """ Preview git commit operation with message validation. Delegates to orchestrator for coordinated preview. Args: message: Commit message to preview sign_off: Whether to add sign-off to commit (default: True) **kwargs: Additional commit options Returns: Dict containing preview results and validation status """ if self.orchestrator: return self.orchestrator.preview_commit_operation( message, sign_off=sign_off, **kwargs ) else: raise RepositoryError( "Git operations not available - not in a git repository", repo_path=self.repo_path, ) @handle_errors(log_errors=True) def execute_commit_operation( self, message: str, force_execute: bool = False, sign_off: bool = True, **kwargs ) -> Dict[str, Any]: """ Execute git commit with full validation pipeline. Delegates to orchestrator for coordinated execution. Args: message: Commit message force_execute: Must be True for actual execution sign_off: Whether to add sign-off to commit (default: True) **kwargs: Additional commit options Returns: Dict containing execution results """ if self.orchestrator: return self.orchestrator.execute_commit_operation( message, force_execute=force_execute, sign_off=sign_off, **kwargs ) else: raise RepositoryError( "Git operations not available - not in a git repository", repo_path=self.repo_path, ) @handle_errors(log_errors=True) def get_git_implementation_info(self) -> Dict[str, Any]: """Get information about the current git implementation.""" return create_success_response( { "git_enabled": self.git_enabled, "implementation": self.git_implementation, "enhanced_features": self.git_enabled, # GitPython always has enhanced features "features": { "basic_operations": self.git_enabled, "enhanced_status": self.git_enabled, "detailed_diffs": self.git_enabled, "commit_statistics": self.git_enabled, "repository_analytics": self.git_enabled, }, } ) @handle_errors(log_errors=True) def get_repository_status(self) -> Dict[str, Any]: """ Get git repository status if available. Returns: Dict containing repository status or error info """ # Get repository path for error responses repo_path = None if self.git_service: repo_path = str(self.git_service.repo_path) if not self.git_enabled: raise RepositoryError( "Git operations not available - not in a git repository", repo_path=repo_path, ) status = self.git_service.get_repository_status() status["git_enabled"] = True status["implementation"] = self.git_implementation return status