Setting the file. One moment.
Fetch Comments · gh Address Comments · openai/skills · Skills Docs
Back to the top of the page scripts/ fetch_comments.py
Python · 237 lines · 6 KB
17
18 import json
19 import subprocess
20 import sys
21 from typing import Any
22
23 QUERY = """ \
24 query(
25 $owner: String!,
26 $repo: String!,
27 $number: Int!,
28 $commentsCursor: String,
29 $reviewsCursor: String,
30 $threadsCursor: String
31 ) {
32 repository(owner: $owner, name: $repo) {
33 pullRequest(number: $number) {
34 number
35 url
36 title
37 state
38
39 # Top-level "Conversation" comments (issue comments on the PR)
40 comments(first: 100, after: $commentsCursor) {
41 pageInfo { hasNextPage endCursor }
42 nodes {
43 id
44 body
45 createdAt
46 updatedAt
47 author { login }
48 }
49 }
50
51 # Review submissions (Approve / Request changes / Comment), with body if present
52 reviews(first: 100, after: $reviewsCursor) {
53 pageInfo { hasNextPage endCursor }
54 nodes {
55 id
56 state
57 body
58 submittedAt
59 author { login }
60 }
61 }
62
63 # Inline review threads (grouped), includes resolved state
64 reviewThreads(first: 100, after: $threadsCursor) {
65 pageInfo { hasNextPage endCursor }
66 nodes {
67 id
68 isResolved
69 isOutdated
70 path
71 line
72 diffSide
73 startLine
74 startDiffSide
75 originalLine
76 originalStartLine
77 resolvedBy { login }
78 comments(first: 100) {
79 nodes {
80 id
81 body
82 createdAt
83 updatedAt
84 author { login }
85 }
86 }
87 }
88 }
89 }
90 }
91 }
92 """
93
94
95 def _run (cmd: list[ str ], stdin: str | None = None ) -> str :
96 p = subprocess.run(cmd, input = stdin, capture_output = True , text = True )
97 if p.returncode != 0 :
98 raise RuntimeError ( f "Command failed: { ' ' .join(cmd) }\n{ p.stderr } " )
99 return p.stdout
100
101
102 def _run_json (cmd: list[ str ], stdin: str | None = None ) -> dict[ str , Any]:
103 out = _run(cmd, stdin = stdin)
104 try :
105 return json.loads(out)
106 except json.JSONDecodeError as e:
107 raise RuntimeError ( f "Failed to parse JSON from command output: { e }\n Raw: \n{ out } " ) from e
108
109
110 def _ensure_gh_authenticated () -> None :
111 try :
112 _run([ "gh" , "auth" , "status" ])
113 except RuntimeError :
114 print ( "run `gh auth login` to authenticate the GitHub CLI" , file = sys.stderr)
115 raise RuntimeError ( "gh auth status failed; run `gh auth login` to authenticate the GitHub CLI" ) from None
116
117
118 def gh_pr_view_json (fields: str ) -> dict[ str , Any]:
119 # fields is a comma-separated list like: "number,headRepositoryOwner,headRepository"
120 return _run_json([ "gh" , "pr" , "view" , "--json" , fields])
121
122
123 def get_current_pr_ref () -> tuple[ str , str , int ]:
124 """
125 Resolve the PR for the current branch (whatever gh considers associated).
126 Works for cross-repo PRs too, by reading head repository owner/name.
127 """
128 pr = gh_pr_view_json( "number,headRepositoryOwner,headRepository" )
129 owner = pr[ "headRepositoryOwner" ][ "login" ]
130 repo = pr[ "headRepository" ][ "name" ]
131 number = int (pr[ "number" ])
132 return owner, repo, number
133
134
135 def gh_api_graphql (
136 owner: str ,
137 repo: str ,
138 number: int ,
139 comments_cursor: str | None = None ,
140 reviews_cursor: str | None = None ,
141 threads_cursor: str | None = None ,
142 ) -> dict[ str , Any]:
143 """
144 Call `gh api graphql` using -F variables, avoiding JSON blobs with nulls.
145 Query is passed via stdin using query=@- to avoid shell newline/quoting issues.
146 """
147 cmd = [
148 "gh" ,
149 "api" ,
150 "graphql" ,
151 "-F" ,
152 "query=@-" ,
153 "-F" ,
154 f "owner= { owner } " ,
155 "-F" ,
156 f "repo= { repo } " ,
157 "-F" ,
158 f "number= { number } " ,
159 ]
160 if comments_cursor:
161 cmd += [ "-F" , f "commentsCursor= { comments_cursor } " ]
162 if reviews_cursor:
163 cmd += [ "-F" , f "reviewsCursor= { reviews_cursor } " ]
164 if threads_cursor:
165 cmd += [ "-F" , f "threadsCursor= { threads_cursor } " ]
166
167 return _run_json(cmd, stdin = QUERY )
168
169
170 def fetch_all (owner: str , repo: str , number: int ) -> dict[ str , Any]:
171 conversation_comments: list[dict[ str , Any]] = []
172 reviews: list[dict[ str , Any]] = []
173 review_threads: list[dict[ str , Any]] = []
174
175 comments_cursor: str | None = None
176 reviews_cursor: str | None = None
177 threads_cursor: str | None = None
178
179 pr_meta: dict[ str , Any] | None = None
180
181 while True :
182 payload = gh_api_graphql(
183 owner = owner,
184 repo = repo,
185 number = number,
186 comments_cursor = comments_cursor,
187 reviews_cursor = reviews_cursor,
188 threads_cursor = threads_cursor,
189 )
190
191 if "errors" in payload and payload[ "errors" ]:
192 raise RuntimeError ( f "GitHub GraphQL errors: \n{ json.dumps(payload[ 'errors' ], indent = 2 ) } " )
193
194 pr = payload[ "data" ][ "repository" ][ "pullRequest" ]
195 if pr_meta is None :
196 pr_meta = {
197 "number" : pr[ "number" ],
198 "url" : pr[ "url" ],
199 "title" : pr[ "title" ],
200 "state" : pr[ "state" ],
201 "owner" : owner,
202 "repo" : repo,
203 }
204
205 c = pr[ "comments" ]
206 r = pr[ "reviews" ]
207 t = pr[ "reviewThreads" ]
208
209 conversation_comments.extend(c.get( "nodes" ) or [])
210 reviews.extend(r.get( "nodes" ) or [])
211 review_threads.extend(t.get( "nodes" ) or [])
212
213 comments_cursor = c[ "pageInfo" ][ "endCursor" ] if c[ "pageInfo" ][ "hasNextPage" ] else None
214 reviews_cursor = r[ "pageInfo" ][ "endCursor" ] if r[ "pageInfo" ][ "hasNextPage" ] else None
215 threads_cursor = t[ "pageInfo" ][ "endCursor" ] if t[ "pageInfo" ][ "hasNextPage" ] else None
216
217 if not (comments_cursor or reviews_cursor or threads_cursor):
218 break
219
220 assert pr_meta is not None
221 return {
222 "pull_request" : pr_meta,
223 "conversation_comments" : conversation_comments,
224 "reviews" : reviews,
225 "review_threads" : review_threads,
226 }
227
228
229 def main () -> None :
230 _ensure_gh_authenticated()
231 owner, repo, number = get_current_pr_ref()
232 result = fetch_all(owner, repo, number)
233 print (json.dumps(result, indent = 2 ))
234
235
236 if __name__ == "__main__" :
237 main()
