diff --git a/cicd_tool/main.py b/cicd_tool/main.py index 38f2bc8..12fe8b3 100644 --- a/cicd_tool/main.py +++ b/cicd_tool/main.py @@ -1,11 +1,15 @@ import argparse import io import json +import mimetypes import os +import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable, List, NamedTuple, Optional from urllib.parse import urlparse +from urllib.parse import quote, urlencode +from urllib.request import Request, urlopen from minio import Minio from minio.error import S3Error @@ -329,6 +333,65 @@ def _publish_target( ) +def _gitea_request(url: str, token: str, data: Optional[bytes] = None, **headers: str): + request = Request(url, data=data, headers={"Authorization": f"token {token}", **headers}) + return urlopen(request, timeout=120) + + +def _upload_gitea_release_asset(version: str, artifact: Path) -> None: + """Attach the installer to the Gitea release that triggered this build.""" + api_url = os.environ.get("GITEA_API", "").rstrip("/") + owner = os.environ.get("OWNER", "") + repo = os.environ.get("REPO", "") + token = os.environ.get("GITEA_TOKEN") or os.environ.get("GITEA_FALLBACK_TOKEN") + if not all((api_url, owner, repo, token)): + raise ConfigurationError( + "GITEA_API, OWNER, REPO and GITEA_TOKEN are required to publish " + "the release asset." + ) + + release_url = ( + f"{api_url}/repos/{quote(owner, safe='')}/{quote(repo, safe='')}" + f"/releases/tags/{quote(version, safe='')}" + ) + with _gitea_request(release_url, token) as response: + release = json.load(response) + + existing_names = { + asset.get("name") for asset in release.get("assets", []) if isinstance(asset, dict) + } + if artifact.name in existing_names: + print(f"Release asset {artifact.name} already exists; skipping upload.") + return + + boundary = f"----gitty-{uuid.uuid4().hex}" + content_type = mimetypes.guess_type(artifact.name)[0] or "application/octet-stream" + body = b"".join( + ( + f"--{boundary}\r\n".encode(), + ( + f'Content-Disposition: form-data; name="attachment"; ' + f'filename="{artifact.name}"\r\n' + ).encode(), + f"Content-Type: {content_type}\r\n\r\n".encode(), + artifact.read_bytes(), + f"\r\n--{boundary}--\r\n".encode(), + ) + ) + upload_url = ( + f"{api_url}/repos/{quote(owner, safe='')}/{quote(repo, safe='')}" + f"/releases/{release['id']}/assets?{urlencode({'name': artifact.name})}" + ) + with _gitea_request( + upload_url, + token, + data=body, + **{"Content-Type": f"multipart/form-data; boundary={boundary}"}, + ): + pass + print(f"Attached {artifact.name} to Gitea release {version}.") + + def main(version: Optional[str] = None, app_dir: Optional[str] = None) -> None: resolved_version = _resolve_version(version) artifact_base_url = os.environ.get("ARTIFACT_BASE_URL", "") @@ -385,6 +448,8 @@ def main(version: Optional[str] = None, app_dir: Optional[str] = None) -> None: artifact_base_url=artifact_base_url, ) + _upload_gitea_release_asset(resolved_version, primary) + if __name__ == "__main__": parser = argparse.ArgumentParser(