121 lines
3.6 KiB
Python
121 lines
3.6 KiB
Python
"""Git log parsing utilities."""
|
|
|
|
import subprocess
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
|
|
|
|
@dataclass
|
|
class Commit:
|
|
"""Represents a git commit."""
|
|
hash: str
|
|
message: str
|
|
author: str
|
|
date: str
|
|
parents: List[str] = field(default_factory=list)
|
|
branch: Optional[str] = None
|
|
|
|
|
|
@dataclass
|
|
class WorkflowStats:
|
|
"""Statistics about a git workflow."""
|
|
commit_count: int = 0
|
|
merge_count: int = 0
|
|
feature_count: int = 0
|
|
branches: List[str] = field(default_factory=list)
|
|
|
|
|
|
class GitLogParser:
|
|
"""Parses git log for visualization."""
|
|
|
|
def __init__(self, repo_path: Path):
|
|
self.repo_path = repo_path
|
|
self.commits: List[Commit] = []
|
|
self.branches: List[str] = []
|
|
self.current_branch: Optional[str] = None
|
|
|
|
def is_git_repo(self) -> bool:
|
|
"""Check if path is a git repository."""
|
|
return (self.repo_path / ".git").exists()
|
|
|
|
def parse_log(self, max_count: int = 50) -> List[Commit]:
|
|
"""Parse git log."""
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "log", f"--max-count={max_count}", "--pretty=format:%h|%s|%an|%ad|%D", "--date=short"],
|
|
cwd=self.repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
for line in result.stdout.strip().split("\n"):
|
|
if line:
|
|
parts = line.split("|", 4)
|
|
if len(parts) >= 4:
|
|
commit = Commit(
|
|
hash=parts[0],
|
|
message=parts[1],
|
|
author=parts[2],
|
|
date=parts[3],
|
|
)
|
|
if len(parts) > 4:
|
|
commit.branch = parts[4].split(", ")[0] if ", " in parts[4] else parts[4]
|
|
self.commits.append(commit)
|
|
|
|
except Exception:
|
|
pass
|
|
|
|
return self.commits
|
|
|
|
def get_commit_parents(self) -> None:
|
|
"""Get parent commits for each commit."""
|
|
try:
|
|
for commit in self.commits:
|
|
result = subprocess.run(
|
|
["git", "rev-list", "--parents", "-n", "1", commit.hash],
|
|
cwd=self.repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if result.stdout.strip():
|
|
parents = result.stdout.strip().split()[1:]
|
|
commit.parents = parents
|
|
|
|
except Exception:
|
|
pass
|
|
|
|
def get_branches(self) -> List[str]:
|
|
"""Get list of branches."""
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "branch", "-a"],
|
|
cwd=self.repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
self.branches = [b.strip().replace("* ", "") for b in result.stdout.strip().split("\n") if b.strip()]
|
|
|
|
for commit in self.commits:
|
|
if commit.branch is None:
|
|
commit.branch = self.current_branch
|
|
|
|
except Exception:
|
|
pass
|
|
|
|
return self.branches
|
|
|
|
def analyze_workflow(self) -> WorkflowStats:
|
|
"""Analyze git workflow patterns."""
|
|
stats = WorkflowStats(commit_count=len(self.commits))
|
|
|
|
for commit in self.commits:
|
|
if "merge" in commit.message.lower():
|
|
stats.merge_count += 1
|
|
if any(kw in commit.message.lower() for kw in ["feature", "feat", "feature-"]):
|
|
stats.feature_count += 1
|
|
|
|
stats.branches = self.get_branches()
|
|
|
|
return stats
|