mrpy.nvim/rplugin/python3/mrpy/main.py

from collections.abc import Callable, Sequence
from pathlib import Path
from typing import Any

from pynvim import Nvim, command, plugin
from pynvim.api import Buffer

from .clickup import ClickupSession, ClickupTask
from .hints import JSONData
from .yaml import load, dump


def usernames_from_objs(objs: JSONData) -> JSONData:
    assert isinstance(objs, list)
    return [k["username"] for k in objs if isinstance(k, dict)]


@plugin
class MrPyPlugin:
    nvim: Nvim
    clickup: ClickupSession
    frontmatter_keys: dict[str, Callable[[JSONData], JSONData]]

    def __init__(self, nvim: Nvim) -> None:
        self.nvim = nvim
        self.clickup = ClickupSession()
        self.frontmatter_keys = {
            "id": str,
            "name": str,
            "assignees": usernames_from_objs,
        }

    def select_task_id(self, tags: set[str] | None = None) -> None:
        tasks = self.clickup.get_tasks(**({"tags[]": "&tags[]=".join(tags)} if tags else {}))
        task_names_by_id = [
            {
                "idx": tix + 1,
                "id": t.id,
                "text": t.short,
                "name": t.name,
                "tags": t.tags,
                "status": t.status,
                "is_child": bool(t.parent),
                "preview": {
                    "text": dump(t.showables),
                    "ft": "yaml",
                },
                "action": f":Mrpy {t.id}",
            }
            for tix, t in enumerate(tasks)
            if not t.name.endswith("Absence")
        ]

        self.nvim.exec_lua(
            """require('snacks').picker.pick({
                title="Select Task",
                format = function(item, _)
				    local ret = {}

					local hl = "SnacksPickerComment"
					if item.status == "in progress" then
					    hl = "@method"
					elseif item.status == "selected for development" then
					    hl = "@constant.builtin"
					elseif item.status == "in review" then
					    hl = "@keyword"
					elseif item.status == "done" then
					    hl = "@variable.builtin"
					end

					if item.is_child == true then
				        ret[#ret + 1] = {  "󰘍 ", "SnacksPickerComment" }
					end

				    ret[#ret + 1] = { item.name, hl }

				    for _, v in ipairs(item.tags) do
				        ret[#ret + 1] = { " #" .. v, "SnacksPickerComment" }
				    end

				    ret[#ret + 1] = {
					    " 󰻾" .. item.id,
					    "SnacksPickerComment"
				    }
				    return ret
			    end,
                preview="preview",
                confirm=function(_, item) vim.cmd(("Mrpy %s"):format(item.id)) end,
                items=...
            })""",
            task_names_by_id,
        )

    def open_task_buffer(self, task_id: str) -> None:
        if " " in task_id:
            *_, last = task_id.split(" ")
            task_id = last.removeprefix("(#").removesuffix(")")

        task = self.clickup.get_task(task_id)
        temp_buf: Buffer = self.nvim.api.create_buf(True, False)
        self.nvim.api.buf_set_name(temp_buf, f"[ClickUp] {task.name}")
        self.nvim.buffers[temp_buf.number].options["filetype"] = "yaml"
        self.nvim.api.create_autocmd(
            ["BufWriteCmd"],
            {"buffer": temp_buf.number, "command": "MrpyPush " + str(temp_buf.number)},
        )

        content = [
            f"# yaml-language-server: $schema={Path(__file__).parent.parent.parent.parent}/schema.json ",
            "---",
        ]
        content.extend(
            dump(
                task.showables,
            ).splitlines()
        )

        self.nvim.api.buf_set_lines(temp_buf, 0, 0, False, content)
        self.nvim.buffers[temp_buf.number].options["modified"] = False
        self.nvim.api.win_set_buf(0, temp_buf)

    @command("Mrpy", nargs="*")
    def entry(self, args: Sequence[str] = ()) -> None:
        match args:
            case (str() as task_id,):
                try:
                    self.open_task_buffer(task_id)
                except:
                    self.select_task_id({task_id})
            case tags:
                self.select_task_id(set(tags))

    @command("MrpyPush", nargs="?")
    def on_vbuf_write(self, args: Sequence[str] = ()) -> None:
        match args:
            case "0", *_:
                return
            case buf_no, *_:
                buf_no = int(buf_no)
            case _:
                return

        try:
            a = self.nvim.buffers[buf_no][0:-1]
            fm = "\n".join(a)
            data = load(fm)
            assert isinstance(data, dict)

            data["markdown_content"] = data["markdown_content"].strip() + "\n"

            self.clickup.update(ClickupTask(**data))
            self.nvim.buffers[buf_no].options["modified"] = False
        except ValueError:
            pass

    @command("MrpyNew", nargs=0)
    def on_new_task(self) -> None:
        task = ClickupTask(name="", status="backlog", markdown_content="", parent_list="")
        temp_buf: Buffer = self.nvim.api.create_buf(True, False)
        self.nvim.api.buf_set_name(temp_buf, f"[ClickUp] New Task")
        self.nvim.buffers[temp_buf.number].options["filetype"] = "yaml"
        self.nvim.api.create_autocmd(
            ["BufWriteCmd"],
            {"buffer": temp_buf.number, "command": "MrpyPushNew " + str(temp_buf.number)},
        )

        content = [
            f"# yaml-language-server: $schema={Path(__file__).parent.parent.parent.parent}/schema.json ",
            "---",
        ]
        content.extend(
            dump(
                task.showables,
            ).splitlines()
        )

        self.nvim.api.buf_set_lines(temp_buf, 0, 0, False, content)
        self.nvim.buffers[temp_buf.number].options["modified"] = False
        self.nvim.api.win_set_buf(0, temp_buf)

    @command("MrpyPushNew", nargs=1)
    def on_new_vbuf_write(self, buf_no: Sequence[Any] = ()) -> None:
        buf_no = int(buf_no[0])

        try:
            a = self.nvim.buffers[buf_no][0:-1]
            fm = "\n".join(a)
            data = load(fm)
            assert isinstance(data, dict)

            data["markdown_content"] = data["markdown_content"].strip() + "\n"

            new_task = ClickupTask(**data)
            new_task_id = self.clickup.create(new_task)

            self.nvim.buffers[buf_no].options["modified"] = False
            self.nvim.command("bdelete " + str(buf_no))

            self.entry([new_task_id])
        except ValueError:
            pass