devx/src/devx/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 the branch name
(e.g., ``DEVX-31-fix-foo`` → ``DEVX-31``), validates the PR title against
the Vikunja task, and squash-merges with a conventional commit message
prefixed by the task ID.

PR title format:    ``{PREFIX}-N: <vikunja task title>``
Merge commit format: ``{PREFIX}-N <conventional commit message>``

The ``{PREFIX}`` is determined by ``DEVX_TASK_PREFIX`` (default: ``DEVX``).
Each project sets its own prefix (e.g., ``GRM``, ``INFRA``).

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:
  CI_GITEA_API_TOKEN=<token> VIKUNJA_TOKEN=<token> python3 -m devx.ci.auto_merge <branch> <pr_title> <repo> <pr_number>
"""

import re
from pathlib import Path
from typing import Any

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

from devx.api_clients import GiteaClient, VikunjaClient
from devx.ci._shared import extract_task_id as _extract_task_id
from devx.config import (
    CONVENTIONAL_RE,
    DEFAULT_PER_PAGE,
    GITEA_API_URL,
    TASK_PREFIX,
    VIKUNJA_API_URL,
    VIKUNJA_PROJECT_ID,
)
from devx.exceptions import APIError
from devx.i18n import _
from devx.tokens import get_ci_token, get_vikunja_token

# Strip leading task ID prefix (e.g. "DEVX-12: " or "OBL-INFRA-364: ") from commit subjects.
_TASK_ID_PREFIX_RE = re.compile(rf"^{TASK_PREFIX}-\d+:\s*")

TASKID_FILE = ".taskid"  # Deprecated, kept for backward-compat warnings
PR_TITLE_RE = re.compile(rf"^{TASK_PREFIX}-\d+:\s+.+")

load_dotenv()


def read_taskid(branch: str) -> str:
    """Read task ID from branch name.

    The branch name is the sole source of truth for the task ID
    (e.g., ``DEVX-31-fix-foo`` → ``DEVX-31``).  Branches must include
    the task ID prefix — there is no ``.taskid`` file fallback.

    If a stale ``.taskid`` file exists and disagrees with the branch
    name, a deprecation warning is printed advising its removal.
    """
    branch_task_id = extract_task_id(branch)
    if branch_task_id:
        # Warn about stale .taskid file if it exists and disagrees
        path = Path(TASKID_FILE)
        if path.exists():
            file_task_id = path.read_text(encoding="utf-8").strip()
            if file_task_id and file_task_id != branch_task_id:
                click.echo(
                    _(
                        "WARNING: .taskid file ({file_id}) is deprecated and disagrees with branch name ({branch_id}). "
                        "Delete .taskid from the repo — branch name is the sole source of truth.",
                        file_id=file_task_id,
                        branch_id=branch_task_id,
                    )
                )
        return branch_task_id
    return ""


def extract_task_id(branch: str) -> str:
    """Extract task identifier from branch name (delegates to shared utility)."""
    return _extract_task_id(branch)


def validate_pr_title(pr_title: str, task_id: str) -> None:
    """Raise ClickException if PR title does not follow the required format.

    Expected: ``{PREFIX}-N: <vikunja task title>``
    """
    if not PR_TITLE_RE.match(pr_title):
        raise click.ClickException(
            _(
                "Oops! PR title must follow format '{prefix}-N: <task title>'.\n"
                "  Expected: {task_id}: <task title>\n"
                "  Got:      {pr_title}",
                prefix=TASK_PREFIX,
                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 DEVX-N identifier.

    Raises ClickException if VIKUNJA_TOKEN is not set or the task is not found.
    """
    try:
        token = get_vikunja_token()
    except click.ClickException:
        raise click.ClickException(
            _("VIKUNJA_TOKEN is not set. This is required in CI to validate PR titles.")
        ) from None
    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.

    Raises ClickException if VIKUNJA_TOKEN is not set, the task is not found,
    or the title doesn't match.
    """
    vikunja_title = get_vikunja_task_title(task_id)
    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.

    Picks the highest-priority conventional commit message from the PR.
    Priority: feat > fix > refactor > docs > chore > other.
    Falls back to the newest commit message if none match.
    """
    priority = {"feat": 5, "fix": 4, "refactor": 3, "docs": 2, "chore": 1, "ci": 1, "style": 1, "test": 1}
    best_msg = ""
    best_score = 0
    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]
        # Strip any leading task ID prefix (e.g. "OBL-INFRA-364: fix: ...") so
        # conventional commit matching works on the remainder.
        stripped = _TASK_ID_PREFIX_RE.sub("", message)
        m = CONVENTIONAL_RE.match(stripped)
        if m:
            prefix = m.group(1).split("(")[0].strip()  # e.g. "feat" from "feat(scope)"
            score = priority.get(prefix, 0)
            if score > best_score:
                best_score = score
                best_msg = stripped
    if best_msg:
        return best_msg
    # Fallback: use the newest commit's first line (strip task ID prefix if present)
    if commits:
        commit_info = commits[-1].get("commit", {})
        raw = str(commit_info.get("message", "") if isinstance(commit_info, dict) else "").split("\n")[0]
        return _TASK_ID_PREFIX_RE.sub("", raw)
    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:
    try:
        token = get_ci_token()
    except click.ClickException:
        raise click.ClickException(_("ERROR: CI_GITEA_TOKEN is not set.")) from None

    # Validate PR number is an integer
    try:
        pr_num = int(pr_number)
    except ValueError:
        raise click.ClickException(_("PR number must be an integer, got: {pr_number}", pr_number=pr_number)) from None

    # Validate repo format
    if "/" not in repo:
        raise click.ClickException(_("Repo must be in 'owner/name' format, got: {repo}", repo=repo))
    owner, repo_name = repo.split("/", 1)
    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 branch name '{branch}'. "
                "Branch names must include the task ID prefix (e.g., DEVX-31-fix-bug).",
                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: DEVX-N: <conventional commit message>
    commits = client.get_pr_commits(pr_num)
    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_num, merge_title)
    except APIError as e:
        if e.status == 405 and "behind" in e.message.lower():
            # Head branch is behind master. Auto-rebase via Gitea API.
            # This triggers a new pull_request synchronize event → new CI run.
            # The next auto-merge attempt will find the branch up-to-date and
            # merge successfully. This is NOT an infinite loop: the rebase
            # resolves the "behind" condition, so the next run merges.
            # If another PR merges in between, the branch may fall behind
            # again, but the process converges as PRs stop merging.
            click.echo(
                _(
                    "Branch is behind master. Auto-rebasing via Gitea API...\n"
                    "A new CI run will start automatically after the rebase.\n"
                    "The next auto-merge attempt will merge this PR.",
                )
            )
            try:
                client.update_pr_branch(pr_num, style="rebase")
            except APIError as rebase_err:
                raise click.ClickException(
                    _(
                        "Auto-rebase failed with HTTP {status}: {message}\n"
                        "Rebase manually:\n"
                        "  git fetch origin master && git rebase origin/master && git push --force-with-lease\n"
                        "Then re-add the ready-to-merge label.",
                        status=rebase_err.status,
                        message=rebase_err.message,
                    )
                ) from None
            # Exit cleanly — the rebase triggers a new CI run that will retry.
            return
        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_num,
            merge_title=merge_title,
        )
    )


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