diff --git a/src/monitor/git.py b/src/monitor/git.py new file mode 100644 index 0000000..6c4d982 --- /dev/null +++ b/src/monitor/git.py @@ -0,0 +1,183 @@ +"""Git operation tracking using subprocess.""" + +import subprocess +from pathlib import Path +from typing import Optional, List, Dict, Any + +from ..storage import Database, GitEvent + + +class GitTracker: + """Tracks git operations in a repository.""" + + def __init__(self, db: Database, session_id: int): + self.db = db + self.session_id = session_id + self._last_commit: Optional[str] = None + self._last_branch: Optional[str] = None + + def _run_git_command( + self, + args: List[str], + cwd: Optional[str] = None + ) -> tuple: + """Run a git command and return output.""" + try: + result = subprocess.run( + ["git"] + args, + cwd=cwd, + capture_output=True, + text=True, + timeout=5 + ) + return result.stdout, result.stderr, result.returncode + except (subprocess.TimeoutExpired, FileNotFoundError): + return "", "git not found", 1 + + def _get_current_branch(self, cwd: Optional[str] = None) -> Optional[str]: + """Get current git branch.""" + stdout, _, rc = self._run_git_command(["branch", "--show-current"], cwd) + if rc == 0: + return stdout.strip() or None + return None + + def _get_current_commit(self, cwd: Optional[str] = None) -> Optional[str]: + """Get current commit hash.""" + stdout, _, rc = self._run_git_command(["rev-parse", "HEAD"], cwd) + if rc == 0: + return stdout.strip() + return None + + def _get_diff(self, cwd: Optional[str] = None) -> Optional[str]: + """Get current git diff.""" + stdout, _, rc = self._run_git_command(["diff", "--stat"], cwd) + if rc == 0: + return stdout.strip() + return None + + def detect_operation(self, cwd: Optional[str] = None) -> Optional[GitEvent]: + """Detect current git state and operations.""" + current_commit = self._get_current_commit(cwd) + current_branch = self._get_current_branch(cwd) + + if current_commit is None: + return None + + if self._last_commit is None: + self._last_commit = current_commit + self._last_branch = current_branch + + return self.db.add_git_event( + session_id=self.session_id, + operation="checkout", + branch=current_branch, + commit_hash=current_commit, + details=f"Started at commit {current_commit[:7]}", + diff=None + ) + + if current_commit != self._last_commit: + self._last_commit = current_commit + + stdout, _, _ = self._run_git_command(["log", "-1", "--format=%s%n%b", current_commit], cwd) + commit_message = stdout.strip().split("\n")[0] if stdout.strip() else "" + + diff_stat = self._get_diff(cwd) + + return self.db.add_git_event( + session_id=self.session_id, + operation="commit", + branch=current_branch, + commit_hash=current_commit, + details=commit_message, + diff=diff_stat + ) + + if current_branch != self._last_branch: + self._last_branch = current_branch + + return self.db.add_git_event( + session_id=self.session_id, + operation="branch", + branch=current_branch, + commit_hash=current_commit, + details=f"Switched to branch {current_branch}", + diff=None + ) + + return None + + def record_commit( + self, + message: str, + cwd: Optional[str] = None, + diff: Optional[str] = None + ) -> int: + """Record a commit event.""" + commit_hash = self._get_current_commit(cwd) + branch = self._get_current_branch(cwd) + + self._last_commit = commit_hash + + return self.db.add_git_event( + session_id=self.session_id, + operation="commit", + branch=branch, + commit_hash=commit_hash, + details=message, + diff=diff + ) + + def record_branch(self, branch_name: str, cwd: Optional[str] = None) -> int: + """Record a branch creation/switch event.""" + commit_hash = self._get_current_commit(cwd) + self._last_branch = branch_name + + return self.db.add_git_event( + session_id=self.session_id, + operation="branch", + branch=branch_name, + commit_hash=commit_hash, + details=f"Created/switched to branch {branch_name}", + diff=None + ) + + def record_push(self, remote: str, branch: str, cwd: Optional[str] = None) -> int: + """Record a push event.""" + commit_hash = self._get_current_commit(cwd) + + return self.db.add_git_event( + session_id=self.session_id, + operation="push", + branch=branch, + commit_hash=commit_hash, + details=f"Pushed to {remote}/{branch}", + diff=None + ) + + def get_recent_commits(self, count: int = 10, cwd: Optional[str] = None) -> List[Dict[str, Any]]: + """Get recent commits.""" + stdout, _, rc = self._run_git_command( + ["log", f"-{count}", "--format=hash:%H|date:%ad|message:%s", "--date=iso"], + cwd + ) + + if rc != 0: + return [] + + commits = [] + for line in stdout.strip().split("\n"): + if line.startswith("hash:") and "|date:" in line: + try: + hash_part = line.split("|")[0].replace("hash:", "") + date_part = line.split("|")[1].replace("date:", "") + msg_part = line.split("|")[2].replace("message:", "") + commits.append({ + "hash": hash_part, + "date": date_part, + "message": msg_part + }) + except IndexError: + continue + + return commits