Add git support

Add git support
* Add git client
* Add git branch tree view

Author: JE-Chen

.idea/discord.xml

@@ -0,0 +1,89 @@
+import logging
+from dataclasses import dataclass, field
+from typing import List, Dict
+
+log = logging.getLogger(__name__)
+
+
+@dataclass
+class CommitNode:
+    sha: str
+    author: str
+    date: str
+    message: str
+    parents: List[str]
+    lane: int = -1  # assigned later
+
+
+@dataclass
+class CommitGraph:
+    nodes: List[CommitNode] = field(default_factory=list)
+    index: Dict[str, int] = field(default_factory=dict)  # sha -> row
+
+    def build(self, commits: List[Dict], refs: Dict[str, str]) -> None:
+        # commits are topo-ordered by git log --topo-order; we keep it.
+        self.nodes = [
+            CommitNode(
+                sha=c["sha"],
+                author=c["author"],
+                date=c["date"],
+                message=c["message"],
+                parents=c["parents"],
+            ) for c in commits
+        ]
+        self.index = {n.sha: i for i, n in enumerate(self.nodes)}
+        self._assign_lanes()
+
+    def _assign_lanes(self) -> None:
+        """
+        Simple lane assignment similar to 'git log --graph' lanes.
+        Greedy: reuse freed lanes; parents may create new lanes.
+        """
+        active: Dict[int, str] = {}  # lane -> sha
+        free_lanes: List[int] = []
+
+        for i, node in enumerate(self.nodes):
+            # If any active lane points to this commit, use that lane
+            lane_found = None
+            for lane, sha in list(active.items()):
+                if sha == node.sha:
+                    lane_found = lane
+                    break
+
+            if lane_found is None:
+                if free_lanes:
+                    node.lane = free_lanes.pop(0)
+                else:
+                    node.lane = 0 if not active else max(active.keys()) + 1
+            else:
+                node.lane = lane_found
+
+            # Update active: current node consumes its lane, parents occupy lanes
+            # Remove the current sha from any lane that pointed to it
+            for lane, sha in list(active.items()):
+                if sha == node.sha:
+                    del active[lane]
+
+            # First parent continues in the same lane; others go to free/new lanes
+            if node.parents:
+                first = node.parents[0]
+                active[node.lane] = first
+                # Side branches
+                for p in node.parents[1:]:
+                    # Pick a free lane or new one
+                    if free_lanes:
+                        pl = free_lanes.pop(0)
+                    else:
+                        pl = 0 if not active else max(active.keys()) + 1
+                    active[pl] = p
+
+            # Any lane whose target no longer appears in the future will be freed later
+            # We approximate by freeing lanes when a target didn't appear in the next rows;
+            # but for minimal viable, free when lane not reassigned by parents this row.
+            used_lanes = set(active.keys())
+            # Collect gaps below max lane as free lanes to reuse
+            max_lane = max(used_lanes) if used_lanes else -1
+            present = set(range(max_lane + 1))
+            missing = sorted(list(present - used_lanes))
+            # Merge missing into free_lanes maintaining order
+            free_lanes = sorted(set(free_lanes + missing))

je_editor/git/commit_graph.py

@@ -0,0 +1,181 @@
+import os
+from datetime import datetime
+
+from PySide6.QtCore import QThread, Signal
+from git import Repo, GitCommandError, InvalidGitRepositoryError, NoSuchPathError
+
+
+# -----------------------
+# Simple audit logger
+# -----------------------
+def audit_log(repo_path: str, action: str, detail: str, ok: bool, err: str = ""):
+    """
+    Append an audit log entry to 'audit.log' in the repo directory.
+    This is useful for compliance and traceability.
+    """
+    try:
+        path = os.path.join(repo_path if repo_path else ".", "audit.log")
+        with open(path, "a", encoding="utf-8") as f:
+            ts = datetime.now().isoformat(timespec="seconds")
+            f.write(f"{ts}\taction={action}\tok={ok}\tdetail={detail}\terr={err}\n")
+    except Exception:
+        pass  # Never let audit logging failure break the UI
+
+
+# -----------------------
+# Git service layer
+# -----------------------
+class GitService:
+    """
+    Encapsulates Git operations using GitPython.
+    Keeps UI logic separate from Git logic.
+    """
+
+    def __init__(self):
+        self.repo: Repo | None = None
+        self.repo_path: str | None = None
+
+    def open_repo(self, path: str):
+        try:
+            self.repo = Repo(path)
+            self.repo_path = path
+            audit_log(path, "open_repo", path, True)
+        except (InvalidGitRepositoryError, NoSuchPathError) as e:
+            audit_log(path, "open_repo", path, False, str(e))
+            raise
+
+    def list_branches(self):
+        self._ensure_repo()
+        branches = [head.name for head in self.repo.heads]
+        audit_log(self.repo_path, "list_branches", ",".join(branches), True)
+        return branches
+
+    def current_branch(self):
+        self._ensure_repo()
+        try:
+            return self.repo.active_branch.name
+        except TypeError:
+            return "(detached HEAD)"
+
+    def checkout(self, branch: str):
+        self._ensure_repo()
+        try:
+            self.repo.git.checkout(branch)
+            audit_log(self.repo_path, "checkout", branch, True)
+        except GitCommandError as e:
+            audit_log(self.repo_path, "checkout", branch, False, str(e))
+            raise
+
+    def list_commits(self, branch: str, max_count: int = 100):
+        self._ensure_repo()
+        commits = list(self.repo.iter_commits(branch, max_count=max_count))
+        data = [
+            {
+                "hexsha": c.hexsha,
+                "summary": c.summary,
+                "author": c.author.name if c.author else "",
+                "date": datetime.fromtimestamp(c.committed_date).isoformat(sep=" ", timespec="seconds"),
+            }
+            for c in commits
+        ]
+        audit_log(self.repo_path, "list_commits", f"{branch}:{len(data)}", True)
+        return data
+
+    def show_diff_of_commit(self, commit_sha: str) -> str:
+        self._ensure_repo()
+        commit = self.repo.commit(commit_sha)
+        parent = commit.parents[0] if commit.parents else None
+        if parent is None:
+            null_tree = self.repo.tree(NULL_TREE)
+            diffs = commit.diff(null_tree, create_patch=True)
+        else:
+            diffs = commit.diff(parent, create_patch=True)
+        text = []
+        for d in diffs:
+            try:
+                text.append(d.diff.decode("utf-8", errors="replace"))
+            except Exception:
+                pass
+        out = "".join(text) if text else "(No patch content)"
+        audit_log(self.repo_path, "show_diff", commit_sha, True)
+        return out
+
+    def stage_all(self):
+        self._ensure_repo()
+        try:
+            self.repo.git.add(all=True)
+            audit_log(self.repo_path, "stage_all", "git add -A", True)
+        except GitCommandError as e:
+            audit_log(self.repo_path, "stage_all", "git add -A", False, str(e))
+            raise
+
+    def commit(self, message: str):
+        self._ensure_repo()
+        if not message.strip():
+            raise ValueError("Commit message is empty.")
+        try:
+            self.repo.index.commit(message)
+            audit_log(self.repo_path, "commit", message, True)
+        except Exception as e:
+            audit_log(self.repo_path, "commit", message, False, str(e))
+            raise
+
+    def pull(self, remote: str = "origin", branch: str | None = None):
+        self._ensure_repo()
+        if branch is None:
+            branch = self.current_branch()
+        try:
+            res = self.repo.git.pull(remote, branch)
+            audit_log(self.repo_path, "pull", f"{remote}/{branch}", True)
+            return res
+        except GitCommandError as e:
+            audit_log(self.repo_path, "pull", f"{remote}/{branch}", False, str(e))
+            raise
+
+    def push(self, remote: str = "origin", branch: str | None = None):
+        self._ensure_repo()
+        if branch is None:
+            branch = self.current_branch()
+        try:
+            res = self.repo.git.push(remote, branch)
+            audit_log(self.repo_path, "push", f"{remote}/{branch}", True)
+            return res
+        except GitCommandError as e:
+            audit_log(self.repo_path, "push", f"{remote}/{branch}", False, str(e))
+            raise
+
+    def remotes(self):
+        self._ensure_repo()
+        return [r.name for r in self.repo.remotes]
+
+    def _ensure_repo(self):
+        if self.repo is None:
+            raise RuntimeError("Repository not opened.")
+
+
+# Null tree constant for initial commit diff
+NULL_TREE = "4b825dc642cb6eb9a060e54bf8d69288fbee4904"
+
+
+# -----------------------
+# Worker thread wrapper
+# -----------------------
+class Worker(QThread):
+    """
+    Runs a function in a separate thread to avoid blocking the UI.
+    Emits (result, error) when done.
+    """
+    done = Signal(object, object)
+
+    def __init__(self, fn, *args, **kwargs):
+        super().__init__()
+        self.fn = fn
+        self.args = args
+        self.kwargs = kwargs
+
+    def run(self):
+        try:
+            res = self.fn(*self.args, **self.kwargs)
+            self.done.emit(res, None)
+        except Exception as e:
+            self.done.emit(None, e)

je_editor/git/git.py

@@ -0,0 +1,66 @@
+import logging
+import subprocess
+from pathlib import Path
+from typing import List, Dict
+
+log = logging.getLogger(__name__)
+
+
+class GitCLI:
+    def __init__(self, repo_path: Path):
+        self.repo_path = Path(repo_path)
+
+    def is_git_repo(self) -> bool:
+        return (self.repo_path / ".git").exists()
+
+    def _run(self, args: List[str]) -> str:
+        log.debug("git %s", " ".join(args))
+        res = subprocess.run(
+            ["git"] + args,
+            cwd=self.repo_path,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            encoding="utf-8",
+        )
+        if res.returncode != 0:
+            log.error("Git failed: %s", res.stderr.strip())
+            raise RuntimeError(res.stderr.strip())
+        return res.stdout
+
+    def get_all_refs(self) -> Dict[str, str]:
+        # returns refname -> commit hash
+        out = self._run(["show-ref", "--heads", "--tags"])
+        refs = {}
+        for line in out.splitlines():
+            if not line.strip():
+                continue
+            sha, ref = line.split(" ", 1)
+            refs[ref.strip()] = sha.strip()
+        return refs
+
+    def get_commits(self, max_count: int = 500) -> List[Dict]:
+        """
+        Return recent commits across all refs, with parents.
+        """
+        fmt = "%H%x01%P%x01%an%x01%ad%x01%s"
+        out = self._run([
+            "log", "--date=short", f"--format={fmt}", "--all",
+            f"--max-count={max_count}", "--topo-order"
+        ])
+        commits = []
+        for line in out.splitlines():
+            if not line.strip():
+                continue
+            parts = line.split("\x01")
+            if len(parts) != 5:
+                continue
+            sha, parents, author, date, msg = parts
+            commits.append({
+                "sha": sha,
+                "parents": [p for p in parents.strip().split() if p],
+                "author": author,
+                "date": date,
+                "message": msg,
+            })
+        return commits

je_editor/git/git_cli.py

@@ -0,0 +1,50 @@
+import datetime
+import traceback
+
+from git import Repo, GitCommandError, InvalidGitRepositoryError, NoSuchPathError
+
+
+class GitCloneHandler:
+    """
+    Handles cloning of remote Git repositories with audit logging.
+    Can be reused in UI or CLI contexts.
+    """
+
+    def __init__(self, audit_log_path: str = "git_clone_audit.log"):
+        self.audit_log_path = audit_log_path
+
+    def clone_repo(self, remote_url: str, local_path: str) -> str:
+        """
+        Clone a remote repository to a local path.
+
+        :param remote_url: The Git repository URL (e.g., https://github.com/user/repo.git)
+        :param local_path: The local directory to clone into
+        :return: The path to the cloned repository
+        :raises: Exception if cloning fails
+        """
+        try:
+            self._log_audit(f"Cloning started: {remote_url} -> {local_path}")
+            Repo.clone_from(remote_url, local_path)
+            self._log_audit(f"Cloning completed: {remote_url} -> {local_path}")
+            return local_path
+        except (GitCommandError, InvalidGitRepositoryError, NoSuchPathError) as e:
+            self._log_audit(
+                f"ERROR: Git operation failed: {remote_url} -> {local_path}\n{str(e)}\nTraceback:\n{traceback.format_exc()}")
+            raise RuntimeError(f"Git operation failed: {str(e)}") from e
+        except Exception as e:
+            self._log_audit(
+                f"ERROR: Unexpected error during clone: {remote_url} -> {local_path}\n{str(e)}\nTraceback:\n{traceback.format_exc()}")
+            raise RuntimeError(f"Unexpected error during clone: {str(e)}") from e
+
+    def _log_audit(self, message: str):
+        """
+        Append an audit log entry with timestamp.
+        """
+        timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        log_entry = f"[{timestamp}] {message}\n"
+        try:
+            with open(self.audit_log_path, "a", encoding="utf-8") as f:
+                f.write(log_entry)
+        except Exception:
+            # Never let audit logging failure break the flow
+            pass