Setting the file. One moment.
Fetch PR Feedback · Iterate PR · getsentry/skills · Skills Docs
Back to the top of the page def extract_feedback_item
— line 270
This file
Number 16.2
Position 2 of 6
Type Python
Size 14 KB
Lines 475 scripts/ fetch_pr_feedback.py
Python · 475 lines · 14 KB
- medium: Should address (m:, standard feedback)
18 - low: Optional suggestions (l:, nit, style)
19 - bot: Informational automated comments (Codecov, Dependabot, etc.)
20 - resolved: Already resolved threads
21
22 Bot classification:
23 - Review bots (Sentry, Warden, Cursor, Bugbot, etc.) provide actionable code
24 feedback. Their comments are categorized by content into high/medium/low with
25 a ``review_bot: true`` flag — they are NOT placed in the ``bot`` bucket.
26 - Info bots (Codecov, Dependabot, Renovate, etc.) post status reports and are
27 placed in the ``bot`` bucket for silent skipping.
28 """
29 from __future__ import annotations
30
31 import argparse
32 import json
33 import re
34 import subprocess
35 import sys
36 from typing import Any
37
38
39 # Bots that provide actionable code review feedback (security issues, lint
40 # violations, bugs). Their comments are categorized by content, not skipped.
41 REVIEW_BOT_PATTERNS = [
42 r " (?i) ^ sentry" ,
43 r " (?i) ^ warden" ,
44 r " (?i) ^ cursor" ,
45 r " (?i) ^ bugbot" ,
46 r " (?i) ^ seer" ,
47 r " (?i) ^ copilot" ,
48 r " (?i) ^ codex" ,
49 r " (?i) ^ claude" ,
50 r " (?i) ^ codeql" ,
51 ]
52
53 # Bots that post informational status reports (coverage, dependency updates).
54 # These are placed in the ``bot`` bucket and skipped silently.
55 INFO_BOT_PATTERNS = [
56 r " (?i) ^ codecov" ,
57 r " (?i) ^ dependabot" ,
58 r " (?i) ^ renovate" ,
59 r " (?i) ^ github-actions" ,
60 r " (?i) ^ mergify" ,
61 r " (?i) ^ semantic-release" ,
62 r " (?i) ^ sonarcloud" ,
63 r " (?i) ^ snyk" ,
64 r " (?i) bot $ " ,
65 r " (?i) \[ bot \] $ " ,
66 ]
67
68
69 def run_gh (args: list[ str ]) -> dict[ str , Any] | list[Any] | None :
70 """Run a gh CLI command and return parsed JSON output."""
71 try :
72 result = subprocess.run(
73 [ "gh" ] + args,
74 capture_output = True ,
75 text = True ,
76 check = True ,
77 )
78 return json.loads(result.stdout) if result.stdout.strip() else None
79 except subprocess.CalledProcessError as e:
80 print ( f "Error running gh { ' ' .join(args) } : { e.stderr } " , file = sys.stderr)
81 return None
82 except json.JSONDecodeError:
83 return None
84
85
86 def get_repo_info () -> tuple[ str , str ] | None :
87 """Get owner and repo name from current directory."""
88 result = run_gh([ "repo" , "view" , "--json" , "owner,name" ])
89 if result:
90 return result.get( "owner" , {}).get( "login" ), result.get( "name" )
91 return None
92
93
94 def get_pr_info (pr_number: int | None = None ) -> dict[ str , Any] | None :
95 """Get PR info, optionally by number or for current branch."""
96 args = [ "pr" , "view" , "--json" , "number,url,headRefName,author,reviews,reviewDecision" ]
97 if pr_number:
98 args.insert( 2 , str (pr_number))
99 return run_gh(args)
100
101
102 def is_review_bot (username: str ) -> bool :
103 """Check if username matches a review bot that posts actionable feedback."""
104 return any (re.search(p, username) for p in REVIEW_BOT_PATTERNS )
105
106
107 def is_info_bot (username: str ) -> bool :
108 """Check if username matches an informational bot (skip silently)."""
109 return any (re.search(p, username) for p in INFO_BOT_PATTERNS )
110
111
112 def is_bot (username: str ) -> bool :
113 """Check if username matches any known bot pattern."""
114 return is_review_bot(username) or is_info_bot(username)
115
116
117 def get_review_comments (owner: str , repo: str , pr_number: int ) -> list[dict[ str , Any]]:
118 """Get inline code review comments via API."""
119 result = run_gh([
120 "api" ,
121 f "repos/ { owner } / { repo } /pulls/ { pr_number } /comments" ,
122 "--paginate" ,
123 ])
124 return result if isinstance (result, list ) else []
125
126
127 def get_issue_comments (owner: str , repo: str , pr_number: int ) -> list[dict[ str , Any]]:
128 """Get PR conversation comments (includes bot comments)."""
129 result = run_gh([
130 "api" ,
131 f "repos/ { owner } / { repo } /issues/ { pr_number } /comments" ,
132 "--paginate" ,
133 ])
134 return result if isinstance (result, list ) else []
135
136
137 def get_review_threads (owner: str , repo: str , pr_number: int ) -> list[dict[ str , Any]]:
138 """Get review threads with resolution status via GraphQL."""
139 query = """
140 query($owner: String!, $repo: String!, $pr: Int!) {
141 repository(owner: $owner, name: $repo) {
142 pullRequest(number: $pr) {
143 reviewThreads(first: 100) {
144 nodes {
145 id
146 isResolved
147 isOutdated
148 path
149 line
150 comments(first: 10) {
151 nodes {
152 id
153 body
154 author {
155 login
156 }
157 createdAt
158 }
159 }
160 }
161 }
162 }
163 }
164 }
165 """
166 try :
167 result = subprocess.run(
168 [
169 "gh" , "api" , "graphql" ,
170 "-f" , f "query= { query } " ,
171 "-F" , f "owner= { owner } " ,
172 "-F" , f "repo= { repo } " ,
173 "-F" , f "pr= { pr_number } " ,
174 ],
175 capture_output = True ,
176 text = True ,
177 check = True ,
178 )
179 data = json.loads(result.stdout)
180 threads = data.get( "data" , {}).get( "repository" , {}).get( "pullRequest" , {}).get( "reviewThreads" , {}).get( "nodes" , [])
181 return threads
182 except (subprocess.CalledProcessError, json.JSONDecodeError):
183 return []
184
185
186 def detect_logaf (body: str ) -> str | None :
187 """Detect LOGAF scale markers in comment body.
188
189 LOGAF scale (https://develop.sentry.dev/engineering-practices/code-review/#logaf-scale):
190 - l: / [l] / low: → low priority (optional)
191 - m: / [m] / medium: → medium priority (should address)
192 - h: / [h] / high: → high priority (must address)
193
194 Returns 'high', 'medium', 'low', or None if no marker found.
195 """
196 # Check for LOGAF markers at start of comment (with optional whitespace)
197 logaf_patterns = [
198 # h: or [h] or high: patterns
199 ( r " ^\s * (?: h: | h \s * : | high: | \[ h \] ) " , "high" ),
200 # m: or [m] or medium: patterns
201 ( r " ^\s * (?: m: | m \s * : | medium: | \[ m \] ) " , "medium" ),
202 # l: or [l] or low: patterns
203 ( r " ^\s * (?: l: | l \s * : | low: | \[ l \] ) " , "low" ),
204 ]
205
206 for pattern, level in logaf_patterns:
207 if re.search(pattern, body, re. IGNORECASE ):
208 return level
209
210 return None
211
212
213 def categorize_comment (comment: dict[ str , Any], body: str ) -> str :
214 """Categorize a comment based on content and author.
215
216 Uses LOGAF scale: high (must fix), medium (should fix), low (optional).
217 """
218 author = comment.get( "author" , {}).get( "login" , "" ) or comment.get( "user" , {}).get( "login" , "" )
219
220 # Info bots are skipped silently; review bots fall through to content
221 # categorization so their actionable feedback is not lost.
222 if is_info_bot(author) and not is_review_bot(author):
223 return "bot"
224
225 # Check for explicit LOGAF markers first
226 logaf_level = detect_logaf(body)
227 if logaf_level:
228 return logaf_level
229
230 # Look for high-priority (blocking) indicators
231 high_patterns = [
232 r " (?i) must \s + ( fix | change | update | address ) " ,
233 r " (?i) this \s + ( is \s + ) ? ( wrong | incorrect | broken | buggy ) " ,
234 r " (?i) security \s + ( issue | vulnerability | concern ) " ,
235 r " (?i) will \s + ( break | cause | fail ) " ,
236 r " (?i) critical" ,
237 r " (?i) blocker" ,
238 ]
239
240 for pattern in high_patterns:
241 if re.search(pattern, body):
242 return "high"
243
244 # Look for low-priority (suggestion) indicators
245 low_patterns = [
246 r " (?i) nit [ : \s] " ,
247 r " (?i) nitpick" ,
248 r " (?i) suggestion [ : \s] " ,
249 r " (?i) consider \s + " ,
250 r " (?i) could \s + ( also \s + ) ? " ,
251 r " (?i) might \s + ( want \s + to | be \s + better ) " ,
252 r " (?i) optional [ : \s] " ,
253 r " (?i) minor [ : \s] " ,
254 r " (?i) style [ : \s] " ,
255 r " (?i) prefer \s + " ,
256 r " (?i) what \s + do \s + you \s + think" ,
257 r " (?i) up \s + to \s + you" ,
258 r " (?i) take \s + it \s + or \s + leave" ,
259 r " (?i) fwiw" ,
260 ]
261
262 for pattern in low_patterns:
263 if re.search(pattern, body):
264 return "low"
265
266 # Default to medium for non-bot comments without clear indicators
267 return "medium"
268
269
270 def extract_feedback_item (
271 body: str ,
272 author: str ,
273 path: str | None = None ,
274 line: int | None = None ,
275 url: str | None = None ,
276 is_resolved: bool = False ,
277 is_outdated: bool = False ,
278 review_bot: bool = False ,
279 thread_id: str | None = None ,
280 ) -> dict[ str , Any]:
281 """Create a standardized feedback item."""
282 # Truncate long bodies for summary
283 summary = body[: 200 ] + "..." if len (body) > 200 else body
284 summary = summary.replace( " \n " , " " ).strip()
285
286 item = {
287 "author" : author,
288 "body" : summary,
289 "full_body" : body,
290 }
291
292 if path:
293 item[ "path" ] = path
294 if line:
295 item[ "line" ] = line
296 if url:
297 item[ "url" ] = url
298 if is_resolved:
299 item[ "resolved" ] = True
300 if is_outdated:
301 item[ "outdated" ] = True
302 if review_bot:
303 item[ "review_bot" ] = True
304 if thread_id:
305 item[ "thread_id" ] = thread_id
306
307 return item
308
309
310 def main ():
311 parser = argparse.ArgumentParser( description = "Fetch and categorize PR feedback" )
312 parser.add_argument( "--pr" , type = int , help = "PR number (defaults to current branch PR)" )
313 args = parser.parse_args()
314
315 # Get repo info
316 repo_info = get_repo_info()
317 if not repo_info:
318 print (json.dumps({ "error" : "Could not determine repository" }))
319 sys.exit( 1 )
320 owner, repo = repo_info
321
322 # Get PR info
323 pr_info = get_pr_info(args.pr)
324 if not pr_info:
325 print (json.dumps({ "error" : "No PR found for current branch" }))
326 sys.exit( 1 )
327
328 pr_number = pr_info[ "number" ]
329 pr_author = pr_info.get( "author" , {}).get( "login" , "" )
330
331 # Get review decision
332 review_decision = pr_info.get( "reviewDecision" , "" )
333
334 # Categorized feedback using LOGAF scale
335 feedback = {
336 "high" : [], # Must address before merge
337 "medium" : [], # Should address
338 "low" : [], # Optional suggestions
339 "bot" : [],
340 "resolved" : [],
341 }
342
343 # Process reviews for overall status
344 reviews = pr_info.get( "reviews" , [])
345 for review in reviews:
346 if review.get( "state" ) == "CHANGES_REQUESTED" :
347 author = review.get( "author" , {}).get( "login" , "" )
348 body = review.get( "body" , "" )
349 if body and author != pr_author:
350 item = extract_feedback_item(body, author)
351 item[ "type" ] = "changes_requested"
352 feedback[ "high" ].append(item)
353
354 # Get review threads (inline comments with resolution status)
355 threads = get_review_threads(owner, repo, pr_number)
356 seen_thread_ids = set ()
357
358 for thread in threads:
359 if not thread.get( "comments" , {}).get( "nodes" ):
360 continue
361
362 first_comment = thread[ "comments" ][ "nodes" ][ 0 ]
363 author = first_comment.get( "author" , {}).get( "login" , "" )
364 body = first_comment.get( "body" , "" )
365
366 # Skip if author is PR author (self-comments)
367 if author == pr_author:
368 continue
369
370 # Skip empty or very short comments
371 if not body or len (body.strip()) < 3 :
372 continue
373
374 is_resolved = thread.get( "isResolved" , False )
375 is_outdated = thread.get( "isOutdated" , False )
376
377 thread_id = thread.get( "id" )
378 item = extract_feedback_item(
379 body = body,
380 author = author,
381 path = thread.get( "path" ),
382 line = thread.get( "line" ),
383 is_resolved = is_resolved,
384 is_outdated = is_outdated,
385 thread_id = thread_id,
386 )
387
388 if thread_id:
389 seen_thread_ids.add(thread_id)
390
391 if is_resolved:
392 feedback[ "resolved" ].append(item)
393 elif is_review_bot(author):
394 category = categorize_comment(first_comment, body)
395 item[ "review_bot" ] = True
396 feedback[category].append(item)
397 elif is_info_bot(author):
398 feedback[ "bot" ].append(item)
399 else :
400 category = categorize_comment(first_comment, body)
401 feedback[category].append(item)
402
403 # Get issue comments (general PR conversation)
404 issue_comments = get_issue_comments(owner, repo, pr_number)
405
406 for comment in issue_comments:
407 author = comment.get( "user" , {}).get( "login" , "" )
408 body = comment.get( "body" , "" )
409
410 # Skip if author is PR author
411 if author == pr_author:
412 continue
413
414 # Skip empty comments
415 if not body or len (body.strip()) < 3 :
416 continue
417
418 item = extract_feedback_item(
419 body = body,
420 author = author,
421 url = comment.get( "html_url" ),
422 )
423
424 if is_review_bot(author):
425 category = categorize_comment(comment, body)
426 item[ "review_bot" ] = True
427 feedback[category].append(item)
428 elif is_info_bot(author):
429 feedback[ "bot" ].append(item)
430 else :
431 category = categorize_comment(comment, body)
432 feedback[category].append(item)
433
434 # Count review bot items across priority buckets
435 review_bot_count = sum (
436 1 for bucket in ( "high" , "medium" , "low" )
437 for item in feedback[bucket]
438 if item.get( "review_bot" )
439 )
440
441 # Build output
442 output = {
443 "pr" : {
444 "number" : pr_number,
445 "url" : pr_info.get( "url" , "" ),
446 "author" : pr_author,
447 "review_decision" : review_decision,
448 },
449 "summary" : {
450 "high" : len (feedback[ "high" ]),
451 "medium" : len (feedback[ "medium" ]),
452 "low" : len (feedback[ "low" ]),
453 "bot_comments" : len (feedback[ "bot" ]),
454 "resolved" : len (feedback[ "resolved" ]),
455 "review_bot_feedback" : review_bot_count,
456 "needs_attention" : len (feedback[ "high" ]) + len (feedback[ "medium" ]),
457 },
458 "feedback" : feedback,
459 }
460
461 # Add actionable summary based on LOGAF priorities
462 if feedback[ "high" ]:
463 output[ "action_required" ] = "Address high-priority feedback before merge"
464 elif feedback[ "medium" ]:
465 output[ "action_required" ] = "Address medium-priority feedback"
466 elif feedback[ "low" ]:
467 output[ "action_required" ] = "Review low-priority suggestions - ask user which to address"
468 else :
469 output[ "action_required" ] = None
470
471 print (json.dumps(output, indent = 2 ))
472
473
474 if __name__ == "__main__" :
475 main()