"""Git file utilities for browsing repository contents.""" import logging import subprocess from dataclasses import dataclass from pathlib import Path from typing import Any @dataclass class FileTreeEntry: """Represents a file or directory in the repository.""" name: str type: str # "file" or "directory" path: str size: int | None = None mode: str | None = None last_commit: dict[str, Any] | None = None @dataclass class BranchInfo: """Represents a git branch.""" name: str is_default: bool last_commit: dict[str, Any] | None = None @dataclass class FileContent: """Represents file content and metadata.""" path: str branch: str content: str size: int encoding: str language: str | None is_binary: bool last_commit: dict[str, Any] | None = None def _run_git_command(repo_path: str, *args: str) -> str: """Run a git command in the repository directory.""" result = subprocess.run( ["git", *args], cwd=repo_path, capture_output=True, text=True, ) if result.returncode != 0: stderr = result.stderr # Handle "dubious ownership" security error if "dubious ownership" in stderr.lower(): logger.warning("Git ownership mismatch for %s, adding to safe.directory", repo_path) # Add this directory to git's safe.directory list subprocess.run( ["git", "config", "--global", "--add", "safe.directory", repo_path], capture_output=True, ) # Retry the command result = subprocess.run( ["git", *args], cwd=repo_path, capture_output=True, text=True, ) if result.returncode == 0: return result.stdout stderr = result.stderr raise RuntimeError(f"Git command failed: {stderr}") return result.stdout logger = logging.getLogger(__name__) def list_tree(repo_path: str, branch: str = "main", path: str = "") -> list[FileTreeEntry]: """List files and directories in a repository path. Args: repo_path: Path to the git repository branch: Branch name to list from path: Directory path within the repository (empty for root) Returns: List of FileTreeEntry objects """ tree_path = f"{branch}:{path}" if path else branch try: output = _run_git_command(repo_path, "ls-tree", "-l", tree_path) except RuntimeError as e: logger.warning("git ls-tree failed for %s with branch '%s': %s", repo_path, tree_path, str(e)) # Try with HEAD if branch doesn't exist tree_path = f"HEAD:{path}" if path else "HEAD" try: output = _run_git_command(repo_path, "ls-tree", "-l", tree_path) except RuntimeError as e: logger.error("git ls-tree failed for %s with HEAD: %s", repo_path, str(e)) # Check if this is an empty repository (no commits yet) error_msg = str(e).lower() if "not a valid object name" in error_msg or "does not exist" in error_msg: # Empty repository - return empty list return [] raise entries = [] for line in output.strip().split("\n"): if not line: continue # Format: \t parts = line.split("\t", 1) if len(parts) != 2: continue meta, name = parts meta_parts = meta.split() if len(meta_parts) < 4: continue mode = meta_parts[0] obj_type = meta_parts[1] _ = meta_parts[2] # object hash, not used size = int(meta_parts[3]) if obj_type == "blob" else None entry_path = f"{path}/{name}" if path else name # Get last commit info for this entry last_commit = _get_last_commit_for_path(repo_path, branch, entry_path) entries.append( FileTreeEntry( name=name, type="directory" if obj_type == "tree" else "file", path=entry_path, size=size, mode=mode, last_commit=last_commit, ) ) return entries def _get_last_commit_for_path(repo_path: str, branch: str, path: str) -> dict[str, Any] | None: """Get the last commit that modified a path.""" try: output = _run_git_command( repo_path, "log", "-1", "--format=%H|%s|%an|%aI", branch, "--", path, ) if not output.strip(): return None parts = output.strip().split("|", 3) if len(parts) != 4: return None return { "hash": parts[0], "message": parts[1], "author": parts[2], "date": parts[3], } except RuntimeError: return None def get_file_content(repo_path: str, branch: str, path: str) -> FileContent: """Get the content of a file. Args: repo_path: Path to the git repository branch: Branch name path: File path within the repository Returns: FileContent with content and metadata """ # Check if file exists try: _run_git_command(repo_path, "cat-file", "-e", f"{branch}:{path}") except RuntimeError: raise FileNotFoundError(f"File '{path}' not found in branch '{branch}'") # Get file size size_output = _run_git_command(repo_path, "cat-file", "-s", f"{branch}:{path}") size = int(size_output.strip()) # Check if binary is_binary = _is_binary_file(repo_path, branch, path) # Get content (only for text files) content = "" if not is_binary: content = _run_git_command(repo_path, "show", f"{branch}:{path}") # Detect language from extension language = _detect_language(path) # Get last commit last_commit = _get_last_commit_for_path(repo_path, branch, path) return FileContent( path=path, branch=branch, content=content, size=size, encoding="utf-8", language=language, is_binary=is_binary, last_commit=last_commit, ) def _is_binary_file(repo_path: str, branch: str, path: str) -> bool: """Check if a file is binary using raw bytes to avoid encoding issues.""" try: result = subprocess.run( ["git", "show", f"{branch}:{path}"], cwd=repo_path, capture_output=True, ) if result.returncode != 0: raise RuntimeError(f"Git command failed: {result.stderr.decode()}") # A file is binary if it contains null bytes return b"\x00" in result.stdout except RuntimeError: return True def _detect_language(path: str) -> str | None: """Detect programming language from file extension.""" ext = Path(path).suffix.lower() language_map = { ".py": "python", ".js": "javascript", ".ts": "typescript", ".jsx": "jsx", ".tsx": "tsx", ".html": "html", ".css": "css", ".scss": "scss", ".json": "json", ".md": "markdown", ".yaml": "yaml", ".yml": "yaml", ".sh": "bash", ".rs": "rust", ".go": "go", ".java": "java", ".c": "c", ".cpp": "cpp", ".h": "c", ".php": "php", ".rb": "ruby", ".sql": "sql", ".dockerfile": "dockerfile", ".vue": "vue", ".svelte": "svelte", } return language_map.get(ext) def list_branches(repo_path: str) -> tuple[list[BranchInfo], str]: """List all branches and identify the default branch. Args: repo_path: Path to the git repository Returns: Tuple of (list of BranchInfo, default branch name) """ # Get all branches try: output = _run_git_command(repo_path, "branch", "-a", "--format=%(refname:short)") except RuntimeError as e: logger.error("Failed to list branches for %s: %s", repo_path, str(e)) raise branches: list[BranchInfo] = [] default_branch = "main" # Get list of remote names to properly filter remote tracking branches try: remote_output = _run_git_command(repo_path, "remote") remote_names = {r.strip() for r in remote_output.strip().split("\n") if r.strip()} except RuntimeError: remote_names = set() for line in output.strip().split("\n"): if not line: continue branch_name = line.strip() # Skip detached HEAD pointer if branch_name == "HEAD": continue # Skip remote tracking branches - they appear as "origin/branch-name" # Check if first part is a remote name if "/" in branch_name: first_part = branch_name.split("/", 1)[0] if first_part in remote_names: # Extract just the branch name part (after "origin/") branch_name = branch_name.split("/", 1)[1] elif branch_name.startswith("remotes/"): # Handle "remotes/origin/branch-name" format parts = branch_name.split("/", 2) if len(parts) >= 3: branch_name = parts[2] else: continue # Skip duplicates if any(b.name == branch_name for b in branches): continue # Check if this is the default branch (HEAD points to it) try: head_output = _run_git_command( repo_path, "symbolic-ref", "HEAD", ) if head_output.strip() == f"refs/heads/{branch_name}": default_branch = branch_name except RuntimeError: pass # Get last commit for branch last_commit = _get_last_commit_for_path(repo_path, branch_name, ".") branches.append( BranchInfo( name=branch_name, is_default=(branch_name == default_branch), last_commit=last_commit, ) ) # If no branches found, try to get HEAD if not branches: try: output = _run_git_command(repo_path, "rev-parse", "--abbrev-ref", "HEAD") branch_name = output.strip() if branch_name and branch_name != "HEAD": last_commit = _get_last_commit_for_path(repo_path, branch_name, ".") branches.append( BranchInfo( name=branch_name, is_default=True, last_commit=last_commit, ) ) default_branch = branch_name except RuntimeError: try: output = _run_git_command(repo_path, "symbolic-ref", "--short", "HEAD") branch_name = output.strip() if branch_name: branches.append( BranchInfo( name=branch_name, is_default=True, last_commit=None, ) ) default_branch = branch_name except RuntimeError: pass return branches, default_branch def commit_file( repo_path: str, branch: str, path: str, content: str, commit_message: str, author_name: str, author_email: str, ) -> str: """Commit a file change. Args: repo_path: Path to the git repository branch: Branch to commit to path: File path within the repository content: New file content commit_message: Commit message author_name: Author name author_email: Author email Returns: Commit hash """ # For bare repositories, we need to use git commands differently # We'll create a temporary worktree, make changes, and commit import tempfile import os # Create a temporary worktree with tempfile.TemporaryDirectory() as worktree_path: # Add worktree _run_git_command( repo_path, "worktree", "add", "--detach", worktree_path, branch, ) try: # Write file content file_path = os.path.join(worktree_path, path) os.makedirs(os.path.dirname(file_path), exist_ok=True) with open(file_path, "w", encoding="utf-8") as f: f.write(content) # Configure git author _run_git_command(worktree_path, "config", "user.name", author_name) _run_git_command(worktree_path, "config", "user.email", author_email) # Stage and commit _run_git_command(worktree_path, "add", path) _run_git_command( worktree_path, "commit", "-m", commit_message, ) # Get commit hash commit_hash = _run_git_command( worktree_path, "rev-parse", "HEAD", ).strip() return commit_hash finally: # Remove worktree _run_git_command(repo_path, "worktree", "remove", worktree_path)