#!/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: `` Merge commit format: ``GRM-N: `` 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= python3 scripts/ci/auto_merge.py """ 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 gitea_runner_manager.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: `` """ if not PR_TITLE_RE.match(pr_title): raise click.ClickException( _( "Oops! PR title must follow format 'GRM-N: '.\n" " Expected: {task_id}: \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 (skip validation). """ 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 return "" 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. """ vikunja_title = get_vikunja_task_title(task_id) if not vikunja_title: click.echo(_("Warning: could not fetch Vikunja task title, 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: 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()