1import os
2from datetime import date, datetime, timedelta
3from typing import Any, Optional
4
5import requests
6import typer
7from pytz import timezone
8from typer import Typer
9
10app: Typer = typer.Typer()
11
12AMERICA_NEW_YORK_TIMEZONE = "America/New_York"
13DATETIME_FORMAT: str = "%B %d, %Y %I:%M %p"
14ISSUES_PER_SECTION: int = 50
15ISSUES_TO_FETCH: int = 100
16
17REPO_OWNER = "zed-industries"
18REPO_NAME = "zed"
19GITHUB_API_BASE_URL = "https://api.github.com"
20
21EXCLUDE_LABEL = "ignore top-ranking issues"
22
23
24@app.command()
25def main(
26 github_token: Optional[str] = None,
27 issue_reference_number: Optional[int] = None,
28 query_day_interval: Optional[int] = None,
29) -> None:
30 script_start_time: datetime = datetime.now()
31 start_date: date | None = None
32
33 if query_day_interval:
34 tz = timezone(AMERICA_NEW_YORK_TIMEZONE)
35 today = datetime.now(tz).date()
36 start_date = today - timedelta(days=query_day_interval)
37
38 # GitHub Workflow will pass in the token as an argument,
39 # but we can place it in our env when running the script locally, for convenience
40 token = github_token or os.getenv("GITHUB_ACCESS_TOKEN")
41 if not token:
42 raise typer.BadParameter(
43 "GitHub token is required. Pass --github-token or set GITHUB_ACCESS_TOKEN env var."
44 )
45
46 headers = {
47 "Authorization": f"token {token}",
48 "Accept": "application/vnd.github+json",
49 }
50
51 section_to_issues = get_section_to_issues(headers, start_date)
52 issue_text: str = create_issue_text(section_to_issues)
53
54 if issue_reference_number:
55 update_reference_issue(headers, issue_reference_number, issue_text)
56 else:
57 print(issue_text)
58
59 run_duration: timedelta = datetime.now() - script_start_time
60 print(f"Ran for {run_duration}")
61
62
63def get_section_to_issues(
64 headers: dict[str, str], start_date: date | None = None
65) -> dict[str, list[dict[str, Any]]]:
66 """Fetch top-ranked issues for each section from GitHub."""
67
68 section_filters = {
69 "Bugs": "type:Bug",
70 "Crashes": "type:Crash",
71 "Features": "type:Feature",
72 "Tracking issues": "type:Tracking",
73 "Meta issues": "type:Meta",
74 "Windows": 'label:"platform:windows"',
75 }
76
77 section_to_issues: dict[str, list[dict[str, Any]]] = {}
78 for section, search_qualifier in section_filters.items():
79 query_parts = [
80 f"repo:{REPO_OWNER}/{REPO_NAME}",
81 "is:issue",
82 "is:open",
83 f'-label:"{EXCLUDE_LABEL}"',
84 search_qualifier,
85 ]
86
87 if start_date:
88 query_parts.append(f"created:>={start_date.strftime('%Y-%m-%d')}")
89
90 query = " ".join(query_parts)
91 url = f"{GITHUB_API_BASE_URL}/search/issues"
92 params = {
93 "q": query,
94 "sort": "reactions-+1",
95 "order": "desc",
96 "per_page": ISSUES_TO_FETCH, # this will work as long as it's ≤ 100
97 }
98
99 # we are only fetching one page on purpose
100 response = requests.get(url, headers=headers, params=params)
101 response.raise_for_status()
102 items = response.json()["items"]
103
104 issues: list[dict[str, Any]] = []
105 for item in items:
106 reactions = item["reactions"]
107 score = reactions["+1"] - reactions["-1"]
108 if score > 0:
109 issues.append(
110 {
111 "url": item["html_url"],
112 "score": score,
113 "created_at": item["created_at"],
114 }
115 )
116
117 if not issues:
118 continue
119
120 issues.sort(key=lambda x: (-x["score"], x["created_at"]))
121 section_to_issues[section] = issues[:ISSUES_PER_SECTION]
122
123 # Sort sections by total score (highest total first)
124 section_to_issues = dict(
125 sorted(
126 section_to_issues.items(),
127 key=lambda item: sum(issue["score"] for issue in item[1]),
128 reverse=True,
129 )
130 )
131 return section_to_issues
132
133
134def update_reference_issue(
135 headers: dict[str, str], issue_number: int, body: str
136) -> None:
137 url = f"{GITHUB_API_BASE_URL}/repos/{REPO_OWNER}/{REPO_NAME}/issues/{issue_number}"
138 response = requests.patch(url, headers=headers, json={"body": body})
139 response.raise_for_status()
140
141
142def create_issue_text(section_to_issues: dict[str, list[dict[str, Any]]]) -> str:
143 tz = timezone(AMERICA_NEW_YORK_TIMEZONE)
144 current_datetime: str = datetime.now(tz).strftime(f"{DATETIME_FORMAT} (%Z)")
145
146 lines: list[str] = [f"*Updated on {current_datetime}*"]
147
148 for section, issues in section_to_issues.items():
149 lines.append(f"\n## {section}\n")
150 for i, issue in enumerate(issues):
151 lines.append(f"{i + 1}. {issue['url']} ({issue['score']} :thumbsup:)")
152
153 lines.append("\n---\n")
154 lines.append(
155 "*For details on how this issue is generated, "
156 "[see the script](https://github.com/zed-industries/zed/blob/main/script/update_top_ranking_issues/main.py)*"
157 )
158
159 return "\n".join(lines)
160
161
162if __name__ == "__main__":
163 app()