timmy-local/tools/git_tools.py

"""
Git Tools for Uni-Wizard
Repository operations and version control
"""

import os
import json
import subprocess
from typing import Dict, List, Optional
from pathlib import Path

from .registry import registry


def run_git_command(args: List[str], cwd: str = None) -> tuple:
    """Execute a git command and return (stdout, stderr, returncode)"""
    try:
        result = subprocess.run(
            ['git'] + args,
            capture_output=True,
            text=True,
            cwd=cwd
        )
        return result.stdout, result.stderr, result.returncode
    except Exception as e:
        return "", str(e), 1


def git_status(repo_path: str = ".") -> str:
    """
    Get git repository status.
    
    Args:
        repo_path: Path to git repository (default: current directory)
        
    Returns:
        Status info including branch, changed files, last commit
    """
    try:
        status = {"repo_path": os.path.abspath(repo_path)}
        
        # Current branch
        stdout, _, rc = run_git_command(['branch', '--show-current'], cwd=repo_path)
        if rc == 0:
            status["branch"] = stdout.strip()
        else:
            return f"Error: Not a git repository at {repo_path}"
        
        # Last commit
        stdout, _, rc = run_git_command(['log', '-1', '--format=%H|%s|%an|%ad', '--date=short'], cwd=repo_path)
        if rc == 0:
            parts = stdout.strip().split('|')
            if len(parts) >= 4:
                status["last_commit"] = {
                    "hash": parts[0][:8],
                    "message": parts[1],
                    "author": parts[2],
                    "date": parts[3]
                }
        
        # Changed files
        stdout, _, rc = run_git_command(['status', '--porcelain'], cwd=repo_path)
        if rc == 0:
            changes = []
            for line in stdout.strip().split('\n'):
                if line:
                    status_code = line[:2]
                    file_path = line[3:]
                    changes.append({
                        "file": file_path,
                        "status": status_code.strip()
                    })
            status["changes"] = changes
            status["has_changes"] = len(changes) > 0
        
        # Remote info
        stdout, _, rc = run_git_command(['remote', '-v'], cwd=repo_path)
        if rc == 0:
            remotes = []
            for line in stdout.strip().split('\n'):
                if line:
                    parts = line.split()
                    if len(parts) >= 2:
                        remotes.append({"name": parts[0], "url": parts[1]})
            status["remotes"] = remotes
        
        return json.dumps(status, indent=2)
        
    except Exception as e:
        return f"Error getting git status: {str(e)}"


def git_log(repo_path: str = ".", count: int = 10) -> str:
    """
    Get recent commit history.
    
    Args:
        repo_path: Path to git repository
        count: Number of commits to show (default: 10)
        
    Returns:
        List of recent commits
    """
    try:
        stdout, stderr, rc = run_git_command(
            ['log', f'-{count}', '--format=%H|%s|%an|%ad', '--date=short'],
            cwd=repo_path
        )
        
        if rc != 0:
            return f"Error: {stderr}"
        
        commits = []
        for line in stdout.strip().split('\n'):
            if line:
                parts = line.split('|')
                if len(parts) >= 4:
                    commits.append({
                        "hash": parts[0][:8],
                        "message": parts[1],
                        "author": parts[2],
                        "date": parts[3]
                    })
        
        return json.dumps({"count": len(commits), "commits": commits}, indent=2)
        
    except Exception as e:
        return f"Error getting git log: {str(e)}"


def git_pull(repo_path: str = ".") -> str:
    """
    Pull latest changes from remote.
    
    Args:
        repo_path: Path to git repository
        
    Returns:
        Pull result
    """
    try:
        stdout, stderr, rc = run_git_command(['pull'], cwd=repo_path)
        
        if rc == 0:
            if 'Already up to date' in stdout:
                return "✓ Already up to date"
            return f"✓ Pull successful:\n{stdout}"
        else:
            return f"✗ Pull failed:\n{stderr}"
            
    except Exception as e:
        return f"Error pulling: {str(e)}"


def git_commit(repo_path: str = ".", message: str = None, files: List[str] = None) -> str:
    """
    Stage and commit changes.
    
    Args:
        repo_path: Path to git repository
        message: Commit message (required)
        files: Specific files to commit (default: all changes)
        
    Returns:
        Commit result
    """
    if not message:
        return "Error: commit message is required"
    
    try:
        # Stage files
        if files:
            for f in files:
                _, stderr, rc = run_git_command(['add', f], cwd=repo_path)
                if rc != 0:
                    return f"✗ Failed to stage {f}: {stderr}"
        else:
            _, stderr, rc = run_git_command(['add', '.'], cwd=repo_path)
            if rc != 0:
                return f"✗ Failed to stage changes: {stderr}"
        
        # Commit
        stdout, stderr, rc = run_git_command(['commit', '-m', message], cwd=repo_path)
        
        if rc == 0:
            return f"✓ Commit successful:\n{stdout}"
        else:
            if 'nothing to commit' in stderr.lower():
                return "✓ Nothing to commit (working tree clean)"
            return f"✗ Commit failed:\n{stderr}"
            
    except Exception as e:
        return f"Error committing: {str(e)}"


def git_push(repo_path: str = ".", remote: str = "origin", branch: str = None) -> str:
    """
    Push to remote repository.
    
    Args:
        repo_path: Path to git repository
        remote: Remote name (default: origin)
        branch: Branch to push (default: current branch)
        
    Returns:
        Push result
    """
    try:
        if not branch:
            # Get current branch
            stdout, _, rc = run_git_command(['branch', '--show-current'], cwd=repo_path)
            if rc == 0:
                branch = stdout.strip()
            else:
                return "Error: Could not determine current branch"
        
        stdout, stderr, rc = run_git_command(['push', remote, branch], cwd=repo_path)
        
        if rc == 0:
            return f"✓ Push successful to {remote}/{branch}"
        else:
            return f"✗ Push failed:\n{stderr}"
            
    except Exception as e:
        return f"Error pushing: {str(e)}"


def git_checkout(repo_path: str = ".", branch: str = None, create: bool = False) -> str:
    """
    Checkout a branch.
    
    Args:
        repo_path: Path to git repository
        branch: Branch name to checkout
        create: Create the branch if it doesn't exist
        
    Returns:
        Checkout result
    """
    if not branch:
        return "Error: branch name is required"
    
    try:
        if create:
            stdout, stderr, rc = run_git_command(['checkout', '-b', branch], cwd=repo_path)
        else:
            stdout, stderr, rc = run_git_command(['checkout', branch], cwd=repo_path)
        
        if rc == 0:
            return f"✓ Checked out branch: {branch}"
        else:
            return f"✗ Checkout failed:\n{stderr}"
            
    except Exception as e:
        return f"Error checking out: {str(e)}"


def git_branch_list(repo_path: str = ".") -> str:
    """
    List all branches.
    
    Args:
        repo_path: Path to git repository
        
    Returns:
        List of branches with current marked
    """
    try:
        stdout, stderr, rc = run_git_command(['branch', '-a'], cwd=repo_path)
        
        if rc != 0:
            return f"Error: {stderr}"
        
        branches = []
        for line in stdout.strip().split('\n'):
            if line:
                branch = line.strip()
                is_current = branch.startswith('*')
                if is_current:
                    branch = branch[1:].strip()
                branches.append({
                    "name": branch,
                    "current": is_current
                })
        
        return json.dumps({"branches": branches}, indent=2)
        
    except Exception as e:
        return f"Error listing branches: {str(e)}"


# Register all git tools
def register_all():
    registry.register(
        name="git_status",
        handler=git_status,
        description="Get git repository status (branch, changes, last commit)",
        parameters={
            "type": "object",
            "properties": {
                "repo_path": {
                    "type": "string",
                    "description": "Path to git repository",
                    "default": "."
                }
            }
        },
        category="git"
    )
    
    registry.register(
        name="git_log",
        handler=git_log,
        description="Get recent commit history",
        parameters={
            "type": "object",
            "properties": {
                "repo_path": {
                    "type": "string",
                    "description": "Path to git repository",
                    "default": "."
                },
                "count": {
                    "type": "integer",
                    "description": "Number of commits to show",
                    "default": 10
                }
            }
        },
        category="git"
    )
    
    registry.register(
        name="git_pull",
        handler=git_pull,
        description="Pull latest changes from remote",
        parameters={
            "type": "object",
            "properties": {
                "repo_path": {
                    "type": "string",
                    "description": "Path to git repository",
                    "default": "."
                }
            }
        },
        category="git"
    )
    
    registry.register(
        name="git_commit",
        handler=git_commit,
        description="Stage and commit changes",
        parameters={
            "type": "object",
            "properties": {
                "repo_path": {
                    "type": "string",
                    "description": "Path to git repository",
                    "default": "."
                },
                "message": {
                    "type": "string",
                    "description": "Commit message (required)"
                },
                "files": {
                    "type": "array",
                    "description": "Specific files to commit (default: all changes)",
                    "items": {"type": "string"}
                }
            },
            "required": ["message"]
        },
        category="git"
    )
    
    registry.register(
        name="git_push",
        handler=git_push,
        description="Push to remote repository",
        parameters={
            "type": "object",
            "properties": {
                "repo_path": {
                    "type": "string",
                    "description": "Path to git repository",
                    "default": "."
                },
                "remote": {
                    "type": "string",
                    "description": "Remote name",
                    "default": "origin"
                },
                "branch": {
                    "type": "string",
                    "description": "Branch to push (default: current)"
                }
            }
        },
        category="git"
    )
    
    registry.register(
        name="git_checkout",
        handler=git_checkout,
        description="Checkout a branch",
        parameters={
            "type": "object",
            "properties": {
                "repo_path": {
                    "type": "string",
                    "description": "Path to git repository",
                    "default": "."
                },
                "branch": {
                    "type": "string",
                    "description": "Branch name to checkout"
                },
                "create": {
                    "type": "boolean",
                    "description": "Create branch if it doesn't exist",
                    "default": False
                }
            },
            "required": ["branch"]
        },
        category="git"
    )
    
    registry.register(
        name="git_branch_list",
        handler=git_branch_list,
        description="List all branches",
        parameters={
            "type": "object",
            "properties": {
                "repo_path": {
                    "type": "string",
                    "description": "Path to git repository",
                    "default": "."
                }
            }
        },
        category="git"
    )


register_all()