#!/usr/bin/env python3 """Fetch Investibook's pinned opencode binary from the fork build artifact.""" from __future__ import annotations import json import os import shutil import stat import sys import tempfile import urllib.error import urllib.parse import urllib.request import zipfile from pathlib import Path ROOT = Path(__file__).resolve().parents[1] DEFAULT_VENDOR_DIR = ROOT / "backend" / "vendor" / "opencode" def _env(name: str, default: str = "") -> str: return os.environ.get(name, default).strip() def _token() -> str: token = _env("GITHUB_TOKEN") or _env("GH_TOKEN") or _env("GIT_ACTION_TOKEN") if not token: raise SystemExit( "Missing GitHub token. Set GITHUB_TOKEN, GH_TOKEN, or GIT_ACTION_TOKEN " "so the private tidbcloud/opencode artifact can be downloaded." ) return token def _request_json(url: str, token: str) -> dict: req = urllib.request.Request( url, headers={ "Accept": "application/vnd.github+json", "Authorization": f"Bearer {token}", "X-GitHub-Api-Version": "2022-11-28", }, ) with urllib.request.urlopen(req, timeout=60) as resp: return json.loads(resp.read().decode("utf-8")) def _download(url: str, token: str, output: Path) -> None: redirect_opener = urllib.request.build_opener(_NoRedirectHandler) req = urllib.request.Request( url, headers={ "Accept": "application/vnd.github+json", "Authorization": f"Bearer {token}", "X-GitHub-Api-Version": "2022-11-28", }, ) try: resp = redirect_opener.open(req, timeout=60) resp.close() download_url = url except urllib.error.HTTPError as exc: if exc.code not in {301, 302, 303, 307, 308}: raise download_url = exc.headers["Location"] # GitHub artifact download redirects to a signed storage URL. Do not forward # the GitHub Authorization header to that second hop or the storage backend # rejects the otherwise-valid signed URL. with urllib.request.urlopen(download_url, timeout=120) as resp, output.open("wb") as fh: shutil.copyfileobj(resp, fh) def _successful_run_ids(api_base: str, repo: str, workflow: str, ref: str, token: str) -> list[int]: encoded_workflow = urllib.parse.quote(workflow, safe="") query = urllib.parse.urlencode({"branch": ref, "status": "success", "per_page": "100"}) data = _request_json( f"{api_base}/repos/{repo}/actions/workflows/{encoded_workflow}/runs?{query}", token, ) runs = data.get("workflow_runs") or [] if not runs: raise SystemExit( f"No successful {repo} {workflow} run found for branch/ref {ref!r}. " "Trigger build-branch-opencode first, then rebuild Investibook." ) return [int(run["id"]) for run in runs] class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None def _select_artifact( api_base: str, repo: str, run_id: int, token: str, name: str, artifact_id: str, ) -> tuple[dict | None, str]: data = _request_json( f"{api_base}/repos/{repo}/actions/runs/{run_id}/artifacts?per_page=100", token, ) artifacts = data.get("artifacts") or [] if artifact_id: matches = [artifact for artifact in artifacts if str(artifact.get("id")) == artifact_id] elif name: matches = [artifact for artifact in artifacts if artifact.get("name") == name] else: matches = [ artifact for artifact in artifacts if str(artifact.get("name", "")).startswith("opencode-linux-x64-") ] if not matches: available = ", ".join(str(artifact.get("name")) for artifact in artifacts) or "(none)" want = artifact_id or name or "opencode-linux-x64-*" return None, f"missing {want!r}; available: {available}" live_matches = [artifact for artifact in matches if not artifact.get("expired", False)] if live_matches: return live_matches[0], "" expired = ", ".join( f"id={artifact.get('id')} name={artifact.get('name')!r} expires_at={artifact.get('expires_at')}" for artifact in matches ) return None, f"all matching artifacts are expired: {expired}" def _find_binary(extract_dir: Path) -> Path: candidates = [ path for path in extract_dir.rglob("opencode") if path.is_file() and path.parent.name == "bin" ] if not candidates: candidates = [path for path in extract_dir.rglob("opencode") if path.is_file()] if not candidates: raise SystemExit("Downloaded opencode artifact did not contain an opencode binary.") return candidates[0] def main() -> None: repo = _env("OPENCODE_ARTIFACT_REPO", "tidbcloud/opencode") workflow = _env("OPENCODE_ARTIFACT_WORKFLOW", "build-branch-opencode.yml") ref = _env("OPENCODE_ARTIFACT_REF", "dev") run_id_text = _env("OPENCODE_ARTIFACT_RUN_ID") artifact_name = _env("OPENCODE_ARTIFACT_NAME") artifact_id = _env("OPENCODE_ARTIFACT_ID") api_base = _env("GITHUB_API_URL", "https://api.github.com").rstrip("/") vendor_dir = Path(_env("OPENCODE_VENDOR_DIR", str(DEFAULT_VENDOR_DIR))).resolve() token = _token() if run_id_text: run_id = int(run_id_text) artifact, reason = _select_artifact( api_base, repo, run_id, token, artifact_name, artifact_id, ) if artifact is None: raise SystemExit(f"Run {run_id} has no downloadable OpenCode artifact: {reason}.") else: skipped: list[str] = [] artifact = None run_id = 0 for candidate_run_id in _successful_run_ids(api_base, repo, workflow, ref, token): candidate, reason = _select_artifact( api_base, repo, candidate_run_id, token, artifact_name, artifact_id, ) if candidate is not None: run_id, artifact = candidate_run_id, candidate break skipped.append(f"run {candidate_run_id}: {reason}") if artifact is None: detail = "; ".join(skipped) or "no successful runs returned" raise SystemExit( f"No downloadable OpenCode artifact found in successful {repo} {workflow} " f"runs for ref {ref!r}: {detail}." ) with tempfile.TemporaryDirectory(prefix="opencode-artifact-") as tmp: tmp_dir = Path(tmp) archive = tmp_dir / "artifact.zip" _download(str(artifact["archive_download_url"]), token, archive) with zipfile.ZipFile(archive) as zf: zf.extractall(tmp_dir / "artifact") binary = _find_binary(tmp_dir / "artifact") bin_dir = vendor_dir / "bin" bin_dir.mkdir(parents=True, exist_ok=True) target = bin_dir / "opencode" shutil.copy2(binary, target) target.chmod(target.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) source = { "repo": repo, "workflow": workflow, "ref": ref, "run_id": run_id, "artifact_id": artifact.get("id"), "artifact_name": artifact.get("name"), } (vendor_dir / "SOURCE.json").write_text(json.dumps(source, indent=2, sort_keys=True) + "\n") print(f"Fetched {artifact.get('name')} from {repo} run {run_id} into {target}") if __name__ == "__main__": try: main() except urllib.error.HTTPError as exc: sys.exit(f"GitHub API request failed: HTTP {exc.code} {exc.reason}")