"""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