"""Git control utilities for repository operations.""" import subprocess from dataclasses import dataclass, field 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: raise RuntimeError(f"Git command failed: {result.stderr}") return result.stdout @dataclass class GitStatus: """Represents the working directory status.""" branch: str modified: list[str] = field(default_factory=list) added: list[str] = field(default_factory=list) deleted: list[str] = field(default_factory=list) untracked: list[str] = field(default_factory=list) renamed: list[str] = field(default_factory=list) ahead: int = 0 behind: int = 0 def get_status(repo_path: str) -> GitStatus: """Get the working directory status. Args: repo_path: Path to the git repository Returns: GitStatus with changes """ # Get current branch try: branch = _run_git_command(repo_path, "rev-parse", "--abbrev-ref", "HEAD").strip() except RuntimeError: try: branch = _run_git_command(repo_path, "symbolic-ref", "--short", "HEAD").strip() except RuntimeError: branch = "HEAD" status = GitStatus(branch=branch) # Get status with porcelain format try: output = _run_git_command(repo_path, "status", "--porcelain", "--branch") except RuntimeError: return status for line in output.strip().split("\n"): if not line: continue # Branch info line starts with ## if line.startswith("## "): branch_info = line[3:] # Parse ahead/behind info if "[ahead " in branch_info: ahead_str = branch_info.split("[ahead ")[1].split("]")[0] status.ahead = int(ahead_str.split(",")[0]) if "[behind " in branch_info: behind_str = branch_info.split("[behind ")[1].split("]")[0] status.behind = int(behind_str.split(",")[0]) continue # Parse status code if len(line) < 3: continue index_status = line[0] worktree_status = line[1] filename = line[3:] # Untracked files if index_status == "?" and worktree_status == "?": status.untracked.append(filename) continue # Added files if index_status == "A" or worktree_status == "A": status.added.append(filename) continue # Deleted files if index_status == "D" or worktree_status == "D": status.deleted.append(filename) continue # Renamed files if index_status == "R" or worktree_status == "R": status.renamed.append(filename) continue # Modified files if index_status == "M" or worktree_status == "M": status.modified.append(filename) continue return status def create_branch(repo_path: str, name: str, base_branch: str = "HEAD") -> None: """Create a new branch. Args: repo_path: Path to the git repository name: Branch name base_branch: Base branch to create from (default: HEAD) Raises: RuntimeError: If branch creation fails """ try: _run_git_command(repo_path, "rev-parse", "--verify", "HEAD^{commit}") except RuntimeError: # No commits yet - empty repository try: _run_git_command(repo_path, "checkout", "--orphan", name) except RuntimeError as e: if "work tree" in str(e).lower(): # Bare repository - use symbolic-ref instead _run_git_command(repo_path, "symbolic-ref", "HEAD", f"refs/heads/{name}") return raise return _run_git_command(repo_path, "branch", name, base_branch) def delete_branch(repo_path: str, name: str, force: bool = False) -> None: """Delete a branch. Args: repo_path: Path to the git repository name: Branch name force: Force delete even if not merged Raises: RuntimeError: If branch deletion fails """ flag = "-D" if force else "-d" _run_git_command(repo_path, "branch", flag, name) def checkout_branch(repo_path: str, name: str) -> None: """Checkout a branch. Args: repo_path: Path to the git repository name: Branch name Raises: RuntimeError: If checkout fails """ try: _run_git_command(repo_path, "checkout", name) except RuntimeError as e: if "work tree" in str(e).lower(): # Bare repository - use symbolic-ref instead _run_git_command(repo_path, "symbolic-ref", "HEAD", f"refs/heads/{name}") return raise def commit_changes( repo_path: str, message: str, author_name: str, author_email: str, files: list[str] | None = None, ) -> str: """Commit changes to the repository. Args: repo_path: Path to the git repository message: Commit message author_name: Author name author_email: Author email files: Specific files to commit (None = all staged) Returns: Commit hash Raises: RuntimeError: If commit fails """ # Stage files if specified if files: for file in files: _run_git_command(repo_path, "add", file) else: _run_git_command(repo_path, "add", "-A") # Commit _run_git_command( repo_path, "commit", "-m", message, f"--author={author_name} <{author_email}>", ) # Return commit hash return _run_git_command(repo_path, "rev-parse", "HEAD").strip() def fetch(repo_path: str) -> None: """Fetch from remote. Args: repo_path: Path to the git repository Raises: RuntimeError: If fetch fails """ _run_git_command(repo_path, "fetch", "--all") def pull(repo_path: str, branch: str | None = None) -> None: """Pull updates from remote. Args: repo_path: Path to the git repository branch: Branch to pull (default: current branch) Raises: RuntimeError: If pull fails """ args = ["pull"] if branch: args.append("origin") args.append(branch) _run_git_command(repo_path, *args) def push(repo_path: str, branch: str | None = None) -> None: """Push changes to remote. Args: repo_path: Path to the git repository branch: Branch to push (default: current branch) Raises: RuntimeError: If push fails """ args = ["push"] if branch: args.extend(["origin", branch]) _run_git_command(repo_path, *args) def merge( repo_path: str, source_branch: str, target_branch: str | None = None, message: str | None = None, ) -> str: """Merge a branch into the current branch. Args: repo_path: Path to the git repository source_branch: Branch to merge from target_branch: Branch to merge into (default: current branch) message: Merge commit message Returns: Merge commit hash Raises: RuntimeError: If merge fails (including conflicts) """ # Checkout target branch if specified if target_branch: checkout_branch(repo_path, target_branch) # Merge args = ["merge", source_branch] if message: args.extend(["-m", message]) _run_git_command(repo_path, *args) # Return merge commit hash return _run_git_command(repo_path, "rev-parse", "HEAD").strip() def get_current_branch(repo_path: str) -> str: """Get the current branch name. Args: repo_path: Path to the git repository Returns: Current branch name """ try: branch = _run_git_command(repo_path, "rev-parse", "--abbrev-ref", "HEAD").strip() if branch != "HEAD": return branch except RuntimeError: pass return _run_git_command(repo_path, "symbolic-ref", "--short", "HEAD").strip()