GitLite/cicd_tool/main.py

import argparse
import io
import json
import os
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable, List, Optional
from urllib.parse import urlparse

from minio import Minio
from minio.error import S3Error


REPO_ROOT = Path(__file__).resolve().parent.parent
PROJECT_DIR = REPO_ROOT / "GitLite"
ARTIFACT_ROOT = PROJECT_DIR / "src-tauri" / "target" / "release" / "bundle"


class ConfigurationError(RuntimeError):
    """Raised when required configuration values are missing or invalid."""


def _resolve_version(version_arg: Optional[str]) -> str:
    """Resolve the version to publish, falling back to tauri.conf.json."""
    if version_arg and version_arg.lower() not in {"auto"}:
        return version_arg

    config_path = PROJECT_DIR / "src-tauri" / "tauri.conf.json"
    if not config_path.exists():
        raise ConfigurationError(f"Cannot locate {config_path}")

    try:
        tauri_conf = json.loads(config_path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise ConfigurationError(f"Invalid JSON in {config_path}") from exc

    version = tauri_conf.get("version")
    if not version:
        raise ConfigurationError("Version not found in tauri.conf.json")

    return str(version)


def _normalise_endpoint(raw_endpoint: str) -> tuple[str, bool]:
    """Return (endpoint, secure) suitable for the MinIO client."""
    if not raw_endpoint:
        raise ConfigurationError("MINIO_ENDPOINT is required")

    if "://" not in raw_endpoint:
        return raw_endpoint, not raw_endpoint.startswith("http://")

    parsed = urlparse(raw_endpoint)
    if not parsed.netloc:
        raise ConfigurationError(f"MINIO_ENDPOINT is invalid: {raw_endpoint}")
    secure = parsed.scheme != "http"
    return parsed.netloc, secure


def _create_minio_client() -> Minio:
    endpoint_raw = os.environ.get("MINIO_ENDPOINT", "")
    endpoint, secure = _normalise_endpoint(endpoint_raw)

    access_key = os.environ.get("S3_ACCESS_KEY") or os.environ.get("AWS_ACCESS_KEY_ID")
    secret_key = os.environ.get("S3_SECRET_KEY") or os.environ.get(
        "AWS_SECRET_ACCESS_KEY"
    )
    if not access_key or not secret_key:
        raise ConfigurationError(
            "Missing S3 credentials. Provide S3_ACCESS_KEY/S3_SECRET_KEY or "
            "AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY."
        )

    region = os.environ.get("S3_REGION") or os.environ.get("AWS_REGION")

    return Minio(
        endpoint=endpoint,
        access_key=access_key,
        secret_key=secret_key,
        secure=secure,
        region=region,
    )


def _ensure_bucket(client: Minio, bucket_name: str) -> None:
    if not bucket_name:
        raise ConfigurationError("S3_BUCKET is required")

    exists = client.bucket_exists(bucket_name)
    if not exists:
        client.make_bucket(bucket_name)


def _collect_artifacts() -> List[Path]:
    artifacts: List[Path] = []
    for directory in (
        ARTIFACT_ROOT / "msi",
        ARTIFACT_ROOT / "nsis",
        ARTIFACT_ROOT / "app",
    ):
        if directory.exists():
            artifacts.extend(p for p in directory.rglob("*") if p.is_file())
    return artifacts


def _select_primary_artifact(artifacts: Iterable[Path]) -> Optional[Path]:
    priority = (".exe", ".msi", ".zip", ".dmg")
    for ext in priority:
        for artifact in artifacts:
            if artifact.suffix.lower() == ext:
                return artifact
    return next(iter(artifacts), None)


def _read_signature(artifact: Path) -> str:
    candidates = [
        artifact.with_suffix(artifact.suffix + ".sig"),
        artifact.with_name(f"{artifact.stem}.sig"),
    ]
    for candidate in candidates:
        if candidate.exists():
            return candidate.read_text(encoding="utf-8").strip()
    raise FileNotFoundError(f"Signature not found for {artifact.name}")


def _build_latest_json(
    version: str,
    primary_artifact: Path,
    signature: str,
    artifact_base_url: str,
    app_dir: str,
) -> str:
    base_url = artifact_base_url.rstrip("/")
    path_parts = [app_dir.strip("/"), version, primary_artifact.name]
    artifact_url = "/".join([base_url, *filter(None, path_parts)])

    pub_date = (
        datetime.now(timezone.utc)
        .replace(microsecond=0)
        .isoformat()
        .replace("+00:00", "Z")
    )

    latest_payload = {
        "version": version,
        "pub_date": pub_date,
        "notes": os.environ.get("RELEASE_NOTES", "Automated release"),
        "platforms": {
            "windows-x86_64": {
                "signature": signature,
                "url": artifact_url,
            }
        },
    }
    return json.dumps(latest_payload, separators=(",", ":"))


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", "")
    if not artifact_base_url:
        raise ConfigurationError("ARTIFACT_BASE_URL is required")

    bucket_name = os.environ.get("S3_BUCKET", "gitlite")
    bucket_prefix = (app_dir or os.environ.get("APP_DIR") or "gitlite").strip("/")

    artifacts = _collect_artifacts()
    if not artifacts:
        raise FileNotFoundError(
            "No release artifacts found in src-tauri/target/release/bundle."
        )

    primary = _select_primary_artifact(artifacts)
    if not primary:
        raise FileNotFoundError("Unable to determine primary artifact to publish.")

    signature = _read_signature(primary)

    latest_json = _build_latest_json(
        version=resolved_version,
        primary_artifact=primary,
        signature=signature,
        artifact_base_url=artifact_base_url,
        app_dir=bucket_prefix,
    )

    local_dist_dir = PROJECT_DIR / "dist"
    local_dist_dir.mkdir(parents=True, exist_ok=True)
    latest_json_path = local_dist_dir / "latest.json"
    latest_json_path.write_text(latest_json + "\n", encoding="utf-8")

    client = _create_minio_client()
    _ensure_bucket(client, bucket_name)

    for artifact in artifacts:
        object_segments = [part for part in (resolved_version, artifact.name) if part]
        object_name = "/".join(object_segments)
        client.fput_object(bucket_name, object_name, str(artifact))
        print(f"Uploaded {artifact.name} -> s3://{bucket_name}/{object_name}")

    latest_object_name = "latest.json"
    latest_bytes = latest_json.encode("utf-8")
    client.put_object(
        bucket_name=bucket_name,
        object_name=latest_object_name,
        data=io.BytesIO(latest_bytes),
        length=len(latest_bytes),
        content_type="application/json",
    )
    print(f"Uploaded latest.json -> s3://{bucket_name}/{latest_object_name}")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Upload GitLite release artifacts to MinIO."
    )
    parser.add_argument(
        "-v",
        "--version",
        default=None,
        help="Version tag to upload (defaults to reading tauri.conf.json).",
    )
    parser.add_argument(
        "--app-dir",
        default=None,
        help="App directory prefix inside the bucket (defaults to gitlite).",
    )
    args = parser.parse_args()

    try:
        main(version=args.version, app_dir=args.app_dir)
    except (ConfigurationError, FileNotFoundError, S3Error) as exc:
        raise SystemExit(f"Upload failed: {exc}") from exc