Automate exact package-set publication

docs/PACKAGE_REGISTRY_RELEASES.md

@@ -37,6 +37,23 @@ changes:
 python tools/gitea/gitea-configure-package-releases.py
 ```
 
+Preview and dispatch the exact wheel/WebUI versions selected by the developer
+meta-package with:
+
+```bash
+python tools/gitea/gitea-dispatch-package-set.py \
+  --env-file ~/.config/gitea/gitea.env
+python tools/gitea/gitea-dispatch-package-set.py \
+  --env-file ~/.config/gitea/gitea.env \
+  --apply
+```
+
+The dispatcher reads exact versions from `packages/govoplan-meta/pyproject.toml`,
+inspects the selected tag to determine whether a WebUI package is expected,
+skips complete registry pairs and does not duplicate an active workflow. Use
+`--repository govoplan-core` for a bounded dispatch or `--verify-existing` to
+rebuild and hash-verify versions already present in both registries.
+
 It builds one wheel and, where applicable, one npm tarball. The workflow records
 the source tag, source commit, filename, size, and SHA-256 in
 `package-artifacts.json` before publishing. Gitea rejects a second upload of the

tests/test_package_set_dispatch.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+import sys
+import unittest
+
+
+META_ROOT = Path(__file__).resolve().parents[1]
+TOOLS_ROOT = META_ROOT / "tools" / "gitea"
+if str(TOOLS_ROOT) not in sys.path:
+    sys.path.insert(0, str(TOOLS_ROOT))
+SCRIPT = TOOLS_ROOT / "gitea-dispatch-package-set.py"
+SPEC = importlib.util.spec_from_file_location("gitea_dispatch_package_set", SCRIPT)
+assert SPEC is not None and SPEC.loader is not None
+MODULE = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = MODULE
+SPEC.loader.exec_module(MODULE)
+
+
+class PackageSetDispatchTests(unittest.TestCase):
+    def test_meta_package_resolves_to_exact_tagged_repository_targets(self) -> None:
+        targets = MODULE.package_targets()
+
+        self.assertEqual(66, len(targets))
+        self.assertEqual(66, len({target.distribution for target in targets}))
+        by_name = {target.distribution: target for target in targets}
+        self.assertEqual("v0.1.14", by_name["govoplan-core"].tag)
+        self.assertEqual("v0.1.8", by_name["govoplan-access"].tag)
+        self.assertTrue(by_name["govoplan-core"].tag_exists)
+        self.assertTrue(by_name["govoplan-access"].has_webui)
+        self.assertEqual(
+            "@govoplan/access-webui",
+            by_name["govoplan-access"].webui_package,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/gitea/gitea-dispatch-package-set.py

@@ -0,0 +1,302 @@
+#!/usr/bin/env python3
+"""Dispatch protected package releases required by the govoplan meta-package."""
+
+from __future__ import annotations
+
+import argparse
+from dataclasses import dataclass
+import json
+from pathlib import Path
+import re
+import subprocess
+import sys
+import tomllib
+
+from gitea_common import (
+    GiteaClient,
+    GiteaError,
+    RepoTarget,
+    load_dotenv,
+    org_path,
+    quote_path,
+    repo_path,
+    require_token,
+)
+
+
+META_ROOT = Path(__file__).resolve().parents[2]
+META_PROJECT = META_ROOT / "packages" / "govoplan-meta" / "pyproject.toml"
+WORKFLOW_ID = "module-package-release.yml"
+EXACT_REQUIREMENT = re.compile(
+    r"^(?P<name>govoplan-[a-z0-9-]+)(?:\[[a-z0-9_,.-]+\])?==(?P<version>[0-9]+\.[0-9]+\.[0-9]+)$"
+)
+ACTIVE_STATES = {"queued", "waiting", "in_progress", "running"}
+
+
+@dataclass(frozen=True, slots=True)
+class PackageTarget:
+    distribution: str
+    version: str
+    repository: str
+    tag_exists: bool
+    has_webui: bool
+
+    @property
+    def tag(self) -> str:
+        return f"v{self.version}"
+
+    @property
+    def webui_package(self) -> str | None:
+        if not self.has_webui:
+            return None
+        return f"@govoplan/{self.distribution.removeprefix('govoplan-')}-webui"
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--url", default="https://git.add-ideas.de")
+    parser.add_argument("--owner", default="GovOPlaN")
+    parser.add_argument("--env-file", type=Path)
+    parser.add_argument(
+        "--repository",
+        action="append",
+        default=[],
+        help="Limit dispatch to one repository; repeat as needed.",
+    )
+    parser.add_argument(
+        "--verify-existing",
+        action="store_true",
+        help="Also rerun exact versions already present in both registries.",
+    )
+    parser.add_argument("--apply", action="store_true")
+    return parser
+
+
+def package_targets(project_path: Path = META_PROJECT) -> tuple[PackageTarget, ...]:
+    project = tomllib.loads(project_path.read_text(encoding="utf-8"))["project"]
+    requirements = list(project.get("dependencies") or [])
+    requirements.extend(project.get("optional-dependencies", {}).get("full") or [])
+    parsed: dict[str, str] = {}
+    for requirement in requirements:
+        match = EXACT_REQUIREMENT.fullmatch(str(requirement))
+        if match is None:
+            raise ValueError(
+                f"Meta-package requirement is not an exact GovOPlaN version: {requirement!r}"
+            )
+        name = match.group("name")
+        version = match.group("version")
+        previous = parsed.setdefault(name, version)
+        if previous != version:
+            raise ValueError(f"Meta-package selects conflicting versions for {name}")
+
+    targets: list[PackageTarget] = []
+    for distribution, version in sorted(parsed.items()):
+        repository = distribution
+        repository_root = META_ROOT.parent / repository
+        if not (repository_root / ".git").is_dir():
+            raise ValueError(f"Package repository is not checked out: {repository}")
+        tag = f"v{version}"
+        tag_exists = _tag_exists(repository_root, tag)
+        has_webui = (
+            _tag_has_path(repository_root, tag, "webui/package.json")
+            if tag_exists
+            else False
+        )
+        targets.append(
+            PackageTarget(
+                distribution=distribution,
+                version=version,
+                repository=repository,
+                tag_exists=tag_exists,
+                has_webui=has_webui,
+            )
+        )
+    return tuple(targets)
+
+
+def _tag_exists(repository: Path, tag: str) -> bool:
+    result = subprocess.run(
+        (
+            "git",
+            "-C",
+            str(repository),
+            "rev-parse",
+            "--verify",
+            "--quiet",
+            f"refs/tags/{tag}",
+        ),
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode not in {0, 1}:
+        raise ValueError(
+            f"Could not inspect {repository.name}:{tag}: {result.stderr.strip()}"
+        )
+    return result.returncode == 0
+
+
+def _tag_has_path(repository: Path, tag: str, path: str) -> bool:
+    result = subprocess.run(
+        ("git", "-C", str(repository), "cat-file", "-e", f"{tag}:{path}"),
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode not in {0, 128}:
+        raise ValueError(
+            f"Could not inspect {repository.name}:{tag}:{path}: {result.stderr.strip()}"
+        )
+    return result.returncode == 0
+
+
+def _published_packages(
+    client: GiteaClient, *, owner: str, package_type: str
+) -> set[tuple[str, str]]:
+    values = client.paginate(
+        f"/packages/{quote_path(owner)}",
+        query={"type": package_type, "q": "govoplan"},
+    )
+    return {
+        (str(item.get("name") or ""), str(item.get("version") or ""))
+        for item in values
+        if item.get("type") == package_type
+    }
+
+
+def _has_active_run(
+    client: GiteaClient, *, owner: str, repository: str
+) -> bool:
+    payload = client.request_json(
+        "GET",
+        repo_path(
+            owner,
+            repository,
+            f"/actions/workflows/{quote_path(WORKFLOW_ID)}/runs",
+        ),
+        query={"limit": 10},
+    )
+    runs = payload.get("workflow_runs") if isinstance(payload, dict) else None
+    return isinstance(runs, list) and any(
+        isinstance(run, dict) and str(run.get("status") or "") in ACTIVE_STATES
+        for run in runs
+    )
+
+
+def dispatch(
+    client: GiteaClient,
+    *,
+    owner: str,
+    targets: tuple[PackageTarget, ...],
+    published_pypi: set[tuple[str, str]],
+    published_npm: set[tuple[str, str]],
+    verify_existing: bool,
+    apply: bool,
+) -> tuple[int, int, int]:
+    dispatched = 0
+    active = 0
+    complete = 0
+    for target in targets:
+        wheel_exists = (target.distribution, target.version) in published_pypi
+        npm_exists = target.webui_package is None or (
+            target.webui_package,
+            target.version,
+        ) in published_npm
+        if wheel_exists and npm_exists and not verify_existing:
+            complete += 1
+            print(f"complete {target.repository}:{target.tag}")
+            continue
+        if _has_active_run(client, owner=owner, repository=target.repository):
+            active += 1
+            print(f"active {target.repository}:{target.tag}")
+            continue
+
+        action = "dispatching" if apply else "would dispatch"
+        print(
+            f"{action} {target.repository}:{target.tag} "
+            f"(wheel={'present' if wheel_exists else 'missing'}, "
+            f"webui={'present' if npm_exists else 'missing'})"
+        )
+        if apply:
+            client.request_json(
+                "POST",
+                repo_path(
+                    owner,
+                    target.repository,
+                    f"/actions/workflows/{quote_path(WORKFLOW_ID)}/dispatches",
+                ),
+                body={"ref": "main", "inputs": {"release_tag": target.tag}},
+            )
+        dispatched += 1
+    return dispatched, active, complete
+
+
+def main() -> int:
+    args = build_parser().parse_args()
+    try:
+        load_dotenv(args.env_file)
+        token = require_token()
+        targets = package_targets()
+        selected = set(args.repository)
+        if selected:
+            known = {target.repository for target in targets}
+            unknown = sorted(selected - known)
+            if unknown:
+                raise ValueError(
+                    "Unknown meta-package repositories: " + ", ".join(unknown)
+                )
+            targets = tuple(
+                target for target in targets if target.repository in selected
+            )
+        missing_tags = [
+            f"{target.repository}:{target.tag}"
+            for target in targets
+            if not target.tag_exists
+        ]
+        if missing_tags:
+            raise ValueError(
+                "Meta-package release tags are missing: " + ", ".join(missing_tags)
+            )
+        target = RepoTarget(base_url=args.url, owner=args.owner, repo="govoplan")
+        with GiteaClient(target, token) as client:
+            secrets = client.request_json(
+                "GET", org_path(args.owner, "/actions/secrets"), query={"limit": 50}
+            )
+            secret_names = {
+                str(item.get("name") or "")
+                for item in secrets
+                if isinstance(item, dict)
+            }
+            required = {"GOVOPLAN_PACKAGE_USERNAME", "GOVOPLAN_PACKAGE_TOKEN"}
+            if not required <= secret_names:
+                raise ValueError(
+                    "Organization package publisher secrets are not configured"
+                )
+            published_pypi = _published_packages(
+                client, owner=args.owner, package_type="pypi"
+            )
+            published_npm = _published_packages(
+                client, owner=args.owner, package_type="npm"
+            )
+            counts = dispatch(
+                client,
+                owner=args.owner,
+                targets=targets,
+                published_pypi=published_pypi,
+                published_npm=published_npm,
+                verify_existing=args.verify_existing,
+                apply=args.apply,
+            )
+        action = "dispatched" if args.apply else "planned"
+        print(
+            f"Package set {action}: {counts[0]}; active: {counts[1]}; "
+            f"already complete: {counts[2]}."
+        )
+        return 0
+    except (GiteaError, OSError, ValueError, json.JSONDecodeError) as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())