main
1#!/usr/bin/env python3
2"""
3Fetch unresolved PR review threads authored by bots for the current branch's PR.
4
5Outputs JSON with:
6 - pull_request metadata (number, url, title, owner, repo)
7 - bot_threads: list of unresolved review threads where the first comment
8 was authored by a bot (login ending in [bot] or type Bot)
9
10Usage:
11 python3 fetch_bot_reviews.py [--pr NUMBER]
12"""
13
14from __future__ import annotations
15
16import argparse
17import json
18import subprocess
19import sys
20from typing import Any
21
22QUERY = """\
23query(
24 $owner: String!,
25 $repo: String!,
26 $number: Int!,
27 $threadsCursor: String
28) {
29 repository(owner: $owner, name: $repo) {
30 pullRequest(number: $number) {
31 number
32 url
33 title
34 state
35 reviewThreads(first: 100, after: $threadsCursor) {
36 pageInfo { hasNextPage endCursor }
37 nodes {
38 id
39 isResolved
40 isOutdated
41 path
42 line
43 startLine
44 comments(first: 50) {
45 nodes {
46 id
47 databaseId
48 body
49 createdAt
50 author {
51 login
52 __typename
53 }
54 url
55 }
56 }
57 }
58 }
59 }
60 }
61}
62"""
63
64
65def _run(cmd: list[str], stdin: str | None = None) -> str:
66 p = subprocess.run(cmd, input=stdin, capture_output=True, text=True)
67 if p.returncode != 0:
68 raise RuntimeError(f"Command failed: {' '.join(cmd)}\n{p.stderr}")
69 return p.stdout
70
71
72def _run_json(cmd: list[str], stdin: str | None = None) -> dict[str, Any]:
73 out = _run(cmd, stdin=stdin)
74 try:
75 return json.loads(out)
76 except json.JSONDecodeError as e:
77 raise RuntimeError(f"Failed to parse JSON: {e}\nRaw:\n{out}") from e
78
79
80def _is_bot(author: dict[str, Any] | None) -> bool:
81 if not author:
82 return False
83 if author.get("__typename") == "Bot":
84 return True
85 login = author.get("login", "")
86 return login.endswith("[bot]")
87
88
89def get_pr_ref(pr_number: int | None = None) -> tuple[str, str, int]:
90 """Resolve owner, repo, number from the PR URL (works for fork PRs too)."""
91 cmd = ["gh", "pr", "view", "--json", "number,url"]
92 if pr_number is not None:
93 cmd.insert(3, str(pr_number))
94 pr = _run_json(cmd)
95 number = int(pr["number"])
96 # URL format: https://github.com/{owner}/{repo}/pull/{number}
97 parts = pr["url"].rstrip("/").split("/")
98 owner = parts[-4]
99 repo = parts[-3]
100 return owner, repo, number
101
102
103def fetch_threads(
104 owner: str, repo: str, number: int
105) -> tuple[dict[str, Any], list[dict[str, Any]]]:
106 """Fetch all review threads, return (pr_meta, all_threads)."""
107 all_threads: list[dict[str, Any]] = []
108 cursor: str | None = None
109 pr_meta: dict[str, Any] | None = None
110
111 while True:
112 cmd = [
113 "gh",
114 "api",
115 "graphql",
116 "-F",
117 "query=@-",
118 "-F",
119 f"owner={owner}",
120 "-F",
121 f"repo={repo}",
122 "-F",
123 f"number={number}",
124 ]
125 if cursor:
126 cmd += ["-F", f"threadsCursor={cursor}"]
127
128 payload = _run_json(cmd, stdin=QUERY)
129 if "errors" in payload and payload["errors"]:
130 raise RuntimeError(
131 f"GraphQL errors:\n{json.dumps(payload['errors'], indent=2)}"
132 )
133
134 pr = payload["data"]["repository"]["pullRequest"]
135 if pr_meta is None:
136 pr_meta = {
137 "number": pr["number"],
138 "url": pr["url"],
139 "title": pr["title"],
140 "state": pr["state"],
141 "owner": owner,
142 "repo": repo,
143 }
144
145 t = pr["reviewThreads"]
146 all_threads.extend(t.get("nodes") or [])
147
148 if t["pageInfo"]["hasNextPage"]:
149 cursor = t["pageInfo"]["endCursor"]
150 else:
151 break
152
153 assert pr_meta is not None
154 return pr_meta, all_threads
155
156
157def filter_bot_unresolved(threads: list[dict[str, Any]]) -> list[dict[str, Any]]:
158 """Return only unresolved threads where the first comment is by a bot."""
159 result = []
160 for thread in threads:
161 if thread.get("isResolved"):
162 continue
163 comments = thread.get("comments", {}).get("nodes", [])
164 if not comments:
165 continue
166 first_author = comments[0].get("author")
167 if _is_bot(first_author):
168 first = comments[0]
169 # Extract suggestion blocks
170 body = first.get("body", "")
171 has_suggestion = "```suggestion" in body
172 result.append(
173 {
174 "thread_id": thread["id"],
175 "file": thread.get("path", ""),
176 "line": thread.get("line"),
177 "start_line": thread.get("startLine"),
178 "is_outdated": thread.get("isOutdated", False),
179 "comment_id": first.get("databaseId"),
180 "comment_url": first.get("url", ""),
181 "author": first_author.get("login", "unknown"),
182 "body": body,
183 "has_suggestion": has_suggestion,
184 "reply_count": len(comments) - 1,
185 }
186 )
187 return result
188
189
190def main() -> None:
191 parser = argparse.ArgumentParser(description="Fetch unresolved bot review threads")
192 parser.add_argument(
193 "--pr", type=int, default=None, help="PR number (defaults to current branch PR)"
194 )
195 args = parser.parse_args()
196
197 try:
198 _run(["gh", "auth", "status"])
199 except RuntimeError:
200 print("gh not authenticated. Run: gh auth login", file=sys.stderr)
201 sys.exit(1)
202
203 owner, repo, number = get_pr_ref(args.pr)
204 pr_meta, threads = fetch_threads(owner, repo, number)
205 bot_threads = filter_bot_unresolved(threads)
206
207 output = {
208 "pull_request": pr_meta,
209 "bot_threads": bot_threads,
210 "total_unresolved_bot_threads": len(bot_threads),
211 }
212 print(json.dumps(output, indent=2))
213
214
215if __name__ == "__main__":
216 main()