jira_client.py

"""
jira_client.py — authenticated Jira REST API client.

Authenticates using an API token (Basic Auth over HTTPS).
Fetches ticket summary, description, and assignee fields.
"""

import logging
import requests
from requests.auth import HTTPBasicAuth

from config import JIRA_BASE_URL, JIRA_EMAIL, JIRA_API_TOKEN

log = logging.getLogger(__name__)


class JiraAuthError(Exception):
    pass


class JiraTicketNotFoundError(Exception):
    pass


class JiraClient:
    def __init__(self):
        if not JIRA_EMAIL or not JIRA_API_TOKEN:
            raise JiraAuthError(
                "JIRA_EMAIL and JIRA_API_TOKEN must be set in your .env file."
            )
        self.base_url = JIRA_BASE_URL.rstrip("/")
        self.auth = HTTPBasicAuth(JIRA_EMAIL, JIRA_API_TOKEN)
        self.headers = {"Accept": "application/json"}

    def _build_url(self, ticket_id: str) -> str:
        return f"{self.base_url}/rest/api/3/issue/{ticket_id}"

    def get_ticket(self, ticket_id: str) -> dict:
        """
        Fetch a Jira ticket and return a simplified dict with:
            - summary   : ticket title (used to identify alarm type)
            - description : raw text (used to extract target node)
            - assignee  : current assignee display name
            - status    : current status name
        """
        url = self._build_url(ticket_id)
        log.debug("GET %s", url)

        resp = requests.get(url, auth=self.auth, headers=self.headers, timeout=10)

        if resp.status_code == 401:
            raise JiraAuthError("Jira authentication failed. Check JIRA_EMAIL and JIRA_API_TOKEN.")
        if resp.status_code == 404:
            raise JiraTicketNotFoundError(f"Ticket {ticket_id} not found in Jira.")
        resp.raise_for_status()

        data = resp.json()
        fields = data.get("fields", {})

        # Jira API v3 stores description as Atlassian Document Format (ADF).
        # Extract plain text from paragraph nodes for simple parsing.
        description_raw = fields.get("description") or {}
        description_text = self._extract_adf_text(description_raw)

        return {
            "id": data["id"],
            "key": data["key"],
            "summary": fields.get("summary", ""),
            "description": description_text,
            "assignee": (fields.get("assignee") or {}).get("displayName", "Unassigned"),
            "status": fields.get("status", {}).get("name", "Unknown"),
        }

    def get_account_id(self, email: str) -> str | None:
        """Look up a Jira account ID by email (used for assignment)."""
        url = f"{self.base_url}/rest/api/3/user/search"
        resp = requests.get(
            url,
            params={"query": email},
            auth=self.auth,
            headers=self.headers,
            timeout=10,
        )
        resp.raise_for_status()
        users = resp.json()
        if users:
            return users[0].get("accountId")
        return None

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

    @staticmethod
    def _extract_adf_text(adf: dict) -> str:
        """Recursively extract plain text from Atlassian Document Format JSON."""
        if not adf or not isinstance(adf, dict):
            return ""
        parts = []
        for node in adf.get("content", []):
            if node.get("type") == "text":
                parts.append(node.get("text", ""))
            else:
                parts.append(JiraClient._extract_adf_text(node))
        return " ".join(p for p in parts if p).strip()