open-design/.github/scripts/lib/github.py

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")