|
# retoor <retoor@molodetz.nl>
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from typing import Any
|
|
|
|
from devplacepy.database import (
|
|
NOTIFICATION_CHANNELS,
|
|
get_notification_prefs,
|
|
reset_notification_prefs,
|
|
set_notification_pref,
|
|
)
|
|
from devplacepy.services.devii.errors import ToolInputError
|
|
|
|
logger = logging.getLogger("devii.notification")
|
|
|
|
|
|
class NotificationController:
|
|
def __init__(self, owner_kind: str, owner_id: str) -> None:
|
|
self._owner_kind = owner_kind
|
|
self._owner_id = owner_id
|
|
|
|
async def dispatch(self, name: str, arguments: dict[str, Any]) -> str:
|
|
if name == "notification_list":
|
|
return self._list()
|
|
if name == "notification_set":
|
|
return self._set(arguments)
|
|
if name == "notification_reset":
|
|
return self._reset()
|
|
raise ToolInputError(f"Unknown notification tool: {name}")
|
|
|
|
def _require_user(self) -> None:
|
|
if self._owner_kind != "user":
|
|
raise ToolInputError(
|
|
"Notification preferences are only available for signed-in users."
|
|
)
|
|
|
|
def _channel(self, arguments: dict[str, Any]) -> str:
|
|
channel = str(arguments.get("channel", "")).strip().lower()
|
|
if channel not in NOTIFICATION_CHANNELS:
|
|
raise ToolInputError("'channel' must be 'in_app', 'push' or 'telegram'.")
|
|
return channel
|
|
|
|
def _value(self, arguments: dict[str, Any]) -> bool:
|
|
raw = arguments.get("value")
|
|
if isinstance(raw, bool):
|
|
return raw
|
|
return str(raw).strip().lower() in ("true", "1", "yes", "on")
|
|
|
|
def _list(self) -> str:
|
|
self._require_user()
|
|
return json.dumps(
|
|
{
|
|
"status": "success",
|
|
"notifications": get_notification_prefs(self._owner_id),
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
def _set(self, arguments: dict[str, Any]) -> str:
|
|
self._require_user()
|
|
notification_type = str(arguments.get("notification_type", "")).strip()
|
|
if not notification_type:
|
|
raise ToolInputError("'notification_type' is required.")
|
|
channel = self._channel(arguments)
|
|
enabled = self._value(arguments)
|
|
try:
|
|
set_notification_pref(
|
|
self._owner_id, notification_type, channel, enabled
|
|
)
|
|
except ValueError as exc:
|
|
raise ToolInputError(str(exc))
|
|
return json.dumps(
|
|
{
|
|
"status": "success",
|
|
"notification_type": notification_type,
|
|
"channel": channel,
|
|
"enabled": enabled,
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
def _reset(self) -> str:
|
|
self._require_user()
|
|
removed = reset_notification_prefs(self._owner_id)
|
|
return json.dumps(
|
|
{"status": "success", "removed": removed},
|
|
ensure_ascii=False,
|
|
)
|