Add commands and parsers modules

.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