headquarter/apps/api/src/utils/git_files.py

"""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: <mode> <type> <hash> <size>\t<name>
        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)