music-assistant-server

1"""
2Support immutable-safe GitHub release workflows.
3
4The command-line interface emits values to ``$GITHUB_OUTPUT`` when requested so
5the release workflows can keep version and artifact validation in tested Python.
6"""
7
8# ruff: noqa: T201
9
10from __future__ import annotations
11
12import argparse
13import hashlib
14import json
15import re
16import subprocess
17import sys
18from collections.abc import Mapping
19from dataclasses import dataclass
20from datetime import UTC, datetime
21from pathlib import Path
22from typing import Any
23
24CHANNEL_PATTERNS = {
25    "stable": re.compile(r"^(?P<base>\d+\.\d+\.\d+)$"),
26    "beta": re.compile(r"^(?P<base>\d+\.\d+\.\d+)b(?P<number>\d+)$"),
27    "rc": re.compile(r"^(?P<base>\d+\.\d+\.\d+)rc(?P<number>\d+)$"),
28    "nightly": re.compile(r"^(?P<base>\d+\.\d+\.\d+)\.dev(?P<number>\d+)$"),
29}
30CHANNEL_BRANCHES = {
31    "stable": "stable",
32    "rc": "stable",
33    "beta": "dev",
34    "nightly": "dev",
35}
36SHA256_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$")
37OCI_REVISION_ANNOTATION = "org.opencontainers.image.revision"
38OCI_WHEEL_ANNOTATION = "io.music-assistant.wheel.sha256"
39FRONTEND_VERSION_PATTERN = re.compile(r"^(?P<base>\d+(?:\.\d+)+)(?:\.post(?P<post>\d+))?$")
40
41
42class ReleaseWorkflowError(RuntimeError):
43    """Report an unsafe or inconsistent release state."""
44
45
46@dataclass(frozen=True)
47class AutoRelease:
48    """Describe the next automatic release decision."""
49
50    version: str
51    should_release: bool
52    previous_tag: str | None
53    commits_since: int
54
55
56@dataclass(frozen=True)
57class Asset:
58    """Describe one exact release asset."""
59
60    name: str
61    size: int
62    sha256: str
63
64
65class GitRepository:
66    """Read release tags and commit relationships from a Git repository."""
67
68    def __init__(self, path: Path) -> None:
69        """
70        Initialize a repository reader.
71
72        :param path: Path to the Git worktree.
73        """
74        self.path = path
75
76    def resolve_commit(self, ref: str) -> str:
77        """
78        Resolve a Git ref to its commit SHA.
79
80        :param ref: Commit, branch, or tag to resolve.
81        """
82        return self._run("rev-parse", f"{ref}^{{commit}}")
83
84    def tags(self, pattern: re.Pattern[str]) -> list[str]:
85        """
86        Return version-sorted tags matching a release pattern.
87
88        :param pattern: Full tag-name pattern to match.
89        """
90        output = self._run("tag", "--sort=-version:refname")
91        return [tag for tag in output.splitlines() if pattern.fullmatch(tag)]
92
93    def all_tags(self) -> set[str]:
94        """Return all tag names."""
95        return set(self._run("tag").splitlines())
96
97    def commits_since(self, tag: str | None, source_sha: str, *, allow_diverged: bool) -> int:
98        """
99        Count source commits after a previous release tag.
100
101        :param tag: Previous release tag, or ``None`` for the full source history.
102        :param source_sha: Exact release source commit.
103        :param allow_diverged: Use the merge base when the tag is on a diverged branch.
104        """
105        if tag is None:
106            return int(self._run("rev-list", "--count", source_sha))
107
108        result = self._run_result("merge-base", "--is-ancestor", tag, source_sha)
109        if result.returncode == 0:
110            base = tag
111        elif result.returncode == 1 and allow_diverged:
112            base = self._run("merge-base", tag, source_sha)
113        elif result.returncode == 1:
114            msg = f"Previous {tag} is not an ancestor of source commit {source_sha}"
115            raise ReleaseWorkflowError(msg)
116        else:
117            raise ReleaseWorkflowError(result.stderr.strip() or f"Unable to compare {tag}")
118        return int(self._run("rev-list", "--count", f"{base}..{source_sha}"))
119
120    def _run(self, *args: str) -> str:
121        result = self._run_result(*args)
122        if result.returncode:
123            raise ReleaseWorkflowError(result.stderr.strip() or f"git {' '.join(args)} failed")
124        return result.stdout.strip()
125
126    def _run_result(self, *args: str) -> subprocess.CompletedProcess[str]:
127        return subprocess.run(  # noqa: S603
128            ["git", "-C", str(self.path), *args],  # noqa: S607
129            check=False,
130            capture_output=True,
131            text=True,
132        )
133
134
135def channel_branch(channel: str) -> str:
136    """
137    Return the source branch for a release channel.
138
139    :param channel: Release channel name.
140    """
141    try:
142        return CHANNEL_BRANCHES[channel]
143    except KeyError as err:
144        raise ReleaseWorkflowError(f"Unsupported release channel: {channel}") from err
145
146
147def validate_version(channel: str, version: str) -> None:
148    """
149    Validate a version against its release channel.
150
151    :param channel: Release channel name.
152    :param version: Version to validate.
153    """
154    try:
155        pattern = CHANNEL_PATTERNS[channel]
156    except KeyError as err:
157        raise ReleaseWorkflowError(f"Unsupported release channel: {channel}") from err
158    if not pattern.fullmatch(version):
159        raise ReleaseWorkflowError(f"Version {version} is invalid for the {channel} channel")
160
161
162def determine_auto_release(
163    repository: GitRepository,
164    channel: str,
165    source_sha: str,
166    *,
167    now: datetime | None = None,
168) -> AutoRelease:
169    """
170    Calculate the next channel version from Git tags and commit ancestry.
171
172    :param repository: Source Git repository.
173    :param channel: Release channel name.
174    :param source_sha: Exact release source commit.
175    :param now: Timestamp used for nightly versions.
176    """
177    source_sha = repository.resolve_commit(source_sha)
178    stable_tags = repository.tags(CHANNEL_PATTERNS["stable"])
179    beta_tags = repository.tags(CHANNEL_PATTERNS["beta"])
180    rc_tags = repository.tags(CHANNEL_PATTERNS["rc"])
181    nightly_tags = repository.tags(CHANNEL_PATTERNS["nightly"])
182    all_tags = repository.all_tags()
183
184    latest_stable = _first(stable_tags)
185    latest_beta = _first(beta_tags)
186    latest_rc = _first(rc_tags)
187    latest_nightly = _first(nightly_tags)
188
189    if channel == "stable":
190        previous_tag = latest_stable
191        commits_since = repository.commits_since(previous_tag, source_sha, allow_diverged=True)
192        version = _next_stable(latest_stable)
193    elif channel == "beta":
194        previous_tag = latest_beta
195        commits_since = repository.commits_since(previous_tag, source_sha, allow_diverged=False)
196        version = _next_beta(latest_beta, latest_stable)
197    elif channel == "rc":
198        version = _next_rc(latest_rc, latest_beta)
199        version_match = CHANNEL_PATTERNS["rc"].fullmatch(version)
200        assert version_match is not None
201        rc_base = version_match.group("base")
202        latest_rc_match = CHANNEL_PATTERNS["rc"].fullmatch(latest_rc or "")
203        if latest_rc_match and latest_rc_match.group("base") == rc_base:
204            previous_tag = latest_rc
205        else:
206            previous_tag = latest_beta
207        commits_since = repository.commits_since(previous_tag, source_sha, allow_diverged=False)
208    elif channel == "nightly":
209        previous_tag = latest_nightly
210        commits_since = repository.commits_since(previous_tag, source_sha, allow_diverged=False)
211        version = _next_nightly(
212            latest_nightly,
213            latest_stable,
214            now or datetime.now(UTC),
215        )
216    else:
217        raise ReleaseWorkflowError(f"Unsupported release channel: {channel}")
218
219    version = _ensure_unique(version, channel, all_tags)
220    return AutoRelease(
221        version=version,
222        should_release=channel != "nightly" or commits_since >= 2,
223        previous_tag=previous_tag,
224        commits_since=commits_since,
225    )
226
227
228def previous_release_tag(
229    repository: GitRepository,
230    channel: str,
231    version: str,
232) -> str | None:
233    """
234    Return the previous tag used as the release-notes base.
235
236    :param repository: Source Git repository.
237    :param channel: Release channel name.
238    :param version: Version currently being prepared.
239    """
240    validate_version(channel, version)
241    if channel == "stable":
242        tags = repository.tags(CHANNEL_PATTERNS["stable"])
243    elif channel == "beta":
244        combined_pattern = re.compile(r"^\d+\.\d+\.\d+(?:b|rc)\d+$")
245        tags = repository.tags(combined_pattern)
246    elif channel == "nightly":
247        tags = repository.tags(CHANNEL_PATTERNS["nightly"])
248    else:
249        match = CHANNEL_PATTERNS["rc"].fullmatch(version)
250        assert match is not None
251        rc_number = int(match.group("number"))
252        if rc_number > 1:
253            previous_rc = f"{match.group('base')}rc{rc_number - 1}"
254            if previous_rc in repository.all_tags():
255                return previous_rc
256        tags = repository.tags(CHANNEL_PATTERNS["beta"])
257    return next((tag for tag in tags if tag != version), None)
258
259
260def is_current_release(
261    repository: GitRepository,
262    channel: str,
263    version: str,
264) -> tuple[bool, str | None]:
265    """
266    Check whether a release is the newest version for its rolling channel.
267
268    :param repository: Source Git repository.
269    :param channel: Release channel name.
270    :param version: Release version to compare.
271    """
272    validate_version(channel, version)
273    if channel == "stable":
274        pattern = CHANNEL_PATTERNS["stable"]
275    elif channel == "nightly":
276        pattern = CHANNEL_PATTERNS["nightly"]
277    else:
278        pattern = re.compile(r"^\d+\.\d+\.\d+(?:b|rc)\d+$")
279    tags = repository.tags(pattern)
280    latest_tag = max(tags, key=_release_version_key, default=None)
281    return latest_tag is None or _release_version_key(version) >= _release_version_key(
282        latest_tag
283    ), latest_tag
284
285
286def compare_release_versions(current: str, requested: str) -> str:
287    """
288    Compare a deployed release version with a requested release version.
289
290    :param current: Version currently deployed to an alias.
291    :param requested: Version requested for promotion.
292    """
293    return _version_relation(
294        _release_version_key(current),
295        _release_version_key(requested),
296    )
297
298
299def compare_frontend_versions(current: str, requested: str) -> str:
300    """
301    Compare a deployed frontend version with a requested frontend version.
302
303    :param current: Frontend version currently configured for the channel.
304    :param requested: Frontend version requested for deployment.
305    """
306    current_base, current_post = _frontend_version_parts(current)
307    requested_base, requested_post = _frontend_version_parts(requested)
308    width = max(len(current_base), len(requested_base))
309    current_key = (*current_base, *((0,) * (width - len(current_base))), current_post)
310    requested_key = (*requested_base, *((0,) * (width - len(requested_base))), requested_post)
311    return _version_relation(current_key, requested_key)
312
313
314def select_release(release_pages: object, version: str) -> dict[str, Any] | None:
315    """
316    Return the only release whose tag exactly matches a version.
317
318    :param release_pages: Paginated responses from the GitHub releases API.
319    :param version: Exact release tag to select.
320    """
321    if not isinstance(release_pages, list):
322        raise ReleaseWorkflowError("GitHub releases response does not contain pages")
323
324    matches: list[dict[str, Any]] = []
325    for page in release_pages:
326        if not isinstance(page, list):
327            raise ReleaseWorkflowError("GitHub releases response contains an invalid page")
328        for release in page:
329            if not isinstance(release, dict):
330                raise ReleaseWorkflowError("GitHub releases response contains invalid metadata")
331            if release.get("tag_name") == version:
332                matches.append(release)
333
334    if len(matches) > 1:
335        release_ids = ", ".join(str(release.get("id", "unknown")) for release in matches)
336        raise ReleaseWorkflowError(f"Multiple releases match exact tag {version}: {release_ids}")
337    if not matches:
338        return None
339
340    release = matches[0]
341    release_id = release.get("id")
342    if not isinstance(release_id, int) or isinstance(release_id, bool) or release_id <= 0:
343        raise ReleaseWorkflowError(f"Release {version} does not contain a valid id")
344    return release
345
346
347def inspect_assets(
348    version: str,
349    *,
350    directory: Path | None = None,
351    release_json: Path | None = None,
352) -> tuple[Asset, Asset]:
353    """
354    Validate and describe the exact wheel and source distribution.
355
356    :param version: Release version.
357    :param directory: Optional directory containing downloaded assets.
358    :param release_json: Optional GitHub release API response.
359    """
360    expected_names = _expected_asset_names(version)
361    api_assets: dict[str, Asset] | None = None
362    local_assets: dict[str, Asset] | None = None
363
364    if release_json is not None:
365        release = json.loads(release_json.read_text(encoding="utf-8"))
366        raw_assets = release.get("assets")
367        if not isinstance(raw_assets, list):
368            raise ReleaseWorkflowError("Release response does not contain an asset list")
369        if len(raw_assets) != len(expected_names):
370            raise ReleaseWorkflowError(
371                f"Release must contain exactly these assets: {', '.join(expected_names)}"
372            )
373        api_assets = {}
374        for raw_asset in raw_assets:
375            if not isinstance(raw_asset, dict):
376                raise ReleaseWorkflowError("Release response contains invalid asset metadata")
377            name = raw_asset.get("name")
378            size = raw_asset.get("size")
379            state = raw_asset.get("state")
380            digest = raw_asset.get("digest")
381            if (
382                not isinstance(name, str)
383                or not isinstance(size, int)
384                or state != "uploaded"
385                or not isinstance(digest, str)
386                or not SHA256_PATTERN.fullmatch(digest)
387                or size <= 0
388            ):
389                raise ReleaseWorkflowError(f"Release asset metadata is invalid: {name!r}")
390            if name in api_assets:
391                raise ReleaseWorkflowError(f"Release contains duplicate asset {name}")
392            api_assets[name] = Asset(name, size, digest.removeprefix("sha256:"))
393        if set(api_assets) != set(expected_names):
394            raise ReleaseWorkflowError(
395                f"Release must contain exactly these assets: {', '.join(expected_names)}"
396            )
397
398    if directory is not None:
399        files = sorted(path for path in directory.iterdir() if path.is_file())
400        if {path.name for path in files} != set(expected_names):
401            raise ReleaseWorkflowError(
402                f"Asset directory must contain exactly: {', '.join(expected_names)}"
403            )
404        local_assets = {
405            path.name: Asset(path.name, path.stat().st_size, _sha256(path)) for path in files
406        }
407        if any(asset.size <= 0 for asset in local_assets.values()):
408            raise ReleaseWorkflowError("Release assets must not be empty")
409
410    assets = local_assets or api_assets
411    if assets is None:
412        raise ReleaseWorkflowError("An asset directory or release response is required")
413    if local_assets is not None and api_assets is not None and local_assets != api_assets:
414        raise ReleaseWorkflowError("Downloaded assets do not match GitHub release metadata")
415    return assets[expected_names[0]], assets[expected_names[1]]
416
417
418def verify_oci_manifest(
419    manifest: dict[str, Any],
420    source_sha: str,
421    wheel_sha256: str,
422) -> tuple[str, list[str]]:
423    """
424    Validate the exact multi-platform image manifest and provenance annotations.
425
426    :param manifest: Formatted Buildx manifest JSON.
427    :param source_sha: Exact source commit expected in the image annotations.
428    :param wheel_sha256: Wheel digest expected in the image annotations.
429    """
430    digest = manifest.get("digest")
431    if not isinstance(digest, str) or not SHA256_PATTERN.fullmatch(digest):
432        raise ReleaseWorkflowError("OCI manifest does not contain a valid digest")
433    annotations = manifest.get("annotations")
434    if not isinstance(annotations, dict):
435        raise ReleaseWorkflowError("OCI manifest does not contain provenance annotations")
436    if annotations.get(OCI_REVISION_ANNOTATION) != source_sha:
437        raise ReleaseWorkflowError("OCI manifest source SHA does not match the release")
438    if annotations.get(OCI_WHEEL_ANNOTATION) != wheel_sha256:
439        raise ReleaseWorkflowError("OCI manifest wheel digest does not match the release")
440
441    raw_manifests = manifest.get("manifests")
442    if not isinstance(raw_manifests, list):
443        raise ReleaseWorkflowError("OCI image is not a multi-platform manifest")
444    runtime_manifests: dict[str, str] = {}
445    for item in raw_manifests:
446        if not isinstance(item, dict):
447            raise ReleaseWorkflowError("OCI manifest contains an invalid descriptor")
448        platform = item.get("platform")
449        child_digest = item.get("digest")
450        if not isinstance(platform, dict) or not isinstance(child_digest, str):
451            raise ReleaseWorkflowError("OCI manifest descriptor is incomplete")
452        operating_system = platform.get("os")
453        architecture = platform.get("architecture")
454        if operating_system == "unknown" and architecture == "unknown":
455            continue
456        if not SHA256_PATTERN.fullmatch(child_digest):
457            raise ReleaseWorkflowError("OCI child manifest digest is invalid")
458        platform_name = f"{operating_system}/{architecture}"
459        if platform_name in runtime_manifests:
460            raise ReleaseWorkflowError(f"OCI image contains duplicate platform {platform_name}")
461        runtime_manifests[platform_name] = child_digest
462    expected_platforms = {"linux/amd64", "linux/arm64"}
463    if set(runtime_manifests) != expected_platforms:
464        raise ReleaseWorkflowError("OCI image must contain linux/amd64 and linux/arm64")
465    return digest, [runtime_manifests[platform] for platform in sorted(expected_platforms)]
466
467
468def update_addon_release(
469    config_path: Path,
470    changelog_path: Path,
471    *,
472    version: str,
473    release_date: str,
474    notes: str,
475    retain: int = 3,
476) -> None:
477    """
478    Update an add-on channel to one canonical release entry.
479
480    :param config_path: Add-on ``config.yaml`` path.
481    :param changelog_path: Add-on ``CHANGELOG.md`` path.
482    :param version: Server version and image tag.
483    :param release_date: Canonical release date in ``DD.MM.YYYY`` format.
484    :param notes: Published server release notes.
485    :param retain: Total number of distinct changelog releases to retain.
486    """
487    if retain < 1:
488        raise ReleaseWorkflowError("At least one changelog release must be retained")
489    config = config_path.read_text(encoding="utf-8")
490    config, replacements = re.subn(
491        r"^version: .+$",
492        f"version: {version}",
493        config,
494        count=1,
495        flags=re.MULTILINE,
496    )
497    if replacements != 1:
498        raise ReleaseWorkflowError(f"Expected one version field in {config_path}")
499    config_path.write_text(config, encoding="utf-8")
500
501    existing = changelog_path.read_text(encoding="utf-8") if changelog_path.exists() else ""
502    blocks = _changelog_blocks(existing)
503    previous_blocks: list[str] = []
504    seen_versions = {version}
505    for block_version, block in blocks:
506        if block_version in seen_versions:
507            continue
508        seen_versions.add(block_version)
509        previous_blocks.append(block)
510    new_block = f"# [{version}] - {release_date}\n\n{notes.strip()}"
511    content = "\n\n\n".join([new_block, *previous_blocks[: retain - 1]]) + "\n"
512    changelog_path.write_text(content, encoding="utf-8")
513
514
515def _next_stable(latest_stable: str | None) -> str:
516    if latest_stable is None:
517        return "0.1.0"
518    major, minor, patch = _version_tuple(latest_stable)
519    return f"{major}.{minor}.{patch + 1}"
520
521
522def _next_beta(latest_beta: str | None, latest_stable: str | None) -> str:
523    if latest_beta is not None:
524        match = CHANNEL_PATTERNS["beta"].fullmatch(latest_beta)
525        assert match is not None
526        beta_base = _version_tuple(match.group("base"))
527        if latest_stable is not None:
528            stable = _version_tuple(latest_stable)
529            if stable[:2] >= beta_base[:2]:
530                return f"{stable[0]}.{stable[1] + 1}.0b1"
531        return f"{match.group('base')}b{int(match.group('number')) + 1}"
532    if latest_stable is not None:
533        major, minor, _patch = _version_tuple(latest_stable)
534        return f"{major}.{minor + 1}.0b1"
535    return "0.1.0b1"
536
537
538def _next_rc(latest_rc: str | None, latest_beta: str | None) -> str:
539    if latest_beta is None:
540        return "0.1.0rc1"
541    beta_match = CHANNEL_PATTERNS["beta"].fullmatch(latest_beta)
542    assert beta_match is not None
543    beta_base = beta_match.group("base")
544    if latest_rc is not None:
545        rc_match = CHANNEL_PATTERNS["rc"].fullmatch(latest_rc)
546        assert rc_match is not None
547        if rc_match.group("base") == beta_base:
548            return f"{beta_base}rc{int(rc_match.group('number')) + 1}"
549    return f"{beta_base}rc1"
550
551
552def _next_nightly(
553    latest_nightly: str | None,
554    latest_stable: str | None,
555    now: datetime,
556) -> str:
557    if latest_stable is None:
558        base = "0.1.0"
559    else:
560        major, minor, _patch = _version_tuple(latest_stable)
561        base = f"{major}.{minor + 1}.0"
562    serial = int(now.astimezone(UTC).strftime("%Y%m%d%H"))
563    if latest_nightly is not None:
564        match = CHANNEL_PATTERNS["nightly"].fullmatch(latest_nightly)
565        assert match is not None
566        if match.group("base") == base:
567            serial = max(serial, int(match.group("number")) + 1)
568    return f"{base}.dev{serial}"
569
570
571def _ensure_unique(version: str, channel: str, tags: set[str]) -> str:
572    while version in tags:
573        match = CHANNEL_PATTERNS[channel].fullmatch(version)
574        assert match is not None
575        base = match.group("base")
576        if channel == "stable":
577            major, minor, patch = _version_tuple(base)
578            version = f"{major}.{minor}.{patch + 1}"
579        elif channel == "beta":
580            version = f"{base}b{int(match.group('number')) + 1}"
581        elif channel == "rc":
582            version = f"{base}rc{int(match.group('number')) + 1}"
583        else:
584            version = f"{base}.dev{int(match.group('number')) + 1}"
585    return version
586
587
588def _version_tuple(version: str) -> tuple[int, int, int]:
589    match = CHANNEL_PATTERNS["stable"].fullmatch(version)
590    if match is None:
591        raise ReleaseWorkflowError(f"Invalid base version: {version}")
592    return tuple(int(part) for part in version.split("."))  # type: ignore[return-value]
593
594
595def _release_version_key(version: str) -> tuple[int, int, int, int, int]:
596    for stage, channel in enumerate(("nightly", "beta", "rc", "stable")):
597        match = CHANNEL_PATTERNS[channel].fullmatch(version)
598        if match is None:
599            continue
600        major, minor, patch = _version_tuple(match.group("base"))
601        number = int(match.groupdict().get("number") or 0)
602        return major, minor, patch, stage, number
603    raise ReleaseWorkflowError(f"Unsupported release version: {version}")
604
605
606def _frontend_version_parts(version: str) -> tuple[tuple[int, ...], int]:
607    match = FRONTEND_VERSION_PATTERN.fullmatch(version)
608    if match is None:
609        raise ReleaseWorkflowError(f"Unsupported frontend version: {version}")
610    base = tuple(int(part) for part in match.group("base").split("."))
611    post = int(match.group("post")) if match.group("post") is not None else -1
612    return base, post
613
614
615def _version_relation(current: tuple[int, ...], requested: tuple[int, ...]) -> str:
616    if current > requested:
617        return "newer"
618    if current < requested:
619        return "older"
620    return "equal"
621
622
623def _first(values: list[str]) -> str | None:
624    return values[0] if values else None
625
626
627def _expected_asset_names(version: str) -> tuple[str, str]:
628    return (
629        f"music_assistant-{version}-py3-none-any.whl",
630        f"music_assistant-{version}.tar.gz",
631    )
632
633
634def _sha256(path: Path) -> str:
635    digest = hashlib.sha256()
636    with path.open("rb") as file_handle:
637        for chunk in iter(lambda: file_handle.read(1024 * 1024), b""):
638            digest.update(chunk)
639    return digest.hexdigest()
640
641
642def _changelog_blocks(content: str) -> list[tuple[str, str]]:
643    header_pattern = re.compile(r"(?m)^# \[([^\]]+)\] - .+$")
644    matches = list(header_pattern.finditer(content))
645    if not matches:
646        if content.strip():
647            raise ReleaseWorkflowError("Existing changelog has no release headings")
648        return []
649    if content[: matches[0].start()].strip():
650        raise ReleaseWorkflowError("Existing changelog has content before its first release")
651    blocks: list[tuple[str, str]] = []
652    for index, match in enumerate(matches):
653        end = matches[index + 1].start() if index + 1 < len(matches) else len(content)
654        blocks.append((match.group(1), content[match.start() : end].strip()))
655    return blocks
656
657
658def _write_outputs(
659    values: Mapping[str, str | int | bool | None],
660    output_path: Path | None,
661) -> None:
662    lines = [
663        f"{key}={str(value).lower() if isinstance(value, bool) else '' if value is None else value}"
664        for key, value in values.items()
665    ]
666    if output_path is not None:
667        with output_path.open("a", encoding="utf-8") as file_handle:
668            file_handle.write("\n".join(lines) + "\n")
669    print("\n".join(lines))
670
671
672def _asset_outputs(assets: tuple[Asset, Asset]) -> dict[str, str | int]:
673    wheel, source = assets
674    return {
675        "wheel_name": wheel.name,
676        "wheel_size": wheel.size,
677        "wheel_sha256": wheel.sha256,
678        "sdist_name": source.name,
679        "sdist_size": source.size,
680        "sdist_sha256": source.sha256,
681    }
682
683
684def _configure_release_parser(parser: argparse.ArgumentParser) -> None:
685    parser.add_argument("--version", required=True)
686    parser.add_argument("--releases-json", type=Path, required=True)
687    parser.add_argument("--release-json", type=Path, required=True)
688    parser.add_argument("--github-output", type=Path)
689
690
691def _build_parser() -> argparse.ArgumentParser:
692    parser = argparse.ArgumentParser(description=__doc__)
693    subparsers = parser.add_subparsers(dest="command", required=True)
694
695    branch_parser = subparsers.add_parser("branch")
696    branch_parser.add_argument("--channel", required=True)
697    branch_parser.add_argument("--github-output", type=Path)
698
699    validate_parser = subparsers.add_parser("validate-version")
700    validate_parser.add_argument("--channel", required=True)
701    validate_parser.add_argument("--version", required=True)
702
703    auto_parser = subparsers.add_parser("auto-version")
704    auto_parser.add_argument("--channel", required=True)
705    auto_parser.add_argument("--source-sha", required=True)
706    auto_parser.add_argument("--repository", type=Path, default=Path.cwd())
707    auto_parser.add_argument("--timestamp")
708    auto_parser.add_argument("--github-output", type=Path)
709
710    previous_parser = subparsers.add_parser("previous-tag")
711    previous_parser.add_argument("--channel", required=True)
712    previous_parser.add_argument("--version", required=True)
713    previous_parser.add_argument("--repository", type=Path, default=Path.cwd())
714    previous_parser.add_argument("--github-output", type=Path)
715
716    current_parser = subparsers.add_parser("current-version")
717    current_parser.add_argument("--channel", required=True)
718    current_parser.add_argument("--version", required=True)
719    current_parser.add_argument("--repository", type=Path, default=Path.cwd())
720    current_parser.add_argument("--github-output", type=Path)
721
722    release_order_parser = subparsers.add_parser("compare-release-versions")
723    release_order_parser.add_argument("--current", required=True)
724    release_order_parser.add_argument("--requested", required=True)
725    release_order_parser.add_argument("--github-output", type=Path)
726
727    frontend_order_parser = subparsers.add_parser("compare-frontend-versions")
728    frontend_order_parser.add_argument("--current", required=True)
729    frontend_order_parser.add_argument("--requested", required=True)
730    frontend_order_parser.add_argument("--github-output", type=Path)
731
732    _configure_release_parser(subparsers.add_parser("select-release"))
733
734    assets_parser = subparsers.add_parser("assets")
735    assets_parser.add_argument("--version", required=True)
736    assets_parser.add_argument("--directory", type=Path)
737    assets_parser.add_argument("--release-json", type=Path)
738    assets_parser.add_argument("--github-output", type=Path)
739
740    manifest_parser = subparsers.add_parser("verify-manifest")
741    manifest_parser.add_argument("--manifest-json", type=Path, required=True)
742    manifest_parser.add_argument("--source-sha", required=True)
743    manifest_parser.add_argument("--wheel-sha256", required=True)
744    manifest_parser.add_argument("--github-output", type=Path)
745
746    addon_parser = subparsers.add_parser("update-addon")
747    addon_parser.add_argument("--config", type=Path, required=True)
748    addon_parser.add_argument("--changelog", type=Path, required=True)
749    addon_parser.add_argument("--version", required=True)
750    addon_parser.add_argument("--release-date", required=True)
751    addon_parser.add_argument("--notes", type=Path, required=True)
752    addon_parser.add_argument("--retain", type=int, default=3)
753    return parser
754
755
756def main() -> int:
757    """Run a release workflow helper command."""
758    args = _build_parser().parse_args()
759    try:
760        if args.command == "branch":
761            _write_outputs({"branch": channel_branch(args.channel)}, args.github_output)
762        elif args.command == "validate-version":
763            validate_version(args.channel, args.version)
764        elif args.command == "auto-version":
765            timestamp = datetime.fromisoformat(args.timestamp) if args.timestamp else None
766            decision = determine_auto_release(
767                GitRepository(args.repository),
768                args.channel,
769                args.source_sha,
770                now=timestamp,
771            )
772            _write_outputs(
773                {
774                    "version": decision.version,
775                    "should_release": decision.should_release,
776                    "previous_tag": decision.previous_tag,
777                    "commits_since": decision.commits_since,
778                },
779                args.github_output,
780            )
781        elif args.command == "previous-tag":
782            tag = previous_release_tag(
783                GitRepository(args.repository),
784                args.channel,
785                args.version,
786            )
787            _write_outputs({"previous_tag": tag}, args.github_output)
788        elif args.command == "current-version":
789            is_current, latest_tag = is_current_release(
790                GitRepository(args.repository),
791                args.channel,
792                args.version,
793            )
794            _write_outputs(
795                {
796                    "is_current": is_current,
797                    "latest_channel_tag": latest_tag,
798                },
799                args.github_output,
800            )
801        elif args.command == "compare-release-versions":
802            _write_outputs(
803                {
804                    "current_relation": compare_release_versions(
805                        args.current,
806                        args.requested,
807                    )
808                },
809                args.github_output,
810            )
811        elif args.command == "compare-frontend-versions":
812            _write_outputs(
813                {
814                    "current_relation": compare_frontend_versions(
815                        args.current,
816                        args.requested,
817                    )
818                },
819                args.github_output,
820            )
821        elif args.command == "select-release":
822            release_pages = json.loads(args.releases_json.read_text(encoding="utf-8"))
823            release = select_release(release_pages, args.version)
824            if release is None:
825                args.release_json.unlink(missing_ok=True)
826                _write_outputs(
827                    {"release_exists": False, "release_id": None},
828                    args.github_output,
829                )
830            else:
831                args.release_json.write_text(
832                    json.dumps(release, indent=2) + "\n",
833                    encoding="utf-8",
834                )
835                _write_outputs(
836                    {"release_exists": True, "release_id": release["id"]},
837                    args.github_output,
838                )
839        elif args.command == "assets":
840            assets = inspect_assets(
841                args.version,
842                directory=args.directory,
843                release_json=args.release_json,
844            )
845            _write_outputs(_asset_outputs(assets), args.github_output)
846        elif args.command == "verify-manifest":
847            manifest = json.loads(args.manifest_json.read_text(encoding="utf-8"))
848            digest, runtime_digests = verify_oci_manifest(
849                manifest,
850                args.source_sha,
851                args.wheel_sha256,
852            )
853            _write_outputs(
854                {
855                    "digest": digest,
856                    "runtime_digests": " ".join(runtime_digests),
857                },
858                args.github_output,
859            )
860        elif args.command == "update-addon":
861            update_addon_release(
862                args.config,
863                args.changelog,
864                version=args.version,
865                release_date=args.release_date,
866                notes=args.notes.read_text(encoding="utf-8"),
867                retain=args.retain,
868            )
869        return 0
870    except (OSError, ReleaseWorkflowError, ValueError, json.JSONDecodeError) as err:
871        print(f"ERROR: {err}", file=sys.stderr)
872        return 1
873
874
875if __name__ == "__main__":
876    sys.exit(main())
877