diff options
Diffstat (limited to 'qutebrowser/browser/webengine/notification.py')
-rw-r--r-- | qutebrowser/browser/webengine/notification.py | 176 |
1 files changed, 82 insertions, 94 deletions
diff --git a/qutebrowser/browser/webengine/notification.py b/qutebrowser/browser/webengine/notification.py index 2b77a5ac4..7ab769659 100644 --- a/qutebrowser/browser/webengine/notification.py +++ b/qutebrowser/browser/webengine/notification.py @@ -49,26 +49,20 @@ import itertools import functools import subprocess from typing import Any, List, Dict, Optional, Iterator, TYPE_CHECKING - -from PyQt5.QtCore import (Qt, QObject, QVariant, QMetaType, QByteArray, pyqtSlot, - pyqtSignal, QTimer, QProcess, QUrl) -from PyQt5.QtGui import QImage, QIcon, QPixmap -from PyQt5.QtDBus import (QDBusConnection, QDBusInterface, QDBus, QDBusServiceWatcher, - QDBusArgument, QDBusMessage, QDBusError) -from PyQt5.QtWidgets import QSystemTrayIcon +from qutebrowser.qt import QtWidgets if TYPE_CHECKING: # putting these behind TYPE_CHECKING also means this module is importable # on installs that don't have these - from PyQt5.QtWebEngineCore import QWebEngineNotification - from PyQt5.QtWebEngineWidgets import QWebEngineProfile + from qutebrowser.qt import QtWebEngineWidgets, QtWebEngineCore, QWebEngineNotification + from qutebrowser.qt import QWebEngineProfile from qutebrowser.config import config from qutebrowser.misc import objects from qutebrowser.utils import ( qtutils, log, utils, debug, message, version, objreg, resources, ) -from qutebrowser.qt import sip +from qutebrowser.qt import QtWebEngine, QtGui, QtDBus, QtCore, sip bridge: Optional['NotificationBridgePresenter'] = None @@ -105,7 +99,7 @@ class Error(Exception): """Raised when something goes wrong with notifications.""" -class AbstractNotificationAdapter(QObject): +class AbstractNotificationAdapter(QtCore.QObject): """An adapter taking notifications and displaying them. @@ -118,14 +112,14 @@ class AbstractNotificationAdapter(QObject): # Emitted by the adapter when the notification with the given ID was closed or # clicked by the user. - close_id = pyqtSignal(int) - click_id = pyqtSignal(int) + close_id = QtCore.pyqtSignal(int) + click_id = QtCore.pyqtSignal(int) # Emitted by the adapter when an error occurred, which should result in the adapter # getting swapped out (potentially initializing the same adapter again, or using a # different one if that fails). - error = pyqtSignal(str) - clear_all = pyqtSignal() + error = QtCore.pyqtSignal(str) + clear_all = QtCore.pyqtSignal() def present( self, @@ -143,7 +137,7 @@ class AbstractNotificationAdapter(QObject): """ raise NotImplementedError - def _should_include_origin(self, origin: QUrl) -> bool: + def _should_include_origin(self, origin: QtCore.QUrl) -> bool: """Check if the origin is useful to include. If we open the page via a file scheme, the origin is QUrl('file:///') which @@ -154,13 +148,13 @@ class AbstractNotificationAdapter(QObject): config.instance.get('content.notifications.show_origin', url=origin), ) - @pyqtSlot(int) + @QtCore.pyqtSlot(int) def on_web_closed(self, notification_id: int) -> None: """Called when a notification was closed by the website.""" raise NotImplementedError -class NotificationBridgePresenter(QObject): +class NotificationBridgePresenter(QtCore.QObject): """Notification presenter which bridges notifications to an adapter. @@ -171,7 +165,7 @@ class NotificationBridgePresenter(QObject): - Switching out adapters if the current one emitted its error signal. """ - def __init__(self, parent: QObject = None) -> None: + def __init__(self, parent: QtCore.QObject = None) -> None: super().__init__(parent) assert _notifications_supported() @@ -235,13 +229,7 @@ class NotificationBridgePresenter(QObject): def install(self, profile: "QWebEngineProfile") -> None: """Set the profile to use this bridge as the presenter.""" - # WORKAROUND for - # https://www.riverbankcomputing.com/pipermail/pyqt/2020-May/042916.html - # Fixed in PyQtWebEngine 5.15.0 - # PYQT_WEBENGINE_VERSION was added with PyQtWebEngine 5.13, but if we're here, - # we already did a version check above. - from PyQt5.QtWebEngine import PYQT_WEBENGINE_VERSION - if PYQT_WEBENGINE_VERSION < 0x050F00: + if QtWebEngine.PYQT_WEBENGINE_VERSION < 0x050F00: # PyQtWebEngine unrefs the callback after it's called, for some # reason. So we call setNotificationPresenter again to *increase* # its refcount to prevent it from getting GC'd. Otherwise, random @@ -323,7 +311,7 @@ class NotificationBridgePresenter(QObject): log.misc.debug("Did not find match") return None - @pyqtSlot(int) + @QtCore.pyqtSlot(int) def _on_adapter_closed(self, notification_id: int) -> None: """A notification was closed by the adapter (usually due to the user). @@ -347,7 +335,7 @@ class NotificationBridgePresenter(QObject): log.misc.debug(f"Ignoring close request for notification {notification_id} " "due to PyQt bug") - @pyqtSlot(int) + @QtCore.pyqtSlot(int) def _on_adapter_clicked(self, notification_id: int) -> None: """A notification was clicked by the adapter (usually due to the user). @@ -377,7 +365,7 @@ class NotificationBridgePresenter(QObject): for win_id in objreg.window_registry: tabbedbrowser = objreg.get("tabbed-browser", window=win_id, scope="window") for idx, tab in enumerate(tabbedbrowser.widgets()): - if tab.url().matches(notification.origin(), QUrl.RemovePath): + if tab.url().matches(notification.origin(), QtCore.QUrl.RemovePath): tabbedbrowser.widget.setCurrentIndex(idx) return log.misc.debug(f"No matching tab found for {notification.origin()}") @@ -395,7 +383,7 @@ class NotificationBridgePresenter(QObject): self._adapter = None self._on_adapter_clear_all() - @pyqtSlot() + @QtCore.pyqtSlot() def _on_adapter_clear_all(self) -> None: """Called when the adapter requests clearing all notifications. @@ -409,7 +397,7 @@ class NotificationBridgePresenter(QObject): for notification_id in list(self._active_notifications): self._on_adapter_closed(notification_id) - @pyqtSlot(str) + @QtCore.pyqtSlot(str) def _on_adapter_error(self, error: str) -> None: """A fatal error happened in the adapter. @@ -440,14 +428,14 @@ class SystrayNotificationAdapter(AbstractNotificationAdapter): NAME = "systray" NOTIFICATION_ID = 1 # only one concurrent notification supported - def __init__(self, parent: QObject = None) -> None: + def __init__(self, parent: QtCore.QObject = None) -> None: super().__init__(parent) - if not QSystemTrayIcon.isSystemTrayAvailable(): + if not QtWidgets.QSystemTrayIcon.isSystemTrayAvailable(): raise Error("No system tray available") - if not QSystemTrayIcon.supportsMessages(): + if not QtWidgets.QSystemTrayIcon.supportsMessages(): raise Error("System tray does not support messages") - self._systray = QSystemTrayIcon(self) + self._systray = QtWidgets.QSystemTrayIcon(self) self._systray.setIcon(objects.qapp.windowIcon()) self._systray.messageClicked.connect(self._on_systray_clicked) @@ -468,27 +456,27 @@ class SystrayNotificationAdapter(AbstractNotificationAdapter): return self.NOTIFICATION_ID - def _convert_icon(self, image: QImage) -> QIcon: + def _convert_icon(self, image: QtGui.QImage) -> QtGui.QIcon: """Convert a QImage to a QIcon.""" if image.isNull(): - return QIcon() - pixmap = QPixmap.fromImage(image, Qt.NoFormatConversion) + return QtGui.QIcon() + pixmap = QtGui.QPixmap.fromImage(image, QtCore.Qt.NoFormatConversion) assert not pixmap.isNull() - icon = QIcon(pixmap) + icon = QtGui.QIcon(pixmap) assert not icon.isNull() return icon - def _format_message(self, text: str, origin: QUrl) -> str: + def _format_message(self, text: str, origin: QtCore.QUrl) -> str: """Format the message to display.""" if not self._should_include_origin(origin): return text return origin.toDisplayString() + '\n\n' + text - @pyqtSlot() + @QtCore.pyqtSlot() def _on_systray_clicked(self) -> None: self.click_id.emit(self.NOTIFICATION_ID) - @pyqtSlot(int) + @QtCore.pyqtSlot(int) def on_web_closed(self, notification_id: int) -> None: assert notification_id == self.NOTIFICATION_ID, notification_id if not sip.isdeleted(self._systray): @@ -509,7 +497,7 @@ class MessagesNotificationAdapter(AbstractNotificationAdapter): NAME = "messages" - def __init__(self, parent: QObject = None) -> None: + def __init__(self, parent: QtCore.QObject = None) -> None: super().__init__(parent) self._id_gen = itertools.count(1) @@ -525,12 +513,12 @@ class MessagesNotificationAdapter(AbstractNotificationAdapter): message.info(markup, replace=f'notifications-{new_id}') # Faking closing, timing might not be 100% accurate - QTimer.singleShot( + QtCore.QTimer.singleShot( config.val.messages.timeout, lambda: self.close_id.emit(new_id)) return new_id - @pyqtSlot(int) + @QtCore.pyqtSlot(int) def on_web_closed(self, _notification_id: int) -> None: """We can't close messages.""" @@ -561,7 +549,7 @@ class HerbeNotificationAdapter(AbstractNotificationAdapter): NAME = "herbe" - def __init__(self, parent: QObject = None) -> None: + def __init__(self, parent: QtCore.QObject = None) -> None: super().__init__(parent) # Also cleans up potentially hanging semaphores from herbe. # https://github.com/dudik/herbe#notifications-dont-show-up @@ -582,7 +570,7 @@ class HerbeNotificationAdapter(AbstractNotificationAdapter): if replaces_id is not None: self.on_web_closed(replaces_id) - proc = QProcess(self) + proc = QtCore.QProcess(self) proc.errorOccurred.connect(self._on_error) lines = list(self._message_lines(qt_notification)) @@ -610,7 +598,7 @@ class HerbeNotificationAdapter(AbstractNotificationAdapter): if not qt_notification.icon().isNull(): yield "(icon not shown)" - def _on_finished(self, pid: int, code: int, status: QProcess.ExitStatus) -> None: + def _on_finished(self, pid: int, code: int, status: QtCore.QProcess.ExitStatus) -> None: """Handle a closing herbe process. From the GitHub page: @@ -623,7 +611,7 @@ class HerbeNotificationAdapter(AbstractNotificationAdapter): signals, we can't do much - emitting self.error would just go use herbe again, so there's no point. """ - if status == QProcess.CrashExit: + if status == QtCore.QProcess.CrashExit: return if code == 0: @@ -635,14 +623,14 @@ class HerbeNotificationAdapter(AbstractNotificationAdapter): stderr = proc.readAllStandardError() raise Error(f'herbe exited with status {code}: {stderr}') - @pyqtSlot(QProcess.ProcessError) - def _on_error(self, error: QProcess.ProcessError) -> None: - if error == QProcess.Crashed: + @QtCore.pyqtSlot(QtCore.QProcess.ProcessError) + def _on_error(self, error: QtCore.QProcess.ProcessError) -> None: + if error == QtCore.QProcess.Crashed: return - name = debug.qenum_key(QProcess.ProcessError, error) + name = debug.qenum_key(QtCore.QProcess.ProcessError, error) raise Error(f'herbe process error: {name}') - @pyqtSlot(int) + @QtCore.pyqtSlot(int) def on_web_closed(self, notification_id: int) -> None: """Handle closing the notification from JS. @@ -689,10 +677,10 @@ class _ServerCapabilities: ) -def _as_uint32(x: int) -> QVariant: +def _as_uint32(x: int) -> QtCore.QVariant: """Convert the given int to an uint32 for DBus.""" - variant = QVariant(x) - successful = variant.convert(QVariant.UInt) + variant = QtCore.QVariant(x) + successful = variant.convert(QtCore.QVariant.UInt) assert successful return variant @@ -734,7 +722,7 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): "org.freedesktop.DBus.Error.Spawn.ChildSignaled", } - def __init__(self, parent: QObject = None) -> None: + def __init__(self, parent: QtCore.QObject = None) -> None: super().__init__(parent) assert _notifications_supported() @@ -745,16 +733,16 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): # possible to run DBus there. raise Error("libnotify is not supported on Windows") - bus = QDBusConnection.sessionBus() + bus = QtDBus.QDBusConnection.sessionBus() if not bus.isConnected(): raise Error( "Failed to connect to DBus session bus: " + self._dbus_error_str(bus.lastError())) - self._watcher = QDBusServiceWatcher( + self._watcher = QtDBus.QDBusServiceWatcher( self.SERVICE, bus, - QDBusServiceWatcher.WatchForUnregistration, + QtDBus.QDBusServiceWatcher.WatchForUnregistration, self, ) self._watcher.serviceUnregistered.connect( # type: ignore[attr-defined] @@ -763,7 +751,7 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): test_service = 'test-notification-service' in objects.debug_flags service = self.TEST_SERVICE if test_service else self.SERVICE - self.interface = QDBusInterface(service, self.PATH, self.INTERFACE, bus) + self.interface = QtDBus.QDBusInterface(service, self.PATH, self.INTERFACE, bus) if not self.interface.isValid(): raise Error( "Could not construct a DBus interface: " + @@ -790,7 +778,7 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): else: self._fetch_capabilities() - @pyqtSlot(str) + @QtCore.pyqtSlot(str) def _on_service_unregistered(self) -> None: """Make sure we know when the notification daemon exits. @@ -867,8 +855,8 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): def _get_server_info(self) -> None: """Query notification server information and set quirks.""" - reply = self.interface.call(QDBus.BlockWithGui, "GetServerInformation") - self._verify_message(reply, "ssss", QDBusMessage.ReplyMessage) + reply = self.interface.call(QtDBus.QDBus.BlockWithGui, "GetServerInformation") + self._verify_message(reply, "ssss", QtDBus.QDBusMessage.ReplyMessage) name, vendor, ver, spec_version = reply.arguments() log.misc.debug( @@ -895,7 +883,7 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): if spec_version in icon_key_overrides: self._quirks.icon_key = icon_key_overrides[spec_version] - def _dbus_error_str(self, error: QDBusError) -> str: + def _dbus_error_str(self, error: QtDBus.QDBusError) -> str: """Get a string for a DBus error.""" if not error.isValid(): return "Unknown error" @@ -903,20 +891,20 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): def _verify_message( self, - msg: QDBusMessage, + msg: QtDBus.QDBusMessage, expected_signature: str, - expected_type: QDBusMessage.MessageType, + expected_type: QtDBus.QDBusMessage.MessageType, ) -> None: """Check the signature/type of a received message. Raises DBusError if the signature doesn't match. """ assert expected_type not in [ - QDBusMessage.ErrorMessage, - QDBusMessage.InvalidMessage, + QtDBus.QDBusMessage.ErrorMessage, + QtDBus.QDBusMessage.InvalidMessage, ], expected_type - if msg.type() == QDBusMessage.ErrorMessage: + if msg.type() == QtDBus.QDBusMessage.ErrorMessage: err = msg.errorName() if err in self._NON_FATAL_ERRORS: self.error.emit(msg.errorMessage()) @@ -932,8 +920,8 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): typ = msg.type() if typ != expected_type: - type_str = debug.qenum_key(QDBusMessage.MessageType, typ) - expected_type_str = debug.qenum_key(QDBusMessage.MessageType, expected_type) + type_str = debug.qenum_key(QtDBus.QDBusMessage.MessageType, typ) + expected_type_str = debug.qenum_key(QtDBus.QDBusMessage.MessageType, expected_type) raise Error( f"Got a message of type {type_str} but expected {expected_type_str}" f"(args: {msg.arguments()})") @@ -951,7 +939,7 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): actions = [] if self._capabilities.actions: actions = ['default', 'Activate'] # key, name - actions_arg = QDBusArgument(actions, QMetaType.QStringList) + actions_arg = QtDBus.QDBusArgument(actions, QtCore.QMetaType.QStringList) origin_url_str = qt_notification.origin().toDisplayString() hints: Dict[str, Any] = { @@ -968,7 +956,7 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): icon = qt_notification.icon() if icon.isNull(): filename = 'icons/qutebrowser-64x64.png' - icon = QImage.fromData(resources.read_file_binary(filename)) + icon = QtGui.QImage.fromData(resources.read_file_binary(filename)) key = self._quirks.icon_key or "image-data" data = self._convert_image(icon) @@ -981,7 +969,7 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): title = html.escape(title, quote=False) reply = self.interface.call( - QDBus.BlockWithGui, + QtDBus.QDBus.BlockWithGui, "Notify", "qutebrowser", # application name _as_uint32(replaces_id), # replaces notification id @@ -992,7 +980,7 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): hints, -1, # timeout; -1 means 'use default' ) - self._verify_message(reply, "u", QDBusMessage.ReplyMessage) + self._verify_message(reply, "u", QtDBus.QDBusMessage.ReplyMessage) notification_id = reply.arguments()[0] @@ -1008,7 +996,7 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): return notification_id - def _convert_image(self, qimage: QImage) -> Optional[QDBusArgument]: + def _convert_image(self, qimage: QtGui.QImage) -> Optional[QtDBus.QDBusArgument]: """Convert a QImage to the structure DBus expects. https://specifications.freedesktop.org/notification-spec/latest/ar01s05.html#icons-and-images-formats @@ -1016,10 +1004,10 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): bits_per_color = 8 has_alpha = qimage.hasAlphaChannel() if has_alpha: - image_format = QImage.Format_RGBA8888 + image_format = QtGui.QImage.Format_RGBA8888 channel_count = 4 else: - image_format = QImage.Format_RGB888 + image_format = QtGui.QImage.Format_RGB888 channel_count = 3 qimage.convertTo(image_format) @@ -1027,7 +1015,7 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): width = qimage.width() height = qimage.height() - image_data = QDBusArgument() + image_data = QtDBus.QDBusArgument() image_data.beginStructure() image_data.add(width) image_data.add(height) @@ -1076,31 +1064,31 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): return None bits = qimage.constBits().asstring(size) - image_data.add(QByteArray(bits)) + image_data.add(QtCore.QByteArray(bits)) image_data.endStructure() return image_data - @pyqtSlot(QDBusMessage) - def _handle_close(self, msg: QDBusMessage) -> None: + @QtCore.pyqtSlot(QtDBus.QDBusMessage) + def _handle_close(self, msg: QtDBus.QDBusMessage) -> None: """Handle NotificationClosed from DBus.""" - self._verify_message(msg, "uu", QDBusMessage.SignalMessage) + self._verify_message(msg, "uu", QtDBus.QDBusMessage.SignalMessage) notification_id, _close_reason = msg.arguments() self.close_id.emit(notification_id) - @pyqtSlot(QDBusMessage) - def _handle_action(self, msg: QDBusMessage) -> None: + @QtCore.pyqtSlot(QtDBus.QDBusMessage) + def _handle_action(self, msg: QtDBus.QDBusMessage) -> None: """Handle ActionInvoked from DBus.""" - self._verify_message(msg, "us", QDBusMessage.SignalMessage) + self._verify_message(msg, "us", QtDBus.QDBusMessage.SignalMessage) notification_id, action_key = msg.arguments() if action_key == "default": self.click_id.emit(notification_id) - @pyqtSlot(int) + @QtCore.pyqtSlot(int) def on_web_closed(self, notification_id: int) -> None: """Send CloseNotification if a notification was closed from JS.""" self.interface.call( - QDBus.NoBlock, + QtDBus.QDBus.NoBlock, "CloseNotification", _as_uint32(notification_id), ) @@ -1108,10 +1096,10 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): def _fetch_capabilities(self) -> None: """Fetch capabilities from the notification server.""" reply = self.interface.call( - QDBus.BlockWithGui, + QtDBus.QDBus.BlockWithGui, "GetCapabilities", ) - self._verify_message(reply, "as", QDBusMessage.ReplyMessage) + self._verify_message(reply, "as", QtDBus.QDBusMessage.ReplyMessage) caplist = reply.arguments()[0] self._capabilities = _ServerCapabilities.from_list(caplist) @@ -1122,7 +1110,7 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): log.misc.debug(f"Notification server capabilities: {self._capabilities}") - def _format_body(self, body: str, origin_url: QUrl) -> str: + def _format_body(self, body: str, origin_url: QtCore.QUrl) -> str: """Format the body according to the server capabilities. If the server doesn't support x-kde-origin-name, we include the origin URL as a @@ -1138,7 +1126,7 @@ class DBusNotificationAdapter(AbstractNotificationAdapter): prefix = None elif self._capabilities.body_markup and self._capabilities.body_hyperlinks: href = html.escape( - origin_url.toString(QUrl.FullyEncoded) # type: ignore[arg-type] + origin_url.toString(QtCore.QUrl.FullyEncoded) # type: ignore[arg-type] ) text = html.escape(urlstr, quote=False) prefix = f'<a href="{href}">{text}</a>' |