Notify 1.0.0

notify_setup.py

1.7 KB · raw

"""Show, or change, where this Aworg's notifications go."""

from __future__ import annotations

import re

from aworg.tools.base import ToolContext, ToolError, ToolResult

from ._topic import load, save, subscribe_text

NAME = "notify_setup"

DESCRIPTION = (
    "Show the ntfy topic this Aworg sends notifications to and how the owner "
    "subscribes to it. Give a topic or server to change them, for example to "
    "an ntfy server the owner runs."
)

INPUT_SCHEMA = {
    "type": "object",
    "properties": {
        "topic": {
            "type": "string",
            "description": "A new topic name: letters, digits, - and _, at most 64.",
        },
        "server": {
            "type": "string",
            "description": "A different ntfy server address. Defaults to https://ntfy.sh.",
        },
    },
}


async def run(context: ToolContext, topic: str = "", server: str = "") -> ToolResult:
    data, _ = load()
    changed = []
    if topic:
        topic = str(topic).strip()
        if not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", topic):
            raise ToolError("A topic is 1-64 letters, digits, '-' or '_'.")
        data["topic"] = topic
        changed.append("topic")
    if server:
        server = str(server).strip().rstrip("/")
        if not server.startswith(("http://", "https://")):
            raise ToolError("The server must be an http:// or https:// address.")
        data["server"] = server
        changed.append("server")
    if changed:
        save(data)
    head = f"Changed the {' and '.join(changed)}. " if changed else ""
    return ToolResult(
        text=f"{head}Notifications go to topic {data['topic']} on {data['server']}. {subscribe_text(data)}",
        summary=data["topic"],
    )