msys2-autobuild/msys2_autobuild/gh.py

import io
import os
import shutil
import sys
import tempfile
import time
import hashlib
from contextlib import contextmanager
from datetime import datetime, UTC
from functools import cache
from pathlib import Path
from typing import Any
from collections.abc import Generator, Callable

import requests
from gitea import Configuration, ApiClient, RepositoryApi, CreateReleaseOption
from gitea import Repository, Release, Attachment
from gitea.rest import ApiException

from .config import REQUESTS_TIMEOUT, BuildType, Config
from .utils import PathLike, get_requests_session


@cache
def _get_repo(name: str) -> Repository:
    gitea = get_gitea()
    split = name.split("/")
    return gitea.repo_get(split[0], split[1])


def get_current_repo() -> Repository:
    repo_full_name = os.environ.get("GITHUB_REPOSITORY", "Befator-Inc-Firmen-Netzwerk/msys2-autobuild")
    return _get_repo(repo_full_name)


def get_repo_for_build_type(build_type: BuildType) -> Repository:
    return _get_repo(Config.RUNNER_CONFIG[build_type]["repo"])


@cache
def get_gitea() -> RepositoryApi:
    configuration = Configuration()
    configuration.host = "https://git.befatorinc.de/api/v1"
    configuration.api_key["Authorization"] = "token 91f6f2e72e6d64fbd0b34133efae4a6c838d0e58"
    gitea = RepositoryApi(ApiClient(configuration))
    return gitea


def download_text_asset(asset: Attachment, cache=False) -> str:
    session = get_requests_session(nocache=not cache)
    with session.get(asset.browser_download_url, timeout=REQUESTS_TIMEOUT) as r:
        r.raise_for_status()
        return r.text


def get_asset_mtime_ns(asset: Attachment) -> int:
    """Returns the mtime of an asset in nanoseconds"""

    return int(asset.created_at.timestamp() * (1000 ** 3))


def download_asset(asset: Attachment, target_path: str,
                   onverify: Callable[[str, str], None] | None = None) -> None:
    session = get_requests_session(nocache=True)
    with session.get(asset.browser_download_url, stream=True, timeout=REQUESTS_TIMEOUT) as r:
        r.raise_for_status()
        fd, temppath = tempfile.mkstemp()
        try:
            os.chmod(temppath, 0o644)
            with os.fdopen(fd, "wb") as h:
                for chunk in r.iter_content(256 * 1024):
                    h.write(chunk)
            mtime_ns = get_asset_mtime_ns(asset)
            os.utime(temppath, ns=(mtime_ns, mtime_ns))
            if onverify is not None:
                onverify(temppath, target_path)
            shutil.move(temppath, target_path)
        finally:
            try:
                os.remove(temppath)
            except OSError:
                pass


def get_gh_asset_name(basename: PathLike, text: bool = False) -> str:
    # GitHub will throw out charaters like '~' or '='. It also doesn't like
    # when there is no file extension and will try to add one
    return hashlib.sha256(str(basename).encode("utf-8")).hexdigest() + (".bin" if not text else ".txt")


def get_asset_filename(asset: Attachment) -> str:
    return asset.name


def get_release_assets(release: Release) -> list[Attachment]:
    assets = []
    for asset in release.assets:
        # We allow uploads from GHA and some special users
        assets.append(asset)
    return assets


def upload_asset(repo: Repository, release: Release, path: PathLike, replace: bool = False,
                 text: bool = False, content: bytes | None = None) -> None:
    gitea = get_gitea()
    path = Path(path)
    basename = os.path.basename(str(path))
    asset_name = get_gh_asset_name(basename, text)
    asset_label = basename

    def can_try_upload_again() -> bool:
        for asset in get_release_assets(release):
            if asset_name == asset.name:
                # We want to treat incomplete assets as if they weren't there
                # so replace them always
                if replace:
                    gitea.repo_delete_release_attachment(repo.owner.login, repo.name, release.id, asset.id)
                    break
                else:
                    print(f"Skipping upload for {asset_name} as {asset_label}, already exists")
                    return False
        return True

    def upload() -> None:
        if content is None:
            with open(path, "rb") as fileobj:
                gitea.repo_create_release_attachment(repo.owner.login, repo.name, release.id, name=asset_label, attachment=path)
        else:
            tmp_path = None
            try:
                with tempfile.NamedTemporaryFile(delete=False) as tf:
                    tf.write(content)
                    tf.flush()
                    tmp_path = tf.name
                    
                    new_asset = gitea.repo_create_release_attachment(repo.owner.login, repo.name, release.id, name=asset_label, attachment=tmp_path)
            finally:
                if tmp_path and os.path.exists(tmp_path):
                    os.remove(tmp_path)

    try:
        upload()
    except (ApiException, requests.RequestException):
        if can_try_upload_again():
            upload()

    print(f"Uploaded {asset_name} as {asset_label}")


def get_release(repo: Repository, name: str, create: bool = True) -> Release:
    """Like Repository.get_release() but creates the referenced release if needed"""

    gitea = get_gitea()
    try:
        return gitea.repo_get_release_by_tag(repo.owner.login, repo.name, name)
    except ApiException:
        if not create:
            raise
        return gitea.repo_create_release(repo.owner.login, repo.name, body=CreateReleaseOption(tag_name = name, prerelease = True))


class CachedAssets:

    def __init__(self) -> None:
        self._assets: dict[BuildType, list[Attachment]] = {}
        self._failed: dict[str, list[Attachment]] = {}

    def get_assets(self, build_type: BuildType) -> list[Attachment]:
        if build_type not in self._assets:
            repo = get_repo_for_build_type(build_type)
            release = get_release(repo, 'staging-' + build_type)
            self._assets[build_type] = get_release_assets(release)
        return self._assets[build_type]

    def get_failed_assets(self, build_type: BuildType) -> list[Attachment]:
        repo = get_repo_for_build_type(build_type)
        key = repo.full_name
        if key not in self._failed:
            release = get_release(repo, 'staging-failed')
            self._failed[key] = get_release_assets(release)
        assets = self._failed[key]
        # XXX: This depends on the format of the filename
        return [a for a in assets if get_asset_filename(a).startswith(build_type + "-")]