ljwx-deploy/scripts/knowledge/github_artifacts.py

#!/usr/bin/env python3
"""GitHub Actions artifact helpers for the knowledge supply chain."""

from __future__ import annotations

import json
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.request import HTTPRedirectHandler, Request, build_opener, urlopen

from .common import JSONMapping, ensure_parent, require_mapping, require_non_empty_str


REQUEST_TIMEOUT_SECONDS = 20
REDIRECT_STATUS_CODES = {301, 302, 303, 307, 308}


class GitHubArtifactError(Exception):
    """Raised when a GitHub artifact request fails."""


class _NoRedirectHandler(HTTPRedirectHandler):
    """Prevent urllib from forwarding GitHub auth headers to signed blob URLs."""

    def redirect_request(
        self,
        req: Request,
        fp: object,
        code: int,
        msg: str,
        headers: object,
        newurl: str,
    ) -> None:
        return None


def _request_json(url: str, token: str) -> JSONMapping:
    request = Request(
        url=url,
        headers={
            "Authorization": f"Bearer {token}",
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": "2022-11-28",
        },
        method="GET",
    )
    try:
        with urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
            body = response.read().decode("utf-8")
    except HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        raise GitHubArtifactError(
            f"GitHub API 请求失败: HTTP {exc.code}: {detail}"
        ) from exc
    except URLError as exc:
        raise GitHubArtifactError(f"GitHub API 请求失败: {exc.reason}") from exc

    try:
        payload = json.loads(body)
    except json.JSONDecodeError as exc:
        raise GitHubArtifactError(f"GitHub API 返回无效 JSON: {body}") from exc
    if not isinstance(payload, dict):
        raise GitHubArtifactError("GitHub API 返回根节点必须是 object")
    return {str(key): value for key, value in payload.items()}


def _download_signed_artifact(url: str) -> bytes:
    request = Request(
        url=url,
        headers={
            "Accept": "application/zip",
        },
        method="GET",
    )
    try:
        with urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
            return response.read()
    except HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        raise GitHubArtifactError(
            f"下载 artifact 失败: HTTP {exc.code}: {detail}"
        ) from exc
    except URLError as exc:
        raise GitHubArtifactError(f"下载 artifact 失败: {exc.reason}") from exc


def download_artifact_zip(
    *,
    repository: str,
    run_id: int,
    artifact_name: str,
    token: str,
    output_path: Path,
) -> Path:
    """Download one named artifact from a workflow run."""

    artifacts_url = (
        f"https://api.github.com/repos/{repository}/actions/runs/{run_id}/artifacts"
    )
    listing = _request_json(artifacts_url, token)
    raw_items = listing.get("artifacts", [])
    if not isinstance(raw_items, list):
        raise GitHubArtifactError("/artifacts 必须是列表")

    artifact_id: int | None = None
    for index, raw_item in enumerate(raw_items):
        item = require_mapping(raw_item, f"artifacts[{index}]")
        candidate_name = require_non_empty_str(
            item.get("name"), f"artifacts[{index}].name"
        )
        if candidate_name != artifact_name:
            continue
        raw_id = item.get("id")
        if not isinstance(raw_id, int) or raw_id <= 0:
            raise GitHubArtifactError(f"artifacts[{index}].id 必须是正整数")
        artifact_id = raw_id
        break

    if artifact_id is None:
        raise GitHubArtifactError(
            f"未找到 artifact: repo={repository} run_id={run_id} name={artifact_name}"
        )

    download_url = (
        f"https://api.github.com/repos/{repository}/actions/artifacts/{artifact_id}/zip"
    )
    request = Request(
        url=download_url,
        headers={
            "Authorization": f"Bearer {token}",
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": "2022-11-28",
        },
        method="GET",
    )
    opener = build_opener(_NoRedirectHandler())
    try:
        with opener.open(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
            payload = response.read()
    except HTTPError as exc:
        if exc.code in REDIRECT_STATUS_CODES:
            redirect_url = exc.headers.get("Location")
            if not redirect_url:
                raise GitHubArtifactError("下载 artifact 失败: 缺少重定向地址") from exc
            payload = _download_signed_artifact(redirect_url)
        else:
            detail = exc.read().decode("utf-8", errors="replace")
            raise GitHubArtifactError(
                f"下载 artifact 失败: HTTP {exc.code}: {detail}"
            ) from exc
    except URLError as exc:
        raise GitHubArtifactError(f"下载 artifact 失败: {exc.reason}") from exc

    if not payload:
        raise GitHubArtifactError("下载 artifact 失败: 返回内容为空")

    ensure_parent(output_path)
    output_path.write_bytes(payload)
    return output_path