kadence/desktop/kadence_desktop/integrations/notifications.py

"""Desktop notifications with buttons, over org.freedesktop.Notifications.

Qt's `QSystemTrayIcon.showMessage` can only show text. The real spec — which
KDE, GNOME and anything using a freedesktop daemon implement — takes *actions*,
so "Pomodoro finished" can carry [Start break] [Skip] [+5 min] and act without
you finding the window. Notifications are also replaced in place (the same id),
so a running timer never stacks up a column of toasts.
"""

from __future__ import annotations

import logging
from collections.abc import Callable

from PyQt6.QtCore import QMetaType, QObject, QVariant, pyqtSlot
from PyQt6.QtDBus import QDBusArgument, QDBusConnection, QDBusInterface, QDBusMessage

log = logging.getLogger(__name__)

_SERVICE = "org.freedesktop.Notifications"
_PATH = "/org/freedesktop/Notifications"
_IFACE = "org.freedesktop.Notifications"

# Urgency levels from the freedesktop spec.
URGENCY_LOW, URGENCY_NORMAL, URGENCY_CRITICAL = 0, 1, 2


class Notifier(QObject):
    """Sends notifications and routes their button presses back to callables."""

    def __init__(self, app_name: str = "Kadence", parent: QObject | None = None) -> None:
        super().__init__(parent)
        self._app_name = app_name
        self._bus = QDBusConnection.sessionBus()
        self._interface = QDBusInterface(_SERVICE, _PATH, _IFACE, self._bus)
        self._handlers: dict[int, dict[str, Callable[[], None]]] = {}
        self._last_id = 0
        self._available = self._interface.isValid()
        if self._available:
            self._bus.connect(_SERVICE, _PATH, _IFACE, "ActionInvoked", self._on_action)
            self._bus.connect(_SERVICE, _PATH, _IFACE, "NotificationClosed", self._on_closed)

    @property
    def available(self) -> bool:
        return self._available

    def capabilities(self) -> list[str]:
        if not self._available:
            return []
        reply = self._interface.call("GetCapabilities")
        arguments = reply.arguments()
        return list(arguments[0]) if arguments else []

    def notify(
        self,
        summary: str,
        body: str = "",
        *,
        actions: dict[str, tuple[str, Callable[[], None]]] | None = None,
        urgency: int = URGENCY_NORMAL,
        timeout_ms: int = 8000,
        replace: bool = True,
        icon: str = "kadence",
        sound: bool = True,
    ) -> int:
        """Show a notification. `actions` maps key -> (label, callback)."""
        if not self._available:
            return 0

        action_list: list[str] = []
        handlers: dict[str, Callable[[], None]] = {}
        for key, (label, callback) in (actions or {}).items():
            action_list += [key, label]
            handlers[key] = callback

        hints = {
            "urgency": QVariant(urgency),
            "desktop-entry": QVariant("kadence"),
            "x-kde-origin-name": QVariant(self._app_name),
        }
        if not sound:
            hints["suppress-sound"] = QVariant(True)

        # The spec's signature is `susssasa{sv}i`: `replaces_id` is a uint32 and
        # `actions` is an array of strings. Python ints marshal as int32 and a
        # Python list as an array of variants, so both need spelling out or the
        # daemon rejects the call as "no such method".
        reply = self._interface.call(
            "Notify",
            self._app_name,
            QDBusArgument(self._last_id if replace else 0, QMetaType.Type.UInt.value),
            icon,
            summary,
            body,
            QDBusArgument(action_list, QMetaType.Type.QStringList.value),
            hints,
            timeout_ms,
        )
        if reply.errorName():
            log.warning("notification refused: %s — %s", reply.errorName(), reply.errorMessage())
            return 0

        arguments = reply.arguments()
        try:
            notification_id = int(arguments[0]) if arguments else 0
        except (TypeError, ValueError):
            log.warning("unexpected Notify reply: %r", arguments)
            return 0

        self._last_id = notification_id
        if handlers:
            self._handlers[notification_id] = handlers
        return notification_id

    def close(self, notification_id: int) -> None:
        if self._available and notification_id:
            self._interface.call("CloseNotification", notification_id)

    @pyqtSlot(QDBusMessage)
    def _on_action(self, message: QDBusMessage) -> None:
        arguments = message.arguments()
        if len(arguments) < 2:
            return
        notification_id, key = int(arguments[0]), str(arguments[1])
        handler = self._handlers.get(notification_id, {}).get(key)
        if handler:
            handler()

    @pyqtSlot(QDBusMessage)
    def _on_closed(self, message: QDBusMessage) -> None:
        arguments = message.arguments()
        if arguments:
            self._handlers.pop(int(arguments[0]), None)