grm/scripts/auto_merge.py

#!/usr/bin/env python3
"""Auto-merge PR by extracting task ID from branch and validating PR title.

Waits for CI checks to complete before attempting the merge.

Usage:
  REPO_TOKEN=<token> python3 scripts/auto_merge.py <branch> <pr_title> <repo> <pr_number> [label_name]
"""

import os
import time

import click
from dotenv import load_dotenv  # pyright: ignore[reportMissingImports,reportUnknownVariableType]

from gitea_runner_manager.api_clients import GiteaClient
from gitea_runner_manager.config import CONVENTIONAL_RE, GITEA_API_URL, TASK_ID_RE
from gitea_runner_manager.exceptions import APIError
from gitea_runner_manager.i18n import _

READY_TO_MERGE = "ready-to-merge"
MAX_WAIT_SECONDS = 900  # 15 minutes
POLL_INTERVAL_SECONDS = 30

load_dotenv(override=True)


def extract_task_id(branch: str) -> str:
    """Extract GRM-N task identifier from branch name."""
    match = TASK_ID_RE.search(branch)
    return match.group(0) if match else ""


def validate_pr_title(pr_title: str) -> None:
    """Raise ClickException if PR title does not follow conventional commits."""
    if not CONVENTIONAL_RE.match(pr_title):
        raise click.ClickException(
            _(
                "Oops! PR title must follow conventional commit format.\n"
                "  Expected: <type>: <description>\n"
                "  Got:      {pr_title}",
                pr_title=pr_title,
            )
        )


def has_ready_to_merge_label(client: GiteaClient, pr_number: str) -> bool:
    """Check whether the PR has the ready-to-merge label via the API."""
    labels = client.get_pr_labels(pr_number)
    return any(label.get("name") == READY_TO_MERGE for label in labels)


def wait_for_ci(
    client: GiteaClient, sha: str, max_wait: int = MAX_WAIT_SECONDS, poll_interval: int = POLL_INTERVAL_SECONDS
) -> bool:
    """Poll commit statuses until all CI checks are complete (not pending).

    Returns True if all checks are successful, False if any failed or timed out.
    """
    elapsed = 0
    while elapsed < max_wait:
        statuses = client.get_commit_status(sha)
        if not statuses:
            click.echo(_("No CI checks reported yet, waiting..."))
            time.sleep(poll_interval)
            elapsed += poll_interval
            continue

        # Deduplicate by context — keep the latest status per context.
        latest: dict[str, dict[str, object]] = {}
        for s in statuses:
            ctx = s.get("context", "")
            if ctx not in latest or s.get("updated_at", "") > latest[ctx].get("updated_at", ""):
                latest[ctx] = s

        ci_statuses = {ctx: s for ctx, s in latest.items() if ctx.startswith("CI /")}
        if not ci_statuses:
            click.echo(_("No CI checks found yet, waiting..."))
            time.sleep(poll_interval)
            elapsed += poll_interval
            continue

        pending = [ctx for ctx, s in ci_statuses.items() if s.get("status") in ("pending", "waiting")]
        if not pending:
            # All CI checks are complete — check if they all succeeded.
            failed = [
                ctx
                for ctx, s in ci_statuses.items()
                if s.get("status") not in ("success", "ok")
            ]
            if failed:
                click.echo(
                    _("CI checks failed: {failed}", failed=", ".join(sorted(failed)))
                )
                return False
            click.echo(_("All CI checks passed."))
            return True

        click.echo(
            _(
                "Waiting for CI checks: {pending} ({elapsed}s elapsed)",
                pending=", ".join(sorted(pending)),
                elapsed=elapsed,
            )
        )
        time.sleep(poll_interval)
        elapsed += poll_interval

    click.echo(_("Timed out waiting for CI checks after {max_wait}s.", max_wait=max_wait))
    return False


@click.command()
@click.argument("branch")
@click.argument("pr_title")
@click.argument("repo")
@click.argument("pr_number")
@click.argument("label_name", required=False, default="")
def main(branch: str, pr_title: str, repo: str, pr_number: str, label_name: str) -> None:
    token = os.environ.get("REPO_TOKEN", "")
    if not token:
        raise click.ClickException(_("ERROR: REPO_TOKEN is not set."))

    owner, repo_name = repo.split("/")
    client = GiteaClient(GITEA_API_URL, token, owner, repo_name)

    # Gitea Actions may not populate github.event.label.name; fall back to API check.
    if label_name != READY_TO_MERGE and not has_ready_to_merge_label(client, pr_number):
        click.echo(_("Label '{label}' is not '{rtm}', skipping.", label=label_name, rtm=READY_TO_MERGE))
        return

    task_id = extract_task_id(branch)
    if not task_id:
        raise click.ClickException(
            _(
                "Oops! No task ID (GRM-N) found in branch name '{branch}'.",
                branch=branch,
            )
        )

    validate_pr_title(pr_title)

    # Wait for CI checks to complete before attempting merge.
    pr = client.get_pr(pr_number)
    sha = pr.get("head", {}).get("sha", "")
    if sha:
        click.echo(_("Waiting for CI checks on commit {sha}...", sha=sha[:8]))
        if not wait_for_ci(client, sha):
            raise click.ClickException(
                _("Cannot merge: CI checks did not pass. Please fix failing checks and re-label.")
            )
    else:
        click.echo(_("Warning: could not determine PR head SHA, proceeding without CI wait."))

    merge_title = f"{task_id}: {pr_title}"

    try:
        client.merge_pr(pr_number, merge_title)
    except APIError as e:
        raise click.ClickException(
            _(
                "Merge failed with HTTP {status}: {message}\n"
                "Please check the PR is ready and you have merge rights.",
                status=e.status,
                message=e.message,
            )
        ) from None

    click.echo(
        _(
            "Nice! PR #{pr_number} squash-merged with title: {merge_title}",
            pr_number=pr_number,
            merge_title=merge_title,
        )
    )


if __name__ == "__main__":  # pragma: no cover
    main()