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))


# Version literals that are deliberately not the package version. Keyed by
# repo-relative path; each entry documents *why* the literal may stay.
VERSION_STRING_ALLOWLIST: dict[str, frozenset[str]] = {
    # Historical release records: rewriting these would destroy the audit trail.
    "docs/publishing.md": frozenset({"0.4.2", "0.4.3", "0.4.4", "0.5.0", "0.5.1", "0.5.2"}),
    # Unrelated third-party/CUDA versions quoted in prose.
    "docs/get_started.md": frozenset({"12.9.0"}),
}

# Paths never scanned: tests deliberately use synthetic versions, reports are
# generated output, and this module's own allowlist/sentinel literals would
# otherwise flag themselves.
VERSION_SCAN_EXCLUDED_PREFIXES = (
    "tests/",
    "build/release/reports/",
    "build/nuitka/",
    "build/lib/",
    "dist/",
    "SDK/",
    "data/",
)
VERSION_SCAN_EXCLUDED_FILES = ("build/release/version.py",)

_VERSION_LITERAL_RE = re.compile(
    r"(?<![<>=!~\w.])v?(\d+\.\d+\.\d+(?:[-+][0-9A-Za-z.]+)?)(?![0-9A-Za-z.])"
)
_SCANNED_SUFFIXES = (".md", ".py", ".txt", ".toml", ".cfg", ".ini", ".rst")


def tracked_text_files(repo_root: Path) -> list[Path]:
    """Tracked files worth scanning for version literals."""
    repo_root = Path(repo_root)
    try:
        completed = subprocess.run(
            ["git", "-C", str(repo_root), "ls-files", "-z"],
            capture_output=True,
            text=True,
            check=False,
        )
        names = [n for n in (completed.stdout or "").split("\0") if n]
    except OSError:
        names = []

    if not names:
        names = [
            str(path.relative_to(repo_root)).replace("\\", "/")
            for path in repo_root.rglob("*")
            if path.is_file()
        ]

    selected = []
    for name in names:
        if name.startswith(VERSION_SCAN_EXCLUDED_PREFIXES):
            continue
        if name in VERSION_SCAN_EXCLUDED_FILES:
            continue
        if not name.endswith(_SCANNED_SUFFIXES):
            continue
        selected.append(Path(name))
    return selected


def find_version_literals(
    repo_root: Path, version: str
) -> list[tuple[str, int, str]]:
    """``(path, lineno, literal)`` for every version-looking string != ``version``.

    Dependency constraints are skipped by the regex lookbehind (``numpy>=1.24.4``
    is not a version reference), and historical/third-party literals require an
    explicit entry in :data:`VERSION_STRING_ALLOWLIST`.
    """
    repo_root = Path(repo_root)
    findings: list[tuple[str, int, str]] = []
    for relative in tracked_text_files(repo_root):
        absolute = repo_root / relative
        try:
            text = absolute.read_text(encoding="utf-8", errors="ignore")
        except OSError:
            continue
        if "\0" in text:
            continue
        allowed = VERSION_STRING_ALLOWLIST.get(relative.as_posix(), frozenset())
        for lineno, line in enumerate(text.splitlines(), start=1):
            for match in _VERSION_LITERAL_RE.finditer(line):
                literal = match.group(1)
                if literal == version or literal in allowed:
                    continue
                findings.append((relative.as_posix(), lineno, literal))
    return findings


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