mcp/hanzo_mcp/tools/vector/git_ingestor.py

"""Git repository ingestor for Hanzo MCP.

This module provides tools for ingesting entire git repositories into the vector store.
It supports chunking and indexing of git repositories with configurable filters and
metadata extraction.
"""

import os
import tempfile
import subprocess
import shutil
from typing import Any, Dict, List, Optional, Set, Tuple, cast

from hanzo_mcp.tools.common.context import ToolContext, create_tool_context
from mcp.server.fastmcp import FastMCP
from mcp.types import Tool

from hanzo_mcp.tools.common.permissions import PermissionManager
from hanzo_mcp.tools.common.validation import validate_parameters
from hanzo_mcp.tools.vector.store_manager import VectorStoreManager


class GitIngestor:
    """Git repository ingestor for Hanzo MCP."""

    def __init__(
        self,
        permission_manager: PermissionManager,
        vector_store_manager: VectorStoreManager,
    ) -> None:
        """Initialize git ingestor.
        
        Args:
            permission_manager: Permission manager for access control
            vector_store_manager: Vector store manager for managing vector databases
        """
        self.permission_manager = permission_manager
        self.vector_store_manager = vector_store_manager

    async def _clone_repo(
        self, 
        repo_url: str, 
        local_path: str, 
        branch: Optional[str] = None,
        tool_ctx: Optional[ToolContext] = None
    ) -> bool:
        """Clone a git repository.
        
        Args:
            repo_url: URL of the git repository
            local_path: Local path to clone to
            branch: Optional branch to checkout
            tool_ctx: Tool context for logging
            
        Returns:
            True if successful, False otherwise
        """
        try:
            clone_cmd = ["git", "clone"]
            
            # Add branch if specified
            if branch:
                clone_cmd.extend(["--branch", branch, "--single-branch"])
                
            # Add repo URL and local path
            clone_cmd.extend([repo_url, local_path])
            
            # Execute clone command
            if tool_ctx:
                await tool_ctx.info(f"Cloning repository {repo_url} to {local_path}")
                
            result = subprocess.run(
                clone_cmd, 
                capture_output=True, 
                text=True,
                check=False
            )
            
            if result.returncode != 0:
                if tool_ctx:
                    await tool_ctx.error(f"Failed to clone repository: {result.stderr}")
                return False
                
            if tool_ctx:
                await tool_ctx.info(f"Successfully cloned repository to {local_path}")
            return True
            
        except Exception as e:
            if tool_ctx:
                await tool_ctx.error(f"Error cloning repository: {str(e)}")
            return False

    async def _get_repo_metadata(
        self, 
        repo_path: str,
        tool_ctx: Optional[ToolContext] = None
    ) -> Dict[str, Any]:
        """Get metadata for a git repository.
        
        Args:
            repo_path: Path to the git repository
            tool_ctx: Tool context for logging
            
        Returns:
            Dictionary of repository metadata
        """
        metadata = {}
        
        try:
            # Get current branch
            branch_cmd = ["git", "-C", repo_path, "rev-parse", "--abbrev-ref", "HEAD"]
            branch_result = subprocess.run(branch_cmd, capture_output=True, text=True, check=False)
            if branch_result.returncode == 0:
                metadata["branch"] = branch_result.stdout.strip()
            
            # Get latest commit hash
            commit_cmd = ["git", "-C", repo_path, "rev-parse", "HEAD"]
            commit_result = subprocess.run(commit_cmd, capture_output=True, text=True, check=False)
            if commit_result.returncode == 0:
                metadata["commit_hash"] = commit_result.stdout.strip()
            
            # Get repository URL
            url_cmd = ["git", "-C", repo_path, "config", "--get", "remote.origin.url"]
            url_result = subprocess.run(url_cmd, capture_output=True, text=True, check=False)
            if url_result.returncode == 0:
                metadata["repo_url"] = url_result.stdout.strip()
            
            # Get commit count
            count_cmd = ["git", "-C", repo_path, "rev-list", "--count", "HEAD"]
            count_result = subprocess.run(count_cmd, capture_output=True, text=True, check=False)
            if count_result.returncode == 0:
                metadata["commit_count"] = int(count_result.stdout.strip())
            
            # Get latest commit date
            date_cmd = ["git", "-C", repo_path, "log", "-1", "--format=%cd", "--date=iso"]
            date_result = subprocess.run(date_cmd, capture_output=True, text=True, check=False)
            if date_result.returncode == 0:
                metadata["last_commit_date"] = date_result.stdout.strip()
            
            return metadata
            
        except Exception as e:
            if tool_ctx:
                await tool_ctx.error(f"Error getting repository metadata: {str(e)}")
            return metadata

    async def register_tools(self, mcp_server: FastMCP) -> None:
        """Register git ingestor tools with the MCP server.
        
        Args:
            mcp_server: The FastMCP server instance
        """
        @mcp_server.tool("git_ingest")
        async def git_ingest(
            ctx: Any, 
            repo_url: str, 
            branch: Optional[str] = None,
            include_patterns: Optional[List[str]] = None,
            exclude_patterns: Optional[List[str]] = None,
            recursive: bool = True,
            temp_dir: Optional[str] = None
        ) -> str:
            """Ingest a git repository into the vector store.
            
            This tool clones a git repository to a temporary directory, indexes its contents,
            and then cleans up the temporary files.
            
            Args:
                repo_url: URL of the git repository to ingest
                branch: Optional branch to checkout (default: main/master branch)
                include_patterns: Optional list of file patterns to include (e.g., ["*.py", "*.md"])
                exclude_patterns: Optional list of file patterns to exclude (e.g., ["*.pyc", "node_modules/*"])
                recursive: Whether to recursively index subdirectories (default: True)
                temp_dir: Optional temporary directory to use (if None, will create one)
                
            Returns:
                Result of the ingestion operation
            """
            tool_ctx = create_tool_context(ctx)
            
            # Validate parameters
            validate_parameters(tool_ctx, {"repo_url": repo_url})
            
            # Create a temporary directory for the repo
            cleanup_temp_dir = False
            if not temp_dir:
                temp_dir = tempfile.mkdtemp(prefix="hanzo_git_")
                cleanup_temp_dir = True
            else:
                # Make sure temp_dir exists and is allowed
                if not os.path.exists(temp_dir):
                    os.makedirs(temp_dir, exist_ok=True)
                
                if not self.permission_manager.is_path_allowed(temp_dir):
                    return await tool_ctx.error(f"Temp directory is not allowed: {temp_dir}")
            
            try:
                # Clone the repository
                repo_name = repo_url.split("/")[-1].replace(".git", "")
                local_repo_path = os.path.join(temp_dir, repo_name)
                
                success = await self._clone_repo(repo_url, local_repo_path, branch, tool_ctx)
                if not success:
                    return await tool_ctx.error(f"Failed to clone repository: {repo_url}")
                
                # Get repository metadata
                repo_metadata = await self._get_repo_metadata(local_repo_path, tool_ctx)
                await tool_ctx.info(f"Repository metadata: {repo_metadata}")
                
                # Create a collection for the repository
                collection = self.vector_store_manager._get_collection(local_repo_path)
                
                # Index the repository
                await tool_ctx.info(f"Indexing repository: {local_repo_path}")
                
                # Index the directory with include/exclude patterns
                file_pattern = None
                if include_patterns:
                    # We can only handle one pattern at a time in _index_directory,
                    # so we'll need to call it multiple times for multiple patterns
                    indexed_files = 0
                    for pattern in include_patterns:
                        files = await self.vector_store_manager._index_directory(
                            collection,
                            local_repo_path,
                            tool_ctx,
                            recursive,
                            pattern
                        )
                        indexed_files += files
                else:
                    # Index all files
                    indexed_files = await self.vector_store_manager._index_directory(
                        collection,
                        local_repo_path,
                        tool_ctx,
                        recursive,
                        None
                    )
                
                # Apply exclude patterns if provided
                if exclude_patterns and indexed_files > 0:
                    excluded_count = 0
                    for pattern in exclude_patterns:
                        # Can't directly delete by pattern, so need to query and delete
                        # Leave as a future enhancement
                        pass
                
                return await tool_ctx.success(
                    f"Successfully ingested repository {repo_url} into vector store",
                    {
                        "repository": repo_url,
                        "branch": branch or repo_metadata.get("branch", "default"),
                        "indexed_files": indexed_files,
                        "local_path": local_repo_path,
                        "metadata": repo_metadata
                    }
                )
                
            except Exception as e:
                await tool_ctx.error(f"Error ingesting repository: {str(e)}")
                return await tool_ctx.error(f"Failed to ingest repository: {str(e)}")
                
            finally:
                # Clean up temporary directory if we created it
                if cleanup_temp_dir and os.path.exists(temp_dir):
                    shutil.rmtree(temp_dir, ignore_errors=True)