mozra.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/.

"""Unified CLI for Jira board helpers."""

import argparse
import sys
import textwrap

from jira_helpers import (
    ConfigError,
    JiraCliError,
    adf_to_text,
    fetch_issue,
    format_issue_list,
    get_board_columns,
    get_board_configuration,
    get_board_filter_query,
    get_default_project_key,
    get_project_override,
    issue_web_url,
    launch_editor,
    print_error,
    run_acli_text,
    search_issues,
    set_debug,
    split_jql_order_clause,
)


def _add_common_limit_options(parser: argparse.ArgumentParser) -> None:
    parser.add_argument(
        "--limit",
        type=int,
        default=50,
        help="Maximum number of issues to return.",
    )
    parser.add_argument(
        "--all",
        action="store_true",
        help="Fetch every matching issue (overrides --limit).",
    )


def _cmd_mine(args: argparse.Namespace) -> int:
    if args.all:
        limit = None
        fetch_all = True
    else:
        limit = args.limit
        fetch_all = False
    try:
        board = get_board_configuration()
        base_jql = get_board_filter_query()
        core_jql, _ = split_jql_order_clause(base_jql)
        clause = f"({core_jql}) AND assignee = currentUser()"
        if not args.include_done:
            clause += " AND resolution = EMPTY"
        jql = f"{clause} ORDER BY updated DESC"
        results = search_issues(
            jql,
            limit=limit,
            fetch_all=fetch_all,
            fields=["summary", "status", "assignee"],
        )
    except (ConfigError, JiraCliError) as exc:
        print_error(str(exc))
        return 1

    heading = f"{board.name} – assigned to you"
    print(heading)
    print("=" * len(heading))
    total = results.total if results.total is not None else len(results.issues)
    print(f"Showing {len(results.issues)} of {total} issues\n")
    print(format_issue_list(results.issues))
    return 0


def _cmd_view(args: argparse.Namespace) -> int:
    try:
        issue = fetch_issue(
            args.issue,
            fields=[
                "summary",
                "status",
                "assignee",
                "description",
                "reporter",
                "issuetype",
                "priority",
            ],
        )
    except (ConfigError, JiraCliError) as exc:
        print_error(str(exc))
        return 1

    fields = issue.get("fields") or {}
    summary = fields.get("summary") or ""
    status = (fields.get("status") or {}).get("name") or ""
    issue_type = (fields.get("issuetype") or {}).get("name") or ""
    priority = (fields.get("priority") or {}).get("name") or ""
    assignee = (fields.get("assignee") or {}).get("displayName") or "Unassigned"
    reporter = (fields.get("reporter") or {}).get("displayName") or ""
    description = adf_to_text(fields.get("description")).strip()

    heading = f"{issue.get('key')} (id {issue.get('id')})"
    print(heading)
    print("=" * len(heading))
    print(f"Status   : {status}")
    if issue_type:
        print(f"Type     : {issue_type}")
    if priority:
        print(f"Priority : {priority}")
    print(f"Assignee : {assignee}")
    if reporter:
        print(f"Reporter : {reporter}")
    print(f"Link     : {issue_web_url(issue.get('key'))}")
    print(f"Summary  : {summary}\n")
    print("Description:")
    if description:
        for paragraph in description.splitlines():
            if paragraph.strip():
                print(textwrap.fill(paragraph, width=100))
            else:
                print()
    else:
        print("(no description)")
    return 0


def _cmd_columns(_: argparse.Namespace) -> int:
    try:
        board = get_board_configuration()
    except (ConfigError, JiraCliError) as exc:
        print_error(str(exc))
        return 1

    heading = f"{board.name} (board {board.board_id})"
    print(heading)
    print("=" * len(heading))
    if not board.columns:
        print("No columns were returned for this board configuration.")
        return 0
    for column in board.columns:
        print(f"\n{column.name}")
        print("-" * len(column.name))
        if column.statuses:
            for status in column.statuses:
                print(f"  • {status}")
        else:
            print("  (no statuses mapped)")
    return 0


def _cmd_column_issues(args: argparse.Namespace) -> int:
    if args.all:
        limit = None
        fetch_all = True
    else:
        limit = args.limit
        fetch_all = False
    try:
        statuses, column_name = _resolve_statuses(args.name)
        base = get_board_filter_query()
        core, _ = split_jql_order_clause(base)
        quoted_parts = []
        for status in statuses:
            safe = status.replace('"', '\\"')
            quoted_parts.append(f'"{safe}"')
        quoted = ", ".join(quoted_parts)
        jql = f"({core}) AND status IN ({quoted}) ORDER BY Rank ASC"
        results = search_issues(
            jql,
            limit=limit,
            fetch_all=fetch_all,
            fields=["summary", "status", "assignee"],
        )
        board = get_board_configuration()
    except (ConfigError, JiraCliError) as exc:
        print_error(str(exc))
        return 1

    if len(statuses) == 1 and statuses[0].casefold() == column_name.casefold():
        descriptor = f"status '{statuses[0]}'"
    elif len(statuses) == 1:
        descriptor = f"status '{statuses[0]}' (column '{column_name}')"
    else:
        descriptor = f"column '{column_name}'"
    heading = f"{board.name} – {descriptor}"
    print(heading)
    print("=" * len(heading))
    total = results.total if results.total is not None else len(results.issues)
    print(f"Showing {len(results.issues)} of {total} issues\n")
    print(format_issue_list(results.issues))
    return 0


def _resolve_statuses(target: str) -> tuple[list[str], str]:
    normalized = target.casefold()
    for column in get_board_columns():
        if column.name.casefold() == normalized:
            if column.statuses:
                return column.statuses, column.name
            raise ConfigError(
                f"Column '{column.name}' does not have any mapped statuses."
            )
        for status in column.statuses:
            if status.casefold() == normalized:
                return [status], column.name
    raise ConfigError(
        f"No column or status matches '{target}'. Run 'mozra columns' to inspect the board."
    )


def _cmd_create(args: argparse.Namespace) -> int:
    try:
        board = get_board_configuration()
        default_project = get_default_project_key()
        project = args.project or get_project_override() or default_project
        if not project:
            raise ConfigError(
                "Specify --project, set project.txt/ACLI_PROJECT, or ensure the board exposes a project key."
            )
        cmd = [
            "jira",
            "workitem",
            "create",
            "--project",
            project,
            "--type",
            args.type,
            "--editor",
        ]
        if args.assignee:
            cmd.extend(["--assignee", args.assignee])
        if args.labels:
            cmd.extend(["--label", args.labels])
        if args.priority:
            cmd.extend(["--priority", args.priority])
        if args.parent:
            cmd.extend(["--parent", args.parent])
        run_acli_text(cmd, stream=True)
    except (ConfigError, JiraCliError) as exc:
        print_error(str(exc))
        return 1

    print(f"Issue creation completed on board '{board.name}'.")
    return 0


def _cmd_edit_description(args: argparse.Namespace) -> int:
    try:
        issue = fetch_issue(args.issue, fields=["summary", "description"])
    except (ConfigError, JiraCliError) as exc:
        print_error(str(exc))
        return 1

    issue_key = issue.get("key") or args.issue
    fields = issue.get("fields") or {}
    summary = fields.get("summary") or ""
    current_text = adf_to_text(fields.get("description")).rstrip()
    print(f"Editing description for {issue_key} – {summary}")
    print("Save and close the editor to push the updated content back to Jira.\n")
    temp_path = None
    try:
        new_text, temp_path = launch_editor(current_text, f"desc-{issue_key}")
    except (ConfigError, JiraCliError) as exc:
        print_error(str(exc))
        return 1

    try:
        if new_text.rstrip() == (current_text or "").rstrip():
            print("No changes detected; skipping update.")
            return 0
        output = run_acli_text(
            [
                "jira",
                "workitem",
                "edit",
                "--key",
                issue_key,
                "--description-file",
                str(temp_path),
                "--yes",
            ]
        )
        print(output)
        return 0
    except (ConfigError, JiraCliError) as exc:
        print_error(str(exc))
        return 1
    finally:
        if temp_path:
            try:
                temp_path.unlink()
            except FileNotFoundError:
                pass


def _cmd_comment(args: argparse.Namespace) -> int:
    cmd = [
        "jira",
        "workitem",
        "comment",
        "create",
        "--key",
        args.issue,
        "--editor",
    ]
    if args.edit_last:
        cmd.append("--edit-last")
    try:
        run_acli_text(cmd, stream=True)
    except (ConfigError, JiraCliError) as exc:
        print_error(str(exc))
        return 1
    print("Comment operation completed.")
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="mozra",
        description="Helper CLI for working with a single Jira board via ACLI.",
    )
    parser.add_argument(
        "--debug",
        action="store_true",
        help="Print verbose diagnostics (also enabled via ACLI_HELPERS_DEBUG=1).",
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    mine = subparsers.add_parser(
        "mine",
        help="List issues on the board assigned to you.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    mine.add_argument(
        "--include-done",
        action="store_true",
        help="Include resolved issues in the listing.",
    )
    _add_common_limit_options(mine)
    mine.set_defaults(func=_cmd_mine)

    view = subparsers.add_parser(
        "view",
        help="Show details for a single issue.",
    )
    view.add_argument("--issue", required=True, help="Issue key, e.g., ENG-123.")
    view.set_defaults(func=_cmd_view)

    columns = subparsers.add_parser(
        "columns",
        help="Display the board's columns and mapped statuses.",
    )
    columns.set_defaults(func=_cmd_columns)

    column_issues = subparsers.add_parser(
        "column-issues",
        help="List issues currently in a specific column or status.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    column_issues.add_argument(
        "--name",
        required=True,
        help="Column or status name to filter on.",
    )
    _add_common_limit_options(column_issues)
    column_issues.set_defaults(func=_cmd_column_issues)

    create = subparsers.add_parser(
        "create",
        help="Create a new issue using your $EDITOR via ACLI.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    create.add_argument(
        "--project",
        help="Project key (defaults to the board's project when available).",
    )
    create.add_argument("--type", default="Task", help="Issue type to create.")
    create.add_argument(
        "--assignee",
        help="Email or account ID to assign. Use '@me' to self-assign.",
    )
    create.add_argument(
        "--labels",
        help="Comma-separated labels to add to the issue.",
    )
    create.add_argument("--priority", help="Priority to set.")
    create.add_argument("--parent", help="Parent issue key for subtasks.")
    create.set_defaults(func=_cmd_create)

    edit_desc = subparsers.add_parser(
        "edit-description",
        help="Edit an issue's description using your $EDITOR.",
    )
    edit_desc.add_argument("--issue", required=True, help="Issue key to edit.")
    edit_desc.set_defaults(func=_cmd_edit_description)

    comment = subparsers.add_parser(
        "comment",
        help="Add or update a comment on an issue using your $EDITOR.",
    )
    comment.add_argument("--issue", required=True, help="Issue key to comment on.")
    comment.add_argument(
        "--edit-last",
        action="store_true",
        help="Edit your most recent comment instead of creating a new one.",
    )
    comment.set_defaults(func=_cmd_comment)

    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    set_debug(args.debug)
    return args.func(args)


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