flake-update-20260505
1#!/usr/bin/env python3
2"""Alfred Script Filter: list GitHub pull requests for a repository."""
3
4import json
5import os
6import platform
7import re
8import subprocess
9import sys
10
11
12def setup_path():
13 """Prepend Homebrew bin to PATH so `gh` is found in Alfred's environment."""
14 machine = platform.machine()
15 prefix = "/opt/homebrew" if machine == "arm64" else "/usr/local"
16 brew_bin = os.path.join(prefix, "bin")
17 os.environ["PATH"] = brew_bin + ":" + os.environ.get("PATH", "")
18
19
20def error_item(title, subtitle=""):
21 """Return Alfred JSON with a single error item."""
22 return json.dumps(
23 {"items": [{"title": title, "subtitle": subtitle, "valid": False}]}
24 )
25
26
27def resolve_alias(query):
28 """If ALIAS_{query} env var exists, return its value; otherwise return query as-is."""
29 alias_key = "ALIAS_" + query
30 return os.environ.get(alias_key, query)
31
32
33ANSI_COLORS = {
34 "SUCCESS": "\033[32m",
35 "FAILURE": "\033[31m",
36 "ERROR": "\033[31m",
37 "PENDING": "\033[33m",
38 "BEHIND": "\033[33m",
39 "DIRTY": "\033[31m",
40 "CONFLICTING": "\033[31m",
41}
42ANSI_RESET = "\033[0m"
43
44
45def get_ci_icon(state, color=False):
46 """Map CI statusCheckRollup state to a nerd font icon and label."""
47 if state == "SUCCESS":
48 icon, label = "\uf058", "CI: Passing"
49 elif state in ("FAILURE", "ERROR"):
50 icon, label = "\uf057", "CI: Failed"
51 elif state == "PENDING":
52 icon, label = "\uf017", "CI: InProgress"
53 else:
54 return "", ""
55 if color and state in ANSI_COLORS:
56 label = f"{ANSI_COLORS[state]}{label}{ANSI_RESET}"
57 return icon, label
58
59
60def get_merge_icon(mergeable, merge_state_status, color=False):
61 """Map mergeable/mergeStateStatus to a nerd font icon and label."""
62 if mergeable == "CONFLICTING":
63 icon, label, ckey = "\uf06a", "Conflicts", "CONFLICTING"
64 elif merge_state_status == "BEHIND":
65 icon, label, ckey = "\uf0ab", "Needs rebase", "BEHIND"
66 elif merge_state_status == "DIRTY":
67 icon, label, ckey = "\uf06a", "Merge conflicts", "DIRTY"
68 elif mergeable == "MERGEABLE" and merge_state_status == "CLEAN":
69 icon, label, ckey = "\uf00c", "Ready to merge", "SUCCESS"
70 elif merge_state_status == "BLOCKED":
71 icon, label, ckey = "\uf023", "Pending", "PENDING"
72 elif merge_state_status == "UNSTABLE":
73 icon, label, ckey = "\uf071", "Unstable", "PENDING"
74 elif mergeable == "UNKNOWN":
75 return "", ""
76 else:
77 return "", ""
78 if color and ckey in ANSI_COLORS:
79 label = f"{ANSI_COLORS[ckey]}{label}{ANSI_RESET}"
80 return icon, label
81
82
83def fetch_my_prs(use_color, filter_term=""):
84 """Fetch open PRs authored by the current user across all repos."""
85 api_host = os.environ.get("API_HOST", "github.com")
86 cache_pulls = os.environ.get("CACHE_PULLS", "10m")
87
88 graphql_query = """
89query($searchQuery: String!, $first: Int!) {
90 search(query: $searchQuery, type: ISSUE, first: $first) {
91 nodes {
92 ... on PullRequest {
93 number
94 title
95 url
96 state
97 mergedAt
98 mergeable
99 mergeStateStatus
100 repository { nameWithOwner }
101 author { login }
102 commits(last: 1) {
103 nodes {
104 commit {
105 statusCheckRollup {
106 state
107 }
108 }
109 }
110 }
111 }
112 }
113 }
114}
115"""
116
117 cmd = [
118 "gh",
119 "api",
120 "graphql",
121 "-f",
122 f"query={graphql_query}",
123 "-f",
124 "searchQuery=is:pr is:open author:@me",
125 "-F",
126 "first=50",
127 "--hostname",
128 api_host,
129 "--cache",
130 cache_pulls,
131 ]
132
133 try:
134 result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
135 except subprocess.TimeoutExpired:
136 print(error_item("Request timed out", "gh api call timed out"))
137 return
138 except FileNotFoundError:
139 print(error_item("GitHub CLI not found", "Install with: brew install gh"))
140 return
141
142 if result.returncode != 0:
143 err_msg = result.stderr.strip() or "Unknown error"
144 print(error_item("GitHub API error", err_msg))
145 return
146
147 try:
148 response = json.loads(result.stdout)
149 except json.JSONDecodeError as exc:
150 print(error_item("Failed to parse response", str(exc)))
151 return
152
153 if "errors" in response:
154 msg = response["errors"][0].get("message", "Unknown GraphQL error")
155 print(error_item("GitHub API error", msg))
156 return
157
158 pulls = response["data"]["search"]["nodes"]
159
160 host = api_host
161 items = [
162 {
163 "title": "Open your pull requests dashboard",
164 "subtitle": f"Open https://{host}/pulls",
165 "arg": f"https://{host}/pulls",
166 }
167 ]
168
169 for pr in pulls:
170 if not pr:
171 continue
172 url = pr.get("url", "")
173 number = pr.get("number", "")
174 title = pr.get("title", "")
175 repo_name = pr.get("repository", {}).get("nameWithOwner", "")
176 author = (pr.get("author") or {}).get("login", "unknown")
177
178 if pr.get("mergedAt"):
179 state = "Merged"
180 elif pr.get("state") == "CLOSED":
181 state = "Closed"
182 else:
183 state = "Open"
184
185 ci_state = None
186 commits = pr.get("commits", {}).get("nodes", [])
187 if commits:
188 rollup = commits[0].get("commit", {}).get("statusCheckRollup")
189 if rollup:
190 ci_state = rollup.get("state")
191
192 ci_icon, ci_label = get_ci_icon(ci_state, color=use_color)
193 merge_icon, merge_label = get_merge_icon(
194 pr.get("mergeable"), pr.get("mergeStateStatus"), color=use_color
195 )
196
197 parts = []
198 if ci_icon:
199 parts.append(f"{ci_icon} {ci_label}")
200 if merge_icon:
201 parts.append(f"{merge_icon} {merge_label}")
202 parts.append(state)
203 subtitle = " - ".join(parts) + f" \u2014 {url}"
204
205 filter_str = f"{number} {title} {repo_name} {author} {state} {ci_label} {merge_label}".lower()
206 if filter_term and filter_term not in filter_str:
207 continue
208
209 items.append(
210 {
211 "title": f"PR #{number}: {title} - {repo_name}",
212 "subtitle": subtitle,
213 "arg": url,
214 "mods": {
215 "cmd": {
216 "subtitle": "Copy PR URL to clipboard",
217 "arg": url,
218 }
219 },
220 "text": {"copy": url},
221 }
222 )
223
224 print(json.dumps({"items": items}))
225
226
227def main():
228 setup_path()
229
230 args = sys.argv[1:]
231 use_color = "--color" in args
232 if use_color:
233 args.remove("--color")
234
235 if not args or not args[0].strip():
236 print(
237 error_item(
238 "No repository specified", "Usage: pr [myprs|owner/repo] [filter]"
239 )
240 )
241 return
242
243 query = args[0].strip()
244 filter_term = args[1].strip().lower() if len(args) > 1 else ""
245
246 if query.lower() == "myprs":
247 fetch_my_prs(use_color, filter_term)
248 return
249 repo = resolve_alias(query)
250
251 repo_pattern = re.compile(r"^[a-zA-Z0-9._-]+/[a-zA-Z0-9._-]+$")
252 if not repo_pattern.match(repo):
253 subtitle = repo if repo != query else ""
254 if subtitle:
255 subtitle = f'Alias "{query}" resolved to invalid repo: {repo}'
256 else:
257 subtitle = "Expected format: owner/repo"
258 print(error_item("Invalid repository format", subtitle))
259 return
260
261 if not _which("gh"):
262 print(
263 error_item(
264 "GitHub CLI not found",
265 "Install with: brew install gh",
266 )
267 )
268 return
269
270 api_host = os.environ.get("API_HOST", "github.com")
271 cache_pulls = os.environ.get("CACHE_PULLS", "10m")
272
273 owner, name = repo.split("/")
274 graphql_query = """
275query($owner: String!, $name: String!, $first: Int!) {
276 repository(owner: $owner, name: $name) {
277 pullRequests(states: OPEN, first: $first, orderBy: {field: CREATED_AT, direction: DESC}) {
278 nodes {
279 number
280 title
281 url
282 state
283 mergedAt
284 mergeable
285 mergeStateStatus
286 author { login }
287 commits(last: 1) {
288 nodes {
289 commit {
290 statusCheckRollup {
291 state
292 }
293 }
294 }
295 }
296 }
297 }
298 }
299}
300"""
301
302 cmd = [
303 "gh",
304 "api",
305 "graphql",
306 "-f",
307 f"query={graphql_query}",
308 "-f",
309 f"owner={owner}",
310 "-f",
311 f"name={name}",
312 "-F",
313 "first=50",
314 "--hostname",
315 api_host,
316 "--cache",
317 cache_pulls,
318 ]
319
320 try:
321 result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
322 except subprocess.TimeoutExpired:
323 print(error_item("Request timed out", f"gh api call for {repo} timed out"))
324 return
325 except FileNotFoundError:
326 print(error_item("GitHub CLI not found", "Install with: brew install gh"))
327 return
328
329 if result.returncode != 0:
330 err_msg = result.stderr.strip() or "Unknown error"
331 print(error_item("GitHub API error", err_msg))
332 return
333
334 try:
335 response = json.loads(result.stdout)
336 except json.JSONDecodeError as exc:
337 print(error_item("Failed to parse response", str(exc)))
338 return
339
340 if "errors" in response:
341 msg = response["errors"][0].get("message", "Unknown GraphQL error")
342 print(error_item("GitHub API error", msg))
343 return
344
345 pulls = response["data"]["repository"]["pullRequests"]["nodes"]
346
347 host = api_host
348 repo_url = f"https://{host}/{repo}"
349
350 items = [
351 {
352 "title": "Open all pull requests page",
353 "subtitle": f"Open {repo_url}/pulls",
354 "arg": f"{repo_url}/pulls",
355 }
356 ]
357
358 for pr in pulls:
359 url = pr.get("url", "")
360 number = pr.get("number", "")
361 title = pr.get("title", "")
362 author = (pr.get("author") or {}).get("login", "unknown")
363
364 if pr.get("mergedAt"):
365 state = "Merged"
366 elif pr.get("state") == "CLOSED":
367 state = "Closed"
368 else:
369 state = "Open"
370
371 ci_state = None
372 commits = pr.get("commits", {}).get("nodes", [])
373 if commits:
374 rollup = commits[0].get("commit", {}).get("statusCheckRollup")
375 if rollup:
376 ci_state = rollup.get("state")
377
378 ci_icon, ci_label = get_ci_icon(ci_state, color=use_color)
379 merge_icon, merge_label = get_merge_icon(
380 pr.get("mergeable"), pr.get("mergeStateStatus"), color=use_color
381 )
382
383 parts = []
384 if ci_icon:
385 parts.append(f"{ci_icon} {ci_label}")
386 if merge_icon:
387 parts.append(f"{merge_icon} {merge_label}")
388 parts.append(state)
389 subtitle = " - ".join(parts) + f" \u2014 {url}"
390
391 filter_str = f"{number} {title} {author} {state} {ci_label} {merge_label}".lower()
392 if filter_term and filter_term not in filter_str:
393 continue
394
395 items.append(
396 {
397 "title": f"PR #{number}: {title} - {author}",
398 "subtitle": subtitle,
399 "arg": url,
400 "mods": {
401 "cmd": {
402 "subtitle": "Copy PR URL to clipboard",
403 "arg": url,
404 }
405 },
406 "text": {"copy": url},
407 }
408 )
409
410 print(json.dumps({"items": items}))
411
412
413def _which(name):
414 """Check if a command exists on PATH."""
415 for directory in os.environ.get("PATH", "").split(os.pathsep):
416 if os.path.isfile(os.path.join(directory, name)):
417 return True
418 return False
419
420
421if __name__ == "__main__":
422 main()