From 58d392ca65fef92411987f85f99e742b9b3770d9 Mon Sep 17 00:00:00 2001 From: Kovid Goyal Date: Thu, 24 Jun 2021 16:40:31 +0530 Subject: [PATCH] Replace python-dbus with jeepney for notifications --- src/calibre/gui2/notify.py | 146 +++++++++++++++++++++++-------------- 1 file changed, 90 insertions(+), 56 deletions(-) diff --git a/src/calibre/gui2/notify.py b/src/calibre/gui2/notify.py index 384ddbf436..85c5bcf760 100644 --- a/src/calibre/gui2/notify.py +++ b/src/calibre/gui2/notify.py @@ -7,15 +7,15 @@ __copyright__ = '2009, Kovid Goyal ' __docformat__ = 'restructuredtext en' -import time +import sys +from contextlib import suppress from functools import lru_cache -from calibre import prints -from calibre.constants import DEBUG, get_osx_version, islinux, ismacos +from calibre.constants import DEBUG, __appname__, get_osx_version, islinux, ismacos from polyglot.builtins import unicode_type -class Notifier(object): +class Notifier: DEFAULT_TIMEOUT = 5000 @@ -37,47 +37,76 @@ def icon(data=False): class DBUSNotifier(Notifier): - def __init__(self, session_bus): - self.ok, self.err = True, None - server, path, interface = self.SERVICE - if DEBUG: - start = time.time() - prints('Looking for desktop notifier support from:', server) + def __init__(self): + self.initialized = False + + def initialize(self): + from jeepney.io.blocking import open_dbus_connection + if self.initialized: + return + self.initialized = True + self.ok = False try: - import dbus - self.dbus = dbus - self._notify = dbus.Interface(session_bus.get_object(server, path), interface) - except Exception as err: - self.ok = False - self.err = unicode_type(err) + self.connection = open_dbus_connection(bus='SESSION') + except Exception: + return + with suppress(Exception): + self.ok = self.initialize_fdo() + if self.ok: + self.notify = self.fdo_notify + return if DEBUG: - prints(server, 'found' if self.ok else 'not found', 'in', '%.1f' % (time.time() - start), 'seconds') + print('Failed to connect to FDO Notifications service', file=sys.stderr) + with suppress(Exception): + self.ok = self.initialize_portal() + if self.ok: + self.notify = self.portal_notify + else: + print('Failed to connect to Portal Notifications service', file=sys.stderr) + def initialize_fdo(self): + from jeepney import DBusAddress, MessageType, new_method_call + self.address = DBusAddress( + '/org/freedesktop/Notifications', + bus_name='org.freedesktop.Notifications', + interface='org.freedesktop.Notifications') -class FDONotifier(DBUSNotifier): + msg = new_method_call(self.address, 'GetCapabilities') + reply = self.connection.send_and_get_reply(msg) + return bool(reply and reply.header.message_type is MessageType.method_return) - SERVICE = 'org.freedesktop.Notifications', '/org/freedesktop/Notifications', 'org.freedesktop.Notifications' + def initialize_portal(self): + from jeepney import DBusAddress, MessageType, Properties + self.address = DBusAddress( + '/org/freedesktop/portal/desktop', + bus_name='org.freedesktop.portal.Desktop', + interface='org.freedesktop.portal.Notification') + p = Properties(self.address) + msg = p.get('version') + reply = self.connection.send_and_get_reply(msg) + return bool(reply and reply.header.message_type is MessageType.method_return and reply.body[0][1] >= 1) - def __call__(self, body, summary=None, replaces_id=None, timeout=0): - if replaces_id is None: - replaces_id = self.dbus.UInt32() + def fdo_notify(self, body, summary=None, replaces_id=None, timeout=0): + from jeepney import new_method_call timeout, body, summary = self.get_msg_parms(timeout, body, summary) + msg = new_method_call( + self.address, 'Notify', 'susssasa{sv}i', + (__appname__, + replaces_id or 0, + icon(), + summary, + body, + [], {}, # Actions, hints + timeout, + )) try: - self._notify.Notify('calibre', replaces_id, icon(), summary, body, - self.dbus.Array(signature='s'), self.dbus.Dictionary({"desktop-entry": "calibre-gui"}, signature='sv'), - timeout) + self.connection.send(msg) except Exception: import traceback traceback.print_exc() - -class XDPNotifier(DBUSNotifier): - - SERVICE = 'org.freedesktop.portal.Desktop', '/org/freedesktop/portal/desktop', 'org.freedesktop.portal.Notification' - - def __call__(self, body, summary=None, replaces_id=None, timeout=0): - if replaces_id is None: - replaces_id = self.dbus.UInt32() + def portal_notify(self, body, summary=None, replaces_id=None, timeout=0): + from jeepney import new_method_call _, body, summary = self.get_msg_parms(timeout, body, summary) # Note: This backend does not natively support the notion of timeouts # @@ -93,27 +122,33 @@ class XDPNotifier(DBUSNotifier): # Doing that however, requires Calibre to first be converted to use # its AppID everywhere and then we still need a fallback for portable # installations. - + msg = new_method_call( + self.address, 'AddNotification', 'sa{sv}', ( + str(replaces_id or 0), + { + "title": ('s', summary), + "body": ('s', body), + "icon": ( + '(sv)', + ( + "bytes", + ('ay', icon(data=True)) + ) + ), + })) try: - self._notify.AddNotification(str(replaces_id), self.dbus.Dictionary({ - "title": self.dbus.String(summary), - "body": self.dbus.String(body), - "icon": self.dbus.Struct(("bytes", self.dbus.ByteArray(icon(data=True), variant_level=1)), signature='sv'), - }, signature='sv')) + self.connection.send(msg) except Exception: import traceback traceback.print_exc() - -def get_dbus_notifier(): - import dbus - session_bus = dbus.SessionBus() - names = frozenset(session_bus.list_names()) - for srv in (FDONotifier, XDPNotifier): - if srv.SERVICE[0] in names: - ans = srv(session_bus) - if ans.ok: - return ans + def __call__(self, body, summary=None, replaces_id=None, timeout=0): + self.initialize() + if not self.ok: + if DEBUG: + print('Failed to connect to any notification service', file=sys.stderr) + return + self.notify(body, summary, replaces_id, timeout) class QtNotifier(Notifier): @@ -178,12 +213,7 @@ class AppleNotifier(Notifier): def get_notifier(systray=None): ans = None if islinux: - try: - ans = get_dbus_notifier() - except Exception: - import traceback - traceback.print_exc() - ans = None + ans = DBUSNotifier() elif ismacos: if get_osx_version() >= (10, 8, 0): ans = AppleNotifier() @@ -200,6 +230,10 @@ def get_notifier(systray=None): return ans -if __name__ == '__main__': +def hello(): n = get_notifier() n('hello') + + +if __name__ == '__main__': + hello()