vibefpmgui.git_log

  1"""The Git panel's *History*: the repository's commits, newest first, with
  2their branches and tags, and the graph of how they join (each commit's
  3lane, the lines to its parents), and the two versions of a file a commit
  4changed.
  5
  6The graph's layout (`graph`): each lane waits for a commit (a parent of one
  7above it). A commit goes in the lane waiting for it (the leftmost, if
  8several are: the others join it there); then that lane waits for its first
  9parent, and its other parents (a merge's) get lanes of their own, the
 10leftmost free one, unless one is already waiting for them. Lanes keep their
 11places (a free one is used again), so a line that passes a row goes
 12straight down; each lane keeps its color from where it started.
 13
 14No Qt: `vibefpmgui.git_log_view` shows it; it's read on a pool thread.
 15"""
 16
 17from __future__ import annotations
 18
 19from dataclasses import dataclass, field
 20from pathlib import Path
 21
 22from .git_blame import FileChange
 23from .git_status import BINARY, MAX_DIFF_BYTES, _check, _git, _text, find_repository
 24
 25MAX_COMMITS = 3000  # read at first, at most (the newest; the rest: the History's Load Older Commits)
 26
 27# Ref kinds, as their badges show them.
 28HEAD, BRANCH, REMOTE, TAG = "head", "branch", "remote", "tag"
 29
 30
 31@dataclass
 32class Ref:
 33    name: str            # "main", "origin/main", "v1.0", "HEAD" (detached)
 34    kind: str            # HEAD, BRANCH, REMOTE or TAG
 35    current: bool = False  # the branch checked out (HEAD -> it)
 36
 37
 38@dataclass
 39class LogCommit:
 40    sha: str
 41    parents: list[str]
 42    author: str
 43    email: str
 44    time: int             # its author's date (seconds since 1970)
 45    subject: str
 46    refs: list[Ref] = field(default_factory=list)
 47
 48
 49@dataclass
 50class GraphRow:
 51    """A commit's row of the graph: lanes are numbered from the left, each
 52    line given as (lane, color number)."""
 53    lane: int                     # the commit's
 54    color: int                    # its lane's
 55    from_above: bool              # a line comes down into it (it's a parent of a commit above)
 56    through: list[tuple[int, int]]  # lanes passing by (top to bottom)
 57    joining: list[tuple[int, int]]  # lanes ending in it (from the top: a branch that started here)
 58    parents: list[tuple[int, int]]  # lanes leaving it (to the bottom: its parents')
 59    width: int                    # lanes in use, above or below it
 60
 61
 62@dataclass
 63class Log:
 64    root: Path
 65    head: str                     # the commit checked out
 66    commits: list[LogCommit]
 67    rows: list[GraphRow]
 68    all_branches: bool
 69    limited: bool                 # more commits than the limit: the newest shown
 70    remote: str = ""              # origin's address (the commits' web pages)
 71    total: int = 0                # the commits there are (more than those listed, if limited)
 72
 73
 74def parse_refs(decoration: str) -> list[Ref]:
 75    """The refs in ``%D`` (with ``--decorate=full``): "HEAD -> refs/heads/main,
 76    tag: refs/tags/v1, refs/remotes/origin/main". The branch checked out
 77    first, then the others, the remote ones, the tags."""
 78    refs = []
 79    for part in (p.strip() for p in decoration.split(",")):
 80        current = part.startswith("HEAD -> ")
 81        part = part.removeprefix("HEAD -> ")
 82        if part == "HEAD":
 83            refs.append(Ref("HEAD", HEAD, True))  # (detached)
 84        elif part.startswith("tag: refs/tags/"):
 85            refs.append(Ref(part.removeprefix("tag: refs/tags/"), TAG))
 86        elif part.startswith("refs/heads/"):
 87            refs.append(Ref(part.removeprefix("refs/heads/"), BRANCH, current))
 88        elif part.startswith("refs/remotes/") and not part.endswith("/HEAD"):
 89            refs.append(Ref(part.removeprefix("refs/remotes/"), REMOTE))
 90    order = {HEAD: 0, BRANCH: 1, REMOTE: 2, TAG: 3}
 91    return sorted(refs, key=lambda r: (not r.current, order[r.kind]))
 92
 93
 94FIELDS = "%H%x1f%P%x1f%an%x1f%ae%x1f%at%x1f%D%x1f%s%x1e"
 95
 96
 97def parse_log(output: bytes) -> list[LogCommit]:
 98    """``git log --format=FIELDS --decorate=full`` read."""
 99    commits = []
100    for record in output.decode("utf-8", "replace").split("\x1e"):
101        parts = record.strip("\n").split("\x1f")
102        if len(parts) != 7 or len(parts[0]) < 7:
103            continue
104        sha, parents, author, email, when, refs, subject = parts
105        commits.append(LogCommit(sha, parents.split(), author, email, int(when) if when.isdigit() else 0, subject,
106                                 parse_refs(refs)))
107    return commits
108
109
110def graph(commits: list[LogCommit]) -> list[GraphRow]:
111    """Each commit's row of the graph (see the module's description)."""
112    lanes: list[str | None] = []  # the commit each lane waits for
113    colors: list[int] = []
114    next_color = 0
115    rows = []
116
117    def free_lane(avoid: int = -1) -> int:
118        for i, waiting in enumerate(lanes):
119            if waiting is None and i != avoid:
120                return i
121        lanes.append(None)
122        colors.append(0)
123        return len(lanes) - 1
124
125    for commit in commits:
126        above = len(lanes)
127        from_above = commit.sha in lanes
128        if from_above:
129            lane = lanes.index(commit.sha)
130        else:
131            lane = free_lane()
132            lanes[lane], colors[lane] = commit.sha, next_color
133            next_color += 1
134        joining = [(i, colors[i]) for i, waiting in enumerate(lanes) if waiting == commit.sha and i != lane]
135        through = [(i, colors[i]) for i, waiting in enumerate(lanes) if waiting is not None and waiting != commit.sha]
136        for i, _color in joining:
137            lanes[i] = None
138        parents = []
139        if commit.parents:
140            lanes[lane] = commit.parents[0]
141            parents.append((lane, colors[lane]))
142            for parent in commit.parents[1:]:
143                if parent in lanes:
144                    i = lanes.index(parent)
145                else:
146                    i = free_lane(avoid=lane)
147                    lanes[i], colors[i] = parent, next_color
148                    next_color += 1
149                parents.append((i, colors[i]))
150        else:
151            lanes[lane] = None  # (the first commit: nothing below)
152        rows.append(GraphRow(lane, colors[lane], from_above, through, joining, parents, max(above, len(lanes))))
153        while lanes and lanes[-1] is None:
154            lanes.pop()
155            colors.pop()
156    return rows
157
158
159def read_log(path: Path, all_branches: bool = True, limit: int | None = MAX_COMMITS) -> Log | None:
160    """The history of the repository ``path`` is in: every branch's (local,
161    remote) and tag's, or only what's checked out (None: not in a
162    repository, or no commits yet); the newest ``limit`` commits (None:
163    all of them, to the first). Raises GitError if git can't read it."""
164    root = find_repository(path)
165    if root is None:
166        return None
167    code, head, _message = _git(root, "rev-parse", "--verify", "-q", "HEAD")
168    if code != 0 or not head or not head.strip():
169        return None
170    which = ["--branches", "--remotes", "--tags", "HEAD"] if all_branches else ["HEAD"]
171    most = [] if limit is None else [f"--max-count={limit + 1}"]
172    output = _check(root, "log", f"--format={FIELDS}", "--decorate=full", "--date-order", *most, *which, "--")
173    commits = parse_log(output)
174    limited = limit is not None and len(commits) > limit
175    total = len(commits)
176    if limited:
177        commits = commits[:limit]
178        total = int(_check(root, "rev-list", "--count", *which, "--").strip() or 0)  # (how many more there are)
179    _code, remote, _message = _git(root, "config", "--get", "remote.origin.url")
180    return Log(root, head.decode().strip(), commits, graph(commits), all_branches, limited,
181               (remote or b"").decode("utf-8", "replace").strip(), total)
182
183
184def file_versions(root: Path, commit: LogCommit, change: FileChange) -> tuple[str | None, str | None]:
185    """(before, after) a commit of a file it changed: the text in its first
186    parent (None: added, or the first commit), and in it (None: deleted).
187    BINARY: it isn't text, or it's huge."""
188
189    def blob(spec: str) -> str | None:
190        code, output, _message = _git(root, "cat-file", "blob", spec, limit=MAX_DIFF_BYTES)
191        if code != 0:
192            return None
193        return BINARY if output is None else _text(output)
194
195    parent = commit.parents[0] if commit.parents else None
196    before = blob(f"{parent}:{change.old_path or change.path}") if parent and change.status != "A" else None
197    after = blob(f"{commit.sha}:{change.path}") if change.status != "D" else None
198    return before, after