Coverage for plugin/scripts/pr_inventory.py: 100%
161 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-10-26 04:21 +0000
1#!/usr/bin/env python3
2"""List open pull/merge requests with the facts a merge order is decided from.
4`/rhiza:history` reads a backlog as a whole: which requests are obsolete, which order
5the rest should merge in, what to pick up next. ``pr_status.py`` already answers *is it
6green*, and ``issue_status.py`` already answers *can this issue be fixed without a
7judgement call*. Neither knows what a request **touches**, and that is what decides an
8order: two requests editing the same file conflict on whichever merges second, so the
9one that rebases should be the one that is cheaper to rebase.
11This script gathers facts and stops. Per request: its branches, draft state, author,
12age and idle time, whether the forge says it merges cleanly, the files it changes, and
13the issues it closes — with the ones already closed called out, since a request whose
14only purpose was fixing an issue someone else already fixed is the commonest kind of
15obsolete. Across requests: every pair that changes a file in common. **Which order to
16merge in is not here** — it weighs risk, size and intent, and that judgement belongs in
17the command's prose.
19The two forges disagree about how much of this a listing carries:
21 gh pr list --json …,files,mergeable,closingIssuesReferences
22 -> everything, in one call
23 glab mr list --output json
24 -> no files and no closing references; `has_conflicts` instead of `mergeable`
26On GitLab the files come from ``glab mr diff <iid> --raw``, one call per request, read
27off the ``diff --git a/X b/Y`` headers — git's own format, not a payload shape that could
28drift. The closing references come from GitLab's closing keywords in the description,
29which is how GitLab itself decides them; ``closes_source`` says which of the two each
30request's list came from, so a caller knows how far to trust an empty one.
32Usage:
33 uv run --python 3.12 --no-project python \\
34 scripts/pr_inventory.py [--target-dir DIR] [--limit N] [--json] [--dry-run]
36Exit codes:
37 0 report printed
38 1 the platform CLI is missing, failed, or answered something unreadable
39 2 the platform could not be determined
40"""
42from __future__ import annotations
44import argparse
45import itertools
46import json
47import re
48import shutil
49import subprocess # nosec B404
50import sys
51from datetime import UTC, datetime
52from pathlib import Path
53from typing import Any
55sys.path.insert(0, str(Path(__file__).resolve().parent))
56from _rhiza_forge import PlatformError, detect_platform # noqa: E402
58EXIT_OK = 0
59EXIT_CLI_FAILED = 1
60EXIT_NO_PLATFORM = 2
62# Spelled out rather than globbed: gh errors on an unknown field, so this list is itself
63# a contract with the CLI. Each one was checked against `gh pr list` on this repo.
64_GH_FIELDS = (
65 "number,title,url,headRefName,baseRefName,isDraft,author,createdAt,updatedAt,"
66 "mergeable,files,closingIssuesReferences,labels"
67)
69# GitLab's closing keywords, the ones that close an issue when the request merges.
70_CLOSING = re.compile(
71 r"\b(?:clos(?:e|es|ed|ing)|fix(?:es|ed|ing)?|resolv(?:e|es|ed|ing)|implement(?:s|ed|ing)?)"
72 r"\s*:?\s+#(\d+)",
73 re.IGNORECASE,
74)
75_DIFF_HEADER = re.compile(r"^diff --git a/(?P<old>.+) b/(?P<new>.+)$")
76_GH_MERGEABLE = {"MERGEABLE": "clean", "CONFLICTING": "conflicting"}
79class InventoryError(Exception):
80 """A platform CLI was absent, failed, or answered something unreadable."""
83def build_list_command(platform: str, *, limit: int) -> list[str]:
84 """Return the argv listing open requests on *platform*.
86 >>> build_list_command("gitlab", limit=5)
87 ['glab', 'mr', 'list', '--output', 'json', '--per-page', '5']
88 """
89 if platform == "github":
90 return [
91 "gh", "pr", "list", "--state", "open",
92 "--json", _GH_FIELDS, "--limit", str(limit),
93 ] # fmt: skip
94 return ["glab", "mr", "list", "--output", "json", "--per-page", str(limit)]
97def build_diff_command(iid: int) -> list[str]:
98 """Return the argv printing GitLab merge request *iid*'s diff in git's own format.
100 >>> build_diff_command(12)
101 ['glab', 'mr', 'diff', '12', '--raw']
102 """
103 return ["glab", "mr", "diff", str(iid), "--raw"]
106def build_issue_state_command(platform: str, number: int) -> list[str]:
107 """Return the argv reading issue *number*'s state.
109 >>> build_issue_state_command("github", 7)
110 ['gh', 'issue', 'view', '7', '--json', 'state']
111 """
112 if platform == "github":
113 return ["gh", "issue", "view", str(number), "--json", "state"]
114 return ["glab", "issue", "view", str(number), "--output", "json"]
117def run_cli(command: list[str], target_dir: Path) -> str:
118 """Run *command* in *target_dir* and return its stdout.
120 Raises:
121 InventoryError: the binary is missing or exited non-zero.
122 """
123 binary = shutil.which(command[0])
124 if binary is None:
125 raise InventoryError(f"{command[0]} is not installed")
126 result = subprocess.run( # nosec B603
127 [binary, *command[1:]], cwd=str(target_dir), capture_output=True, text=True, check=False
128 )
129 if result.returncode != 0:
130 detail = (result.stderr or result.stdout).strip().splitlines()
131 raise InventoryError(detail[0] if detail else f"{command[0]} failed")
132 return result.stdout
135def run_json(command: list[str], target_dir: Path) -> Any:
136 """Run *command* and parse its stdout as JSON.
138 Raises:
139 InventoryError: as `run_cli`, or the output was not JSON.
140 """
141 stdout = run_cli(command, target_dir)
142 try:
143 return json.loads(stdout or "null")
144 except json.JSONDecodeError as exc:
145 raise InventoryError(f"{command[0]} did not answer JSON: {exc}") from exc
148def closing_references(text: str) -> list[int]:
149 """The issue numbers *text* closes with one of GitLab's closing keywords, in order.
151 A bare mention is not a closing reference — only a keyword in front of it is:
153 >>> closing_references("Closes #12. Fixes: #3, see #9; resolved #12")
154 [12, 3]
155 """
156 seen: dict[int, None] = {}
157 for match in _CLOSING.finditer(text):
158 seen.setdefault(int(match.group(1)), None)
159 return list(seen)
162def parse_diff(diff: str) -> list[dict[str, Any]]:
163 """Reduce a git-format *diff* to ``[{path, additions, deletions}]``, one per file.
165 The path is the *new* side's, so a rename is reported where it lands:
167 >>> text = "diff --git a/x.py b/y.py\\n--- a/x.py\\n+++ b/y.py\\n+new\\n-old\\n+more\\n"
168 >>> parse_diff(text)
169 [{'path': 'y.py', 'additions': 2, 'deletions': 1}]
170 """
171 files: list[dict[str, Any]] = []
172 for line in diff.splitlines():
173 header = _DIFF_HEADER.match(line)
174 if header:
175 files.append({"path": header.group("new"), "additions": 0, "deletions": 0})
176 elif files and not line.startswith(("+++", "---")):
177 if line.startswith("+"):
178 files[-1]["additions"] += 1
179 elif line.startswith("-"):
180 files[-1]["deletions"] += 1
181 return files
184def days_between(earlier: str, now: datetime) -> int | None:
185 """Whole days from the ISO timestamp *earlier* to *now*, or None when unparseable.
187 Both forges' spellings — GitHub's ``Z`` and GitLab's milliseconds — are read:
189 >>> now = datetime(2026, 10, 9, tzinfo=UTC)
190 >>> days_between("2026-10-01T12:00:00Z", now), days_between("2026-10-02T00:00:00.123Z", now)
191 (7, 6)
192 >>> days_between("", now) is None
193 True
194 """
195 try:
196 then = datetime.fromisoformat(earlier.replace("Z", "+00:00"))
197 except ValueError:
198 return None
199 return (now - then).days
202def _text(raw: dict[str, Any], key: str) -> str:
203 """*raw*'s value at *key* as a string, with a missing or null value as ``''``.
205 >>> _text({"a": None}, "a"), _text({"a": 3}, "a")
206 ('', '3')
207 """
208 return str(raw.get(key) or "")
211def _count(raw: dict[str, Any], key: str) -> int:
212 """*raw*'s value at *key* as an int, with a missing or null value as ``0``."""
213 return int(raw.get(key) or 0)
216def _common(raw: dict[str, Any], keys: dict[str, str]) -> dict[str, Any]:
217 """The text fields both forges carry, read through *keys* (shared name -> forge key)."""
218 return {shared: _text(raw, key) for shared, key in keys.items()}
221_GH_KEYS = {
222 "title": "title",
223 "url": "url",
224 "branch": "headRefName",
225 "base": "baseRefName",
226 "created": "createdAt",
227 "updated": "updatedAt",
228}
229_GL_KEYS = {
230 "title": "title",
231 "url": "web_url",
232 "branch": "source_branch",
233 "base": "target_branch",
234 "created": "created_at",
235 "updated": "updated_at",
236}
239def normalize_github(raw: dict[str, Any]) -> dict[str, Any]:
240 """One `gh pr list` object in the shared vocabulary."""
241 files = [
242 {
243 "path": _text(f, "path"),
244 "additions": _count(f, "additions"),
245 "deletions": _count(f, "deletions"),
246 }
247 for f in raw.get("files") or []
248 ]
249 return {
250 "id": raw.get("number"),
251 **_common(raw, _GH_KEYS),
252 "draft": bool(raw.get("isDraft")),
253 "author": _text(raw.get("author") or {}, "login"),
254 "mergeable": _GH_MERGEABLE.get(_text(raw, "mergeable").upper(), "unknown"),
255 "labels": [_text(label, "name") for label in raw.get("labels") or []],
256 "files": files,
257 "closes": [ref.get("number") for ref in raw.get("closingIssuesReferences") or []],
258 "closes_source": "forge",
259 }
262def _gitlab_mergeable(raw: dict[str, Any]) -> str:
263 """``has_conflicts`` read as `mergeable`; absent, ``unknown`` rather than a guess.
265 >>> [_gitlab_mergeable(r) for r in ({"has_conflicts": True}, {"has_conflicts": False}, {})]
266 ['conflicting', 'clean', 'unknown']
267 """
268 conflicts = raw.get("has_conflicts")
269 if conflicts is None:
270 return "unknown"
271 return "conflicting" if conflicts else "clean"
274def normalize_gitlab(raw: dict[str, Any], files: list[dict[str, Any]]) -> dict[str, Any]:
275 """One `glab mr list` object in the shared vocabulary, with *files* from its diff.
277 ``has_conflicts`` is the nearest GitLab has to GitHub's ``mergeable``.
278 """
279 return {
280 "id": raw.get("iid"),
281 **_common(raw, _GL_KEYS),
282 "draft": bool(raw.get("draft") or raw.get("work_in_progress")),
283 "author": _text(raw.get("author") or {}, "username"),
284 "mergeable": _gitlab_mergeable(raw),
285 "labels": list(raw.get("labels") or []),
286 "files": files,
287 "closes": closing_references(_text(raw, "description")),
288 "closes_source": "description",
289 }
292def overlaps(requests: list[dict[str, Any]]) -> list[dict[str, Any]]:
293 """Every pair of requests changing a file in common, with the files they share.
295 >>> a = {"id": 1, "files": [{"path": "x"}, {"path": "y"}]}
296 >>> b = {"id": 2, "files": [{"path": "y"}]}
297 >>> c = {"id": 3, "files": [{"path": "z"}]}
298 >>> overlaps([a, b, c])
299 [{'a': 1, 'b': 2, 'files': ['y']}]
300 """
301 found = []
302 for first, second in itertools.combinations(requests, 2):
303 shared = {f["path"] for f in first["files"]} & {f["path"] for f in second["files"]}
304 if shared:
305 found.append({"a": first["id"], "b": second["id"], "files": sorted(shared)})
306 return found
309def issue_state(platform: str, number: int, target_dir: Path) -> str:
310 """Issue *number*'s state as ``open``, ``closed``, or ``unknown`` when it cannot be read.
312 Unreadable is not closed: a reference to an issue in another project, or one the
313 token cannot see, must not make a request look obsolete.
314 """
315 try:
316 payload = run_json(build_issue_state_command(platform, number), target_dir) or {}
317 except InventoryError:
318 return "unknown"
319 state = str(payload.get("state") or "").lower()
320 return {"opened": "open", "open": "open", "closed": "closed"}.get(state, "unknown")
323def _fetch(platform: str, target_dir: Path, limit: int) -> list[dict[str, Any]]:
324 """Ask the forge for its open requests, normalised, files included."""
325 listing = run_json(build_list_command(platform, limit=limit), target_dir) or []
326 if platform == "github":
327 return [normalize_github(raw) for raw in listing]
328 return [
329 normalize_gitlab(raw, parse_diff(run_cli(build_diff_command(raw["iid"]), target_dir)))
330 for raw in listing
331 ]
334def _annotate(
335 request: dict[str, Any], platform: str, target_dir: Path, now: datetime
336) -> dict[str, Any]:
337 """Add the derived facts: size, age, idle time, and which closing references closed."""
338 request["additions"] = sum(f["additions"] for f in request["files"])
339 request["deletions"] = sum(f["deletions"] for f in request["files"])
340 request["age_days"] = days_between(request["created"], now)
341 request["idle_days"] = days_between(request["updated"], now)
342 request["closes_closed"] = [
343 number
344 for number in request["closes"]
345 if issue_state(platform, number, target_dir) == "closed"
346 ]
347 return request
350def collect(
351 target_dir: Path, *, limit: int, dry_run: bool = False, now: datetime | None = None
352) -> dict[str, Any]:
353 """Return the inventory for *target_dir*'s open requests.
355 Raises:
356 PlatformError: the platform could not be determined.
357 InventoryError: the platform CLI failed.
358 """
359 platform = detect_platform(target_dir)
360 command = build_list_command(platform, limit=limit)
361 report: dict[str, Any] = {"platform": platform, "command": command, "requests": []}
362 report["overlaps"] = []
363 if dry_run:
364 return report
365 moment = now or datetime.now(UTC)
366 requests = [
367 _annotate(request, platform, target_dir, moment)
368 for request in _fetch(platform, target_dir, limit)
369 ]
370 report["requests"] = requests
371 report["overlaps"] = overlaps(requests)
372 return report
375def _render_request(request: dict[str, Any]) -> list[str]:
376 """The text lines for one request."""
377 flags = [request["mergeable"]] + (["draft"] if request["draft"] else [])
378 lines = [
379 f"#{request['id']} {request['title']}",
380 f" {request['branch']} -> {request['base']} by {request['author'] or '?'}"
381 f" [{', '.join(flags)}]",
382 f" age {request['age_days']}d, idle {request['idle_days']}d, "
383 f"{len(request['files'])} files, +{request['additions']}/-{request['deletions']}",
384 ]
385 if request["closes"]:
386 closed = request["closes_closed"]
387 suffix = f" (already closed: {', '.join(f'#{n}' for n in closed)})" if closed else ""
388 lines.append(f" closes {', '.join(f'#{n}' for n in request['closes'])}{suffix}")
389 return lines
392def render(report: dict[str, Any]) -> str:
393 """Return *report* as text."""
394 lines = [f"platform {report['platform']}", f"command {' '.join(report['command'])}", ""]
395 if not report["requests"]:
396 lines.append("no open requests")
397 for request in report["requests"]:
398 lines.extend(_render_request(request))
399 if report["overlaps"]:
400 lines += ["", "overlapping files:"]
401 for pair in report["overlaps"]:
402 lines.append(f" #{pair['a']} & #{pair['b']}: {', '.join(pair['files'])}")
403 return "\n".join(lines)
406def main(argv: list[str] | None = None) -> int:
407 """Parse arguments, gather the inventory, print it, and return an exit code."""
408 parser = argparse.ArgumentParser(
409 description="List open requests with the files, size and state a merge order needs."
410 )
411 parser.add_argument("--target-dir", default=".", help="Repository root (default: cwd).")
412 parser.add_argument("--limit", type=int, default=50, help="Maximum requests to fetch.")
413 parser.add_argument(
414 "--json", dest="json_output", action="store_true", help="Emit the report as JSON."
415 )
416 parser.add_argument("--dry-run", action="store_true", help="Print the argv without running it.")
417 args = parser.parse_args(argv)
419 try:
420 report = collect(Path(args.target_dir).resolve(), limit=args.limit, dry_run=args.dry_run)
421 except PlatformError as exc:
422 print(f"[ERROR] {exc}", file=sys.stderr)
423 return EXIT_NO_PLATFORM
424 except InventoryError as exc:
425 print(f"[ERROR] {exc}", file=sys.stderr)
426 return EXIT_CLI_FAILED
428 print(json.dumps(report, indent=2) if args.json_output else render(report))
429 return EXIT_OK
432if __name__ == "__main__": # pragma: no cover
433 raise SystemExit(main())