Implement filtering of desktop notifications

Fixes #7670

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`)

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

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)

kitty/options/definition.py

@@ -3035,6 +3035,29 @@ The value of :code:`VAR2` will be :code:`<path to home directory>/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 <https://github.com/kovidgoyal/kitty/blob/master/docs/notifications.py>`.
+''')
+
 opt('+watcher', '',
     option_type='store_multiple',
     add_to_default=False,

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': {},

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 = {}

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: