235 lines
7.0 KiB
Python
235 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Fetch GitHub PR review comments and output JSON for insert_code_review_comments.
|
|
|
|
Requires: gh CLI (authenticated), git.
|
|
Must be run from within a git repository whose current branch has an open PR.
|
|
|
|
Prints JSON to stdout matching the insert_code_review_comments tool schema.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
from trim_diff_hunk import trim_diff_hunk, line_in_hunk, last_reachable_line
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def run_command(args, error_msg="Command failed"):
|
|
"""Run a command and return stdout. Exits on failure."""
|
|
result = subprocess.run(
|
|
args,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
env={**os.environ, "GH_PAGER": ""},
|
|
)
|
|
if result.returncode != 0:
|
|
print(f"{error_msg}: {result.stderr.strip()}", file=sys.stderr)
|
|
sys.exit(1)
|
|
return result.stdout
|
|
|
|
|
|
def run_gh_api(endpoint):
|
|
"""Run ``gh api --paginate`` and return a list of JSON objects.
|
|
|
|
Handles the case where ``gh`` outputs multiple concatenated JSON arrays
|
|
(one per page).
|
|
"""
|
|
text = run_command(
|
|
["gh", "api", endpoint, "--paginate"],
|
|
error_msg=f"gh api {endpoint} failed",
|
|
).strip()
|
|
if not text:
|
|
return []
|
|
|
|
items = []
|
|
decoder = json.JSONDecoder()
|
|
pos = 0
|
|
try:
|
|
while pos < len(text):
|
|
while pos < len(text) and text[pos] in " \t\n\r":
|
|
pos += 1
|
|
if pos >= len(text):
|
|
break
|
|
obj, end = decoder.raw_decode(text, pos)
|
|
items.extend(obj if isinstance(obj, list) else [obj])
|
|
pos = end
|
|
except json.JSONDecodeError as exc:
|
|
print(f"Failed to parse API response: {exc}", file=sys.stderr)
|
|
sys.exit(1)
|
|
return items
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Comment building
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _comment(cid, author, ts, body, url, location=None, reply_to=None):
|
|
"""Build a dict matching the insert_code_review_comments comment schema."""
|
|
c = {
|
|
"comment_id": cid,
|
|
"author": author,
|
|
"last_modified_timestamp": ts,
|
|
"comment_body": body,
|
|
"html_url": url,
|
|
}
|
|
if reply_to:
|
|
c["reply_metadata"] = {"parent_comment_id": reply_to}
|
|
elif location:
|
|
c["location_metadata"] = location
|
|
return c
|
|
|
|
|
|
def _resolve_line(hunk, line, original_line, side):
|
|
"""Pick the first line number that is reachable in the hunk on *side*.
|
|
|
|
Tries *line* first (current diff position), then *original_line*
|
|
(position when the comment was placed). If neither is reachable,
|
|
falls back to the last reachable line in the hunk on *side*.
|
|
|
|
Returns the resolved line number, or ``None`` if nothing is reachable.
|
|
"""
|
|
if line and line_in_hunk(hunk, line, side):
|
|
return line
|
|
if original_line and line_in_hunk(hunk, original_line, side):
|
|
return original_line
|
|
return last_reachable_line(hunk, side)
|
|
|
|
|
|
def _resolve_comment_line(comment, hunk):
|
|
"""Resolve validated (end_line, start_line, side) for a diff comment.
|
|
|
|
Uses ``side`` from the GitHub API as the authoritative diff side.
|
|
Returns ``(end_line, start_line | None, side)`` or ``None`` when the
|
|
comment cannot be attached to any line in the hunk.
|
|
"""
|
|
side = comment.get("side") or "RIGHT"
|
|
|
|
end_line = _resolve_line(
|
|
hunk,
|
|
comment.get("line"),
|
|
comment.get("original_line"),
|
|
side,
|
|
)
|
|
if end_line is None:
|
|
return None
|
|
|
|
raw_start = comment.get("start_line")
|
|
raw_original_start = comment.get("original_start_line")
|
|
if raw_start or raw_original_start:
|
|
start_line = _resolve_line(hunk, raw_start, raw_original_start, side)
|
|
if start_line is not None and start_line > end_line:
|
|
start_line = None
|
|
else:
|
|
start_line = None
|
|
|
|
return (end_line, start_line, side)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def main():
|
|
repo_root = run_command(
|
|
["git", "rev-parse", "--show-toplevel"],
|
|
"Not a git repository",
|
|
).strip()
|
|
|
|
pr = json.loads(
|
|
run_command(
|
|
[
|
|
"gh", "pr", "view",
|
|
"--json", "number,headRepository,headRepositoryOwner,baseRefName",
|
|
],
|
|
"Failed to get PR info (is there an open PR on this branch?)",
|
|
)
|
|
)
|
|
number = pr["number"]
|
|
owner = pr["headRepositoryOwner"]["login"]
|
|
repo = pr["headRepository"]["name"]
|
|
base = pr["baseRefName"]
|
|
|
|
api = f"/repos/{owner}/{repo}"
|
|
|
|
issue_comments = run_gh_api(f"{api}/issues/{number}/comments")
|
|
diff_comments = run_gh_api(f"{api}/pulls/{number}/comments")
|
|
reviews = run_gh_api(f"{api}/pulls/{number}/reviews")
|
|
|
|
comments = []
|
|
|
|
# -- Issue comments (PR-level, no location or reply metadata) -----------
|
|
for c in issue_comments:
|
|
comments.append(
|
|
_comment(
|
|
str(c["id"]),
|
|
c["user"]["login"] if c.get("user") else "[deleted]",
|
|
c["updated_at"],
|
|
c["body"],
|
|
c["html_url"],
|
|
)
|
|
)
|
|
|
|
# -- Diff comments (line-level, with location or reply) -----------------
|
|
for c in diff_comments:
|
|
cid = str(c["id"])
|
|
author = c["user"]["login"] if c.get("user") else "[deleted]"
|
|
ts = c["updated_at"]
|
|
body = c["body"]
|
|
url = c["html_url"]
|
|
|
|
reply_to_id = c.get("in_reply_to_id")
|
|
if reply_to_id:
|
|
comments.append(
|
|
_comment(cid, author, ts, body, url, reply_to=str(reply_to_id))
|
|
)
|
|
continue
|
|
|
|
hunk = c.get("diff_hunk", "")
|
|
resolved = _resolve_comment_line(c, hunk)
|
|
|
|
loc = {"filepath": c["path"]}
|
|
if resolved:
|
|
end_line, start_line, side = resolved
|
|
if hunk:
|
|
loc["diff_hunk"] = trim_diff_hunk(
|
|
hunk, end_line, side=side, start_line=start_line
|
|
)
|
|
loc["end_line"] = end_line
|
|
if start_line:
|
|
loc["start_line"] = start_line
|
|
loc["side"] = side
|
|
|
|
comments.append(_comment(cid, author, ts, body, url, location=loc))
|
|
|
|
# -- Reviews (PR-level, no location) ------------------------------------
|
|
for r in reviews:
|
|
if not r.get("body"):
|
|
continue
|
|
comments.append(
|
|
_comment(
|
|
str(r["id"]),
|
|
r["user"]["login"] if r.get("user") else "[deleted]",
|
|
r.get("submitted_at", ""),
|
|
r["body"],
|
|
r["html_url"],
|
|
)
|
|
)
|
|
|
|
result = {
|
|
"local_repository_path": repo_root,
|
|
"base_branch": base,
|
|
"comments": comments,
|
|
}
|
|
|
|
json.dump(result, sys.stdout, indent=2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|