Commit Helper MCP

""" Plugin Adapters for Commitizen MCP Connector Provides adapter pattern for different Commitizen plugins to handle field mapping and validation consistently. """ import re import logging from typing import Dict, Any, Optional from abc import ABC, abstractmethod from .errors import handle_errors, ValidationError, PluginError logger = logging.getLogger(__name__) class PluginAdapter(ABC): """Base adapter for Commitizen plugins.""" @abstractmethod def map_fields(self, answers: Dict[str, Any]) -> Dict[str, Any]: """Map generic fields to plugin-specific fields.""" pass @abstractmethod def validate_message(self, message: str) -> bool: """Validate message using plugin-specific rules.""" pass @abstractmethod def get_validation_pattern(self) -> Optional[str]: """Get the validation regex pattern for this plugin.""" pass class ConventionalCommitsAdapter(PluginAdapter): """Adapter for ConventionalCommitsCz plugin.""" # Conventional commits pattern - matches the standard format PATTERN = r"^(feat|fix|docs|style|refactor|perf|test|build|ci|chore|revert)(\(.+\))?: .{1,50}" @handle_errors(log_errors=True) def map_fields(self, answers: Dict[str, Any]) -> Dict[str, Any]: """Map to conventional commits format.""" # The ConventionalCommitsCz plugin expects these specific fields mapped = { "prefix": answers.get("type", answers.get("prefix", "")), "subject": answers.get("subject", ""), "body": answers.get("body", ""), "scope": answers.get("scope", ""), "footer": answers.get("footer", ""), "is_breaking_change": answers.get( "breaking", answers.get("is_breaking_change", False) ), } # Ensure all required fields are present (even if empty) for field in ["prefix", "subject", "body", "scope", "footer"]: if field not in mapped: mapped[field] = "" # Ensure boolean field is properly set if "is_breaking_change" not in mapped: mapped["is_breaking_change"] = False logger.debug(f"Mapped fields for ConventionalCommits: {mapped}") return mapped @handle_errors(log_errors=True) def validate_message(self, message: str) -> bool: """Validate using conventional commits pattern.""" if not message or not message.strip(): raise ValidationError( "Commit message cannot be empty", validation_type="commit_message", invalid_value=message, ) # Use the conventional commits pattern is_valid = bool(re.match(self.PATTERN, message.strip())) logger.debug(f"ConventionalCommits validation for '{message}': {is_valid}") return is_valid @handle_errors(log_errors=True) def get_validation_pattern(self) -> str: """Get the validation regex pattern.""" return self.PATTERN class GenericAdapter(PluginAdapter): """Generic adapter for unknown plugins.""" @handle_errors(log_errors=True) def map_fields(self, answers: Dict[str, Any]) -> Dict[str, Any]: """Pass through fields as-is for generic plugins.""" # For unknown plugins, just pass through the fields # but ensure common fields are present mapped = dict(answers) # Add common fields if missing common_fields = ["prefix", "subject", "body", "scope", "footer"] for field in common_fields: if field not in mapped: mapped[field] = "" logger.debug(f"Generic field mapping: {mapped}") return mapped @handle_errors(log_errors=True) def validate_message(self, message: str) -> bool: """Basic validation - just check if message is not empty.""" if not message or not message.strip(): raise ValidationError( "Commit message cannot be empty", validation_type="commit_message", invalid_value=message, ) return True @handle_errors(log_errors=True) def get_validation_pattern(self) -> Optional[str]: """No specific pattern for generic adapter.""" return None class PluginAdapterFactory: """Factory for creating appropriate plugin adapters.""" # Map plugin class names to their adapters ADAPTER_MAP = { "ConventionalCommitsCz": ConventionalCommitsAdapter, "Cz": ConventionalCommitsAdapter, # Alternative name "ConventionalCommits": ConventionalCommitsAdapter, # Another variant } @classmethod @handle_errors(log_errors=True) def create_adapter(cls, plugin_name: str) -> PluginAdapter: """Create appropriate adapter for the given plugin.""" if not plugin_name: raise PluginError("Plugin name cannot be empty", plugin_name="unknown") adapter_class = cls.ADAPTER_MAP.get(plugin_name, GenericAdapter) adapter = adapter_class() logger.info(f"Created {adapter.__class__.__name__} for plugin '{plugin_name}'") return adapter @classmethod @handle_errors(log_errors=True) def get_supported_plugins(cls) -> list: """Get list of supported plugin names.""" return list(cls.ADAPTER_MAP.keys())