jira_helpers.py

#!/usr/bin/env python3
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.

"""Shared helpers for the Jira board automation scripts."""

from __future__ import annotations

import base64
import contextlib
import itertools
import json
import os
import shlex
import subprocess
import sys
import tempfile
import textwrap
import threading
from collections.abc import Iterable, Sequence
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
from typing import TextIO
from urllib import error, parse, request

ROOT = Path(__file__).resolve().parent
DEBUG = False


class ConfigError(RuntimeError):
    """Raised when local configuration is incomplete."""


class JiraCliError(RuntimeError):
    """Raised when Atlassian CLI invocations fail."""


@dataclass
class IssueRecord:
    key: str
    issue_id: str
    summary: str
    status: str
    assignee: str | None
    url: str


@dataclass
class SearchResults:
    issues: list[IssueRecord]
    total: int | None


@dataclass
class BoardColumn:
    name: str
    statuses: list[str]


@dataclass
class BoardConfig:
    board_id: str
    name: str
    filter_query: str
    columns: list[BoardColumn]
    project_key: str | None


_AUTH_READY = False


def set_debug(enabled: bool) -> None:
    global DEBUG
    DEBUG = bool(enabled)


def debug(message: str) -> None:
    if not DEBUG:
        return
    try:
        sys.stdout.flush()
    except OSError:
        pass
    try:
        sys.stderr.write("\x1b[2K\r")  # clear current line, return to column 0
    except OSError:
        pass
    print(f"[debug] {message}", file=sys.stderr, flush=True)


def auto_enable_debug_from_env() -> None:
    """Turn on debug logging when ACLI_HELPERS_DEBUG is truthy."""
    if os.environ.get("ACLI_HELPERS_DEBUG"):
        set_debug(True)


class _StatusSpinner:
    """Lightweight terminal spinner shown while network calls are pending."""

    def __init__(self, stream: TextIO, message: str) -> None:
        self.stream = stream
        self.message = message
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None

    def start(self) -> None:
        if self._thread is not None:
            return
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()

    def stop(self) -> None:
        self._stop.set()
        thread = self._thread
        if thread is not None:
            thread.join()
            self._thread = None
        self._clear_line()

    def _run(self) -> None:
        for frame in itertools.cycle("|/-\\"):
            if self._stop.is_set():
                break
            self._render(frame)
            if self._stop.wait(0.1):
                break

    def _render(self, frame: str) -> None:
        try:
            self.stream.write(f"\r{frame} {self.message}")
            self.stream.flush()
        except OSError:
            self._stop.set()

    def _clear_line(self) -> None:
        width = len(self.message) + 2
        try:
            self.stream.write("\r" + " " * width + "\r")
            self.stream.flush()
        except OSError:
            pass


def _spinner_stream() -> TextIO | None:
    if DEBUG:
        return None
    stream = getattr(sys, "stderr", None)
    if stream is None:
        return None
    isatty = getattr(stream, "isatty", None)
    if not callable(isatty):
        return None
    try:
        if not isatty():
            return None
    except OSError:
        return None
    if not hasattr(stream, "write") or not hasattr(stream, "flush"):
        return None
    return stream


@contextlib.contextmanager
def _spinner(message: str):
    stream = _spinner_stream()
    if not stream:
        yield
        return
    spinner = _StatusSpinner(stream=stream, message=message)
    spinner.start()
    try:
        yield
    finally:
        spinner.stop()


def _read_file_value(filename: str, required: bool = True) -> str:
    path = ROOT / filename
    if not path.exists():
        if required:
            raise ConfigError(
                f"Missing {filename}. Copy the matching *.example file and fill in your values."
            )
        return ""
    for raw_line in path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        return line
    if required:
        raise ConfigError(
            f"{filename} is empty; add the required value on the first non-comment line."
        )
    return ""


@lru_cache(maxsize=1)
def get_api_token() -> str:
    token = os.environ.get("ACLI_TOKEN")
    if token:
        return token.strip()
    token = _read_file_value("token.txt", required=True)
    return token.strip()


@lru_cache(maxsize=1)
def get_email() -> str:
    value = os.environ.get("ACLI_EMAIL")
    if value:
        return value.strip()
    return _read_file_value("email.txt", required=True)


@lru_cache(maxsize=1)
def get_site_base_url() -> str:
    site = os.environ.get("ACLI_SITE")
    if not site:
        site = _read_file_value("site.txt", required=True)
    site = site.strip()
    if not site:
        raise ConfigError("Site host cannot be blank.")
    if site.startswith("http://"):
        raise ConfigError(
            "Only HTTPS Jira hosts are supported; update site.txt to use https://"
        )
    if not site.startswith("https://"):
        site = f"https://{site}"
    parsed = parse.urlparse(site)
    if not parsed.scheme or not parsed.netloc:
        raise ConfigError(f"Unable to parse Jira site URL from '{site}'.")
    normalized = f"https://{parsed.netloc}"
    debug(f"Using Jira site base URL: {normalized.rstrip('/')}")
    return normalized.rstrip("/")


@lru_cache(maxsize=1)
def get_site_host_for_cli() -> str:
    parsed = parse.urlparse(get_site_base_url())
    return parsed.netloc


def get_board_id() -> str:
    raw = _read_file_value("board_id.txt", required=True)
    board_id = raw.strip()
    if not board_id.isdigit():
        raise ConfigError("Board ID should be the numeric ID from the Jira board URL.")
    return board_id


def _basic_auth_header() -> str:
    token = get_api_token()
    email = get_email()
    auth = base64.b64encode(f"{email}:{token}".encode()).decode("ascii")
    return f"Basic {auth}"


def _jira_request_json(path: str) -> dict:
    base_url = get_site_base_url()
    if path.startswith("http://") or path.startswith("https://"):
        url = path
    else:
        if not path.startswith("/"):
            path = f"/{path}"
        url = f"{base_url}{path}"
    debug(f"Requesting Jira API: {url}")
    req = request.Request(url)
    req.add_header("Accept", "application/json")
    req.add_header("Authorization", _basic_auth_header())
    try:
        with _spinner("Contacting Jira…"), request.urlopen(req) as resp:
            payload = resp.read()
            debug(f"Jira API {url} responded with {resp.status} ({len(payload)} bytes)")
    except error.HTTPError as exc:
        body = exc.read().decode("utf-8", "replace") if DEBUG else ""
        debug(f"Jira API error {exc.code}: {body.strip() or exc.reason}")
        raise ConfigError(
            f"Jira API request failed with status {exc.code}: {exc.reason}"
        ) from exc
    except error.URLError as exc:
        raise ConfigError(f"Unable to contact Jira: {exc.reason}") from exc
    try:
        return json.loads(payload)
    except json.JSONDecodeError as exc:
        raise ConfigError("Jira API response was not valid JSON.") from exc


def _fetch_board_metadata(board_id: str) -> dict:
    return _jira_request_json(f"/rest/agile/1.0/board/{board_id}")


def _derive_filter_query(config_data: dict, board_meta: dict) -> str:
    filter_obj = config_data.get("filter") or {}
    filter_query = (filter_obj.get("query") or filter_obj.get("jql") or "").strip()
    if filter_query:
        debug("Board configuration already includes filter JQL.")
        return filter_query

    filter_id = filter_obj.get("id") or board_meta.get("filterId")
    if filter_id:
        debug(f"Fetching filter {filter_id} to obtain JQL…")
        try:
            filter_details = _jira_request_json(f"/rest/api/3/filter/{filter_id}")
            jql = (filter_details.get("jql") or "").strip()
            if jql:
                return jql
            debug("Filter lookup succeeded but did not include JQL.")
        except ConfigError as exc:
            debug(f"Filter lookup failed: {exc}")

    # Fall back to the project key if available (team-managed boards)
    location = board_meta.get("location") or config_data.get("location") or {}
    project_key = location.get("projectKey")
    if project_key:
        debug(f"Falling back to project filter for {project_key}.")
        return f"project = {project_key}"

    raise ConfigError(
        "Unable to determine the board filter query from the Jira API response."
    )


@lru_cache(maxsize=1)
def get_board_configuration() -> BoardConfig:
    board_id = get_board_id()
    config_data = _jira_request_json(f"/rest/agile/1.0/board/{board_id}/configuration")
    board_meta = _fetch_board_metadata(board_id)
    filter_query = _derive_filter_query(config_data, board_meta)
    columns_data = (config_data.get("columnConfig") or {}).get("columns") or []
    columns: list[BoardColumn] = []
    for column in columns_data:
        name = column.get("name") or "Unnamed"
        statuses = [
            status.get("name")
            for status in column.get("statuses") or []
            if status.get("name")
        ]
        columns.append(BoardColumn(name=name, statuses=statuses))
    location = config_data.get("location") or board_meta.get("location") or {}
    project_key = location.get("projectKey") or location.get("projectId")
    if project_key is not None:
        project_key = str(project_key)
    return BoardConfig(
        board_id=board_id,
        name=config_data.get("name") or board_meta.get("name") or f"Board {board_id}",
        filter_query=filter_query,
        columns=columns,
        project_key=project_key,
    )


def get_board_filter_query() -> str:
    return get_board_configuration().filter_query


def get_board_columns() -> list[BoardColumn]:
    return get_board_configuration().columns


def get_default_project_key() -> str | None:
    return get_board_configuration().project_key


@lru_cache(maxsize=1)
def get_project_override() -> str | None:
    env_value = os.environ.get("ACLI_PROJECT")
    if env_value:
        cleaned = env_value.strip()
        if cleaned:
            return cleaned
    value = _read_file_value("project.txt", required=False).strip()
    return value or None


def split_jql_order_clause(jql: str) -> tuple[str, str]:
    """Return (query_without_order, order_clause)."""
    text = (jql or "").strip()
    if not text:
        return "", ""
    lower = text.lower()
    idx = lower.find(" order by ")
    if idx == -1:
        return text, ""
    core = text[:idx].rstrip()
    order_clause = text[idx:].strip()
    debug(f"Stripped ORDER BY clause '{order_clause}' from board filter.")
    return core, order_clause


def issue_web_url(issue_key: str) -> str:
    return f"{get_site_base_url()}/browse/{issue_key}"


def launch_editor(
    initial_text: str, filename_prefix: str, extension: str = ".md"
) -> tuple[str, Path]:
    safe_prefix = filename_prefix.replace("/", "-") or "acli"
    try:
        fd, temp_path = tempfile.mkstemp(prefix=f"{safe_prefix}-", suffix=extension)
    except OSError as exc:
        raise ConfigError(
            "Unable to create a secure temporary file for the editor."
        ) from exc
    path = Path(temp_path)
    try:
        os.chmod(path, 0o600)
    except OSError:
        # Best effort; continue even if chmod fails (umask already gave 600 on POSIX).
        pass
    with os.fdopen(fd, "w", encoding="utf-8") as handle:
        handle.write(initial_text or "")
    editor = os.environ.get("VISUAL") or os.environ.get("EDITOR") or "vi"
    try:
        cmd = shlex.split(editor)
    except ValueError as exc:
        raise ConfigError(f"Unable to parse EDITOR value '{editor}': {exc}") from exc
    if not cmd:
        raise ConfigError("EDITOR command resolved to an empty value.")
    cmd.append(str(path))
    debug(f"Launching editor command: {shlex.join(cmd)}")
    try:
        subprocess.run(cmd, check=True)
    except FileNotFoundError as exc:
        raise ConfigError(f"Editor '{cmd[0]}' not found on PATH.") from exc
    except subprocess.CalledProcessError as exc:
        raise JiraCliError(
            f"The editor command exited with status {exc.returncode}."
        ) from exc
    new_text = path.read_text(encoding="utf-8")
    return new_text, path


def ensure_authenticated() -> None:
    global _AUTH_READY
    if _AUTH_READY:
        return
    debug("Checking ACLI authentication status…")
    with _spinner("Checking ACLI authentication…"):
        status = subprocess.run(
            ["acli", "jira", "auth", "status"],
            cwd=ROOT,
            capture_output=True,
            text=True,
        )
    if status.returncode == 0:
        debug("ACLI auth already configured.")
        _AUTH_READY = True
        return
    site = get_site_host_for_cli()
    email = get_email()
    token = get_api_token()
    debug("ACLI auth missing; attempting login via token…")
    with _spinner("Logging in to ACLI…"):
        login = subprocess.run(
            [
                "acli",
                "jira",
                "auth",
                "login",
                "--site",
                site,
                "--email",
                email,
                "--token",
            ],
            cwd=ROOT,
            input=token,
            text=True,
            capture_output=True,
        )
    if login.returncode != 0:
        message = (
            login.stderr.strip()
            or login.stdout.strip()
            or "Unknown authentication error"
        )
        debug(f"ACLI login failed: {message}")
        raise JiraCliError(f"ACLI authentication failed: {message}")
    debug("ACLI login succeeded.")
    _AUTH_READY = True


def _run_acli(
    args: Sequence[str], *, expect_json: bool = False, stream: bool = False
) -> str | dict:
    ensure_authenticated()
    if stream and expect_json:
        raise ValueError("Cannot stream output when JSON is required.")
    cmd = ["acli", *args]
    quoted = shlex.join(cmd)
    debug(f"Running ACLI command: {quoted}")
    desc = " ".join(cmd[1:4]).strip()
    spinner_message = (
        f"Running Atlassian CLI ({desc})…" if desc else "Running Atlassian CLI…"
    )
    try:
        if stream:
            result = subprocess.run(cmd, cwd=ROOT)
            stdout = ""
            stderr = ""
        else:
            with _spinner(spinner_message):
                result = subprocess.run(
                    cmd,
                    cwd=ROOT,
                    capture_output=True,
                    text=True,
                )
            stdout = result.stdout
            stderr = result.stderr
    except FileNotFoundError as exc:
        raise ConfigError("The 'acli' executable was not found on PATH.") from exc
    debug(f"ACLI exited with {result.returncode}")
    if result.returncode != 0:
        message = (
            (stderr.strip() if stderr else "")
            or (stdout.strip() if stdout else "")
            or "ACLI command failed"
        )
        if not stream:
            debug(f"ACLI stderr: {stderr.strip() if stderr else ''}")
            debug(f"ACLI stdout: {stdout.strip() if stdout else ''}")
        raise JiraCliError(message)
    if stream:
        return ""
    output = (stdout or "").strip()
    debug(
        f"ACLI stdout ({len(output)} chars): {output if len(output) < 500 else output[:500] + '…'}"
    )
    if not expect_json:
        return output
    try:
        return json.loads(output)
    except json.JSONDecodeError as exc:
        raise JiraCliError("ACLI did not return valid JSON output.") from exc


def run_acli_text(args: Sequence[str], *, stream: bool = False) -> str:
    output = _run_acli(args, expect_json=False, stream=stream)
    if stream:
        return ""
    assert isinstance(output, str)
    return output


def search_issues(
    jql: str,
    *,
    limit: int | None = None,
    fetch_all: bool = False,
    fields: Iterable[str] | None = None,
) -> SearchResults:
    debug(f"Searching issues with JQL: {jql}")
    args: list[str] = ["jira", "workitem", "search", "--jql", jql, "--json"]
    if fields:
        field_list = ",".join(sorted(set(fields)))
        args.extend(["--fields", field_list])
    if fetch_all:
        args.append("--paginate")
    elif limit is not None:
        if limit <= 0:
            raise ConfigError("Limit must be greater than zero when provided.")
        args.extend(["--limit", str(limit)])
    data = _run_acli(args, expect_json=True)
    if isinstance(data, list):
        issue_nodes = data
        total = len(data)
    else:
        issue_nodes = data.get("issues") or []
        total = data.get("total")
    records = [issue_from_raw(node) for node in issue_nodes]
    return SearchResults(issues=records, total=total)


def issue_from_raw(node: dict) -> IssueRecord:
    fields = node.get("fields") or {}
    summary = fields.get("summary") or ""
    status_name = ""
    status_field = fields.get("status") or {}
    if isinstance(status_field, dict):
        status_name = (
            status_field.get("name")
            or status_field.get("statusCategory", {}).get("name")
            or ""
        )
    assignee = None
    assignee_field = fields.get("assignee")
    if isinstance(assignee_field, dict):
        assignee = assignee_field.get("displayName") or assignee_field.get("name")
    issue_key = node.get("key") or ""
    issue_id = node.get("id")
    return IssueRecord(
        key=issue_key,
        issue_id=str(issue_id) if issue_id is not None else "?",
        summary=summary,
        status=status_name or "(unknown)",
        assignee=assignee,
        url=issue_web_url(issue_key) if issue_key else "",
    )


def format_issue_list(records: Sequence[IssueRecord]) -> str:
    if not records:
        return "No matching issues."
    lines: list[str] = []
    for record in records:
        summary = textwrap.shorten(
            record.summary or "(no summary)", width=88, placeholder="…"
        )
        assignee = f" · {record.assignee}" if record.assignee else ""
        lines.append(
            f"- {record.key} (id {record.issue_id}) [{record.status}]{assignee}"
        )
        lines.append(f"  {summary}")
        lines.append(f"  {record.url}")
    return "\n".join(lines)


def fetch_issue(key: str, fields: Iterable[str] | None = None) -> dict:
    if not key:
        raise ConfigError("Issue key is required.")
    debug(f"Fetching issue {key}")
    args: list[str] = ["jira", "workitem", "view", key, "--json"]
    if fields:
        args.extend(["--fields", ",".join(sorted(set(fields)))])
    data = _run_acli(args, expect_json=True)
    if isinstance(data, list) and data:
        return data[0]
    if isinstance(data, dict):
        if data.get("key"):
            return data
        issues = data.get("issues")
        if isinstance(issues, list) and issues:
            return issues[0]
    raise JiraCliError("Unexpected response when fetching the issue details.")


def adf_to_text(payload) -> str:
    if payload is None:
        return ""
    if isinstance(payload, str):
        return payload
    if isinstance(payload, dict):
        node_type = payload.get("type")
        content = payload.get("content") or []
        if node_type in {"doc", "panel", "blockquote"}:
            return "\n".join(filter(None, (adf_to_text(item) for item in content)))
        if node_type == "paragraph":
            inner = "".join(adf_to_text(item) for item in content)
            return inner
        if node_type in {"heading", "listItem"}:
            return "".join(adf_to_text(item) for item in content)
        if node_type == "bulletList":
            return "\n".join(f"- {adf_to_text(item)}".strip() for item in content)
        if node_type == "orderedList":
            parts = []
            for idx, item in enumerate(content, start=1):
                text = adf_to_text(item).strip()
                parts.append(f"{idx}. {text}")
            return "\n".join(parts)
        if node_type == "text":
            return payload.get("text") or ""
        if node_type == "hardBreak":
            return "\n"
        return "".join(adf_to_text(item) for item in content)
    if isinstance(payload, list):
        return "\n".join(filter(None, (adf_to_text(item) for item in payload)))
    return ""


def print_error(message: str) -> None:
    print(f"Error: {message}", file=sys.stderr)


auto_enable_debug_from_env()