tenant.openagents/omega

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