grm/scripts/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 click
import requests

from gitea_runner_manager.i18n import _

VIKUNJA_API = "https://work.oblachno.oblachno.fyi/api/v1"
TASK_ID_RE = re.compile(r"GRM-\d+")
PROJECT_ID = 6


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."""
    first_line = commit_msg.split("\n")[0]
    return re.sub(r"^GRM-\d+:\s*", "", first_line)


def _handle_http_error(e: requests.HTTPError) -> None:
    """Raise a user-friendly Click exception for HTTP errors."""
    response = e.response
    status = response.status_code if response else 0
    try:
        body = response.json() if response else {}
        message = body.get("message", str(e))
    except Exception:
        message = str(e)
    raise click.ClickException(
        _(
            "Vikunja API error: HTTP {status} — {message}",
            status=status,
            message=message,
        )
    )


def resolve_task_id(token: str, task_id: str) -> int:
    """Resolve GRM-N identifier to Vikunja numeric task ID."""
    url = f"{VIKUNJA_API}/tasks/all"
    headers = {"Authorization": f"Bearer {token}"}
    params = {"per_page": 50}
    response = requests.get(url, headers=headers, params=params, timeout=30)
    response.raise_for_status()
    tasks = response.json()
    matches = [t for t in tasks if t.get("project_id") == PROJECT_ID and t.get("identifier") == task_id]
    if not matches:
        raise click.ClickException(
            _(
                "Could not find Vikunja task for {task_id} in project {project_id}.",
                task_id=task_id,
                project_id=PROJECT_ID,
            )
        )
    return int(matches[0]["id"])


def post_comment(token: str, task_id: int, html: str) -> None:
    """Post an HTML comment to a Vikunja task."""
    url = f"{VIKUNJA_API}/tasks/{task_id}/comments"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }
    payload = {"comment": html}
    response = requests.post(url, headers=headers, json=payload, timeout=30)
    response.raise_for_status()


def mark_task_done(token: str, task_id: int) -> None:
    """Mark a Vikunja task as done."""
    url = f"{VIKUNJA_API}/tasks/{task_id}"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }
    payload = {"done": True}
    response = requests.put(url, headers=headers, json=payload, timeout=30)
    response.raise_for_status()


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")
@click.option("--commit-sha", default="", help="Commit SHA")
def main(commit_msg: str, commit_sha: str) -> None:
    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:
        click.echo(_("No task ID in commit message, skipping Vikunja update. All good — nothing to do here!"))
        return

    vikunja_task_id = 0
    try:
        vikunja_task_id = resolve_task_id(token, task_id)
        conv_msg = extract_conventional_msg(commit_msg)
        sha = commit_sha or "unknown"
        html = build_comment(task_id, conv_msg, sha)

        post_comment(token, vikunja_task_id, html)
        mark_task_done(token, vikunja_task_id)
    except requests.HTTPError as e:
        _handle_http_error(e)

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