agent-skills/skills/g3fo-commit-jira/scripts/jira_cli.py

#!/usr/bin/env python3
"""
Jira REST API CLI — same credentials as upload_attachment.py (jira_upload.env).
Use when Atlassian MCP is unavailable (other IDEs).

Commands:
  myself          Current user (accountId for assignee)
  search          JQL search
  issue           Get one issue by key
  create          Create Task (G3SF by default), assign to self
  comment         Add comment (ADF) to issue

Examples:
  python jira_cli.py --config ../jira_upload.env search --limit 5
  python jira_cli.py --config ../jira_upload.env issue G3SF-123
  python jira_cli.py --config ../jira_upload.env create --summary "[G3SF] fix foo" --description-file report.md
  python jira_cli.py --config ../jira_upload.env comment --issue G3SF-123 --body-file impact.md
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional

try:
    import requests
except ImportError:
    print("ERROR: pip install requests", file=sys.stderr)
    sys.exit(1)

from jira_env import CONFIG_FILENAME, load_config, require_credentials, session_headers

DEFAULT_JQL = (
    "project = G3SF AND assignee = currentUser() AND statusCategory != Done "
    "ORDER BY updated DESC"
)


def _empty_doc() -> Dict[str, Any]:
    return {
        "type": "doc",
        "version": 1,
        "content": [{"type": "paragraph", "content": [{"type": "text", "text": " "}]}],
    }


def _text_nodes(s: str) -> List[Dict[str, Any]]:
    """Single plain text node (non-empty)."""
    if not s:
        return []
    return [{"type": "text", "text": s}]


def parse_inline_adf(s: str) -> List[Dict[str, Any]]:
    """
    Parse inline **bold**, ~~strikethrough~~, `code`, *italic*, [text](url), ***bold italic***
    into ADF text nodes with marks.
    """
    if not s:
        return []
    # Order: links, bold+italic, bold, strikethrough, italic, code (non-greedy).
    pattern = re.compile(
        r"\[([^\]]+)\]\(([^)]+)\)"  # [label](url)
        r"|\*\*\*(.+?)\*\*\*"  # ***bold italic***
        r"|\*\*(.+?)\*\*"  # **bold**
        r"|~~(.+?)~~"  # ~~strikethrough~~
        r"|(?<!\*)\*(?!\*)(.+?)(?<!\*)\*(?!\*)"  # *italic* (avoid matching **)
        r"|(?<!`)`([^`\n]+)`(?!`)"  # `code`
    )
    nodes: List[Dict[str, Any]] = []
    last = 0
    for m in pattern.finditer(s):
        if m.start() > last:
            chunk = s[last : m.start()]
            if chunk:
                nodes.append({"type": "text", "text": chunk})
        if m.group(1) is not None and m.group(2) is not None:
            nodes.append(
                {
                    "type": "text",
                    "text": m.group(1),
                    "marks": [{"type": "link", "attrs": {"href": m.group(2).strip()}}],
                }
            )
        elif m.group(3) is not None:
            nodes.append(
                {
                    "type": "text",
                    "text": m.group(3),
                    "marks": [{"type": "strong"}, {"type": "em"}],
                }
            )
        elif m.group(4) is not None:
            nodes.append(
                {"type": "text", "text": m.group(4), "marks": [{"type": "strong"}]}
            )
        elif m.group(5) is not None:
            nodes.append(
                {"type": "text", "text": m.group(5), "marks": [{"type": "strike"}]}
            )
        elif m.group(6) is not None:
            nodes.append(
                {"type": "text", "text": m.group(6), "marks": [{"type": "em"}]}
            )
        elif m.group(7) is not None:
            nodes.append(
                {"type": "text", "text": m.group(7), "marks": [{"type": "code"}]}
            )
        last = m.end()
    if last < len(s):
        tail = s[last:]
        if tail:
            nodes.append({"type": "text", "text": tail})
    return nodes


def _split_table_row(line: str) -> List[str]:
    """Split a pipe-delimited table row into cells, stripping whitespace."""
    parts = line.split("|")
    cells = []
    for p in parts:
        c = p.strip()
        if c:
            cells.append(c)
    return cells


def _is_separator_row(cells: List[str]) -> bool:
    """Check if a row is a GFM table separator (e.g., |---|---|)."""
    return bool(cells) and all(re.fullmatch(r"-{3,}", c.strip()) for c in cells)


def _is_table_row(line: str) -> bool:
    """Detect if a line looks like a GFM table row."""
    stripped = line.strip()
    if not stripped.startswith("|"):
        return False
    cells = _split_table_row(stripped)
    return len(cells) >= 1


def _parse_table(lines: List[str], start: int, n: int):
    """
    Parse a GFM table starting at index start.
    Returns (adf_table_dict, next_index).
    """
    header_cells = _split_table_row(lines[start].strip())
    num_cols = len(header_cells)
    i = start + 1

    # Skip separator row if present
    if i < n and _is_separator_row(_split_table_row(lines[i].strip())):
        i += 1

    data_rows: List[List[str]] = []
    while i < n:
        s = lines[i].strip()
        if not s or not _is_table_row(s):
            break
        row_cells = _split_table_row(s)
        # Pad or trim to match header column count
        while len(row_cells) < num_cols:
            row_cells.append("")
        data_rows.append(row_cells[:num_cols])
        i += 1

    # Build ADF table
    def _make_cell(text: str, is_header: bool = False) -> Dict[str, Any]:
        cell_type = "tableHeader" if is_header else "tableCell"
        inline_nodes = parse_inline_adf(text) or _text_nodes(text or " ")
        return {
            "type": cell_type,
            "content": [{"type": "paragraph", "content": inline_nodes}],
        }

    def _make_row(cell_texts: List[str], is_header: bool = False) -> Dict[str, Any]:
        return {
            "type": "tableRow",
            "content": [_make_cell(c, is_header=is_header) for c in cell_texts],
        }

    table_content: List[Dict[str, Any]] = [_make_row(header_cells, is_header=True)]
    for dr in data_rows:
        table_content.append(_make_row(dr))

    return {"type": "table", "content": table_content}, i


def _paragraph_from_buffer(lines: List[str]) -> Optional[Dict[str, Any]]:
    if not lines:
        return None
    inner: List[Dict[str, Any]] = []
    for i, line in enumerate(lines):
        if i > 0:
            inner.append({"type": "hardBreak"})
        inner.extend(parse_inline_adf(line) or _text_nodes(line))
    if not inner:
        return None
    return {"type": "paragraph", "content": inner}


def markdown_to_adf(text: str) -> Dict[str, Any]:
    """
    Convert common Markdown to Atlassian Document Format (headings, lists, bold, code, links).
    Jira REST API stores comments/descriptions as ADF; plain paragraphs showed ### literally.
    """
    text = text or ""
    if not text.strip():
        return _empty_doc()

    lines = text.split("\n")
    content: List[Dict[str, Any]] = []
    n = len(lines)
    i = 0
    para_buf: List[str] = []

    def flush_paragraph() -> None:
        nonlocal para_buf
        if not para_buf:
            return
        p = _paragraph_from_buffer(para_buf)
        para_buf = []
        if p:
            content.append(p)

    while i < n:
        raw = lines[i]
        stripped = raw.strip()

        if not stripped:
            flush_paragraph()
            i += 1
            continue

        # ATX heading # .. ######
        hm = re.match(r"^(#{1,6})\s+(.+)$", stripped)
        if hm and len(hm.group(1)) <= 6:
            flush_paragraph()
            level = len(hm.group(1))
            title = hm.group(2).strip()
            title_nodes = parse_inline_adf(title) or _text_nodes(title)
            content.append(
                {"type": "heading", "attrs": {"level": level}, "content": title_nodes}
            )
            i += 1
            continue

        # Horizontal rule
        if re.fullmatch(r"[-*_]{3,}", stripped):
            flush_paragraph()
            content.append({"type": "rule"})
            i += 1
            continue

        # GFM Table: | col1 | col2 |
        if _is_table_row(stripped):
            flush_paragraph()
            table_adf, i = _parse_table(lines, i, n)
            content.append(table_adf)
            continue

        # Fenced code block
        if stripped.startswith("```"):
            flush_paragraph()
            lang = stripped[3:].strip() or "plaintext"
            code_lines: List[str] = []
            i += 1
            while i < n:
                if lines[i].strip().startswith("```"):
                    i += 1
                    break
                code_lines.append(lines[i])
                i += 1
            code_text = "\n".join(code_lines)
            content.append(
                {
                    "type": "codeBlock",
                    "attrs": {"language": lang},
                    "content": [{"type": "text", "text": code_text or " "}],
                }
            )
            continue

        # Bullet list (consecutive - or *)
        if re.match(r"^[-*]\s+", stripped):
            flush_paragraph()
            items: List[str] = []
            while i < n:
                s = lines[i].strip()
                if not s:
                    break
                bm = re.match(r"^[-*]\s+(.*)$", s)
                if not bm:
                    break
                items.append(bm.group(1))
                i += 1
            if items:
                content.append(
                    {
                        "type": "bulletList",
                        "content": [
                            {
                                "type": "listItem",
                                "content": [
                                    {
                                        "type": "paragraph",
                                        "content": parse_inline_adf(it)
                                    or _text_nodes(it),
                                    }
                                ],
                            }
                            for it in items
                        ],
                    }
                )
            continue

        # Ordered list
        om = re.match(r"^(\d+)\.\s+(.*)$", stripped)
        if om:
            flush_paragraph()
            start_order = int(om.group(1))
            items = [om.group(2)]
            i += 1
            while i < n:
                s = lines[i].strip()
                if not s:
                    break
                m = re.match(r"^\d+\.\s+(.*)$", s)
                if not m:
                    break
                items.append(m.group(1))
                i += 1
            content.append(
                {
                    "type": "orderedList",
                    "attrs": {"order": start_order},
                    "content": [
                        {
                            "type": "listItem",
                            "content": [
                                {
                                    "type": "paragraph",
                                    "content": parse_inline_adf(it)
                                    or _text_nodes(it),
                                }
                            ],
                        }
                        for it in items
                    ],
                }
            )
            continue

        # Blockquote: single line > text
        if stripped.startswith("> "):
            flush_paragraph()
            quote_lines: List[str] = [stripped[2:].strip()]
            i += 1
            while i < n and lines[i].strip().startswith("> "):
                quote_lines.append(lines[i].strip()[2:].strip())
                i += 1
            q_inner: List[Dict[str, Any]] = []
            for j, ql in enumerate(quote_lines):
                if j > 0:
                    q_inner.append({"type": "hardBreak"})
                q_inner.extend(parse_inline_adf(ql) or _text_nodes(ql))
            content.append(
                {
                    "type": "blockquote",
                    "content": [{"type": "paragraph", "content": q_inner}],
                }
            )
            continue

        para_buf.append(raw)
        i += 1

    flush_paragraph()

    if not content:
        return _empty_doc()
    return {"type": "doc", "version": 1, "content": content}


def plain_to_adf(text: str) -> Dict[str, Any]:
    """Backward-compatible name: Markdown-aware conversion for Jira ADF."""
    return markdown_to_adf(text)


def unescape_text(text: str) -> str:
    """
    Unescape common escape sequences that may appear as literal strings
    in CLI arguments or generated content.
    Converts \\n -> newline, \\t -> tab, \\\\ -> backslash, etc.
    """
    if not text:
        return text
    result = text.replace("\\\\", "\x00ESCAPED_BACKSLASH\x00")
    result = result.replace("\\n", "\n")
    result = result.replace("\\t", "\t")
    result = result.replace("\\r", "\r")
    result = result.replace("\x00ESCAPED_BACKSLASH\x00", "\\")
    return result


def _out(data: Any, fmt: str) -> None:
    if fmt == "json":
        print(json.dumps(data, ensure_ascii=False, indent=2))
    else:
        print(data)


def cmd_myself(base: str, email: str, token: str, fmt: str) -> int:
    url = f"{base}/rest/api/3/myself"
    r = requests.get(url, headers=session_headers(email, token, json_body=False), timeout=60)
    if r.status_code != 200:
        print(f"ERROR: HTTP {r.status_code}", file=sys.stderr)
        print(r.text[:800], file=sys.stderr)
        return 1
    j = r.json()
    if fmt == "json":
        _out(j, "json")
    else:
        print(f"accountId: {j.get('accountId')}")
        print(f"displayName: {j.get('displayName')}")
        print(f"email: {j.get('emailAddress')}")
    return 0


def cmd_search(base: str, email: str, token: str, jql: str, limit: int, fmt: str) -> int:
    url = f"{base}/rest/api/3/search/jql"
    payload = {
        "jql": jql,
        "maxResults": limit,
        "fields": ["key", "summary", "status", "assignee", "updated"],
    }
    r = requests.post(
        url,
        headers=session_headers(email, token),
        json=payload,
        timeout=60,
    )
    if r.status_code != 200:
        print(f"ERROR: HTTP {r.status_code}", file=sys.stderr)
        print(r.text[:800], file=sys.stderr)
        return 1
    data = r.json()
    issues = data.get("issues") or []
    if fmt == "json":
        _out(data, "json")
        return 0
    if not issues:
        print("(no issues)")
        return 0
    for i, iss in enumerate(issues, 1):
        f = iss.get("fields") or {}
        st = (f.get("status") or {}).get("name") or "?"
        summ = (f.get("summary") or "")[:80]
        print(f"{i}. {iss.get('key')}  [{summ}]  status: {st}")
    return 0


def cmd_issue(base: str, email: str, token: str, key: str, fmt: str) -> int:
    url = f"{base}/rest/api/3/issue/{key}"
    params = {"fields": "summary,status,assignee,project,description,created,updated"}
    r = requests.get(
        url,
        headers=session_headers(email, token, json_body=False),
        params=params,
        timeout=60,
    )
    if r.status_code != 200:
        print(f"ERROR: HTTP {r.status_code}", file=sys.stderr)
        print(r.text[:800], file=sys.stderr)
        return 1
    j = r.json()
    if fmt == "json":
        _out(j, "json")
        return 0
    f = j.get("fields") or {}
    print(f"key: {j.get('key')}")
    print(f"summary: {f.get('summary')}")
    print(f"status: {(f.get('status') or {}).get('name')}")
    proj = f.get("project") or {}
    print(f"project: {proj.get('key')} {proj.get('name')}")
    asn = f.get("assignee")
    print(f"assignee: {(asn or {}).get('displayName') or asn}")
    return 0


def cmd_create(
    base: str,
    email: str,
    token: str,
    project: str,
    summary: str,
    description: str,
    issuetype: str,
    assign_self: bool,
    fmt: str,
) -> int:
    fields: Dict[str, Any] = {
        "project": {"key": project},
        "summary": summary.strip(),
        "description": plain_to_adf(description),
        "issuetype": {"name": issuetype},
    }
    if assign_self:
        mr = requests.get(
            f"{base}/rest/api/3/myself",
            headers=session_headers(email, token, json_body=False),
            timeout=30,
        )
        if mr.status_code == 200:
            aid = mr.json().get("accountId")
            if aid:
                fields["assignee"] = {"accountId": aid}
    payload = {"fields": fields}
    url = f"{base}/rest/api/3/issue"
    r = requests.post(url, headers=session_headers(email, token), json=payload, timeout=60)
    if r.status_code not in (200, 201):
        # Retry without assignee if permission error
        if r.status_code == 400 and assign_self and "assignee" in fields:
            del fields["assignee"]
            r2 = requests.post(
                url, headers=session_headers(email, token), json={"fields": fields}, timeout=60
            )
            if r2.status_code in (200, 201):
                key = r2.json().get("key")
                print(f"WARN: created without assignee (API rejected assignee). key={key}", file=sys.stderr)
                if fmt == "json":
                    _out(r2.json(), "json")
                else:
                    print(key)
                return 0
        print(f"ERROR: HTTP {r.status_code}", file=sys.stderr)
        print(r.text[:1200], file=sys.stderr)
        return 1
    j = r.json()
    if fmt == "json":
        _out(j, "json")
    else:
        print(j.get("key", j))
    return 0


def cmd_comment(base: str, email: str, token: str, issue: str, body: str, fmt: str) -> int:
    url = f"{base}/rest/api/3/issue/{issue.strip()}/comment"
    payload = {"body": plain_to_adf(body)}
    r = requests.post(url, headers=session_headers(email, token), json=payload, timeout=60)
    if r.status_code not in (200, 201):
        print(f"ERROR: HTTP {r.status_code}", file=sys.stderr)
        print(r.text[:800], file=sys.stderr)
        return 1
    j = r.json()
    if fmt == "json":
        _out(j, "json")
    else:
        print(f"OK comment id={j.get('id')}")
    return 0


def read_body(description: Optional[str], path: Optional[str]) -> str:
    if path:
        p = Path(path)
        if not p.is_file():
            raise FileNotFoundError(f"File not found: {p}")
        return unescape_text(p.read_text(encoding="utf-8"))
    return unescape_text(description or "")


def main() -> int:
    parser = argparse.ArgumentParser(description="Jira REST CLI (same token as upload_attachment)")
    parser.add_argument(
        "--config",
        default=None,
        help=f"path to {CONFIG_FILENAME} (default: cwd / script dir / skill dir)",
    )
    parser.add_argument("--format", choices=("text", "json"), default="text", dest="fmt")

    sub = parser.add_subparsers(dest="cmd", required=True)

    p_my = sub.add_parser("myself", help="GET /myself (accountId, email)")
    p_my.set_defaults(func="myself")

    p_se = sub.add_parser("search", help="JQL search")
    p_se.add_argument("--jql", default=DEFAULT_JQL, help="JQL (default: G3SF in-progress for current user)")
    p_se.add_argument("--limit", type=int, default=5)
    p_se.set_defaults(func="search")

    p_is = sub.add_parser("issue", help="Get issue by key")
    p_is.add_argument("key", help="e.g. G3SF-123")
    p_is.set_defaults(func="issue")

    p_cr = sub.add_parser("create", help="Create issue (Task)")
    p_cr.add_argument("--project", default="G3SF")
    p_cr.add_argument("--summary", required=True)
    p_cr.add_argument(
        "--description",
        default="",
        help="Markdown → ADF (headings, lists, bold, code, links)",
    )
    p_cr.add_argument("--description-file", dest="description_file", default=None)
    p_cr.add_argument("--issuetype", default="Task", dest="issuetype")
    p_cr.add_argument(
        "--no-assign-self",
        action="store_true",
        help="do not set assignee to API token user",
    )
    p_cr.set_defaults(func="create")

    p_co = sub.add_parser("comment", help="Add comment to issue")
    p_co.add_argument("--issue", required=True)
    p_co.add_argument("--body", default="")
    p_co.add_argument("--body-file", dest="body_file", default=None)
    p_co.set_defaults(func="comment")

    args = parser.parse_args()
    script_dir = Path(__file__).resolve().parent
    try:
        cfg = load_config(args.config, script_dir)
        base, email, token = require_credentials(cfg)
    except ValueError as e:
        print(f"ERROR: {e}", file=sys.stderr)
        return 1
    except FileNotFoundError as e:
        print(f"ERROR: {e}", file=sys.stderr)
        return 1

    fmt = args.fmt
    if args.func == "myself":
        return cmd_myself(base, email, token, fmt)
    if args.func == "search":
        return cmd_search(base, email, token, args.jql, args.limit, fmt)
    if args.func == "issue":
        return cmd_issue(base, email, token, args.key.strip(), fmt)
    if args.func == "create":
        try:
            desc = read_body(args.description, args.description_file)
        except FileNotFoundError as e:
            print(f"ERROR: {e}", file=sys.stderr)
            return 1
        return cmd_create(
            base,
            email,
            token,
            args.project,
            args.summary,
            desc,
            args.issuetype,
            assign_self=not args.no_assign_self,
            fmt=fmt,
        )
    if args.func == "comment":
        try:
            body = read_body(args.body, args.body_file)
        except FileNotFoundError as e:
            print(f"ERROR: {e}", file=sys.stderr)
            return 1
        if not body.strip():
            print("ERROR: --body or --body-file required", file=sys.stderr)
            return 1
        return cmd_comment(base, email, token, args.issue, body, fmt)
    return 1


if __name__ == "__main__":
    sys.exit(main())