aaronrene / overseer-kit public
github.py python
262 lines 9.2 KB
Raw
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 1 day ago
1 """GitHub Contents + meta + optional checks adapters (§HGD.4.1)."""
2
3 from __future__ import annotations
4
5 import base64
6 from dataclasses import dataclass
7 from typing import Any
8 from urllib.parse import quote
9
10 from tools.hosted_dashboard.cache import EphemeralByteCache
11 from tools.hosted_dashboard.http_client import UpstreamClient, UpstreamError
12 from tools.hosted_dashboard.validators import (
13 DEFAULT_HANDOVER_PATH,
14 DEFAULT_ORG_ENUMERATION_CAP,
15 DEFAULT_ROADMAP_PATH,
16 MARKER_PATH,
17 )
18
19 GITHUB_API = "https://api.github.com"
20
21
22 @dataclass(frozen=True)
23 class RepoMeta:
24 """Repository metadata from ``github_meta``."""
25
26 owner: str
27 name: str
28 full_name: str
29 default_branch: str
30 private: bool
31
32
33 @dataclass(frozen=True)
34 class FileContent:
35 """Fetched file bytes with digest."""
36
37 path: str
38 text: str
39 raw: bytes
40 sha256: str
41 ref: str
42 source_id: str
43
44
45 @dataclass(frozen=True)
46 class MarkerSummary:
47 """Redacted marker parse (§HGD.5.7)."""
48
49 present: bool
50 roadmap_path: str | None
51 handover_path: str | None
52 vcs_regime: str | None
53
54
55 class GitHubAdapters:
56 """Baseline ``github_contents`` + ``github_meta`` (+ optional checks)."""
57
58 def __init__(
59 self,
60 client: UpstreamClient,
61 *,
62 cache: EphemeralByteCache | None = None,
63 checks_advisory: bool = False,
64 enumeration_cap: int = DEFAULT_ORG_ENUMERATION_CAP,
65 max_doc_bytes: int = 2_000_000,
66 ) -> None:
67 self._client = client
68 self._cache = cache or EphemeralByteCache()
69 self._checks_advisory = checks_advisory
70 self._enumeration_cap = enumeration_cap
71 self._max_doc_bytes = max_doc_bytes
72
73 @property
74 def cache(self) -> EphemeralByteCache:
75 return self._cache
76
77 def get_repo_meta(self, owner: str, repo: str) -> RepoMeta:
78 data = self._client.get_json(f"{GITHUB_API}/repos/{quote(owner)}/{quote(repo)}")
79 if not isinstance(data, dict):
80 raise UpstreamError("upstream_error", detail="invalid repo meta")
81 default_branch = data.get("default_branch")
82 if not isinstance(default_branch, str) or not default_branch.strip():
83 raise UpstreamError("not_found", detail="missing default_branch")
84 return RepoMeta(
85 owner=str(data.get("owner", {}).get("login", owner)),
86 name=str(data.get("name", repo)),
87 full_name=str(data.get("full_name", f"{owner}/{repo}")),
88 default_branch=default_branch.strip(),
89 private=bool(data.get("private", False)),
90 )
91
92 def list_org_repos(self, owner: str) -> list[RepoMeta]:
93 """Enumerate repos under ``owner`` capped at ``enumeration_cap``."""
94 out: list[RepoMeta] = []
95 page = 1
96 while len(out) < self._enumeration_cap:
97 url = (
98 f"{GITHUB_API}/orgs/{quote(owner)}/repos"
99 f"?per_page=100&page={page}&type=all"
100 )
101 try:
102 data = self._client.get_json(url)
103 except UpstreamError:
104 # Fallback to user repos listing when orgs endpoint fails.
105 url = (
106 f"{GITHUB_API}/users/{quote(owner)}/repos"
107 f"?per_page=100&page={page}&type=all"
108 )
109 data = self._client.get_json(url)
110 if not isinstance(data, list) or not data:
111 break
112 for item in data:
113 if len(out) >= self._enumeration_cap:
114 break
115 if not isinstance(item, dict):
116 continue
117 default_branch = item.get("default_branch")
118 if not isinstance(default_branch, str) or not default_branch.strip():
119 continue
120 out.append(
121 RepoMeta(
122 owner=str(item.get("owner", {}).get("login", owner)),
123 name=str(item.get("name", "")),
124 full_name=str(item.get("full_name", "")),
125 default_branch=default_branch.strip(),
126 private=bool(item.get("private", False)),
127 )
128 )
129 if len(data) < 100:
130 break
131 page += 1
132 return out
133
134 def fetch_file(self, owner: str, repo: str, path: str, *, ref: str) -> FileContent:
135 cache_key = f"gh:{owner}/{repo}:{ref}:{path}"
136 cached = self._cache.get(cache_key)
137 if cached is not None and isinstance(cached.payload, FileContent):
138 return cached.payload
139
140 url = (
141 f"{GITHUB_API}/repos/{quote(owner)}/{quote(repo)}/contents/"
142 f"{quote(path, safe='/')}?ref={quote(ref)}"
143 )
144 # Prefer raw accept to avoid base64 decode complexity when available.
145 try:
146 raw = self._client.get_bytes(url, accept="application/vnd.github.raw")
147 except UpstreamError:
148 data = self._client.get_json(url)
149 if not isinstance(data, dict):
150 raise UpstreamError("upstream_error", detail="invalid contents")
151 encoding = data.get("encoding")
152 content = data.get("content")
153 if encoding == "base64" and isinstance(content, str):
154 raw = base64.b64decode(content)
155 elif isinstance(content, str):
156 raw = content.encode("utf-8")
157 else:
158 raise UpstreamError("not_found", detail="empty contents")
159
160 if len(raw) > self._max_doc_bytes:
161 raw = raw[: self._max_doc_bytes]
162 text = raw.decode("utf-8", errors="replace")
163 entry = self._cache.put(cache_key, raw)
164 result = FileContent(
165 path=path,
166 text=text,
167 raw=raw,
168 sha256=entry.sha256,
169 ref=ref,
170 source_id="github_contents",
171 )
172 self._cache.put(cache_key, raw, payload=result)
173 return result
174
175 def fetch_marker_summary(self, owner: str, repo: str, *, ref: str) -> MarkerSummary:
176 try:
177 file = self.fetch_file(owner, repo, MARKER_PATH, ref=ref)
178 except UpstreamError as exc:
179 if exc.token == "not_found":
180 return MarkerSummary(
181 present=False,
182 roadmap_path=None,
183 handover_path=None,
184 vcs_regime=None,
185 )
186 raise
187 return parse_marker_yaml(file.text)
188
189 def doc_paths_from_marker(self, marker: MarkerSummary) -> tuple[str, str]:
190 roadmap = marker.roadmap_path or DEFAULT_ROADMAP_PATH
191 handover = marker.handover_path or DEFAULT_HANDOVER_PATH
192 return roadmap, handover
193
194 def advisory_checks(self, owner: str, repo: str, *, ref: str) -> dict[str, Any] | None:
195 if not self._checks_advisory:
196 return None
197 url = f"{GITHUB_API}/repos/{quote(owner)}/{quote(repo)}/commits/{quote(ref)}/check-runs"
198 try:
199 data = self._client.get_json(url)
200 except UpstreamError as exc:
201 return {
202 "ok": False,
203 "label": "Advisory — not kit hard gates",
204 "items": [],
205 "error": exc.token,
206 }
207 items: list[dict[str, str]] = []
208 check_runs = data.get("check_runs") if isinstance(data, dict) else None
209 if isinstance(check_runs, list):
210 for run in check_runs:
211 if not isinstance(run, dict):
212 continue
213 name = str(run.get("name") or "check")
214 conclusion = str(run.get("conclusion") or run.get("status") or "unknown")
215 items.append({"name": name, "conclusion": conclusion})
216 return {
217 "ok": True,
218 "label": "Advisory — not kit hard gates",
219 "items": items,
220 }
221
222
223 def parse_marker_yaml(text: str) -> MarkerSummary:
224 """Parse living-doc paths + regime from marker YAML; never returns raw secrets."""
225 try:
226 import yaml
227
228 data = yaml.safe_load(text)
229 except Exception:
230 return MarkerSummary(present=True, roadmap_path=None, handover_path=None, vcs_regime=None)
231
232 if not isinstance(data, dict):
233 return MarkerSummary(present=True, roadmap_path=None, handover_path=None, vcs_regime=None)
234
235 docs = data.get("docs") if isinstance(data.get("docs"), dict) else {}
236 roadmap = docs.get("roadmap") if isinstance(docs, dict) else None
237 handover = docs.get("handover") if isinstance(docs, dict) else None
238 # Paths in config are often filenames under docs/; compose defaults.
239 roadmap_path = _compose_doc_path(roadmap)
240 handover_path = _compose_doc_path(handover)
241
242 vcs = data.get("vcs") if isinstance(data.get("vcs"), dict) else {}
243 regime = vcs.get("regime") if isinstance(vcs, dict) else None
244 regime_str = regime if isinstance(regime, str) else None
245
246 return MarkerSummary(
247 present=True,
248 roadmap_path=roadmap_path,
249 handover_path=handover_path,
250 vcs_regime=regime_str,
251 )
252
253
254 def _compose_doc_path(name: Any) -> str | None:
255 if not isinstance(name, str) or not name.strip():
256 return None
257 text = name.strip().lstrip("./")
258 if text.startswith("docs/"):
259 return text
260 if "/" in text:
261 return text
262 return f"docs/{text}"
File History 1 commit
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 1 day ago