Setting the file. One moment.
Fetch Review Requests · gh Review Requests · getsentry/skills · Skills Docs
Back to the top of the page scripts/ fetch_review_requests.py
Python · 113 lines · 3 KB
17
"""
18
19 import argparse
20 import json
21 import subprocess
22 import sys
23
24
25 def gh (path: str , paginate: bool = False ) -> list | dict :
26 cmd = [ "gh" , "api" , path]
27 if paginate:
28 cmd.append( "--paginate" )
29 result = subprocess.run(cmd, capture_output = True , text = True )
30 if result.returncode != 0 or not result.stdout:
31 print ( f "Error running gh { ' ' .join(cmd) } : { result.stderr } " , file = sys.stderr)
32 return [] if paginate else {}
33 return json.loads(result.stdout)
34
35
36 def main ():
37 parser = argparse.ArgumentParser()
38 parser.add_argument( "--org" , default = "getsentry" )
39 parser.add_argument( "--teams" , required = True , help = "Comma-separated team slugs" )
40 args = parser.parse_args()
41
42 team_slugs = [t.strip() for t in args.teams.split( "," )]
43
44 # Resolve team members for all specified teams
45 members: set[ str ] = set ()
46 team_display_names: dict[ str , str ] = {}
47 for slug in team_slugs:
48 data = gh( f "orgs/ { args.org } /teams/ { slug } /members" , paginate = True )
49 for m in data:
50 members.add(m[ "login" ])
51 # Get display name
52 team_data = gh( f "orgs/ { args.org } /teams/ { slug } " )
53 team_display_names[slug] = team_data.get( "name" , slug)
54
55 # Fetch unread notifications (GitHub API default: unread only)
56 all_notifs = gh( "notifications" , paginate = True )
57 review_notifs = [
58 n for n in all_notifs
59 if n[ "reason" ] == "review_requested" and n[ "unread" ]
60 ]
61
62 prs = []
63 for n in review_notifs:
64 url = n[ "subject" ][ "url" ]
65 repo_path = url.replace( "https://api.github.com/repos/" , "" )
66 repo = repo_path.rsplit( "/pulls/" , 1 )[ 0 ]
67 pr_num = repo_path.rsplit( "/" , 1 )[ - 1 ]
68 html_url = f "https://github.com/ { repo } /pull/ { pr_num } "
69
70 pr_data = gh( f "repos/ { repo } /pulls/ { pr_num } " )
71 if not pr_data:
72 continue
73
74 # Skip merged or closed PRs
75 if pr_data.get( "merged_at" ) or pr_data.get( "state" ) == "closed" :
76 continue
77
78 author = pr_data[ "user" ][ "login" ]
79
80 reviewers_data = gh( f "repos/ { repo } /pulls/ { pr_num } /requested_reviewers" )
81 requested_team_names = [t[ "slug" ] for t in reviewers_data.get( "teams" , [])]
82 matching_teams = [
83 t for t in requested_team_names
84 if any (slug.lower() == t.lower() for slug in team_slugs)
85 ]
86
87 by_team_member = author in members
88 review_from_team = len (matching_teams) > 0
89
90 if not (by_team_member or review_from_team):
91 continue
92
93 reasons = []
94 if review_from_team:
95 reasons.append( f "review requested from: { ', ' .join(matching_teams) } " )
96 if by_team_member:
97 reasons.append( f "opened by: { author } " )
98
99 prs.append({
100 "notification_id" : n[ "id" ],
101 "title" : n[ "subject" ][ "title" ],
102 "url" : html_url,
103 "repo" : repo,
104 "pr_number" : int (pr_num),
105 "author" : author,
106 "reasons" : reasons,
107 })
108
109 print (json.dumps({ "total" : len (prs), "prs" : prs}, indent = 2 ))
110
111
112 if __name__ == "__main__" :
113 main()