SyncPlayer/publish.py

"""Release/publish helper for SyncPlayer.

What it does (in order):
1) Push current git branch to origin.
2) Create an annotated git tag with the provided version and push it.
3) Build a Docker image tagged with:
     - git.alphen.cloud/bramval/SyncPlayer:<version>
     - git.alphen.cloud/bramval/SyncPlayer:latest
   Then push both tags.

Usage:
  python publish.py 1.2.3
  python publish.py v1.2.3 --dry-run
"""

from __future__ import annotations

import argparse
import re
import subprocess
import sys
from typing import Iterable, Sequence


DOCKER_REPO = "git.alphen.cloud/bramval/syncplayer"


def _run(cmd: Sequence[str], *, dry_run: bool = False) -> None:
    printable = " ".join(_quote_arg(a) for a in cmd)
    print(f"\n$ {printable}")
    if dry_run:
        return

    subprocess.run(list(cmd), check=True)


def _capture(cmd: Sequence[str]) -> str:
    return subprocess.check_output(list(cmd), text=True, stderr=subprocess.STDOUT).strip()


def _quote_arg(s: str) -> str:
    # Minimal quoting for readability in logs.
    if re.search(r"[\s\"]", s):
        return '"' + s.replace('"', '\\"') + '"'
    return s


def _ensure_git_repo() -> None:
    try:
        inside = _capture(["git", "rev-parse", "--is-inside-work-tree"])
    except Exception as e:  # noqa: BLE001
        raise SystemExit("Not a git repository (or git is not installed).") from e

    if inside.lower() != "true":
        raise SystemExit("Not a git repository.")


def _ensure_clean_worktree(*, allow_dirty: bool) -> None:
    status = _capture(["git", "status", "--porcelain"])  # empty == clean
    if status and not allow_dirty:
        print("\nERROR: Git working tree is not clean. Commit/stash changes first.")
        print(status)
        raise SystemExit(2)


def _current_branch() -> str:
    branch = _capture(["git", "rev-parse", "--abbrev-ref", "HEAD"])
    if branch == "HEAD":
        raise SystemExit("Detached HEAD; checkout a branch before publishing.")
    return branch


def _check_origin_remote() -> None:
    try:
        url = _capture(["git", "config", "--get", "remote.origin.url"])
    except Exception:
        url = ""

    # Not a hard failure: the user may use SSH or a different URL form.
    if "git.alphen.cloud/bramval/SyncPlayer" not in url:
        print(
            "\nWARNING: remote.origin.url does not look like the expected repo.\n"
            f"  origin = {url or '(missing)'}\n"
            "  expected to contain: git.alphen.cloud/bramval/SyncPlayer"
        )


_DOCKER_TAG_RE = re.compile(r"^[A-Za-z0-9_][A-Za-z0-9_.-]{0,127}$")


def _validate_version(version: str) -> None:
    # Use docker tag constraints so the same string works for both git tag and docker tag.
    if not _DOCKER_TAG_RE.match(version):
        raise SystemExit(
            "Invalid version/tag. Use only letters/digits/underscore/dot/dash (max 128 chars).\n"
            f"Got: {version!r}"
        )


def _ensure_tag_not_exists(tag: str) -> None:
    # `git show-ref --tags --verify` exits non-zero when missing.
    try:
        subprocess.run(
            ["git", "show-ref", "--tags", "--verify", "--quiet", f"refs/tags/{tag}"],
            check=True,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
    except subprocess.CalledProcessError:
        return

    raise SystemExit(f"Git tag already exists: {tag}")


def publish(version: str, *, dry_run: bool, allow_dirty: bool) -> None:
    _validate_version(version)
    _ensure_git_repo()
    _check_origin_remote()
    # In dry-run we want to show the full command sequence even if the tree is dirty.
    # For real publishes we keep the strict guard by default.
    if dry_run:
        try:
            _ensure_clean_worktree(allow_dirty=True)
        except SystemExit:
            # Shouldn't happen because allow_dirty=True, but keep behavior explicit.
            pass
    else:
        _ensure_clean_worktree(allow_dirty=allow_dirty)

    branch = _current_branch()
    tag = version

    # --- Git ---
    print(f"\n== Git: pushing branch '{branch}' to origin ==")
    _run(["git", "push", "origin", branch], dry_run=dry_run)

    print(f"\n== Git: tagging '{tag}' and pushing tag ==")
    if not dry_run:
        _ensure_tag_not_exists(tag)
    _run(["git", "tag", "-a", tag, "-m", f"Release {tag}"], dry_run=dry_run)
    _run(["git", "push", "origin", tag], dry_run=dry_run)

    # --- Docker ---
    image_version = f"{DOCKER_REPO}:{version}"
    image_latest = f"{DOCKER_REPO}:latest"

    print("\n== Docker: build ==")
    _run(["docker", "build", "-t", image_version, "-t", image_latest, "."], dry_run=dry_run)

    print("\n== Docker: push ==")
    _run(["docker", "push", image_version], dry_run=dry_run)
    _run(["docker", "push", image_latest], dry_run=dry_run)

    print("\nDone.")


def _parse_args(argv: Iterable[str]) -> argparse.Namespace:
    p = argparse.ArgumentParser(description="Push git + build/push docker images (version + latest).")
    p.add_argument("version", help="Version/tag to publish (e.g. 1.2.3 or v1.2.3)")
    p.add_argument(
        "--dry-run",
        action="store_true",
        help="Print commands but do not execute them.",
    )
    p.add_argument(
        "--allow-dirty",
        action="store_true",
        help="Allow publishing with uncommitted git changes (not recommended).",
    )
    return p.parse_args(list(argv))


def main(argv: Sequence[str] | None = None) -> int:
    ns = _parse_args(sys.argv[1:] if argv is None else argv)
    publish(ns.version, dry_run=ns.dry_run, allow_dirty=ns.allow_dirty)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())