|
16 | 16 | import shutil |
17 | 17 | import subprocess |
18 | 18 | import sys |
| 19 | +import urllib.error |
| 20 | +import urllib.request |
19 | 21 | from datetime import datetime, timezone |
20 | 22 | from pathlib import Path |
21 | 23 |
|
22 | 24 | REPO = os.environ.get("STAR_HISTORY_REPO", "cobusgreyling/loop-engineering") |
23 | 25 | OUT_DIR = Path(os.environ.get("STAR_HISTORY_OUT_DIR", "assets/visuals")) |
24 | 26 |
|
25 | 27 |
|
26 | | -def fetch_star_timestamps() -> list[datetime]: |
27 | | - endpoint = f"repos/{REPO}/stargazers?per_page=100" |
28 | | - proc = subprocess.run( |
29 | | - [ |
30 | | - "gh", |
31 | | - "api", |
32 | | - endpoint, |
33 | | - "--paginate", |
34 | | - "-H", |
35 | | - "Accept: application/vnd.github.v3.star+json", |
36 | | - "--jq", |
37 | | - ".[].starred_at", |
38 | | - ], |
39 | | - check=True, |
40 | | - capture_output=True, |
41 | | - text=True, |
42 | | - env={**os.environ}, |
| 28 | +def _api_token() -> str: |
| 29 | + token = os.environ.get("GH_TOKEN") or os.environ.get("GITHUB_TOKEN") |
| 30 | + if token: |
| 31 | + return token |
| 32 | + if shutil.which("gh"): |
| 33 | + proc = subprocess.run( |
| 34 | + ["gh", "auth", "token"], |
| 35 | + capture_output=True, |
| 36 | + text=True, |
| 37 | + check=False, |
| 38 | + ) |
| 39 | + if proc.returncode == 0 and proc.stdout.strip(): |
| 40 | + return proc.stdout.strip() |
| 41 | + raise RuntimeError( |
| 42 | + "GH_TOKEN/GITHUB_TOKEN or gh auth required " |
| 43 | + "(CI: set STAR_HISTORY_TOKEN repo secret)" |
43 | 44 | ) |
| 45 | + |
| 46 | + |
| 47 | +def fetch_star_timestamps() -> list[datetime]: |
| 48 | + """Paginate the stargazers REST API (gh CLI auth is unreliable in Actions).""" |
| 49 | + token = _api_token() |
44 | 50 | timestamps: list[datetime] = [] |
45 | | - for line in proc.stdout.splitlines(): |
46 | | - starred_at = line.strip().strip('"') |
47 | | - if starred_at: |
48 | | - timestamps.append( |
49 | | - datetime.fromisoformat(starred_at.replace("Z", "+00:00")) |
50 | | - ) |
| 51 | + page = 1 |
| 52 | + while True: |
| 53 | + url = f"https://api.github.com/repos/{REPO}/stargazers?per_page=100&page={page}" |
| 54 | + req = urllib.request.Request( |
| 55 | + url, |
| 56 | + headers={ |
| 57 | + "Accept": "application/vnd.github.v3.star+json", |
| 58 | + "Authorization": f"Bearer {token}", |
| 59 | + "X-GitHub-Api-Version": "2022-11-28", |
| 60 | + }, |
| 61 | + ) |
| 62 | + try: |
| 63 | + with urllib.request.urlopen(req, timeout=120) as resp: |
| 64 | + payload = json.loads(resp.read().decode()) |
| 65 | + except urllib.error.HTTPError as exc: |
| 66 | + body = exc.read().decode(errors="replace") |
| 67 | + raise RuntimeError( |
| 68 | + f"GitHub stargazers API failed (HTTP {exc.code}): {body}" |
| 69 | + ) from exc |
| 70 | + if not payload: |
| 71 | + break |
| 72 | + for item in payload: |
| 73 | + starred_at = item.get("starred_at") |
| 74 | + if starred_at: |
| 75 | + timestamps.append( |
| 76 | + datetime.fromisoformat(starred_at.replace("Z", "+00:00")) |
| 77 | + ) |
| 78 | + if len(payload) < 100: |
| 79 | + break |
| 80 | + page += 1 |
51 | 81 | timestamps.sort() |
52 | 82 | return timestamps |
53 | 83 |
|
@@ -172,13 +202,10 @@ def y_pos(count: int) -> float: |
172 | 202 |
|
173 | 203 |
|
174 | 204 | def main() -> int: |
175 | | - token = os.environ.get("GH_TOKEN") or os.environ.get("GITHUB_TOKEN") |
176 | | - if not (token or shutil.which("gh")): |
177 | | - print( |
178 | | - "GH_TOKEN/GITHUB_TOKEN or gh CLI required " |
179 | | - "(CI: set STAR_HISTORY_TOKEN repo secret)", |
180 | | - file=sys.stderr, |
181 | | - ) |
| 205 | + try: |
| 206 | + _api_token() |
| 207 | + except RuntimeError as exc: |
| 208 | + print(str(exc), file=sys.stderr) |
182 | 209 | return 1 |
183 | 210 |
|
184 | 211 | print(f"Fetching stargazers for {REPO}...") |
|
0 commit comments