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"],
)