grm/scripts/auto_merge.py

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

Waits for CI checks to complete before attempting the merge.

PR title format:  ``GRM-N: <vikunja task title>``
Merge commit format: ``GRM-N <conventional commit message>``

The conventional commit message is taken from the first commit on the PR
branch (the branch HEAD).  This allows the PR title to be a human-friendly
Vikunja task title while the squashed commit follows conventional commits.

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

import os
import re
import time
from typing import Any

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

# PR title: GRM-N: <vikunja task title>
PR_TITLE_RE = re.compile(r"^GRM-\d+:\s+.+")

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, task_id: str) -> None:
    """Raise ClickException if PR title does not follow the required format.

    Expected: ``GRM-N: <vikunja task title>``
    """
    if not PR_TITLE_RE.match(pr_title):
        raise click.ClickException(
            _(
                "Oops! PR title must follow format 'GRM-N: <task title>'.\n"
                "  Expected: {task_id}: <task title>\n"
                "  Got:      {pr_title}",
                task_id=task_id,
                pr_title=pr_title,
            )
        )
    if not pr_title.startswith(f"{task_id}:"):
        raise click.ClickException(
            _(
                "Oops! PR title task ID mismatch.\n"
                "  Branch task ID: {task_id}\n"
                "  PR title:       {pr_title}",
                task_id=task_id,
                pr_title=pr_title,
            )
        )


def extract_conventional_msg(commits: list[dict[str, Any]]) -> str:
    """Extract the conventional commit message from PR commits.

    Iterates commits in reverse order (newest first) to find the first
    message matching the conventional commit format.  Falls back to the
    newest commit message if none match.
    """
    for commit in reversed(commits):
        commit_info = commit.get("commit", {})
        message = str(commit_info.get("message", "") if isinstance(commit_info, dict) else "").split("\n")[0]
        if CONVENTIONAL_RE.match(message):
            return message
    # Fallback: use the newest commit's first line
    if commits:
        commit_info = commits[-1].get("commit", {})
        return str(commit_info.get("message", "") if isinstance(commit_info, dict) else "").split("\n")[0]
    return ""


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, task_id)

    # 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."))

    # Build merge title: GRM-N <conventional commit message>
    commits = client.get_pr_commits(pr_number)
    conv_msg = extract_conventional_msg(commits)
    if not conv_msg:
        raise click.ClickException(
            _("Could not extract conventional commit message from PR commits.")
        )
    merge_title = f"{task_id} {conv_msg}"

    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()