# #region GitServiceStatusMixin [C:4] [TYPE Module] [SEMANTICS git, status, diff, log, history, lock] # @LAYER Infrastructure # @BRIEF Status, diff, and commit history operations for GitService (all concurrent-safe via per-dashboard locks). # @RELATION CALLED_BY -> [GitService] from datetime import datetime from src.core.logger import belief_scope, logger # #region GitServiceStatusMixin [C:3] [TYPE Class] # @BRIEF Mixin providing repository status, diff, and commit history for GitService. class GitServiceStatusMixin: # region _parse_status_porcelain [TYPE Function] # @PURPOSE: Parse git status --porcelain output into staged, modified, and untracked file lists. # @PRE `repo` is an open GitPython Repo instance. # @POST Returns (staged, modified, untracked) tuple of file path lists. # @RATIONALE Avoids repo.is_dirty() / repo.index.diff("HEAD") which internally # call git diff --cached, a flag unsupported in some Git environments (exit 129). # Using git status --porcelain is self-contained and avoids the --cached flag entirely. def _parse_status_porcelain(self, repo) -> tuple[list[str], list[str], list[str]]: with belief_scope("GitService._parse_status_porcelain"): staged: list[str] = [] modified: list[str] = [] untracked: list[str] = [] try: output = repo.git.status("--porcelain") except Exception: logger.warning("[_parse_status_porcelain][Coherence:Failed] git status --porcelain failed") return staged, modified, untracked for line in output.split("\n"): if not line: continue if line.startswith("??"): untracked.append(line[2:].strip()) continue if line.startswith("!!"): continue if len(line) < 3: continue x = line[0] y = line[1] rest = line[3:] path = rest.split(" -> ")[-1].strip() if " -> " in rest else rest.strip() if x != " " and x != "?": staged.append(path) if y != " " and y != "?": if path not in modified: modified.append(path) return staged, modified, untracked # endregion _parse_status_porcelain # region get_status [C:4] [TYPE Function] [SEMANTICS git,status,lock] # @PURPOSE: Get current repository status (concurrent-safe). # @PRE Repository for dashboard_id exists. # @POST Returns a dictionary representing the Git status. # @RETURN dict def get_status(self, dashboard_id: int) -> dict: with self._locked(dashboard_id): with belief_scope("GitService.get_status"): repo = self.get_repo(dashboard_id) has_commits = False try: repo.head.commit has_commits = True except (ValueError, Exception): has_commits = False current_branch = repo.active_branch.name tracking_branch = None has_upstream = False ahead_count = 0 behind_count = 0 try: tracking_branch = repo.active_branch.tracking_branch() has_upstream = tracking_branch is not None except Exception: tracking_branch = None has_upstream = False if has_upstream and tracking_branch is not None: try: ahead_count = sum(1 for _ in repo.iter_commits(f"{tracking_branch.name}..{current_branch}")) behind_count = sum(1 for _ in repo.iter_commits(f"{current_branch}..{tracking_branch.name}")) except Exception: ahead_count = 0 behind_count = 0 staged_files, modified_files, untracked_files = self._parse_status_porcelain(repo) is_dirty = bool(staged_files or modified_files or untracked_files) is_diverged = ahead_count > 0 and behind_count > 0 if is_diverged: sync_state = "DIVERGED" elif behind_count > 0: sync_state = "BEHIND_REMOTE" elif ahead_count > 0: sync_state = "AHEAD_REMOTE" elif is_dirty or modified_files or staged_files or untracked_files: sync_state = "CHANGES" else: sync_state = "SYNCED" return { "is_dirty": is_dirty, "untracked_files": untracked_files, "modified_files": modified_files, "staged_files": staged_files, "current_branch": current_branch, "upstream_branch": tracking_branch.name if tracking_branch is not None else None, "has_upstream": has_upstream, "ahead_count": ahead_count, "behind_count": behind_count, "is_diverged": is_diverged, "sync_state": sync_state, } # endregion get_status # region get_diff [C:4] [TYPE Function] [SEMANTICS git,diff,lock] # @PURPOSE: Generate diff for a file or the whole repository (concurrent-safe). # @PARAM file_path (str) - Optional specific file. # @PARAM staged (bool) - Whether to show staged changes. # @PRE Repository for dashboard_id exists. # @POST Returns the diff text as a string. # @RETURN str def get_diff(self, dashboard_id: int, file_path: str = None, staged: bool = False) -> str: with self._locked(dashboard_id): with belief_scope("GitService.get_diff"): repo = self.get_repo(dashboard_id) diff_args = [] if staged: diff_args.append("--staged") if file_path: return repo.git.diff(*diff_args, "--", file_path) return repo.git.diff(*diff_args) # endregion get_diff # region get_commit_history [C:4] [TYPE Function] [SEMANTICS git,history,lock] # @PURPOSE: Retrieve commit history for a repository (concurrent-safe). # @PARAM limit (int) - Max number of commits to return. # @PRE Repository for dashboard_id exists. # @POST Returns a list of dictionaries for each commit in history. # @RETURN List[dict] def get_commit_history(self, dashboard_id: int, limit: int = 50) -> list[dict]: with self._locked(dashboard_id): with belief_scope("GitService.get_commit_history"): repo = self.get_repo(dashboard_id) commits = [] try: if not repo.heads and not repo.remotes: return [] for commit in repo.iter_commits(max_count=limit): commits.append({ "hash": commit.hexsha, "author": commit.author.name, "email": commit.author.email, "timestamp": datetime.fromtimestamp(commit.committed_date), "message": commit.message.strip(), "files_changed": list(commit.stats.files.keys()) }) except Exception as e: logger.explore(f"Could not retrieve commit history for dashboard {dashboard_id}: {e}", extra={"src": "get_commit_history"}) return [] return commits # endregion get_commit_history # #endregion GitServiceStatusMixin # #endregion GitServiceStatusMixin