From de21e5e488f0ce5d825673f900d029c3f879453b Mon Sep 17 00:00:00 2001 From: Kovid Goyal Date: Sun, 28 Jul 2024 08:41:24 +0530 Subject: [PATCH] Implement filtering of desktop notifications Fixes #7670 --- docs/changelog.rst | 2 + docs/notifications.py | 49 +++++++++++++++++++++ kitty/notifications.py | 87 ++++++++++++++++++++++++++++--------- kitty/options/definition.py | 23 ++++++++++ kitty/options/parse.py | 23 ++++++---- kitty/options/types.py | 3 ++ kitty/options/utils.py | 4 ++ 7 files changed, 162 insertions(+), 29 deletions(-) create mode 100644 docs/notifications.py diff --git a/docs/changelog.rst b/docs/changelog.rst index 064c643cb..d62aeb043 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -82,6 +82,8 @@ Detailed list of changes - Desktop notifications protocol: Add support for closing notifications and querying if the terminal emulator supports the protocol (:iss:`7658`, :iss:`7659`) +- A new option :opt:`filter_notification` to filter out or perform arbitrary actions on desktop notifications based on sophisticated criteria (:iss:`7670`) + - A new protocol to allow terminal applications to change colors in the terminal more robustly than with the legacy XTerm protocol (:ref:`color_control`) - Sessions: A new command ``focus_matching_window`` to shift focus to a specific window, useful when creating complex layouts with splits (:disc:`7635`) diff --git a/docs/notifications.py b/docs/notifications.py new file mode 100644 index 000000000..215bae805 --- /dev/null +++ b/docs/notifications.py @@ -0,0 +1,49 @@ +#!/usr/bin/env python + +# A sample script to process notifications. Save it as +# ~/.config/kitty/notifications.py + +import subprocess + +from kitty.notifications import NotificationCommand, Urgency + + +def log_notification(nc: NotificationCommand) -> None: + # Log notifications to /tmp/notifications-log.txt + with open('/tmp/notifications-log.txt', 'a') as log: + print(f'title: {nc.title}', file=log) + print(f'body: {nc.body}', file=log) + print(f'app: {nc.application_name}', file=log) + print(f'type: {nc.notification_type}', file=log) + print('\n', file=log) + + +def main(nc: NotificationCommand) -> bool: + ''' + This function should return True to filter out the notification + ''' + log_notification(nc) + + # filter out notifications with 'unwanted' in their titles + if 'unwanted' in nc.title.lower(): + return True + + # filter out notifications from the application badapp + if nc.application_name == 'badapp': + return True + + # filter out low urgency notifications + if nc.urgency is Urgency.Low: + return True + + # replace some bad text in the notification body + nc.body = nc.body.replace('bad text', 'good text') + + # run a script if this notification is from myapp and has + # type foo, passing in the title and body as command line args + # to the script. + if nc.application_name == 'myapp' and nc.notification_type == 'foo': + subprocess.Popen(['/path/to/my/script', nc.title, nc.body]) + + # dont filter out this notification + return False diff --git a/kitty/notifications.py b/kitty/notifications.py index 5e04e4c79..39079c0ec 100644 --- a/kitty/notifications.py +++ b/kitty/notifications.py @@ -7,11 +7,11 @@ from collections import OrderedDict from contextlib import suppress from enum import Enum from itertools import count -from typing import Any, Callable, Dict, FrozenSet, Iterator, List, NamedTuple, Optional, Tuple, Union +from typing import Any, Callable, Dict, FrozenSet, Iterator, List, NamedTuple, Optional, Set, Tuple, Union from weakref import ReferenceType, ref -from .constants import cache_dir, is_macos, logo_png_file -from .fast_data_types import ESC_OSC, StreamingBase64Decoder, base64_decode, current_focused_os_window_id, get_boss +from .constants import cache_dir, config_dir, is_macos, logo_png_file +from .fast_data_types import ESC_OSC, StreamingBase64Decoder, base64_decode, current_focused_os_window_id, get_boss, get_options from .types import run_once from .typing import WindowType from .utils import get_custom_window_icon, log_error, sanitize_control_codes @@ -190,22 +190,29 @@ def limit_size(x: str) -> str: class NotificationCommand: - done: bool = True - identifier: str = '' - channel_id: int = 0 - desktop_notification_id: int = -1 + # data received from client and eventually displayed/processed title: str = '' body: str = '' actions: FrozenSet[Action] = frozenset((Action.focus,)) only_when: OnlyWhen = OnlyWhen.unset urgency: Optional[Urgency] = None - close_response_requested: Optional[bool] = None icon_data_key: str = '' - icon_path: str = '' icon_name: str = '' application_name: str = '' notification_type: str = '' + # event callbacks + on_activation: Optional[Callable[['NotificationCommand'], None]] = None + on_close: Optional[Callable[['NotificationCommand'], None]] = None + + # metadata + identifier: str = '' + done: bool = True + channel_id: int = 0 + desktop_notification_id: int = -1 + close_response_requested: Optional[bool] = None + icon_path: str = '' + # payload handling current_payload_type: PayloadType = PayloadType.title current_payload_buffer: Optional[EncodedDataStore] = None @@ -214,9 +221,6 @@ class NotificationCommand: created_by_desktop: bool = False activation_token: str = '' - # event callbacks - on_activation: Optional[Callable[['NotificationCommand'], None]] = None - def __init__(self, icon_data_cache: 'ReferenceType[IconDataCache]', log: 'Log') -> None: self.icon_data_cache_ref = icon_data_cache self.log = log @@ -369,6 +373,25 @@ class NotificationCommand: else: self.title = sanitize_text(self.body) self.body = '' + self.urgency = Urgency.Normal if self.urgency is None else self.urgency + + def matches_rule_item(self, location:str, query:str) -> bool: + import re + pat = re.compile(query) + val = {'title': self.title, 'body': self.body, 'app': self.application_name, 'type': self.notification_type}[location] + return pat.search(val) is not None + + def matches_rule(self, rule: str) -> bool: + if rule == 'all': + return True + from .search_query_parser import search + def get_matches(location: str, query: str, candidates: Set['NotificationCommand']) -> Set['NotificationCommand']: + return {x for x in candidates if x.matches_rule_item(location, query)} + try: + return self in search(rule, ('title', 'body', 'app', 'type'), {self}, get_matches) + except Exception as e: + self.log(f'Ignoring invalid filter_notification rule: {rule} with error: {e}') + return False class DesktopIntegration: @@ -426,8 +449,8 @@ class MacOSIntegration(DesktopIntegration): # so dont double % # for %% escaping. body = (nc.body or ' ') - urgency = Urgency.Normal if nc.urgency is None else nc.urgency - cocoa_send_notification(str(desktop_notification_id), nc.title, body, '', urgency.value) + assert nc.urgency is not None + cocoa_send_notification(str(desktop_notification_id), nc.title, body, '', nc.urgency.value) return desktop_notification_id def notification_activated(self, event: str, ident: str) -> None: @@ -499,10 +522,10 @@ class FreeDesktopIntegration(DesktopIntegration): def notify(self, nc: NotificationCommand) -> int: from .fast_data_types import dbus_send_notification app_icon = nc.icon_name or nc.icon_path or get_custom_window_icon()[1] or logo_png_file - body = nc.body.replace('<', '<\u200c') # prevent HTML tags from being recognized - urgency = Urgency.Normal if nc.urgency is None else nc.urgency + body = nc.body.replace('<', '<\u200c').replace('&', '&\u200c') # prevent HTML markup from being recognized + assert nc.urgency is not None desktop_notification_id = dbus_send_notification( - app_name=nc.application_name or 'kitty', app_icon=app_icon, title=nc.title, body=body, timeout=-1, urgency=urgency.value) + app_name=nc.application_name or 'kitty', app_icon=app_icon, title=nc.title, body=body, timeout=-1, urgency=nc.urgency.value) if debug_desktop_integration: log_error(f'Created notification with {desktop_notification_id=}') return desktop_notification_id @@ -581,6 +604,15 @@ class NotificationManager: self.base_cache_dir = base_cache_dir self.log = log self.icon_data_cache = IconDataCache(base_cache_dir=self.base_cache_dir) + script_path = os.path.join(config_dir, 'notifications.py') + self.filter_script: Callable[[NotificationCommand], bool] = lambda nc: False + if os.path.exists(script_path): + import runpy + try: + m = runpy.run_path(script_path) + self.filter_script = m['main'] + except Exception as e: + self.log(f'Failed to load {script_path} with error: {e}') self.reset() def reset(self) -> None: @@ -610,13 +642,18 @@ class NotificationManager: try: n.on_activation(n) except Exception as e: - self.log(e) + self.log('Notification on_activation handler failed with error:', e) def notification_closed(self, desktop_notification_id: int) -> None: if n := self.in_progress_notification_commands.get(desktop_notification_id): self.purge_notification(n) if n.close_response_requested: self.send_closed_response(n.channel_id, n.identifier) + if n.on_close is not None: + try: + n.on_close(n) + except Exception as e: + self.log('Notification on_close handler failed with error:', e) def create_notification_cmd(self) -> NotificationCommand: return NotificationCommand(ref(self.icon_data_cache), self.log) @@ -639,7 +676,7 @@ class NotificationManager: cmd.on_activation = self.desktop_integration.on_new_version_notification_activation self.notify_with_command(cmd, 0) - def is_notification_allowed(self, cmd: NotificationCommand, channel_id: int) -> bool: + def is_notification_allowed(self, cmd: NotificationCommand, channel_id: int, apply_filter_rules: bool = True) -> bool: if cmd.only_when is not OnlyWhen.always and cmd.only_when is not OnlyWhen.unset: ui_state = self.channel.ui_state(channel_id) if ui_state.has_keyboard_focus: @@ -648,10 +685,20 @@ class NotificationManager: return False return True + def is_notification_filtered(self, cmd: NotificationCommand) -> bool: + if self.filter_script(cmd): + self.log(f'Notification {cmd.title!r} filtered out by script') + return True + for rule in get_options().filter_notification: + if cmd.matches_rule(rule): + self.log(f'Notification {cmd.title!r} filtered out by filter_notification rule: {rule}') + return True + return False + def notify_with_command(self, cmd: NotificationCommand, channel_id: int) -> Optional[int]: cmd.channel_id = channel_id cmd.finalise() - if not cmd.title or not self.is_notification_allowed(cmd, channel_id): + if not cmd.title or not self.is_notification_allowed(cmd, channel_id) or self.is_notification_filtered(cmd): return None desktop_notification_id = self.desktop_integration.notify(cmd) self.register_in_progress_notification(cmd, desktop_notification_id) diff --git a/kitty/options/definition.py b/kitty/options/definition.py index 75e9c09c1..a5dbc2169 100644 --- a/kitty/options/definition.py +++ b/kitty/options/definition.py @@ -3035,6 +3035,29 @@ The value of :code:`VAR2` will be :code:`/a/b`. ''' ) +opt('+filter_notification', '', option_type='filter_notification', add_to_default=False, long_text=''' +Specify rules to filter out notifications sent by applications running in kitty. +Can be specified multiple times to create multiple filter rules. A rule specification +is of the form :code:`field:regexp`. A filter rule +can match on any of the fields: :code:`title`, :code:`body`, :code:`app`, :code:`type`. +The special value of :code:`all` filters out all notifications. Rules can be combined +using Boolean operators. Some examples:: + + filter_notification title:hello or body:"abc.*def" + # filter out notification from vim except for ones about updates, (?i) + # makes matching case insesitive. + filter_notification app:"[ng]?vim" and not body:"(?i)update" + # filter out all notifications + filter_notification all + +The field :code:`app` is the name of the application sending the notification and :code:`type` +is the type of the notification. Not all applications will send these fields, so you can also +match on the title and body of the notification text. More sophisticated programmatic filtering +and custom actions on notifications can be done by creating a notifications.py file in the +kitty config directory (:file:`~/.config/kitty`). An annotated sample is +:link:`available `. +''') + opt('+watcher', '', option_type='store_multiple', add_to_default=False, diff --git a/kitty/options/parse.py b/kitty/options/parse.py index b7b011cd9..2132f13d7 100644 --- a/kitty/options/parse.py +++ b/kitty/options/parse.py @@ -12,15 +12,15 @@ from kitty.options.utils import ( config_or_absolute_path, copy_on_select, cursor_blink_interval, cursor_text_color, deprecated_adjust_line_height, deprecated_hide_window_decorations_aliases, deprecated_macos_show_window_title_in_menubar_alias, deprecated_send_text, disable_ligatures, - edge_width, env, font_features, hide_window_decorations, macos_option_as_alt, macos_titlebar_color, - menu_map, modify_font, narrow_symbols, notify_on_cmd_finish, optional_edge_width, parse_font_spec, - parse_map, parse_mouse_map, paste_actions, remote_control_password, resize_debounce_time, - scrollback_lines, scrollback_pager_history_size, shell_integration, store_multiple, symbol_map, - tab_activity_symbol, tab_bar_edge, tab_bar_margin_height, tab_bar_min_tabs, tab_fade, - tab_font_style, tab_separator, tab_title_template, titlebar_color, to_cursor_shape, - to_cursor_unfocused_shape, to_font_size, to_layout_names, to_modifiers, url_prefixes, url_style, - visual_bell_duration, visual_window_select_characters, window_border_width, window_logo_scale, - window_size + edge_width, env, filter_notification, font_features, hide_window_decorations, macos_option_as_alt, + macos_titlebar_color, menu_map, modify_font, narrow_symbols, notify_on_cmd_finish, + optional_edge_width, parse_font_spec, parse_map, parse_mouse_map, paste_actions, + remote_control_password, resize_debounce_time, scrollback_lines, scrollback_pager_history_size, + shell_integration, store_multiple, symbol_map, tab_activity_symbol, tab_bar_edge, + tab_bar_margin_height, tab_bar_min_tabs, tab_fade, tab_font_style, tab_separator, + tab_title_template, titlebar_color, to_cursor_shape, to_cursor_unfocused_shape, to_font_size, + to_layout_names, to_modifiers, url_prefixes, url_style, visual_bell_duration, + visual_window_select_characters, window_border_width, window_logo_scale, window_size ) @@ -976,6 +976,10 @@ class Parser: def file_transfer_confirmation_bypass(self, val: str, ans: typing.Dict[str, typing.Any]) -> None: ans['file_transfer_confirmation_bypass'] = str(val) + def filter_notification(self, val: str, ans: typing.Dict[str, typing.Any]) -> None: + for k, v in filter_notification(val, ans["filter_notification"]): + ans["filter_notification"][k] = v + def focus_follows_mouse(self, val: str, ans: typing.Dict[str, typing.Any]) -> None: ans['focus_follows_mouse'] = to_bool(val) @@ -1448,6 +1452,7 @@ def create_result_dict() -> typing.Dict[str, typing.Any]: 'action_alias': {}, 'env': {}, 'exe_search_path': {}, + 'filter_notification': {}, 'font_features': {}, 'kitten_alias': {}, 'menu_map': {}, diff --git a/kitty/options/types.py b/kitty/options/types.py index af3757dfe..3fa0c38f8 100644 --- a/kitty/options/types.py +++ b/kitty/options/types.py @@ -347,6 +347,7 @@ option_names = ( # {{{ 'env', 'exe_search_path', 'file_transfer_confirmation_bypass', + 'filter_notification', 'focus_follows_mouse', 'font_family', 'font_features', @@ -633,6 +634,7 @@ class Options: action_alias: typing.Dict[str, str] = {} env: typing.Dict[str, str] = {} exe_search_path: typing.Dict[str, str] = {} + filter_notification: typing.Dict[str, str] = {} font_features: typing.Dict[str, typing.Tuple[kitty.fast_data_types.ParsedFontFeature, ...]] = {} kitten_alias: typing.Dict[str, str] = {} menu_map: typing.Dict[typing.Tuple[str, ...], str] = {} @@ -755,6 +757,7 @@ defaults = Options() defaults.action_alias = {} defaults.env = {} defaults.exe_search_path = {} +defaults.filter_notification = {} defaults.font_features = {} defaults.kitten_alias = {} defaults.menu_map = {} diff --git a/kitty/options/utils.py b/kitty/options/utils.py index 987c31bfb..876bf6381 100644 --- a/kitty/options/utils.py +++ b/kitty/options/utils.py @@ -787,6 +787,10 @@ def config_or_absolute_path(x: str, env: Optional[Dict[str, str]] = None) -> Opt return resolve_abs_or_config_path(x, env) +def filter_notification(val: str, current_val: Dict[str, str]) -> Iterable[Tuple[str, str]]: + yield val, '' + + def remote_control_password(val: str, current_val: Dict[str, str]) -> Iterable[Tuple[str, Sequence[str]]]: val = val.strip() if val: