-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathgenerate_grain_index.py
More file actions
172 lines (141 loc) · 5.04 KB
/
Copy pathgenerate_grain_index.py
File metadata and controls
172 lines (141 loc) · 5.04 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
#!/usr/bin/env python3
"""
Generate _data/grain-index.json from grain.yaml manifests across all
feastorg repos that contain one.
Discovery enumerates all public org repos and attempts to fetch grain.yaml
from each. This avoids code search indexing lag.
Usage:
python3 scripts/generate_grain_index.py
Environment:
GITHUB_TOKEN — required; a token with read access to the feastorg org.
GITHUB_TOKEN is available automatically in Actions.
Output:
_data/grain-index.json
"""
from __future__ import annotations
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
try:
import requests
except ImportError:
sys.exit("Error: 'requests' not installed. Run: pip install requests")
try:
import yaml
except ImportError:
sys.exit("Error: 'pyyaml' not installed. Run: pip install pyyaml")
ORG = "feastorg"
OUTPUT = Path("_data/grain-index.json")
API = "https://api.github.com"
# Fields to extract from each grain.yaml
EXTRACT = [
("id", lambda m: m.get("id")),
("name", lambda m: m.get("name")),
("status", lambda m: m.get("status")),
("category", lambda m: m.get("category")),
("summary", lambda m: m.get("summary")),
("hw_version", lambda m: (m.get("version") or {}).get("hardware")),
("form_factor", lambda m: (m.get("hardware") or {}).get("form_factor")),
("license_hw", lambda m: (m.get("license") or {}).get("hardware")),
("compatibility", lambda m: m.get("compatibility")),
("related_slices",lambda m: m.get("related_slices")),
("tags", lambda m: (m.get("metadata") or {}).get("tags", [])),
("updated", lambda m: (m.get("metadata") or {}).get("updated")),
]
STATUS_ORDER = ["released", "prototype", "concept", "deprecated"]
CATEGORY_ORDER = ["shield", "card", "adapter", "module"]
def gh_session() -> requests.Session:
token = os.environ.get("GITHUB_TOKEN", "")
if not token:
sys.exit("Error: GITHUB_TOKEN environment variable not set.")
s = requests.Session()
s.headers.update(
{
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
}
)
return s
def list_all_repos(session: requests.Session) -> list[str]:
"""Return names of all public repos in the org."""
repos = []
page = 1
while True:
r = session.get(
f"{API}/orgs/{ORG}/repos",
params={"type": "public", "per_page": 100, "page": page},
)
r.raise_for_status()
batch = r.json()
if not batch:
break
repos.extend(repo["name"] for repo in batch)
page += 1
return sorted(repos)
def fetch_manifest(session: requests.Session, repo: str) -> dict | None:
"""Fetch and parse grain.yaml from the repo's main branch."""
url = f"https://raw.githubusercontent.com/{ORG}/{repo}/main/grain.yaml"
r = session.get(url)
if r.status_code == 404:
print(f" SKIP {repo}: no grain.yaml on main", flush=True)
return None
r.raise_for_status()
try:
return yaml.safe_load(r.text)
except yaml.YAMLError as e:
print(f" WARN {repo}: YAML parse error — {e}", flush=True)
return None
def extract(manifest: dict, repo: str) -> dict:
entry = {"repo": repo}
for key, fn in EXTRACT:
try:
entry[key] = fn(manifest)
except Exception:
entry[key] = None
entry["url"] = f"https://feastorg.github.io/{repo}/"
return entry
def sort_key(entry: dict) -> tuple:
status_rank = (
STATUS_ORDER.index(entry["status"])
if entry["status"] in STATUS_ORDER
else 99
)
cat_rank = (
CATEGORY_ORDER.index(entry["category"])
if entry["category"] in CATEGORY_ORDER
else 99
)
return (status_rank, cat_rank, entry.get("id") or "")
def main() -> None:
session = gh_session()
print(f"Listing all public repos in {ORG}...", flush=True)
repos = list_all_repos(session)
print(f"Found {len(repos)} repos. Probing for grain.yaml...", flush=True)
grains = []
for repo in repos:
manifest = fetch_manifest(session, repo)
if manifest is None:
continue
print(f" FOUND {repo}", flush=True)
grains.append(extract(manifest, repo))
grains.sort(key=sort_key)
summary: dict[str, int] = {}
for g in grains:
key = g.get("status") or "unknown"
summary[key] = summary.get(key, 0) + 1
output = {
"generated": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
"total": len(grains),
"summary": summary,
"grains": grains,
}
OUTPUT.parent.mkdir(parents=True, exist_ok=True)
OUTPUT.write_text(json.dumps(output, indent=2, ensure_ascii=False) + "\n")
print(f"\nWrote {len(grains)} grain(s) to {OUTPUT}", flush=True)
for status, count in sorted(summary.items()):
print(f" {status}: {count}", flush=True)
if __name__ == "__main__":
main()