grm/scripts/ci/post_merge.py

#!/usr/bin/env python3
"""Update Vikunja task after a merge to master.

Usage:
  VIKUNJA_TOKEN=<token> python3 scripts/post_merge.py <commit_msg> [--commit-sha <sha>]
"""

import os
import re
import subprocess  # nosec B404

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

from gitea_runner_manager.api_clients import VikunjaClient
from gitea_runner_manager.config import DEFAULT_PER_PAGE, TASK_ID_RE, VIKUNJA_API_URL, VIKUNJA_PROJECT_ID
from gitea_runner_manager.exceptions import APIError
from gitea_runner_manager.i18n import _

load_dotenv(override=True)


def _get_git_commit_message() -> str:
    """Get the full commit message of the latest commit."""
    result = subprocess.run(  # nosec B603 B607
        ["git", "log", "-1", "--pretty=%B"],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        raise click.ClickException(f"git log failed: {result.stderr.strip()}")
    return result.stdout.strip()


def _get_git_commit_sha() -> str:
    """Get the SHA of the latest commit."""
    result = subprocess.run(  # nosec B603 B607
        ["git", "rev-parse", "HEAD"],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        raise click.ClickException(f"git rev-parse failed: {result.stderr.strip()}")
    return result.stdout.strip()


def extract_task_id(commit_msg: str) -> str:
    """Extract GRM-N task identifier from the first line of commit message."""
    first_line = commit_msg.split("\n")[0]
    match = TASK_ID_RE.search(first_line)
    return match.group(0) if match else ""


def extract_conventional_msg(commit_msg: str) -> str:
    """Strip the GRM-N prefix from the commit subject.

    Handles both formats:
    - ``GRM-N: <message>`` (legacy, colon-separated)
    - ``GRM-N <message>`` (current, space-separated)
    """
    first_line = commit_msg.split("\n")[0]
    return re.sub(r"^GRM-\d+[:\s]\s*", "", first_line)


def resolve_task_id(client: VikunjaClient, task_id: str) -> int:
    """Resolve GRM-N identifier to Vikunja numeric task ID.

    Paginates through the project's tasks to handle projects with more
    than 50 tasks.  Raises ClickException if the task is not found.
    """
    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 int(matches[0]["id"])
        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 build_comment(task_id: str, conv_msg: str, commit_sha: str) -> str:
    """Build HTML comment body for Vikunja."""
    return f"<p><strong>{task_id}</strong>: {conv_msg}</p><p>Commit: <code>{commit_sha}</code></p>"


@click.command()
@click.argument("commit_msg", required=False)
@click.option("--commit-sha", default="", help="Commit SHA")
@click.option("--from-git", is_flag=True, default=False, help="Read commit message and SHA from git.")
@click.option(
    "--git-sha",
    default="",
    help="Read commit message from a specific git SHA (avoids race condition with parallel jobs).",
)
def main(commit_msg: str | None, commit_sha: str, from_git: bool, git_sha: str) -> None:
    if git_sha:
        # Read commit message from a specific SHA — this avoids the race
        # condition where a parallel job (e.g., release) pushes a new commit
        # to master before this job reads HEAD.
        result = subprocess.run(  # nosec B603 B607
            ["git", "log", "-1", "--pretty=%B", git_sha],
            capture_output=True,
            text=True,
            check=False,
        )
        if result.returncode != 0:
            raise click.ClickException(f"git log failed for SHA {git_sha}: {result.stderr.strip()}")
        commit_msg = result.stdout.strip()
        if not commit_sha:
            commit_sha = git_sha
    elif from_git:
        commit_msg = _get_git_commit_message()
        if not commit_sha:
            commit_sha = _get_git_commit_sha()
    if not commit_msg:
        raise click.ClickException("commit_msg argument is required (or use --from-git or --git-sha)")
    token = os.environ.get("VIKUNJA_TOKEN", "")
    if not token:
        raise click.ClickException(_("ERROR: VIKUNJA_TOKEN is not set."))

    task_id = extract_task_id(commit_msg)
    if not task_id:
        first_line = commit_msg.split("\n")[0]
        # Skip gracefully for infrastructure commits that don't follow
        # the GRM-N convention: release commits, reverts, bot commits, etc.
        infra_patterns = [
            r"^release: v\d+\.\d+\.\d+",  # release commits
            r"^revert: ",  # git revert commits
            r"^Merge ",  # merge commits
            r"^\[skip ci\]",  # skip-ci commits
        ]
        for pattern in infra_patterns:
            if re.match(pattern, first_line):
                click.echo(
                    _(
                        "Infrastructure commit (no GRM-N task ID), skipping Vikunja update: {msg}",
                        msg=first_line,
                    )
                )
                return
        # Non-infrastructure commits without GRM-N prefix — warn but don't fail
        click.echo(
            _(
                "Warning: No task ID (GRM-N) found in commit message: {msg}. Skipping Vikunja update.",
                msg=first_line,
            )
        )
        return

    client = VikunjaClient(VIKUNJA_API_URL, token)
    vikunja_task_id = resolve_task_id(client, task_id)
    conv_msg = extract_conventional_msg(commit_msg)
    sha = commit_sha or "unknown"
    html = build_comment(task_id, conv_msg, sha)

    try:
        client.post_comment(vikunja_task_id, html)
        client.update_task(vikunja_task_id, done=True)
    except APIError as e:
        # Vikunja is a project management tool — if it's down, the merge
        # still succeeded. Warn but don't fail the post-merge workflow.
        click.echo(
            _(
                "Warning: Vikunja API error (HTTP {status}): {message}. "
                "Task {task_id} was NOT updated. The merge succeeded — "
                "please update the Vikunja task manually.",
                status=e.status,
                message=e.message,
                task_id=task_id,
            )
        )
        return

    click.echo(
        _(
            "Nice! Vikunja task {task_id} (ID {vikunja_id}) updated and marked done.",
            task_id=task_id,
            vikunja_id=vikunja_task_id,
        )
    )


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