diff --git a/.termflow/parsers/git_parser.py b/.termflow/parsers/git_parser.py new file mode 100644 index 0000000..76655c1 --- /dev/null +++ b/.termflow/parsers/git_parser.py @@ -0,0 +1,120 @@ +"""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