DEVX-92: feat: add distribute_items CI tool for parallel VM deployment

src/devx/ci/distribute_items.py

@@ -0,0 +1,210 @@
+#!/usr/bin/env python3
+"""Distribute a list of items across N parallel runners using LPT scheduling.
+
+Generic item distribution for CI matrix jobs. Items are read from a JSON
+array on stdin (or from a file via --items-file), sorted for deterministic
+ordering, then assigned to *max_runners* groups using LPT (Longest
+Processing Time first) scheduling.
+
+Each item is a string (e.g. an Ansible ``--limit`` pattern like
+``observability`` or ``infra-314-vm``). Optionally, items can be objects
+with ``{"id": "...", "weight": N}`` to provide explicit weights.
+
+The assigned group for *runner_index* is written to ``$GITHUB_ENV`` as
+``ASSIGNED_ITEMS`` (space-delimited) for use by subsequent steps.
+
+Usage::
+
+    echo '["observability", "infra-314-vm"]' | \\
+        python3 -m devx.ci.distribute_items \\
+        --runner-index 1 --max-runners 3 \\
+        --github-env --skip-if-excess
+
+    # With weights:
+    echo '[{"id": "observability", "weight": 5}, {"id": "customer-1", "weight": 3}]' | \\
+        python3 -m devx.ci.distribute_items \\
+        --runner-index 1 --max-runners 3 --github-env
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+
+import click
+
+from devx.i18n import _
+
+DEFAULT_MAX_RUNNERS = 3
+DEFAULT_WEIGHT = 1
+
+
+def parse_items(raw: str) -> list[str]:
+    """Parse a JSON array into a list of item identifier strings.
+
+    Accepts both plain string arrays (``["a", "b"]``) and object arrays
+    (``[{"id": "a", "weight": 2}]``). Returns just the identifier strings.
+    """
+    data = json.loads(raw)
+    if not isinstance(data, list):
+        raise click.ClickException(_("Items input must be a JSON array, got {type}", type=type(data).__name__))
+    items: list[str] = []
+    for entry in data:
+        if isinstance(entry, str):
+            items.append(entry)
+        elif isinstance(entry, dict) and "id" in entry:
+            items.append(str(entry["id"]))
+        else:
+            raise click.ClickException(
+                _("Each item must be a string or an object with 'id', got {type}", type=type(entry).__name__)
+            )
+    return items
+
+
+def parse_weighted_items(raw: str) -> tuple[list[str], list[int]]:
+    """Parse a JSON array into (items, weights) lists.
+
+    For plain string arrays, all items get ``DEFAULT_WEIGHT``.
+    For object arrays, the ``weight`` field is used (default: ``DEFAULT_WEIGHT``).
+    """
+    data = json.loads(raw)
+    if not isinstance(data, list):
+        raise click.ClickException(_("Items input must be a JSON array, got {type}", type=type(data).__name__))
+    items: list[str] = []
+    weights: list[int] = []
+    for entry in data:
+        if isinstance(entry, str):
+            items.append(entry)
+            weights.append(DEFAULT_WEIGHT)
+        elif isinstance(entry, dict) and "id" in entry:
+            items.append(str(entry["id"]))
+            weights.append(int(entry.get("weight", DEFAULT_WEIGHT)))
+        else:
+            raise click.ClickException(
+                _("Each item must be a string or an object with 'id', got {type}", type=type(entry).__name__)
+            )
+    return items, weights
+
+
+def distribute(items: list[str], weights: list[int], max_runners: int) -> list[list[str]]:
+    """Split *items* into *max_runners* balanced groups using LPT scheduling.
+
+    Items are sorted by weight (descending), then assigned to the runner
+    with the least total weight.
+    """
+    groups: list[list[str]] = [[] 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
+
+
+def items_for_runner(items: list[str], weights: list[int], runner_index: int, max_runners: int) -> list[str]:
+    """Return the subset of items assigned to *runner_index* (0-based)."""
+    groups = distribute(items, weights, max_runners)
+    if runner_index < 0 or runner_index >= len(groups):
+        raise click.ClickException(
+            _("Runner index {index} out of range (0..{max})", index=runner_index, max=max_runners - 1)
+        )
+    return groups[runner_index]
+
+
+def _write_github_env(key: str, value: str) -> None:
+    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") 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")
+
+
+@click.command()
+@click.option(
+    "--items-file",
+    type=click.Path(exists=True, file_okay=True, path_type=None),
+    default=None,
+    help="Read items from a JSON file instead of stdin.",
+)
+@click.option(
+    "--runner-index",
+    type=int,
+    default=None,
+    help="One-based runner index. If omitted, prints all groups.",
+)
+@click.option(
+    "--max-runners",
+    type=int,
+    default=DEFAULT_MAX_RUNNERS,
+    show_default=True,
+    help="Total number of parallel runners.",
+)
+@click.option(
+    "--github-env",
+    is_flag=True,
+    default=False,
+    help="Write ASSIGNED_ITEMS and SKIP to $GITHUB_ENV.",
+)
+@click.option(
+    "--skip-if-excess",
+    is_flag=True,
+    default=False,
+    help="With --github-env: write SKIP=true when runner-index exceeds max-runners.",
+)
+def main(
+    items_file: str | None,
+    runner_index: int | None,
+    max_runners: int,
+    github_env: bool,
+    skip_if_excess: bool,
+) -> None:
+    # Read items from file or stdin
+    if items_file is not None:
+        with open(items_file) as f:  # noqa: PTH123
+            raw = f.read()
+    else:
+        raw = sys.stdin.read()
+
+    raw = raw.strip()
+    if not raw:
+        raw = "[]"
+
+    items, weights = parse_weighted_items(raw)
+
+    if runner_index is None:
+        groups = distribute(items, weights, max_runners)
+        for i, group in enumerate(groups):
+            labels = " ".join(group) if group else "(none)"
+            click.echo(f"Runner {i}: {labels}")
+        return
+
+    if skip_if_excess and github_env and runner_index > max_runners:
+        click.echo(f"Skipping — runner index {runner_index} > max runners {max_runners}")
+        _write_github_env("ASSIGNED_ITEMS", "")
+        _write_github_env("SKIP", "true")
+        return
+
+    if runner_index < 1:
+        raise click.ClickException(f"Runner index {runner_index} is out of range (must be >= 1)")
+
+    zero_based = runner_index - 1
+    assigned = items_for_runner(items, weights, zero_based, max_runners)
+    encoded = " ".join(assigned)
+
+    if github_env:
+        _write_github_env("ASSIGNED_ITEMS", encoded)
+        _write_github_env("SKIP", "false")
+        click.echo(f"Assigned {len(assigned)} items to runner {runner_index}: {encoded}")
+        return
+
+    click.echo(encoded)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()

src/devx/translations.json

@@ -639,6 +639,14 @@
     "ru": "Dockerfile not found: {path}",
     "zh": "Dockerfile not found: {path}"
   },
+  "Each item must be a string or an object with 'id', got {type}": {
+    "bg": "Всеки елемент трябва да е низ или обект с 'id', получено {type}",
+    "de": "Jedes Element muss ein String oder ein Objekt mit 'id' sein, erhalten {type}",
+    "en": "Each item must be a string or an object with 'id', got {type}",
+    "pl": "Każdy element musi być ciągiem lub obiektem z 'id', otrzymano {type}",
+    "ru": "Каждый элемент должен быть строкой или объектом с 'id', получено {type}",
+    "zh": "每个元素必须是字符串或带有 'id' 的对象，得到 {type}"
+  },
   "Dry-run mode: on branch '{branch}' (not master). Some checks may behave differently.": {
     "bg": "Dry-run mode: on branch '{branch}' (not master). Some checks may behave differently.",
     "de": "Dry-run mode: on branch '{branch}' (not master). Some checks may behave differently.",
@@ -2247,6 +2255,14 @@
     "ru": "Invalid checklist category: {cat}. Must be numbers.",
     "zh": "Invalid checklist category: {cat}. Must be numbers."
   },
+  "Items input must be a JSON array, got {type}": {
+    "bg": "Входните данни трябва да са JSON масив, получено {type}",
+    "de": "Eingabe muss ein JSON-Array sein, erhalten {type}",
+    "en": "Items input must be a JSON array, got {type}",
+    "pl": "Dane wejściowe muszą być tablicą JSON, otrzymano {type}",
+    "ru": "Входные данные должны быть JSON-массивом, получено {type}",
+    "zh": "输入必须是 JSON 数组，得到 {type}"
+  },
   "Review body must be at least 50 characters.": {
     "en": "Review body must be at least 50 characters.",
     "bg": "Review body must be at least 50 characters.",