jira_updater.py

"""
jira_updater.py — posts mitigation comments, assigns tickets,
and transitions them to In Progress via the Jira REST API.
"""

import logging
import requests

from config import JIRA_BASE_URL, JIRA_EMAIL, JIRA_API_TOKEN, ENGINEER_USERNAME, JIRA_TRANSITION_IN_PROGRESS
from jira_client import JiraClient

log = logging.getLogger(__name__)


class JiraUpdater:
    def __init__(self, client: JiraClient):
        self.client = client
        self.base_url = JIRA_BASE_URL.rstrip("/")
        self.auth = client.auth
        self.headers = {
            "Accept": "application/json",
            "Content-Type": "application/json",
        }

    # ── Public methods ────────────────────────────────────────────────────────

    def post_mitigation_comment(
        self,
        ticket_id: str,
        alarm_type: str,
        node: str,
        playbook: str,
        result: dict,
    ) -> None:
        """Post a structured mitigation comment to the Jira ticket."""
        status = "✅ SUCCESS" if result.get("success") else "❌ FAILED"
        body = self._build_comment_adf(alarm_type, node, playbook, status, result)
        url = f"{self.base_url}/rest/api/3/issue/{ticket_id}/comment"

        log.debug("Posting comment to %s", ticket_id)
        resp = requests.post(url, auth=self.auth, headers=self.headers, json=body, timeout=10)
        resp.raise_for_status()
        log.info("Comment posted to %s", ticket_id)

    def assign_to_current_user(self, ticket_id: str) -> None:
        """Assign the ticket to the engineer running the CLI."""
        account_id = self.client.get_account_id(JIRA_EMAIL)
        if not account_id:
            log.warning("Could not resolve Jira account ID for %s — skipping assignment.", JIRA_EMAIL)
            return

        url = f"{self.base_url}/rest/api/3/issue/{ticket_id}/assignee"
        resp = requests.put(
            url,
            auth=self.auth,
            headers=self.headers,
            json={"accountId": account_id},
            timeout=10,
        )
        resp.raise_for_status()
        log.info("Ticket %s assigned to %s (%s)", ticket_id, ENGINEER_USERNAME, JIRA_EMAIL)

    def transition_in_progress(self, ticket_id: str) -> None:
        """Move the ticket to the 'In Progress' status."""
        url = f"{self.base_url}/rest/api/3/issue/{ticket_id}/transitions"
        body = {"transition": {"id": JIRA_TRANSITION_IN_PROGRESS}}

        resp = requests.post(url, auth=self.auth, headers=self.headers, json=body, timeout=10)
        resp.raise_for_status()
        log.info("Ticket %s transitioned to In Progress.", ticket_id)

    # ── Helpers ───────────────────────────────────────────────────────────────

    @staticmethod
    def _build_comment_adf(
        alarm_type: str,
        node: str,
        playbook: str,
        status: str,
        result: dict,
    ) -> dict:
        """Build an Atlassian Document Format (ADF) comment body."""
        output_snippet = (result.get("stdout") or "No output captured.")[:2000]

        return {
            "body": {
                "version": 1,
                "type": "doc",
                "content": [
                    {
                        "type": "heading",
                        "attrs": {"level": 3},
                        "content": [{"type": "text", "text": f"🤖 Automated Mitigation — {status}"}],
                    },
                    _adf_paragraph(f"Alarm Type : {alarm_type}"),
                    _adf_paragraph(f"Target Node: {node}"),
                    _adf_paragraph(f"Playbook   : {playbook}"),
                    _adf_paragraph(f"Engineer   : {ENGINEER_USERNAME}"),
                    {
                        "type": "heading",
                        "attrs": {"level": 4},
                        "content": [{"type": "text", "text": "Playbook Output"}],
                    },
                    {
                        "type": "codeBlock",
                        "attrs": {"language": "text"},
                        "content": [{"type": "text", "text": output_snippet}],
                    },
                ],
            }
        }


def _adf_paragraph(text: str) -> dict:
    return {
        "type": "paragraph",
        "content": [{"type": "text", "text": text}],
    }