vibefpmgui.git_status

  1"""Git: a project's repository's status, and the two versions of a changed
  2file to compare, from the git program (``git status --porcelain=v2``, a
  3format made for programs; ``git cat-file``). Git must be installed (on the
  4PATH); without it the Git panel says so.
  5
  6git runs as a process of its own (see `vibefpmgui.console.run`): this
  7process only waits for it, without Python's lock, so the window goes on.
  8
  9No Qt: `vibefpmgui.git_panel` shows it, `vibefpmgui.window.git` reads the
 10status on a pool thread.
 11"""
 12
 13from __future__ import annotations
 14
 15import os
 16import shutil
 17from dataclasses import dataclass, field
 18from pathlib import Path
 19
 20from . import console
 21from .source_text import decode
 22
 23# Where a change is: what `git status` calls "Changes to be committed",
 24# "Changes not staged for commit", "Untracked files" and "Unmerged paths".
 25STAGED, CHANGED, UNTRACKED, CONFLICT = "staged", "changed", "untracked", "conflict"
 26AREAS = (CONFLICT, STAGED, CHANGED, UNTRACKED)  # (the order they're listed in)
 27AREA_TITLES = {CONFLICT: "Conflicts", STAGED: "Staged Changes", CHANGED: "Changes", UNTRACKED: "Untracked"}
 28
 29# What happened to the file, and its letter (as in `git status --short`).
 30LETTERS = {"modified": "M", "added": "A", "deleted": "D", "renamed": "R", "typechange": "T",
 31           "untracked": "U", "conflict": "!"}
 32KIND_WORDS = {"modified": "modified", "added": "added", "deleted": "deleted", "renamed": "renamed",
 33              "typechange": "type changed", "untracked": "untracked", "conflict": "conflict"}
 34
 35MAX_DIFF_BYTES = 4 * 1024 * 1024  # (a bigger version isn't shown: see versions)
 36TIMEOUT = 120  # seconds a git command may take
 37BINARY = "\0binary"  # versions() marker: the file isn't text
 38
 39
 40@dataclass(frozen=True)
 41class Change:
 42    path: str           # relative to the repository's top folder, with "/"
 43    area: str           # STAGED, CHANGED, UNTRACKED or CONFLICT
 44    kind: str           # a key of LETTERS
 45    old_path: str | None = None  # a renamed file's name before
 46
 47    @property
 48    def letter(self) -> str:
 49        return LETTERS[self.kind]
 50
 51
 52@dataclass
 53class RepoStatus:
 54    root: Path                  # the repository's top folder (its work tree)
 55    branch: str | None = None   # None: detached HEAD (see head)
 56    head: str | None = None     # the commit checked out (short), None before the first commit
 57    upstream: str | None = None  # e.g. "origin/main"
 58    ahead: int = 0              # commits here that the upstream doesn't have
 59    behind: int = 0             # and the other way around
 60    changes: list[Change] = field(default_factory=list)
 61
 62    def in_area(self, area: str) -> list[Change]:
 63        return [c for c in self.changes if c.area == area]
 64
 65    def summary(self) -> str:
 66        """ "main → origin/main · 2 ahead, 1 behind" and the like."""
 67        if self.branch is None:
 68            where = f"detached at {self.head}" if self.head else "detached HEAD"
 69        else:
 70            where = self.branch + ("" if self.head else " (no commits yet)")
 71        if self.upstream:
 72            where += f" → {self.upstream}"
 73            counts = [f"{n} {word}" for n, word in ((self.ahead, "ahead"), (self.behind, "behind")) if n]
 74            where += f" · {', '.join(counts) if counts else 'up to date'}"
 75        return where
 76
 77
 78class GitError(RuntimeError):
 79    """git couldn't do it (its message)."""
 80
 81
 82def executable() -> str | None:
 83    """The git program, or None if it isn't installed (on the PATH)."""
 84    return shutil.which("git")
 85
 86
 87def available() -> bool:
 88    """Whether git is installed (the Git panel says so otherwise)."""
 89    return executable() is not None
 90
 91
 92def _git(folder: Path, *args: str, limit: int | None = None) -> tuple[int, bytes | None, str]:
 93    """Run ``git -C folder args`` (no optional locks: it never holds the
 94    index lock that your own git commands need); (exit code, its output,
 95    or None if over ``limit`` bytes, its error message)."""
 96    git = executable()
 97    if git is None:
 98        raise GitError("git isn't installed")
 99    env = dict(os.environ, GIT_OPTIONAL_LOCKS="0", GIT_TERMINAL_PROMPT="0")
100    try:
101        code, output, message = console.run([git, "-C", str(folder), *args], TIMEOUT, limit, env)
102    except OSError as error:
103        raise GitError(f"couldn't run git: {error}") from None
104    except TimeoutError as error:
105        raise GitError(str(error)) from None
106    return code, output, message
107
108
109def _check(folder: Path, *args: str) -> bytes:
110    code, output, message = _git(folder, *args)
111    if code != 0:
112        lines = message.splitlines()
113        raise GitError(lines[-1].removeprefix("fatal: ") if lines else f"git {args[0]} failed ({code})")
114    return output or b""
115
116
117def find_repository(path: Path) -> Path | None:
118    """The top folder of the git repository ``path`` is in, or None (not in
119    one, a bare one, or no git)."""
120    if not available() or not Path(path).is_dir():
121        return None
122    code, output, _message = _git(Path(path), "rev-parse", "--show-toplevel")
123    if code != 0 or not output or not output.strip():
124        return None
125    return Path(output.decode("utf-8", "replace").strip()).resolve()
126
127
128# Each side of porcelain v2's "XY": what happened (index side: staged; work tree side: not staged).
129KINDS = {"M": "modified", "A": "added", "D": "deleted", "R": "renamed", "C": "added", "T": "typechange"}
130
131
132def read_status(path: Path) -> RepoStatus | None:
133    """The status of the repository ``path`` is in (None: not in one). A file
134    can be in two areas: some of its changes staged, some not. Raises
135    GitError if git can't read it (its message)."""
136    root = find_repository(path)
137    if root is None:
138        return None
139    output = _check(root, "status", "--porcelain=v2", "--branch", "-z", "--untracked-files=all")
140    return parse_status(root, output)
141
142
143def discard(root: Path, change: Change) -> list[Path]:
144    """Discard a file's changes (the Git panel's Discard Changes…). In
145    Changes: back to its staged version (or the last commit's), what's staged
146    kept. In Staged Changes: back to the last commit's version, staged and
147    not (a file new in it unstaged, a renamed one back under its old name).
148    The files then to remove, which git has no copy of (an untracked file, a
149    new one unstaged): returned, not deleted here (see the window: they go
150    to the Trash). Raises GitError (git's message)."""
151    if change.area == CONFLICT:
152        raise GitError("a file with a conflict: resolve it, or abort the merge, first")
153    if change.area == UNTRACKED:
154        return [root / change.path]
155    if change.area == CHANGED:
156        _check(root, "restore", "--worktree", "--", change.path)
157        return []
158    if change.kind in ("added", "renamed"):  # (not in the last commit under this name)
159        _check(root, "rm", "--cached", "--quiet", "--", change.path)
160        if change.kind == "renamed" and change.old_path:
161            _check(root, "restore", "--source=HEAD", "--staged", "--worktree", "--", change.old_path)
162        return [root / change.path]
163    _check(root, "restore", "--source=HEAD", "--staged", "--worktree", "--", change.path)
164    return []
165
166
167def parse_status(root: Path, output: bytes) -> RepoStatus:
168    """``git status --porcelain=v2 --branch -z`` read."""
169    status = RepoStatus(root)
170    fields = output.decode("utf-8", "surrogateescape").split("\0")
171    i = 0
172    while i < len(fields):
173        entry = fields[i]
174        i += 1
175        if entry.startswith("# "):
176            key, _, value = entry[2:].partition(" ")
177            if key == "branch.oid":
178                status.head = None if value == "(initial)" else value[:7]
179            elif key == "branch.head":
180                status.branch = None if value == "(detached)" else value
181            elif key == "branch.upstream":
182                status.upstream = value
183            elif key == "branch.ab":
184                ahead, _, behind = value.partition(" ")
185                status.ahead, status.behind = abs(int(ahead)), abs(int(behind))
186        elif entry.startswith(("1 ", "2 ")):
187            parts = entry.split(" ", 9 if entry[0] == "2" else 8)
188            file, old = parts[-1], None
189            if entry[0] == "2":  # (renamed or copied: the next field is its name before)
190                old = fields[i]
191                i += 1
192            staged, changed = parts[1][0], parts[1][1]
193            if staged != ".":
194                status.changes.append(Change(file, STAGED, KINDS.get(staged, "modified"),
195                                             old if staged in "RC" else None))
196            if changed != ".":
197                status.changes.append(Change(file, CHANGED, KINDS.get(changed, "modified")))
198        elif entry.startswith("u "):
199            status.changes.append(Change(entry.split(" ", 10)[-1], CONFLICT, "conflict"))
200        elif entry.startswith("? "):
201            status.changes.append(Change(entry[2:], UNTRACKED, "untracked"))
202    status.changes.sort(key=lambda c: c.path)
203    return status
204
205
206def _text(data: bytes | None) -> str | None:
207    if data is None:
208        return None
209    if len(data) > MAX_DIFF_BYTES or b"\0" in data[:8192]:
210        return BINARY
211    return decode(data)[0].replace("\r\n", "\n")
212
213
214def versions(root: Path, change: Change) -> tuple[str | None, str | None, str, str]:
215    """(old text, new text, old label, new label) of a change: what `git diff`
216    compares. Staged: the last commit and the index; not staged: the index
217    and the file; untracked: nothing and the file; a conflict: the last
218    commit and the file (with its conflict markers). None: there's no such
219    version (added, deleted); BINARY: it isn't text, or it's huge."""
220
221    def blob(spec: str) -> bytes | str | None:
222        code, output, _message = _git(root, "cat-file", "blob", spec, limit=MAX_DIFF_BYTES)
223        if code != 0:
224            return None  # (not in that version)
225        return BINARY if output is None else output
226
227    def committed() -> bytes | str | None:
228        return blob(f"HEAD:{change.old_path or change.path}")
229
230    def indexed() -> bytes | str | None:
231        return blob(f":{change.path}")
232
233    def on_disk() -> bytes | str | None:
234        file = root / change.path
235        try:
236            if not file.is_file():
237                return None
238            return BINARY if file.stat().st_size > MAX_DIFF_BYTES else file.read_bytes()
239        except OSError:
240            return None
241
242    if change.area == STAGED:
243        old, new, labels = committed(), indexed(), ("last commit", "staged")
244    elif change.area == CHANGED:
245        old, new, labels = indexed(), on_disk(), ("staged", "working tree")
246    elif change.area == UNTRACKED:
247        old, new, labels = None, on_disk(), ("", "working tree")
248    else:
249        old, new, labels = committed(), on_disk(), ("last commit", "working tree")
250    return (*(v if isinstance(v, str) else _text(v) for v in (old, new)), *labels)