from __future__ import annotations import json import os import tempfile import urllib.error import urllib.parse import urllib.request from pathlib import Path from typing import Any API_ROOT = "https://api.github.com" API_VERSION = "2022-11-28" USER_AGENT = "open-design-actions/1" class GitHubError(RuntimeError): pass def require_env(name: str) -> str: value = os.environ.get(name, "") if not value: raise GitHubError(f"{name} is required") return value def event_payload(path: Path | None = None) -> dict[str, Any]: event_path = path or Path(require_env("GITHUB_EVENT_PATH")) try: value = json.loads(event_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: raise GitHubError(f"cannot load GitHub event payload: {error}") from error if not isinstance(value, dict): raise GitHubError("GitHub event payload must be an object") return value def api_request(path: str, *, token: str | None = None) -> urllib.request.Request: if not path.startswith("/"): path = f"/{path}" headers = { "Accept": "application/vnd.github+json", "X-GitHub-Api-Version": API_VERSION, "User-Agent": USER_AGENT, } resolved_token = token or os.environ.get("GH_TOKEN") or os.environ.get("GITHUB_TOKEN") if resolved_token: headers["Authorization"] = f"Bearer {resolved_token}" return urllib.request.Request(f"{API_ROOT}{path}", headers=headers) def api_json(path: str, *, token: str | None = None) -> Any: try: with urllib.request.urlopen(api_request(path, token=token), timeout=30) as response: return json.load(response) except (OSError, json.JSONDecodeError, urllib.error.URLError) as error: raise GitHubError(f"GitHub API request failed for {path}: {error}") from error def run_artifacts(repository: str, run_id: int) -> list[dict[str, Any]]: quoted_repository = "/".join(urllib.parse.quote(part, safe="") for part in repository.split("/")) artifacts: list[dict[str, Any]] = [] page = 1 while True: value = api_json(f"/repos/{quoted_repository}/actions/runs/{run_id}/artifacts?per_page=100&page={page}") if not isinstance(value, dict) or not isinstance(value.get("artifacts"), list): raise GitHubError("GitHub artifacts response has an invalid shape") batch = value["artifacts"] for artifact in batch: if not isinstance(artifact, dict): raise GitHubError("GitHub artifact entry has an invalid shape") artifacts.append(artifact) if len(batch) < 100: return artifacts page += 1 def run_jobs(repository: str, run_id: int, attempt: int) -> list[dict[str, Any]]: """Read exactly one attempt; never borrow a previous attempt's success.""" if type(run_id) is not int or run_id <= 0 or type(attempt) is not int or attempt <= 0: raise GitHubError("job lookup requires a positive run id and attempt") quoted_repository = "/".join(urllib.parse.quote(part, safe="") for part in repository.split("/")) jobs: list[dict[str, Any]] = [] page = 1 while True: value = api_json(f"/repos/{quoted_repository}/actions/runs/{run_id}/attempts/{attempt}/jobs?per_page=100&page={page}") if not isinstance(value, dict) or not isinstance(value.get("jobs"), list): raise GitHubError("GitHub jobs response has an invalid shape") batch = value["jobs"] if any(not isinstance(job, dict) for job in batch): raise GitHubError("GitHub job entry has an invalid shape") jobs.extend(batch) if len(batch) < 100: return jobs page += 1 def latest_run_artifact(repository: str, run_id: int, name: str) -> dict[str, Any] | None: """Select the newest surviving artifact across workflow run attempts. Failed-job reruns retain artifacts from earlier attempts and may upload the same name again. Consumers still validate the downloaded handoff's attempt, so an older artifact can never become accepted evidence for a newer attempt. """ matches = [ artifact for artifact in run_artifacts(repository, run_id) if artifact.get("name") == name and artifact.get("expired") is False ] if not matches: return None for artifact in matches: artifact_id = artifact.get("id") if not isinstance(artifact_id, int) or artifact_id <= 0: raise GitHubError(f"artifact {name!r} has an invalid id") if not isinstance(artifact.get("created_at"), str) or not artifact["created_at"]: raise GitHubError(f"artifact {name!r} has an invalid creation time") return max(matches, key=lambda artifact: (artifact["created_at"], artifact["id"])) def download_artifact(repository: str, artifact_id: int, destination: Path) -> None: quoted_repository = "/".join(urllib.parse.quote(part, safe="") for part in repository.split("/")) destination.parent.mkdir(parents=True, exist_ok=True) handle, temporary = tempfile.mkstemp(prefix=f".{destination.name}.", dir=str(destination.parent)) try: with os.fdopen(handle, "wb") as output: try: request = api_request(f"/repos/{quoted_repository}/actions/artifacts/{artifact_id}/zip") opener = urllib.request.build_opener(_NoRedirect()) try: response = opener.open(request, timeout=30) except urllib.error.HTTPError as redirect: if redirect.code not in {301, 302, 303, 307, 308}: raise location = redirect.headers.get("Location") if not location: raise GitHubError(f"artifact {artifact_id} download redirect omitted Location") response = urllib.request.urlopen( urllib.request.Request(location, headers={"User-Agent": USER_AGENT}), timeout=60, ) with response: while chunk := response.read(1024 * 1024): output.write(chunk) except (OSError, urllib.error.URLError) as error: raise GitHubError(f"cannot download GitHub artifact {artifact_id}: {error}") from error os.replace(temporary, destination) except BaseException: try: os.unlink(temporary) except OSError: pass raise class _NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, request, file_pointer, code, message, headers, new_url): return None def append_outputs(values: dict[str, str]): output_path = os.environ.get("GITHUB_OUTPUT") lines = [f"{key}={value}" for key, value in values.items()] if output_path: with Path(output_path).open("a", encoding="utf-8") as output: output.write("\n".join(lines) + "\n") else: print("\n".join(lines)) def append_summary(markdown: str): summary_path = os.environ.get("GITHUB_STEP_SUMMARY") if summary_path: with Path(summary_path).open("a", encoding="utf-8") as summary: summary.write(markdown.rstrip() + "\n")