devx/src/devx/ci/_shared.py

"""Shared utilities for CI modules."""

from __future__ import annotations

import os
import subprocess  # nosec B404

import click

from devx.config import TASK_ID_RE
from devx.i18n import _


def get_latest_tag() -> str:
    """Get the latest git tag, or empty string if none exists."""
    result = subprocess.run(  # nosec B603 B607
        ["git", "describe", "--tags", "--abbrev=0"],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        return ""
    return result.stdout.strip()


def run_cmd(
    args: list[str],
    check: bool = True,
    capture: bool = True,
) -> subprocess.CompletedProcess[str]:
    """Run a command and return the completed process.

    Args:
        args: Command and arguments as a list.
        check: If True, raise :class:`click.ClickException` on non-zero exit.
        capture: If True, capture stdout/stderr. If False, inherit parent's.
    """
    result = subprocess.run(  # nosec B603
        args,
        capture_output=capture,
        text=True,
        check=False,
    )
    if check and result.returncode != 0:
        raise click.ClickException(
            _(
                "Command failed ({cmd}): {stderr}",
                cmd=" ".join(args),
                stderr=result.stderr.strip() if result.stderr else result.stdout.strip(),
            )
        )
    return result


def extract_task_id(text: str) -> str:
    """Extract the ``{PREFIX}-N`` task identifier from *text*.

    Returns the matched string (e.g. ``DEVX-42``) or an empty string if
    no task ID is found.
    """
    match = TASK_ID_RE.search(text)
    return match.group(0) if match else ""


def write_github_env(key: str, value: str) -> None:
    """Append a key=value line to the ``$GITHUB_ENV`` file.

    Multi-line values use the heredoc syntax required by Gitea Actions.
    Raises :class:`click.ClickException` if ``GITHUB_ENV`` is not set.
    """
    gh_env = os.environ.get("GITHUB_ENV")
    if not gh_env:
        raise click.ClickException("GITHUB_ENV environment variable is not set")
    with open(gh_env, "a", encoding="utf-8") as f:  # noqa: PTH123
        if "\n" in value:
            delimiter = "EOF"
            f.write(f"{key}<<{delimiter}\n{value}\n{delimiter}\n")
        else:
            f.write(f"{key}={value}\n")


def write_github_output(key: str, value: str) -> None:
    """Append a key=value line to the ``$GITHUB_OUTPUT`` file.

    Raises :class:`click.ClickException` if ``GITHUB_OUTPUT`` is not set.
    """
    gh_output = os.environ.get("GITHUB_OUTPUT")
    if not gh_output:
        raise click.ClickException("GITHUB_OUTPUT environment variable is not set")
    with open(gh_output, "a", encoding="utf-8") as f:  # noqa: PTH123
        f.write(f"{key}={value}\n")


def lpt_distribute[T](items: list[T], weights: list[int], max_runners: int) -> list[list[T]]:
    """Distribute *items* across *max_runners* using LPT scheduling.

    Sorts items by weight (descending), then assigns each to the runner
    with the least total weight. This produces a more balanced distribution
    than naive round-robin when items have varying costs.

    Args:
        items: Items to distribute.
        weights: Parallel list of integer weights (higher = heavier).
        max_runners: Number of runner groups to create.

    Returns:
        A list of ``max_runners`` lists, each containing the items assigned
        to that runner.
    """
    groups: list[list[T]] = [[] for _ in range(max_runners)]
    loads = [0] * max_runners
    indexed = sorted(enumerate(items), key=lambda x: (-weights[x[0]], x[0]))
    for orig_idx, item in indexed:
        min_runner = min(range(max_runners), key=lambda r: loads[r])
        groups[min_runner].append(item)
        loads[min_runner] += weights[orig_idx]
    return groups