orisys-sdk/build/release/version.py

"""Version consistency and duplicate-release guards for the release gate."""

from __future__ import annotations

import json
import os
import re
import subprocess
import urllib.error
import urllib.request
from pathlib import Path

PYPI_JSON_URL = "https://pypi.org/pypi/orisys/json"

# Sibling repository that carries the public GitHub releases and their tags.
RELEASE_REPO_DIRNAME = "orisys-sdk-release"

# orisys-<version>-<pytag>-<abitag>-<platformtag>.whl
_WHEEL_NAME_RE = re.compile(r"^orisys-(?P<version>[^-]+)-[^-]+-[^-]+-.+\.whl$")


def wheel_version(wheel_path: Path) -> str | None:
    match = _WHEEL_NAME_RE.match(Path(wheel_path).name)
    return match.group("version") if match else None


def sdk_versions(repo_root: Path) -> list[str]:
    """Version directory names present under ``SDK/``, in ascending order.

    Only digit-leading names count as versions, so stray directories (``tmp``,
    ``latest``) never appear as releases.
    """
    sdk_root = Path(repo_root) / "SDK"
    if not sdk_root.is_dir():
        return []

    return sorted(
        (
            entry.name
            for entry in sdk_root.iterdir()
            if entry.is_dir() and entry.name[:1].isdigit()
        ),
        key=_version_sort_key,
    )


def sdk_wheels(repo_root: Path, version: str) -> list[Path]:
    """Wheels already published into ``SDK/<version>/``."""
    version_dir = Path(repo_root) / "SDK" / version
    if not version_dir.is_dir():
        return []
    return sorted(version_dir.glob("*.whl"))


def dynamic_version_attr(repo_root: Path) -> str | None:
    """Return pyproject's configured dynamic version attr (e.g. orisys.__version__)."""
    from common.pyproject_utils import load_pyproject

    config = (
        load_pyproject(Path(repo_root))
        .get("tool", {})
        .get("setuptools", {})
        .get("dynamic", {})
        .get("version")
    )
    if isinstance(config, dict):
        attr = config.get("attr")
        if isinstance(attr, str):
            return attr
    return None


def pypi_released_versions(timeout: float = 20.0) -> tuple[list[str] | None, str | None]:
    """Read the published version list from PyPI.

    Returns ``(versions, error)``; ``versions`` is ``None`` when the lookup
    failed. This is the only network call in the release gate and it is
    read-only, so it is safe to run before an irreversible upload.
    """
    request = urllib.request.Request(
        PYPI_JSON_URL,
        headers={"User-Agent": "orisys-release-preflight", "Accept": "application/json"},
    )
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            payload = json.load(response)
    except urllib.error.HTTPError as exc:
        if exc.code == 404:
            return [], None
        return None, f"PyPI returned HTTP {exc.code}"
    except Exception as exc:  # network, DNS, TLS, JSON
        return None, f"{type(exc).__name__}: {exc}"

    releases = payload.get("releases")
    if not isinstance(releases, dict):
        return None, "unexpected PyPI response shape (no 'releases' mapping)"
    listed = [
        str(version)
        for version, files in releases.items()
        if isinstance(files, list) and files
    ]
    return sorted(listed, key=_version_sort_key), None


def _version_sort_key(name: str):
    parts = re.findall(r"\d+", name)
    return ([int(part) for part in parts] or [0], name)


def git_tags(repo_root: Path) -> list[str]:
    """Local git tags from this repo and the sibling release repo.

    Tags are the only offline evidence that a version was already released
    (``v0.5.1`` in ``orisys-sdk-release`` marks a published GitHub release), so
    the duplicate guard works without network access.
    """
    roots = [Path(repo_root)]
    sibling = Path(repo_root).parent / RELEASE_REPO_DIRNAME
    if sibling.is_dir():
        roots.append(sibling)

    tags: set[str] = set()
    for root in roots:
        try:
            completed = subprocess.run(
                ["git", "-C", str(root), "tag", "--list"],
                capture_output=True,
                text=True,
                check=False,
            )
        except OSError:
            continue
        if completed.returncode != 0:
            continue
        tags.update(
            line.strip() for line in (completed.stdout or "").splitlines() if line.strip()
        )
    return sorted(tags)


def released_tags_for(version: str, tags: list[str]) -> list[str]:
    """Tag spellings that mark ``version`` as already released."""
    return sorted({version, f"v{version}"} & set(tags))


def conda_envs_root() -> Path | None:
    """Locate the conda envs directory the same way build/wheel/win.bat does."""
    override = os.getenv("CONDA_ENVS_ROOT")
    if override:
        path = Path(override)
        return path if path.is_dir() else None

    home = Path.home()
    candidates = (
        home / "anaconda3" / "envs",
        home / "miniconda3" / "envs",
        Path(os.getenv("LOCALAPPDATA", "")) / "anaconda3" / "envs",
        Path(os.getenv("LOCALAPPDATA", "")) / "miniconda3" / "envs",
    )
    for candidate in candidates:
        if candidate.is_dir():
            return candidate
    return None


def python_for_tag(tag: str, envs_root: Path | None = None) -> Path | None:
    """Resolve an interpreter for a wheel tag like ``cp310`` (env ``py310``)."""
    digits = tag[2:] if tag.startswith("cp") else tag
    env_name = f"py{digits}"
    root = envs_root or conda_envs_root()
    if root is None:
        return None

    for relative in (
        Path(env_name) / "python.exe",
        Path(env_name) / "bin" / "python",
    ):
        candidate = root / relative
        if candidate.is_file():
            return candidate
    return None