391 lines
12 KiB
Python
391 lines
12 KiB
Python
"""
|
|
Git operations tools for Clover - A terminal assistant for AI-powered project management
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional, Tuple
|
|
import difflib
|
|
|
|
def git_status(repo_path: str = ".") -> Dict[str, List[str]]:
|
|
"""
|
|
Query repository status and return changes.
|
|
|
|
Args:
|
|
repo_path (str): Path to the git repository
|
|
|
|
Returns:
|
|
Dict containing staged, unstaged, and untracked files
|
|
"""
|
|
try:
|
|
# Check if we're in a git repository
|
|
result = subprocess.run(
|
|
["git", "rev-parse", "--is-inside-work-tree"],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
|
|
if result.stdout.strip() != "true":
|
|
return {"error": "Not a git repository"}
|
|
|
|
# Get status
|
|
status_result = subprocess.run(
|
|
["git", "status", "--porcelain"],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
|
|
lines = status_result.stdout.strip().split('\n') if status_result.stdout.strip() else []
|
|
|
|
# Parse git status output
|
|
staged = [] # Files that are staged for commit
|
|
unstaged = [] # Files that have changes but aren't staged
|
|
untracked = [] # New files not tracked by git
|
|
|
|
for line in lines:
|
|
if not line:
|
|
continue
|
|
status_code = line[:2].strip()
|
|
filepath = line[3:].strip()
|
|
|
|
if status_code.startswith('A') or status_code.startswith('M'):
|
|
staged.append(filepath)
|
|
elif status_code.startswith(' M') or status_code.startswith(' D'):
|
|
unstaged.append(filepath)
|
|
elif status_code.startswith('?'):
|
|
untracked.append(filepath)
|
|
|
|
return {
|
|
"staged": staged,
|
|
"unstaged": unstaged,
|
|
"untracked": untracked
|
|
}
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
return {"error": f"Git command failed: {e.stderr}"}
|
|
except Exception as e:
|
|
return {"error": f"Error getting git status: {str(e)}"}
|
|
|
|
def git_diff(repo_path: str = ".", staged_only: bool = True) -> Dict[str, List[Dict]]:
|
|
"""
|
|
Generate JSON diff of changes for staged files.
|
|
|
|
Args:
|
|
repo_path (str): Path to the git repository
|
|
staged_only (bool): Whether to show only staged changes
|
|
|
|
Returns:
|
|
Dict containing file diffs and overall structure information
|
|
"""
|
|
try:
|
|
# Check if we're in a git repository
|
|
result = subprocess.run(
|
|
["git", "rev-parse", "--is-inside-work-tree"],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
|
|
if result.stdout.strip() != "true":
|
|
return {"error": "Not a git repository"}
|
|
|
|
# Get diff command
|
|
cmd = ["git", "diff"]
|
|
if staged_only:
|
|
cmd.append("--cached") # Show only staged changes
|
|
|
|
diff_result = subprocess.run(
|
|
cmd,
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
|
|
if not diff_result.stdout.strip():
|
|
return {"files": [], "message": "No changes to show"}
|
|
|
|
# Parse the diff output and organize by files
|
|
files = []
|
|
current_file = None
|
|
file_content = ""
|
|
|
|
for line in diff_result.stdout.split('\n'):
|
|
if line.startswith('diff --git'):
|
|
# Save previous file content
|
|
if current_file:
|
|
files.append({
|
|
"filename": current_file,
|
|
"diff": file_content.strip()
|
|
})
|
|
|
|
# Extract filename from diff line
|
|
parts = line.split(' ')
|
|
if len(parts) >= 3:
|
|
current_file = parts[2][2:] # Remove 'a/' prefix
|
|
file_content = ""
|
|
elif line.startswith('@@'):
|
|
file_content += line + '\n'
|
|
elif line.startswith('+') or line.startswith('-') or line.startswith(' '):
|
|
file_content += line + '\n'
|
|
|
|
# Don't forget the last file
|
|
if current_file:
|
|
files.append({
|
|
"filename": current_file,
|
|
"diff": file_content.strip()
|
|
})
|
|
|
|
return {"files": files}
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
return {"error": f"Git diff command failed: {e.stderr}"}
|
|
except Exception as e:
|
|
return {"error": f"Error generating git diff: {str(e)}"}
|
|
|
|
def git_commit(repo_path: str = ".", message: Optional[str] = None) -> Dict[str, str]:
|
|
"""
|
|
Commit changes with auto-generated commit message.
|
|
|
|
Args:
|
|
repo_path (str): Path to the git repository
|
|
message (str): Optional custom commit message
|
|
|
|
Returns:
|
|
Dict containing commit result status and info
|
|
"""
|
|
try:
|
|
# Check if we're in a git repository
|
|
result = subprocess.run(
|
|
["git", "rev-parse", "--is-inside-work-tree"],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
|
|
if result.stdout.strip() != "true":
|
|
return {"error": "Not a git repository"}
|
|
|
|
# Check if there are staged changes
|
|
status_result = subprocess.run(
|
|
["git", "diff", "--cached", "--name-only"],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
|
|
if not status_result.stdout.strip():
|
|
return {"error": "No changes to commit"}
|
|
|
|
# Generate commit message if not provided
|
|
if not message:
|
|
diff = git_diff(repo_path, staged_only=True)
|
|
if "error" in diff:
|
|
message = "Update project files"
|
|
else:
|
|
# Simple auto-generating logic - could be expanded with LLM integration in the future
|
|
message = "Auto-commit: changes made to repository"
|
|
|
|
# Stage all changes and commit
|
|
commit_result = subprocess.run(
|
|
["git", "commit", "-m", message],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
|
|
return {
|
|
"status": "success",
|
|
"message": f"Committed with message: {message}",
|
|
"output": commit_result.stdout.strip()
|
|
}
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
return {"error": f"Git commit failed: {e.stderr}"}
|
|
except Exception as e:
|
|
return {"error": f"Error committing changes: {str(e)}"}
|
|
|
|
def git_push(repo_path: str = ".", remote: str = "origin", branch: str = "main") -> Dict[str, str]:
|
|
"""
|
|
Push committed changes to remote repository.
|
|
|
|
Args:
|
|
repo_path (str): Path to the git repository
|
|
remote (str): Remote name (default: origin)
|
|
branch (str): Branch name (default: main)
|
|
|
|
Returns:
|
|
Dict containing push result status and info
|
|
"""
|
|
try:
|
|
# Check if we're in a git repository
|
|
result = subprocess.run(
|
|
["git", "rev-parse", "--is-inside-work-tree"],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
|
|
if result.stdout.strip() != "true":
|
|
return {"error": "Not a git repository"}
|
|
|
|
# Push changes
|
|
push_result = subprocess.run(
|
|
["git", "push", remote, branch],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
|
|
return {
|
|
"status": "success",
|
|
"message": f"Pushed to {remote}/{branch}",
|
|
"output": push_result.stdout.strip()
|
|
}
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
# Handle common errors more appropriately
|
|
if "Could not resolve host" in e.stderr:
|
|
return {"error": "Network error - could not reach remote repository"}
|
|
elif "Authentication failed" in e.stderr:
|
|
return {"error": "Authentication failed - check your credentials"}
|
|
else:
|
|
return {"error": f"Git push failed: {e.stderr}"}
|
|
except Exception as e:
|
|
return {"error": f"Error pushing changes: {str(e)}"}
|
|
|
|
def git_log(repo_path: str = ".", limit: int = 10) -> Dict[str, List[Dict]]:
|
|
"""
|
|
Show commit history with structured output.
|
|
|
|
Args:
|
|
repo_path (str): Path to the git repository
|
|
limit (int): Number of commits to show
|
|
|
|
Returns:
|
|
Dict containing commit history
|
|
"""
|
|
try:
|
|
# Check if we're in a git repository
|
|
result = subprocess.run(
|
|
["git", "rev-parse", "--is-inside-work-tree"],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
|
|
if result.stdout.strip() != "true":
|
|
return {"error": "Not a git repository"}
|
|
|
|
# Get commit history
|
|
log_result = subprocess.run(
|
|
["git", "log", f"--oneline", f"-{limit}"],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
|
|
commits = []
|
|
for line in log_result.stdout.strip().split('\n'):
|
|
if line:
|
|
# Format: <hash> <message>
|
|
parts = line.split(' ', 1)
|
|
if len(parts) >= 2:
|
|
commits.append({
|
|
"hash": parts[0],
|
|
"message": parts[1]
|
|
})
|
|
else:
|
|
commits.append({
|
|
"hash": parts[0],
|
|
"message": ""
|
|
})
|
|
|
|
return {"commits": commits}
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
return {"error": f"Git log failed: {e.stderr}"}
|
|
except Exception as e:
|
|
return {"error": f"Error getting git log: {str(e)}"}
|
|
|
|
def git_add(repo_path: str = ".", files: List[str] = None) -> Dict[str, str]:
|
|
"""
|
|
Add files to staging area.
|
|
|
|
Args:
|
|
repo_path (str): Path to the git repository
|
|
files (List[str]): Files to add
|
|
|
|
Returns:
|
|
Dict containing add result status and info
|
|
"""
|
|
try:
|
|
# Check if we're in a git repository
|
|
result = subprocess.run(
|
|
["git", "rev-parse", "--is-inside-work-tree"],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
|
|
if result.stdout.strip() != "true":
|
|
return {"error": "Not a git repository"}
|
|
|
|
# Add files
|
|
cmd = ["git", "add"]
|
|
if files:
|
|
cmd.extend(files)
|
|
else:
|
|
cmd.append(".")
|
|
|
|
add_result = subprocess.run(
|
|
cmd,
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
|
|
return {
|
|
"status": "success",
|
|
"message": "Files added to staging area"
|
|
}
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
return {"error": f"Git add failed: {e.stderr}"}
|
|
except Exception as e:
|
|
return {"error": f"Error adding files: {str(e)}"}
|
|
|
|
# Example usage function
|
|
def example_usage():
|
|
"""
|
|
Example of how to use the Git tools.
|
|
"""
|
|
print("Git Tools Examples:")
|
|
|
|
# Example 1: Get status
|
|
result = git_status()
|
|
print(f"Status: {result}")
|
|
|
|
# Example 2: Git diff (this would show what changes exist)
|
|
# result = git_diff(staged_only=True)
|
|
# print(f"Diff: {result}")
|
|
|
|
if __name__ == "__main__":
|
|
example_usage()
|