plane-agentd/agentd/plane.py

"""Minimal Plane REST client (API v1, header X-API-Key) plus HTML<->text helpers.

One client per identity: agentd posts with the token of the terminal that executes the run."""
import html
import json
import re
import urllib.error
import urllib.request


class PlaneError(Exception):
    pass


def html_to_text(h):
    h = re.sub(r"(?i)<li[^>]*>", "- ", h or "")
    h = re.sub(r"(?i)<br\s*/?>|</p>|</div>|</li>|</pre>|</h\d>|</tr>", "\n", h)
    text = html.unescape(re.sub(r"<[^>]+>", "", h))
    text = re.sub(r"[ \t]+\n", "\n", text)
    return re.sub(r"\n{3,}", "\n\n", text).strip()


def text_to_html(text, pre=None):
    out = "<p>%s</p>" % html.escape(text.strip(), quote=False).replace("\n", "<br>")
    if pre:
        out += "<pre>%s</pre>" % html.escape(pre, quote=False)
    return out


def brief_to_html(text):
    """Work-item description from a brief file: raw HTML passes through, anything else goes in <pre>."""
    return text if text.lstrip().startswith("<") else "<pre>%s</pre>" % html.escape(text, quote=False)


class Plane:
    def __init__(self, base_url, workspace, token, timeout=60):
        if not token:
            raise PlaneError("no Plane token for this identity (is the vault key injected?)")
        self.root = "%s/api/v1/workspaces/%s/" % (base_url.rstrip("/"), workspace)
        self._token = token
        self.timeout = timeout

    def __repr__(self):
        return "Plane(%s)" % self.root

    def request(self, method, path, body=None):
        url = path if path.startswith("http") else self.root + path
        data = json.dumps(body).encode() if body is not None else None
        req = urllib.request.Request(url, data=data, method=method)
        req.add_header("X-API-Key", self._token)
        req.add_header("Content-Type", "application/json")
        try:
            with urllib.request.urlopen(req, timeout=self.timeout) as resp:
                raw = resp.read().decode()
        except urllib.error.HTTPError as exc:
            detail = exc.read().decode(errors="replace")[:300]
            raise PlaneError("%s %s -> HTTP %s %s" % (method, path, exc.code, detail)) from None
        except urllib.error.URLError as exc:
            raise PlaneError("%s %s -> %s" % (method, path, exc.reason)) from None
        return json.loads(raw) if raw else None

    def paged(self, path):
        """Follow Plane cursor pagination and return every result."""
        out, cursor = [], None
        while True:
            sep = "&" if "?" in path else "?"
            r = self.request("GET", path + sep + "per_page=100" + ("&cursor=" + cursor if cursor else ""))
            if isinstance(r, list):
                return r
            out += r.get("results", [])
            if not r.get("next_page_results") or not r.get("next_cursor"):
                return out
            cursor = r["next_cursor"]

    # ---------------------------------------------------------------- lookups (name -> id)
    def states(self, project):
        return {s["name"]: s["id"] for s in self.paged("projects/%s/states/" % project)}

    def labels(self, project):
        return {l["name"]: l["id"] for l in self.paged("projects/%s/labels/" % project)}

    def members(self):
        m = self.request("GET", "members/")
        if isinstance(m, dict):
            m = m.get("results", [])
        return {x["display_name"]: x["id"] for x in m if x.get("display_name")}

    def modules(self, project):
        return {m["name"]: m["id"] for m in self.paged("projects/%s/modules/" % project)}

    # ---------------------------------------------------------------- work items
    def list_issues(self, project):
        return self.paged("projects/%s/issues/" % project)

    def get_issue(self, project, issue):
        return self.request("GET", "projects/%s/issues/%s/" % (project, issue))

    def find_issue(self, project, seq):
        for i in self.list_issues(project):
            if i.get("sequence_id") == seq:
                return i
        raise PlaneError("no work item with sequence %s" % seq)

    def create_issue(self, project, body):
        return self.request("POST", "projects/%s/issues/" % project, body)

    def patch_issue(self, project, issue, **fields):
        return self.request("PATCH", "projects/%s/issues/%s/" % (project, issue), fields)

    def set_state(self, project, issue, state_id, assignee_ids=None):
        fields = {"state": state_id}
        if assignee_ids is not None:
            fields["assignees"] = assignee_ids
        return self.patch_issue(project, issue, **fields)

    def set_assignees(self, project, issue, assignee_ids):
        return self.patch_issue(project, issue, assignees=assignee_ids)

    def add_to_module(self, project, module, issue):
        return self.request("POST", "projects/%s/modules/%s/module-issues/" % (project, module), {"issues": [issue]})

    def create_module(self, project, name):
        return self.request("POST", "projects/%s/modules/" % project, {"name": name})

    # ---------------------------------------------------------------- comments
    def list_comments(self, project, issue):
        cs = self.paged("projects/%s/issues/%s/comments/" % (project, issue))
        return sorted(cs, key=lambda c: c.get("created_at", ""))

    def post_comment(self, project, issue, comment_html):
        return self.request("POST", "projects/%s/issues/%s/comments/" % (project, issue), {"comment_html": comment_html})