grm/scripts/ci/auto_merge.py

#!/usr/bin/env python3
"""Auto-merge PR when all CI checks pass.

Runs as the final job in ci.yml. Reads the task ID from ``.taskid`` file
(falling back to branch name extraction for backwards compatibility),
validates the PR title, and squash-merges with a conventional commit
message prefixed by the task ID.

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

The conventional commit message is extracted from the PR commits.
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/ci/auto_merge.py <branch> <pr_title> <repo> <pr_number>
"""

import os
import re
import subprocess  # nosec B404
from pathlib import Path
from typing import Any

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

from gitea_runner_manager.api_clients import GiteaClient, VikunjaClient
from gitea_runner_manager.config import (
    CONVENTIONAL_RE,
    DEFAULT_PER_PAGE,
    GITEA_API_URL,
    TASK_ID_RE,
    VIKUNJA_API_URL,
    VIKUNJA_PROJECT_ID,
)
from gitea_runner_manager.exceptions import APIError
from scripts.i18n import _

TASKID_FILE = ".taskid"
PR_TITLE_RE = re.compile(r"^GRM-\d+:\s+.+")

load_dotenv(override=True)


def run_cmd(args: list[str], check: bool = True) -> subprocess.CompletedProcess[str]:
    """Run a command and return the completed process."""
    result = subprocess.run(args, capture_output=True, text=True, check=False)  # nosec B603
    if check and result.returncode != 0:
        raise click.ClickException(
            _(
                "Command failed ({cmd}): {stderr}",
                cmd=" ".join(args),
                stderr=result.stderr.strip() or result.stdout.strip(),
            )
        )
    return result


def read_taskid(branch: str) -> str:
    """Read task ID from .taskid file, falling back to branch name extraction.

    The .taskid file is a simple text file containing just the task ID
    (e.g., ``GRM-60``).  If the file doesn't exist, extract from the
    branch name as a backwards-compatibility fallback.
    """
    path = Path(TASKID_FILE)
    if path.exists():
        task_id = path.read_text(encoding="utf-8").strip()
        if task_id:
            return task_id
    # Fallback: extract from branch name
    match = TASK_ID_RE.search(branch)
    return match.group(0) if match else ""


def extract_task_id(branch: str) -> str:
    """Extract GRM-N task identifier from branch name (legacy fallback)."""
    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 get_vikunja_task_title(task_id: str) -> str:
    """Fetch the Vikunja task title for the given GRM-N identifier.

    Returns empty string if VIKUNJA_TOKEN is not set (local dev without token).
    Raises ClickException if the token is set but the task is not found.
    """
    token = os.environ.get("VIKUNJA_TOKEN", "")
    if not token:
        return ""
    client = VikunjaClient(VIKUNJA_API_URL, token)
    page = 1
    while True:
        tasks = client.list_project_tasks(VIKUNJA_PROJECT_ID, page=page, per_page=DEFAULT_PER_PAGE)
        if not tasks:
            break
        matches = [t for t in tasks if t.get("identifier") == task_id]
        if matches:
            return str(matches[0].get("title", ""))
        if len(tasks) < DEFAULT_PER_PAGE:
            break
        page += 1
    raise click.ClickException(
        _(
            "Could not find Vikunja task {task_id} in project {project_id}. "
            "Every PR must have a corresponding Vikunja task.",
            task_id=task_id,
            project_id=VIKUNJA_PROJECT_ID,
        )
    )


def validate_pr_title_matches_vikunja(pr_title: str, task_id: str) -> None:
    """Validate that PR title matches the Vikunja task title.

    Skips validation if VIKUNJA_TOKEN is not set (local dev).
    Raises ClickException if the task is not found or the title doesn't match.
    """
    vikunja_title = get_vikunja_task_title(task_id)
    if not vikunja_title:
        # VIKUNJA_TOKEN not set — skip validation (local dev)
        click.echo(_("Warning: VIKUNJA_TOKEN not set, skipping title match validation."))
        return
    expected = f"{task_id}: {vikunja_title}"
    if pr_title != expected:
        raise click.ClickException(
            _(
                "PR title does not match Vikunja task title.\n  Expected: {expected}\n  Got:      {pr_title}",
                expected=expected,
                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 ""


@click.command()
@click.argument("branch")
@click.argument("pr_title")
@click.argument("repo")
@click.argument("pr_number")
def main(branch: str, pr_title: str, repo: str, pr_number: 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)

    task_id = read_taskid(branch)
    if not task_id:
        raise click.ClickException(
            _(
                "Oops! No task ID found in .taskid file or branch name '{branch}'.",
                branch=branch,
            )
        )
    click.echo(_("Task ID: {task_id}", task_id=task_id))

    validate_pr_title(pr_title, task_id)
    validate_pr_title_matches_vikunja(pr_title, task_id)

    # 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:
        if e.status == 405 and "behind" in e.message.lower():
            # Head branch is behind master — pull master and rebase, then retry
            click.echo(_("Head branch is behind master. Pulling and rebasing..."))
            try:
                run_cmd(["git", "fetch", "origin", "master"])
                run_cmd(["git", "rebase", "origin/master"])
                run_cmd(["git", "push", "--force-with-lease"])
                click.echo(_("Rebased and pushed. Retrying merge..."))
                client.merge_pr(pr_number, merge_title)
            except (APIError, Exception) as retry_err:
                raise click.ClickException(
                    _(
                        "Merge failed after rebase retry: {error}\nPlease rebase the PR manually.",
                        error=str(retry_err),
                    )
                ) from None
        else:
            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()