initial
This commit is contained in:
23 files changed
+2117
No files matched your search
@@ -0,0 +1,231 @@
|
||||
"""
|
||||
Shared Git service for devtools.
|
||||
"""
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Optional, Tuple
|
||||
import os
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
class GitService:
|
||||
"""Base Git service that can be extended by specific tools."""
|
||||
|
||||
def __init__(self, repo_path: str):
|
||||
"""Initialize GitService with repository path."""
|
||||
self.repo_path = repo_path
|
||||
if not os.path.exists(os.path.join(repo_path, '.git')):
|
||||
raise ValueError(f"Not a git repository: {repo_path}")
|
||||
|
||||
def _is_git_repo(self) -> bool:
|
||||
"""Check if the current directory is a git repository."""
|
||||
try:
|
||||
subprocess.run(
|
||||
["git", "rev-parse", "--is-inside-work-tree"],
|
||||
cwd=self.repo_path,
|
||||
check=True,
|
||||
capture_output=True
|
||||
)
|
||||
return True
|
||||
except subprocess.CalledProcessError:
|
||||
return False
|
||||
|
||||
def _run_git_command(self, args: List[str]) -> Tuple[str, str]:
|
||||
"""Run a git command and return its output."""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git"] + args,
|
||||
cwd=self.repo_path,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True
|
||||
)
|
||||
return result.stdout.strip(), result.stderr.strip()
|
||||
except subprocess.CalledProcessError as e:
|
||||
return e.stdout.strip(), e.stderr.strip()
|
||||
|
||||
def get_staged_files(self) -> List[str]:
|
||||
"""Get list of staged files."""
|
||||
stdout, _ = self._run_git_command(["diff", "--cached", "--name-only"])
|
||||
return stdout.splitlines() if stdout else []
|
||||
|
||||
def get_unstaged_files(self) -> List[str]:
|
||||
"""Get list of unstaged files."""
|
||||
stdout, _ = self._run_git_command(["ls-files", "--modified", "--others", "--exclude-standard"])
|
||||
return stdout.splitlines() if stdout else []
|
||||
|
||||
def get_diff(self, file_path: str, staged: bool = True) -> str:
|
||||
"""Get diff for a specific file."""
|
||||
args = ["diff", "--cached" if staged else ""]
|
||||
if file_path:
|
||||
args.append(file_path)
|
||||
stdout, _ = self._run_git_command(args)
|
||||
return stdout
|
||||
|
||||
def get_all_diffs(self, staged: bool = True) -> Dict[str, str]:
|
||||
"""Get diffs for all changed files."""
|
||||
files = self.get_staged_files() if staged else self.get_unstaged_files()
|
||||
return {file: self.get_diff(file, staged) for file in files}
|
||||
|
||||
def commit(self, message: str, sign: bool = False) -> bool:
|
||||
"""Create a commit with the given message."""
|
||||
args = ["commit"]
|
||||
if sign:
|
||||
args.append("-S")
|
||||
args.extend(["-m", message])
|
||||
stdout, stderr = self._run_git_command(args)
|
||||
return not stderr
|
||||
|
||||
def get_repo_name(self) -> str:
|
||||
"""Get the repository name."""
|
||||
stdout, _ = self._run_git_command(["rev-parse", "--show-toplevel"])
|
||||
return Path(stdout).name
|
||||
|
||||
def get_current_branch(self) -> str:
|
||||
"""Get the current branch name."""
|
||||
stdout, _ = self._run_git_command(["rev-parse", "--abbrev-ref", "HEAD"])
|
||||
return stdout
|
||||
|
||||
def push(self, branch: Optional[str] = None) -> bool:
|
||||
"""Push changes to remote."""
|
||||
if not branch:
|
||||
branch = self.get_current_branch()
|
||||
stdout, stderr = self._run_git_command(["push", "origin", branch])
|
||||
return not stderr
|
||||
|
||||
def get_staged_changes(self, files: Optional[List[str]] = None) -> Dict[str, str]:
|
||||
"""Get staged changes for specified files or all staged changes."""
|
||||
try:
|
||||
diffs = {} # Initialize diffs at the start
|
||||
|
||||
if files:
|
||||
for file in files:
|
||||
result = subprocess.run(
|
||||
['git', 'diff', '--staged', '--', file],
|
||||
cwd=self.repo_path,
|
||||
capture_output=True,
|
||||
text=True
|
||||
)
|
||||
if result.stdout:
|
||||
diffs[file] = result.stdout
|
||||
else:
|
||||
result = subprocess.run(
|
||||
['git', 'diff', '--staged'],
|
||||
cwd=self.repo_path,
|
||||
capture_output=True,
|
||||
text=True
|
||||
)
|
||||
if result.stdout:
|
||||
# Parse the diff to get individual file changes
|
||||
current_file = None
|
||||
current_diff = []
|
||||
|
||||
for line in result.stdout.split('\n'):
|
||||
if line.startswith('diff --git'):
|
||||
if current_file and current_diff:
|
||||
diffs[current_file] = '\n'.join(current_diff)
|
||||
current_file = line.split('b/')[-1].strip()
|
||||
current_diff = [line]
|
||||
elif current_file:
|
||||
current_diff.append(line)
|
||||
|
||||
if current_file and current_diff:
|
||||
diffs[current_file] = '\n'.join(current_diff)
|
||||
|
||||
return diffs
|
||||
except subprocess.CalledProcessError as e:
|
||||
raise Exception(f"Failed to get staged changes: {str(e)}")
|
||||
|
||||
def stage_all_changes(self) -> None:
|
||||
"""Stage all changes in the repository."""
|
||||
try:
|
||||
subprocess.run(
|
||||
['git', 'add', '.'],
|
||||
cwd=self.repo_path,
|
||||
check=True
|
||||
)
|
||||
except subprocess.CalledProcessError as e:
|
||||
raise Exception(f"Failed to stage changes: {str(e)}")
|
||||
|
||||
def get_commit_history(self, since: Optional[str] = None, limit: Optional[int] = None) -> List[Dict[str, str]]:
|
||||
"""Get commit history.
|
||||
|
||||
Args:
|
||||
since: Get commits since this reference (tag, commit, etc.)
|
||||
limit: Maximum number of commits to return
|
||||
"""
|
||||
args = [
|
||||
"log",
|
||||
"--pretty=format:%H%n%an%n%ad%n%B%n==END==",
|
||||
"--date=format:%Y-%m-%d %H:%M:%S %z"
|
||||
]
|
||||
|
||||
if since:
|
||||
args.append(f"{since}..HEAD")
|
||||
if limit:
|
||||
args.append(f"-n {limit}")
|
||||
|
||||
stdout, _ = self._run_git_command(args)
|
||||
|
||||
commits = []
|
||||
current_commit = []
|
||||
|
||||
for line in stdout.split('\n'):
|
||||
if line == "==END==":
|
||||
if current_commit:
|
||||
commit_hash, author, date, *message_lines = current_commit
|
||||
commits.append({
|
||||
'hash': commit_hash,
|
||||
'author': author,
|
||||
'date': date,
|
||||
'message': '\n'.join(message_lines)
|
||||
})
|
||||
current_commit = []
|
||||
else:
|
||||
current_commit.append(line)
|
||||
|
||||
return commits
|
||||
|
||||
def get_commits_since_tag(self, tag: str) -> List[Dict[str, str]]:
|
||||
"""Get commits since a tag.
|
||||
|
||||
Args:
|
||||
tag: Git tag to use as reference point
|
||||
|
||||
Returns:
|
||||
List of commit dictionaries with hash and message
|
||||
"""
|
||||
cmd = ["git", "log", f"{tag}..HEAD", "--pretty=format:%H|%s"]
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, cwd=self.repo_path)
|
||||
result.check_returncode()
|
||||
|
||||
commits = []
|
||||
for line in result.stdout.splitlines():
|
||||
hash_, message = line.split("|", 1)
|
||||
commits.append({"hash": hash_, "message": message})
|
||||
|
||||
return commits
|
||||
|
||||
def get_commits_since_date(self, days: int) -> List[Dict[str, str]]:
|
||||
"""Get commits since a date.
|
||||
|
||||
Args:
|
||||
days: Number of days to look back
|
||||
|
||||
Returns:
|
||||
List of commit dictionaries with hash and message
|
||||
"""
|
||||
# Calculate date
|
||||
date = datetime.now() - timedelta(days=days)
|
||||
date_str = date.strftime("%Y-%m-%d")
|
||||
|
||||
# Get commits
|
||||
cmd = ["git", "log", f"--since={date_str}", "--pretty=format:%H|%s"]
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, cwd=self.repo_path)
|
||||
result.check_returncode()
|
||||
|
||||
commits = []
|
||||
for line in result.stdout.splitlines():
|
||||
hash_, message = line.split("|", 1)
|
||||
commits.append({"hash": hash_, "message": message})
|
||||
|
||||
return commits
|
||||
Reference in new issue
Block a user