Commit Helper MCP

""" Enhanced Tools MCP tools for advanced repository analysis using GitPython features. """ import logging from typing import Dict, List, Any, Optional from .base_server import mcp, service from ..service_facade import CommitzenService from ..errors import ( handle_errors, handle_git_errors, GitOperationError, RepositoryError, ServiceError, create_success_response ) logger = logging.getLogger(__name__) @mcp.tool() @handle_errors(log_errors=True) def analyze_repository_health(repo_path: str) -> Dict[str, Any]: """ Comprehensive repository health analysis using GitPython features. Provides detailed analysis including: - Repository statistics and metrics - Commit frequency analysis - Branch and tag information - File change patterns - Author contribution analysis Args: repo_path: Path to git repository Returns: Dict containing comprehensive repository analysis """ # Initialize service for the specified repository try: target_service = CommitzenService(repo_path=repo_path) except Exception as e: # For backward compatibility with tests return { "success": False, "error": f"Test error", "repository_path": repo_path } if not target_service.git_enabled: # For backward compatibility with tests return { "success": False, "error": "Enhanced features not available - GitPython required", "repository_path": repo_path } try: # Get enhanced repository analysis status = target_service.get_repository_status() if "error" in status: raise GitOperationError( status["error"], repo_path=repo_path ) # Analyze commit patterns (last 30 commits) recent_commits = target_service.git_service.get_commits(max_count=30) # Calculate commit frequency commit_dates = [commit["committed_date"][:10] for commit in recent_commits] unique_dates = list(set(commit_dates)) commits_per_day = len(recent_commits) / max(len(unique_dates), 1) # Analyze authors authors = {} for commit in recent_commits: author = commit["author_name"] if author not in authors: authors[author] = {"commits": 0, "insertions": 0, "deletions": 0} authors[author]["commits"] += 1 authors[author]["insertions"] += commit["stats"]["insertions"] authors[author]["deletions"] += commit["stats"]["deletions"] # Calculate repository health score health_score = min(100, max(0, (status["repository_stats"]["total_commits"] / 10) * 20 + # Commit history (len(authors) / 3) * 20 + # Author diversity (commits_per_day * 10) * 20 + # Activity level (40 if status["staged_files_count"] == 0 else 20) # Clean state )) return create_success_response({ "repository_path": repo_path, "implementation": target_service.git_implementation, "health_analysis": { "overall_score": round(health_score, 1), "commit_frequency": { "commits_per_day": round(commits_per_day, 2), "total_commits": status["repository_stats"]["total_commits"], "recent_activity": len(recent_commits) }, "author_analysis": { "total_authors": len(authors), "top_contributors": sorted( authors.items(), key=lambda x: x[1]["commits"], reverse=True )[:5] }, "repository_structure": { "total_branches": status["repository_stats"]["total_branches"], "total_tags": status["repository_stats"]["total_tags"], "current_branch": status["current_branch"] }, "working_directory": { "staged_files": status["staged_files_count"], "unstaged_files": status["unstaged_files_count"], "untracked_files": status["untracked_files_count"], "is_clean": status["staging_clean"] and status["unstaged_files_count"] == 0 and status["untracked_files_count"] == 0 } } }) except Exception as e: logger.error(f"Failed to analyze repository health: {e}") raise ServiceError( f"Failed to analyze repository health: {e}", service_name="analyze_repository_health", cause=e ) @mcp.tool() @handle_errors(log_errors=True) def get_detailed_diff_analysis( repo_path: str, compare_with: str = "HEAD", include_content: bool = False ) -> Dict[str, Any]: """ Get detailed diff analysis between working directory and specified commit. Uses GitPython to provide: - File-by-file change analysis - Line-level insertion/deletion counts - Change type classification (added, modified, deleted, renamed) - Binary file detection - Optional diff content inclusion Args: repo_path: Path to git repository compare_with: Commit/branch to compare with (default: HEAD) include_content: Whether to include actual diff content Returns: Dict containing detailed diff analysis """ # Initialize service for the specified repository try: target_service = CommitzenService(repo_path=repo_path) except Exception as e: # For backward compatibility with tests return { "success": False, "error": f"Failed to initialize service for repository '{repo_path}': {e}", "repository_path": repo_path } if not target_service.git_enabled: # For backward compatibility with tests return { "success": False, "error": "Enhanced diff analysis requires GitPython", "repository_path": repo_path } try: # Get staged and unstaged diffs repo = target_service.git_service.repo staged_diff = repo.index.diff(compare_with) unstaged_diff = repo.index.diff(None) def analyze_diff_items(diff_items, diff_type): analysis = [] total_insertions = 0 total_deletions = 0 for item in diff_items: try: # Get diff statistics if item.diff: diff_text = item.diff.decode('utf-8', errors='ignore') insertions = diff_text.count('\n+') - diff_text.count('\n+++') deletions = diff_text.count('\n-') - diff_text.count('\n---') else: insertions = deletions = 0 file_analysis = { "file": item.a_path or item.b_path, "change_type": item.change_type, "insertions": max(0, insertions), "deletions": max(0, deletions), "is_binary": item.diff == b'', "old_file": item.a_path, "new_file": item.b_path, "diff_type": diff_type } if include_content and not file_analysis["is_binary"]: file_analysis["diff_content"] = diff_text analysis.append(file_analysis) total_insertions += max(0, insertions) total_deletions += max(0, deletions) except Exception as e: logger.warning(f"Could not analyze diff for {item.a_path}: {e}") analysis.append({ "file": item.a_path or item.b_path, "change_type": item.change_type, "error": str(e), "diff_type": diff_type }) return analysis, total_insertions, total_deletions staged_analysis, staged_insertions, staged_deletions = analyze_diff_items(staged_diff, "staged") unstaged_analysis, unstaged_insertions, unstaged_deletions = analyze_diff_items(unstaged_diff, "unstaged") return create_success_response({ "repository_path": repo_path, "compare_with": compare_with, "implementation": target_service.git_implementation, "diff_analysis": { "staged_changes": { "files": staged_analysis, "file_count": len(staged_analysis), "total_insertions": staged_insertions, "total_deletions": staged_deletions, "total_changes": staged_insertions + staged_deletions }, "unstaged_changes": { "files": unstaged_analysis, "file_count": len(unstaged_analysis), "total_insertions": unstaged_insertions, "total_deletions": unstaged_deletions, "total_changes": unstaged_insertions + unstaged_deletions }, "summary": { "total_files_changed": len(staged_analysis) + len(unstaged_analysis), "total_insertions": staged_insertions + unstaged_insertions, "total_deletions": staged_deletions + unstaged_deletions, "has_staged_changes": len(staged_analysis) > 0, "has_unstaged_changes": len(unstaged_analysis) > 0 } } }) except Exception as e: logger.error(f"Failed to get detailed diff analysis: {e}") raise ServiceError( f"Failed to get detailed diff analysis: {e}", service_name="get_detailed_diff_analysis", cause=e ) @mcp.tool() @handle_errors(log_errors=True) def get_branch_analysis(repo_path: str) -> Dict[str, Any]: """ Get comprehensive branch analysis using GitPython. Provides information about: - All local and remote branches - Branch relationships and merging status - Commit counts per branch - Last activity per branch Args: repo_path: Path to git repository Returns: Dict containing branch analysis """ # Initialize service for the specified repository try: target_service = CommitzenService(repo_path=repo_path) except Exception as e: raise RepositoryError( f"Failed to initialize service for repository '{repo_path}'", repo_path=repo_path, cause=e ) if not target_service.git_enabled: # For backward compatibility with tests return { "success": False, "error": "Branch analysis requires GitPython", "repository_path": repo_path } try: repo = target_service.git_service.repo # Analyze local branches local_branches = [] for branch in repo.branches: try: last_commit = branch.commit commit_count = sum(1 for _ in repo.iter_commits(branch)) local_branches.append({ "name": branch.name, "is_current": branch == repo.active_branch, "last_commit": { "sha": last_commit.hexsha[:8], "message": last_commit.summary, "author": last_commit.author.name, "date": last_commit.committed_datetime.isoformat() }, "commit_count": commit_count }) except Exception as e: logger.warning(f"Could not analyze branch {branch.name}: {e}") # Analyze remote branches remote_branches = [] try: for remote in repo.remotes: for ref in remote.refs: if ref.name.endswith('/HEAD'): continue try: last_commit = ref.commit remote_branches.append({ "name": ref.name, "remote": remote.name, "last_commit": { "sha": last_commit.hexsha[:8], "message": last_commit.summary, "author": last_commit.author.name, "date": last_commit.committed_datetime.isoformat() } }) except Exception as e: logger.warning(f"Could not analyze remote branch {ref.name}: {e}") except Exception as e: logger.warning(f"Could not analyze remote branches: {e}") return create_success_response({ "repository_path": repo_path, "implementation": target_service.git_implementation, "branch_analysis": { "current_branch": repo.active_branch.name if repo.active_branch else "HEAD (detached)", "local_branches": { "branches": local_branches, "count": len(local_branches) }, "remote_branches": { "branches": remote_branches, "count": len(remote_branches) }, "summary": { "total_branches": len(local_branches) + len(remote_branches), "local_count": len(local_branches), "remote_count": len(remote_branches) } } }) except Exception as e: logger.error(f"Failed to get branch analysis: {e}") raise ServiceError( f"Failed to get branch analysis: {e}", service_name="get_branch_analysis", cause=e ) @mcp.tool() @handle_errors(log_errors=True) def smart_commit_suggestion( repo_path: str, analyze_changes: bool = True, suggest_type: bool = True, suggest_scope: bool = True ) -> Dict[str, Any]: """ Intelligent commit message suggestions based on repository changes. Uses GitPython to analyze staged changes and suggest: - Appropriate commit type based on file patterns - Scope based on affected directories/modules - Subject line based on change patterns Args: repo_path: Path to git repository analyze_changes: Whether to analyze file changes for suggestions suggest_type: Whether to suggest commit type suggest_scope: Whether to suggest scope Returns: Dict containing intelligent commit suggestions """ # Initialize service for the specified repository try: target_service = CommitzenService(repo_path=repo_path) except Exception as e: raise RepositoryError( f"Failed to initialize service for repository '{repo_path}'", repo_path=repo_path, cause=e ) try: # Get current staged files status = target_service.get_repository_status() if "error" in status: raise GitOperationError( status["error"], repo_path=repo_path ) if status["staging_clean"]: # For backward compatibility with tests return { "success": False, "error": "No staged changes to analyze", "repository_path": repo_path } suggestions = { "types": [], "scopes": [], "subjects": [], "analysis": {} } if analyze_changes and target_service.git_enabled: # Analyze file patterns for intelligent suggestions staged_files = status["staged_files"] # Analyze file types and patterns file_patterns = { "docs": [".md", ".rst", ".txt", "README", "CHANGELOG"], "tests": ["test_", "_test", ".test.", "/tests/", "/test/"], "config": [".json", ".yaml", ".yml", ".toml", ".ini", ".cfg"], "frontend": [".js", ".ts", ".jsx", ".tsx", ".vue", ".css", ".scss"], "backend": [".py", ".java", ".go", ".rs", ".cpp", ".c"], "build": ["Makefile", "setup.py", "pyproject.toml", "package.json", "Dockerfile"], "ci": [".github/", ".gitlab-ci", "Jenkinsfile", ".travis"] } detected_patterns = set() affected_dirs = set() for file_path in staged_files: # Detect file patterns for pattern_type, patterns in file_patterns.items(): if any(pattern in file_path for pattern in patterns): detected_patterns.add(pattern_type) # Extract directory for scope suggestion if "/" in file_path: dir_parts = file_path.split("/") if len(dir_parts) > 1: affected_dirs.add(dir_parts[0]) # Suggest commit types based on patterns if suggest_type: if "docs" in detected_patterns: suggestions["types"].append({"type": "docs", "confidence": 0.9, "reason": "Documentation files modified"}) if "tests" in detected_patterns: suggestions["types"].append({"type": "test", "confidence": 0.8, "reason": "Test files modified"}) if "config" in detected_patterns: suggestions["types"].append({"type": "chore", "confidence": 0.7, "reason": "Configuration files modified"}) if "build" in detected_patterns or "ci" in detected_patterns: suggestions["types"].append({"type": "ci", "confidence": 0.8, "reason": "Build/CI files modified"}) if "frontend" in detected_patterns or "backend" in detected_patterns: suggestions["types"].append({"type": "feat", "confidence": 0.6, "reason": "Code files modified (likely feature)"}) suggestions["types"].append({"type": "fix", "confidence": 0.5, "reason": "Code files modified (possibly fix)"}) # Suggest scopes based on affected directories if suggest_scope: for dir_name in affected_dirs: suggestions["scopes"].append({ "scope": dir_name, "confidence": 0.7, "reason": f"Files in {dir_name}/ directory modified" }) # Generate subject suggestions if len(staged_files) == 1: file_name = staged_files[0].split("/")[-1] suggestions["subjects"].append({ "subject": f"update {file_name}", "confidence": 0.6, "reason": "Single file modification" }) elif "docs" in detected_patterns: suggestions["subjects"].append({ "subject": "update documentation", "confidence": 0.8, "reason": "Documentation files modified" }) elif "tests" in detected_patterns: suggestions["subjects"].append({ "subject": "add/update tests", "confidence": 0.7, "reason": "Test files modified" }) suggestions["analysis"] = { "staged_files_count": len(staged_files), "detected_patterns": list(detected_patterns), "affected_directories": list(affected_dirs), "file_types": list(set(f.split(".")[-1] for f in staged_files if "." in f)) } # Sort suggestions by confidence suggestions["types"].sort(key=lambda x: x["confidence"], reverse=True) suggestions["scopes"].sort(key=lambda x: x["confidence"], reverse=True) suggestions["subjects"].sort(key=lambda x: x["confidence"], reverse=True) return create_success_response({ "repository_path": repo_path, "implementation": target_service.git_implementation, "enhanced_features_used": target_service.git_enabled, "suggestions": suggestions, "staged_files": status["staged_files"] }) except Exception as e: logger.error(f"Failed to generate smart commit suggestions: {e}") raise ServiceError( f"Failed to generate smart commit suggestions: {e}", service_name="smart_commit_suggestion", cause=e ) @mcp.tool() @handle_errors(log_errors=True) def batch_commit_analysis( repo_path: str, file_groups: List[Dict[str, Any]], generate_messages: bool = True ) -> Dict[str, Any]: """ Analyze multiple file groups for batch commit operations. Helps organize staged changes into logical commit groups with appropriate commit messages for each group. Args: repo_path: Path to git repository file_groups: List of file groups with metadata generate_messages: Whether to generate commit messages for each group Returns: Dict containing batch commit analysis and suggestions """ # Initialize service for the specified repository try: target_service = CommitzenService(repo_path=repo_path) except Exception as e: raise RepositoryError( f"Failed to initialize service for repository '{repo_path}'", repo_path=repo_path, cause=e ) try: batch_analysis = [] for i, group in enumerate(file_groups): group_files = group.get("files", []) group_description = group.get("description", f"Group {i+1}") if not group_files: continue # Analyze this group of files group_analysis = { "group_id": i, "description": group_description, "files": group_files, "file_count": len(group_files) } if generate_messages: # Generate smart suggestions for this group group_analysis["suggested_messages"] = [ { "type": "feat", "subject": f"implement {group_description}", "confidence": 0.7 } ] batch_analysis.append(group_analysis) return create_success_response({ "repository_path": repo_path, "implementation": target_service.git_implementation, "batch_analysis": batch_analysis, "total_groups": len(batch_analysis), "total_files": sum(group["file_count"] for group in batch_analysis) }) except Exception as e: logger.error(f"Failed to analyze batch commits: {e}") raise ServiceError( f"Failed to analyze batch commits: {e}", service_name="batch_commit_analysis", cause=e )