a9bcc3319a
Currently notifications emitted with `notify_on_cmd_finish` and unfocused left withing notification bar forever and user have to clear them by hand. This patch automatically clears notification when emitted window gained its focus back assuming that user is aware about action and not needs for notification any more. Change-Id: I8c9f58eeed696eb473aa4a966436aeb1a15fd3d6
1069 lines
46 KiB
Python
1069 lines
46 KiB
Python
#!/usr/bin/env python
|
|
# License: GPLv3 Copyright: 2024, Kovid Goyal <kovid at kovidgoyal.net>
|
|
|
|
import os
|
|
import re
|
|
from collections import OrderedDict
|
|
from collections.abc import Iterator, Sequence
|
|
from contextlib import suppress
|
|
from enum import Enum
|
|
from functools import partial
|
|
from itertools import count
|
|
from typing import Any, Callable, NamedTuple, Optional, Set, Union
|
|
from weakref import ReferenceType, ref
|
|
|
|
from .constants import cache_dir, config_dir, is_macos, logo_png_file, standard_icon_names, standard_sound_names
|
|
from .fast_data_types import ESC_OSC, StreamingBase64Decoder, add_timer, 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
|
|
|
|
debug_desktop_integration = False
|
|
|
|
|
|
def image_type(data: bytes) -> str:
|
|
if data[:8] == b"\211PNG\r\n\032\n":
|
|
return 'png'
|
|
if data[:6] in (b'GIF87a', b'GIF89a'):
|
|
return 'gif'
|
|
if data[:2] == b'\xff\xd8':
|
|
return 'jpeg'
|
|
return 'unknown'
|
|
|
|
|
|
class IconDataCache:
|
|
|
|
def __init__(self, base_cache_dir: str = '', max_cache_size: int = 128 * 1024 * 1024):
|
|
self.max_cache_size = max_cache_size
|
|
self.key_map: dict[str, str] = {}
|
|
self.hash_map: 'OrderedDict[str, Set[str]]' = OrderedDict()
|
|
self.base_cache_dir = base_cache_dir
|
|
self.cache_dir = ''
|
|
self.total_size = 0
|
|
import struct
|
|
self.seed: int = struct.unpack("!Q", os.urandom(8))[0]
|
|
|
|
def _ensure_state(self) -> str:
|
|
if not self.cache_dir:
|
|
self.cache_dir = os.path.join(self.base_cache_dir or cache_dir(), 'notifications-icons', str(os.getpid()))
|
|
os.makedirs(self.cache_dir, exist_ok=True, mode=0o700)
|
|
return self.cache_dir
|
|
|
|
def __del__(self) -> None:
|
|
if self.cache_dir:
|
|
import shutil
|
|
with suppress(FileNotFoundError):
|
|
shutil.rmtree(self.cache_dir)
|
|
self.cache_dir = ''
|
|
|
|
def keys(self) -> Iterator[str]:
|
|
yield from self.key_map.keys()
|
|
|
|
def hash(self, data: bytes) -> str:
|
|
from kittens.transfer.rsync import xxh128_hash_with_seed
|
|
d = xxh128_hash_with_seed(data, self.seed)
|
|
return d.hex() + '.' + image_type(data)
|
|
|
|
def add_icon(self, key: str, data: bytes) -> str:
|
|
self._ensure_state()
|
|
data_hash = self.hash(data)
|
|
path = os.path.join(self.cache_dir, data_hash)
|
|
if not os.path.exists(path):
|
|
with open(path, 'wb') as f:
|
|
f.write(data)
|
|
self.total_size += len(data)
|
|
self.hash_map[data_hash] = self.hash_map.pop(data_hash, set()) | {key} # mark this data as being used recently
|
|
if key:
|
|
self.key_map[key] = data_hash
|
|
self.prune()
|
|
return path
|
|
|
|
def get_icon(self, key: str) -> str:
|
|
self._ensure_state()
|
|
data_hash = self.key_map.get(key)
|
|
if data_hash:
|
|
self.hash_map[data_hash] = self.hash_map.pop(data_hash, set()) | {key} # mark this data as being used recently
|
|
return os.path.join(self.cache_dir, data_hash)
|
|
return ''
|
|
|
|
def clear(self) -> None:
|
|
while self.hash_map:
|
|
data_hash, keys = self.hash_map.popitem(False)
|
|
for key in keys:
|
|
self.key_map.pop(key, None)
|
|
self._remove_data_file(data_hash)
|
|
|
|
def prune(self) -> None:
|
|
self._ensure_state()
|
|
while self.total_size > self.max_cache_size and self.hash_map:
|
|
data_hash, keys = self.hash_map.popitem(False)
|
|
for key in keys:
|
|
self.key_map.pop(key, None)
|
|
self._remove_data_file(data_hash)
|
|
|
|
def _remove_data_file(self, data_hash: str) -> None:
|
|
path = os.path.join(self.cache_dir, data_hash)
|
|
with suppress(FileNotFoundError):
|
|
sz = os.path.getsize(path)
|
|
os.remove(path)
|
|
self.total_size -= sz
|
|
|
|
def remove_icon(self, key: str) -> None:
|
|
self._ensure_state()
|
|
data_hash = self.key_map.pop(key, None)
|
|
if data_hash:
|
|
for key in self.hash_map.pop(data_hash, set()):
|
|
self.key_map.pop(key, None)
|
|
self._remove_data_file(data_hash)
|
|
|
|
|
|
class Urgency(Enum):
|
|
Low: int = 0
|
|
Normal: int = 1
|
|
Critical: int = 2
|
|
|
|
|
|
class PayloadType(Enum):
|
|
unknown = ''
|
|
title = 'title'
|
|
body = 'body'
|
|
query = '?'
|
|
close = 'close'
|
|
icon = 'icon'
|
|
alive = 'alive'
|
|
buttons = 'buttons'
|
|
|
|
@property
|
|
def is_text(self) -> bool:
|
|
return self in (PayloadType.title, PayloadType.body, PayloadType.buttons)
|
|
|
|
|
|
class OnlyWhen(Enum):
|
|
unset = ''
|
|
always = 'always'
|
|
unfocused = 'unfocused'
|
|
invisible = 'invisible'
|
|
|
|
|
|
class Action(Enum):
|
|
focus = 'focus'
|
|
report = 'report'
|
|
|
|
|
|
class DataStore:
|
|
|
|
def __init__(self, max_size: int = 4 * 1024 * 1024) -> None:
|
|
self.buf: list[bytes] = []
|
|
self.current_size = 0
|
|
self.max_size = max_size
|
|
self.truncated = 0
|
|
|
|
def __call__(self, data: bytes) -> None:
|
|
if data:
|
|
if self.current_size > self.max_size:
|
|
self.truncated += len(data)
|
|
else:
|
|
self.current_size += len(data)
|
|
self.buf.append(data)
|
|
|
|
def finalise(self) -> bytes:
|
|
return b''.join(self.buf)
|
|
|
|
|
|
class EncodedDataStore:
|
|
|
|
def __init__(self, data_store: DataStore) -> None:
|
|
self.decoder = StreamingBase64Decoder()
|
|
self.data_store = data_store
|
|
|
|
@property
|
|
def truncated(self) -> int:
|
|
return self.data_store.truncated
|
|
|
|
def add_unencoded_data(self, data: Union[str, bytes]) -> None:
|
|
if isinstance(data, str):
|
|
data = data.encode('utf-8')
|
|
self.flush_encoded_data()
|
|
self.data_store(data)
|
|
|
|
def add_base64_data(self, data: Union[str, bytes]) -> None:
|
|
if isinstance(data, str):
|
|
data = data.encode('ascii')
|
|
try:
|
|
decoded = self.decoder.decode(data)
|
|
except ValueError:
|
|
log_error('Ignoring invalid base64 encoded data in notification request')
|
|
else:
|
|
self.data_store(decoded)
|
|
|
|
def flush_encoded_data(self) -> None:
|
|
if self.decoder.needs_more_data():
|
|
log_error('Received incomplete encoded data for notification request')
|
|
self.decoder.reset()
|
|
|
|
def finalise(self) -> bytes:
|
|
self.flush_encoded_data()
|
|
return self.data_store.finalise()
|
|
|
|
|
|
def limit_size(x: str, limit: int = 1024) -> str:
|
|
if len(x) > limit:
|
|
x = x[:limit]
|
|
return x
|
|
|
|
|
|
class NotificationCommand:
|
|
|
|
# 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
|
|
icon_data_key: str = ''
|
|
icon_names: tuple[str, ...] = ()
|
|
application_name: str = ''
|
|
notification_types: tuple[str, ...] = ()
|
|
timeout: int = -2
|
|
buttons: tuple[str, ...] = ()
|
|
sound_name: str = ''
|
|
|
|
# event callbacks
|
|
on_activation: Optional[Callable[['NotificationCommand', int], None]] = None
|
|
on_close: Optional[Callable[['NotificationCommand'], None]] = None
|
|
on_update: Optional[Callable[['NotificationCommand', '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
|
|
|
|
# desktop integration specific fields
|
|
created_by_desktop: bool = False
|
|
activation_token: str = ''
|
|
|
|
def __init__(self, icon_data_cache: 'ReferenceType[IconDataCache]', log: 'Log') -> None:
|
|
self.icon_data_cache_ref = icon_data_cache
|
|
self.log = log
|
|
|
|
@property
|
|
def report_requested(self) -> bool:
|
|
return Action.report in self.actions
|
|
|
|
@property
|
|
def focus_requested(self) -> bool:
|
|
return Action.focus in self.actions
|
|
|
|
def __repr__(self) -> str:
|
|
fields = {}
|
|
for x in ('title', 'body', 'identifier', 'actions', 'urgency', 'done'):
|
|
val = getattr(self, x)
|
|
if val:
|
|
fields[x] = val
|
|
return f'NotificationCommand{fields}'
|
|
|
|
def parse_metadata(self, metadata: str, prev: 'NotificationCommand') -> tuple[PayloadType, bool]:
|
|
payload_type = PayloadType.title
|
|
payload_is_encoded = False
|
|
if metadata:
|
|
for part in metadata.split(':'):
|
|
if not part:
|
|
continue
|
|
k, v = part.split('=', 1)
|
|
if k == 'p':
|
|
try:
|
|
payload_type = PayloadType(v)
|
|
except ValueError:
|
|
payload_type = PayloadType.unknown
|
|
elif k == 'i':
|
|
self.identifier = sanitize_id(v)
|
|
elif k == 'e':
|
|
payload_is_encoded = v == '1'
|
|
elif k == 'd':
|
|
self.done = v != '0'
|
|
elif k == 'a':
|
|
for ax in v.split(','):
|
|
if remove := ax.startswith('-'):
|
|
ax = ax.lstrip('+-')
|
|
try:
|
|
ac = Action(ax)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
if remove:
|
|
self.actions -= {ac}
|
|
else:
|
|
self.actions = self.actions.union({ac})
|
|
elif k == 'o':
|
|
with suppress(ValueError):
|
|
self.only_when = OnlyWhen(v)
|
|
elif k == 'u':
|
|
with suppress(Exception):
|
|
self.urgency = Urgency(int(v))
|
|
elif k == 'c':
|
|
self.close_response_requested = v != '0'
|
|
elif k == 'g':
|
|
self.icon_data_key = sanitize_id(v)
|
|
elif k == 'n':
|
|
try:
|
|
self.icon_names += (base64_decode(v).decode('utf-8'),)
|
|
except Exception:
|
|
self.log('Ignoring invalid icon name in notification: {v!r}')
|
|
elif k == 'f':
|
|
try:
|
|
self.application_name = base64_decode(v).decode('utf-8')
|
|
except Exception:
|
|
self.log('Ignoring invalid application_name in notification: {v!r}')
|
|
elif k == 't':
|
|
try:
|
|
self.notification_types += (base64_decode(v).decode('utf-8'),)
|
|
except Exception:
|
|
self.log('Ignoring invalid notification type in notification: {v!r}')
|
|
elif k == 'w':
|
|
try:
|
|
self.timeout = max(-1, int(v))
|
|
except Exception:
|
|
self.log('Ignoring invalid timeout in notification: {v!r}')
|
|
elif k == 's':
|
|
try:
|
|
self.sound_name = base64_decode(v).decode('utf-8')
|
|
except Exception:
|
|
self.log('Ignoring invalid sound name in notification: {v!r}')
|
|
if not prev.done and prev.identifier == self.identifier:
|
|
self.merge_metadata(prev)
|
|
return payload_type, payload_is_encoded
|
|
|
|
def merge_metadata(self, prev: 'NotificationCommand') -> None:
|
|
self.actions = prev.actions.union(self.actions)
|
|
self.title = prev.title
|
|
self.body = prev.body
|
|
if self.only_when is OnlyWhen.unset:
|
|
self.only_when = prev.only_when
|
|
if self.urgency is None:
|
|
self.urgency = prev.urgency
|
|
if self.close_response_requested is None:
|
|
self.close_response_requested = prev.close_response_requested
|
|
if not self.icon_data_key:
|
|
self.icon_data_key = prev.icon_data_key
|
|
if prev.icon_names:
|
|
self.icon_names = prev.icon_names + self.icon_names
|
|
if not self.application_name:
|
|
self.application_name = prev.application_name
|
|
if prev.notification_types:
|
|
self.notification_types = prev.notification_types + self.notification_types
|
|
if prev.buttons:
|
|
self.buttons += prev.buttons
|
|
if not self.sound_name:
|
|
self.sound_name = prev.sound_name
|
|
if self.timeout < -1:
|
|
self.timeout = prev.timeout
|
|
self.icon_path = prev.icon_path
|
|
|
|
def create_payload_buffer(self, payload_type: PayloadType) -> EncodedDataStore:
|
|
self.current_payload_type = payload_type
|
|
return EncodedDataStore(DataStore())
|
|
|
|
def set_payload(self, payload_type: PayloadType, payload_is_encoded: bool, payload: str, prev_cmd: 'NotificationCommand') -> None:
|
|
if prev_cmd.current_payload_type is payload_type:
|
|
self.current_payload_type = payload_type
|
|
self.current_payload_buffer = prev_cmd.current_payload_buffer
|
|
prev_cmd.current_payload_buffer = None
|
|
else:
|
|
if prev_cmd.current_payload_buffer:
|
|
self.current_payload_type = prev_cmd.current_payload_type
|
|
self.commit_data(prev_cmd.current_payload_buffer.finalise(), prev_cmd.current_payload_buffer.truncated)
|
|
if self.current_payload_buffer is None:
|
|
self.current_payload_buffer = self.create_payload_buffer(payload_type)
|
|
if payload_is_encoded:
|
|
self.current_payload_buffer.add_base64_data(payload)
|
|
else:
|
|
self.current_payload_buffer.add_unencoded_data(payload)
|
|
|
|
def commit_data(self, data: bytes, truncated: int) -> None:
|
|
if not data:
|
|
return
|
|
if self.current_payload_type.is_text:
|
|
if truncated:
|
|
text = ' too long, truncated'
|
|
else:
|
|
text = data.decode('utf-8', 'replace')
|
|
if self.current_payload_type is PayloadType.title:
|
|
self.title = limit_size(self.title + text)
|
|
elif self.current_payload_type is PayloadType.body:
|
|
self.body = limit_size(self.body + text)
|
|
elif self.current_payload_type is PayloadType.icon:
|
|
if truncated:
|
|
self.log('Ignoring too long notification icon data')
|
|
else:
|
|
icd = self.icon_data_cache_ref()
|
|
if icd:
|
|
self.icon_path = icd.add_icon(self.icon_data_key, data)
|
|
elif self.current_payload_type is PayloadType.buttons:
|
|
self.buttons += tuple(limit_size(x, 256) for x in text.split('\u2028') if x)
|
|
self.buttons = self.buttons[:8]
|
|
|
|
def finalise(self) -> None:
|
|
if self.current_payload_buffer:
|
|
self.commit_data(self.current_payload_buffer.finalise(), self.current_payload_buffer.truncated)
|
|
self.current_payload_buffer = None
|
|
if self.icon_data_key and not self.icon_path:
|
|
icd = self.icon_data_cache_ref()
|
|
if icd:
|
|
self.icon_path = icd.get_icon(self.icon_data_key)
|
|
if self.title:
|
|
self.title = sanitize_text(self.title)
|
|
self.body = sanitize_text(self.body)
|
|
else:
|
|
self.title = sanitize_text(self.body)
|
|
self.body = ''
|
|
self.urgency = Urgency.Normal if self.urgency is None else self.urgency
|
|
self.close_response_requested = bool(self.close_response_requested)
|
|
self.timeout = max(-1, self.timeout)
|
|
self.sound_name = self.sound_name or 'system'
|
|
|
|
def matches_rule_item(self, location:str, query:str) -> bool:
|
|
import re
|
|
pat = re.compile(query)
|
|
if location == 'type':
|
|
for x in self.notification_types:
|
|
if pat.search(x) is not None:
|
|
return True
|
|
val = {'title': self.title, 'body': self.body, 'app': self.application_name}[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:
|
|
|
|
supports_close_events: bool = True
|
|
supports_body: bool = True
|
|
supports_buttons: bool = True
|
|
supports_sound: bool = True
|
|
supports_sound_names: str = 'xdg-names'
|
|
supports_timeout_natively: bool = True
|
|
|
|
def __init__(self, notification_manager: 'NotificationManager'):
|
|
self.notification_manager = notification_manager
|
|
self.initialize()
|
|
|
|
def initialize(self) -> None:
|
|
pass
|
|
|
|
def query_live_notifications(self, channel_id: int, identifier: str) -> None:
|
|
raise NotImplementedError('Implement me in subclass')
|
|
|
|
def close_notification(self, desktop_notification_id: int) -> bool:
|
|
raise NotImplementedError('Implement me in subclass')
|
|
|
|
def notify(self, nc: NotificationCommand, existing_desktop_notification_id: Optional[int]) -> int:
|
|
raise NotImplementedError('Implement me in subclass')
|
|
|
|
def on_new_version_notification_activation(self, cmd: NotificationCommand, which: int) -> None:
|
|
from .update_check import notification_activated
|
|
notification_activated()
|
|
|
|
def payload_type_supported(self, x: PayloadType) -> bool:
|
|
if x is PayloadType.body and not self.supports_body:
|
|
return False
|
|
if x is PayloadType.buttons and not self.supports_buttons:
|
|
return False
|
|
return True
|
|
|
|
def query_response(self, identifier: str) -> str:
|
|
actions = ','.join(x.value for x in Action)
|
|
when = ','.join(x.value for x in OnlyWhen if x.value)
|
|
urgency = ','.join(str(x.value) for x in Urgency)
|
|
i = f'i={identifier or "0"}:'
|
|
p = ','.join(x.value for x in PayloadType if x.value and self.payload_type_supported(x))
|
|
c = ':c=1' if self.supports_close_events else ''
|
|
s = 'system,silent,' + ','.join(sorted(standard_sound_names))
|
|
return f'99;{i}p=?;a={actions}:o={when}:u={urgency}:p={p}{c}:w=1:s={s}'
|
|
|
|
|
|
class MacOSNotificationCategory(NamedTuple):
|
|
id: str
|
|
buttons: tuple[str, ...] = ()
|
|
button_ids: tuple[str, ...] = ()
|
|
|
|
|
|
class MacOSIntegration(DesktopIntegration):
|
|
|
|
supports_close_events: bool = False
|
|
supports_sound_names: str = ''
|
|
supports_timeout_natively: bool = False
|
|
|
|
def initialize(self) -> None:
|
|
from .fast_data_types import cocoa_set_notification_activated_callback
|
|
self.id_counter = count(start=1)
|
|
self.live_notification_queries: list[tuple[int, str]] = []
|
|
self.failed_icons: OrderedDict[str, bool] = OrderedDict()
|
|
self.icd_key_prefix = os.urandom(16).hex()
|
|
self.category_cache: OrderedDict[tuple[str, ...], MacOSNotificationCategory] = OrderedDict()
|
|
self.category_id_counter = count(start=2)
|
|
self.buttons_id_counter = count(start=1)
|
|
self.default_category = MacOSNotificationCategory('1')
|
|
self.current_categories: frozenset[MacOSNotificationCategory] = frozenset()
|
|
cocoa_set_notification_activated_callback(self.notification_activated)
|
|
|
|
def query_live_notifications(self, channel_id: int, identifier: str) -> None:
|
|
from .fast_data_types import cocoa_live_delivered_notifications
|
|
if not cocoa_live_delivered_notifications():
|
|
self.notification_manager.send_live_response(channel_id, identifier, ())
|
|
else:
|
|
self.live_notification_queries.append((channel_id, identifier))
|
|
|
|
def close_notification(self, desktop_notification_id: int) -> bool:
|
|
from .fast_data_types import cocoa_remove_delivered_notification
|
|
close_succeeded = cocoa_remove_delivered_notification(str(desktop_notification_id))
|
|
if debug_desktop_integration:
|
|
log_error(f'Close request for {desktop_notification_id=} {"succeeded" if close_succeeded else "failed"}')
|
|
return close_succeeded
|
|
|
|
def get_icon_for_name(self, name: str) -> str:
|
|
from .fast_data_types import cocoa_bundle_image_as_png
|
|
if name in self.failed_icons:
|
|
return ''
|
|
image_type, image_name = 1, name
|
|
if sic := standard_icon_names.get(name):
|
|
image_name = sic[1]
|
|
image_type = 2
|
|
icd = self.notification_manager.icon_data_cache
|
|
icd_key = self.icd_key_prefix + name
|
|
ans = icd.get_icon(icd_key)
|
|
if ans:
|
|
return ans
|
|
try:
|
|
data = cocoa_bundle_image_as_png(image_name, image_type=image_type)
|
|
except Exception as err:
|
|
if debug_desktop_integration:
|
|
self.notification_manager.log(f'Failed to get icon for {name} with error: {err}')
|
|
self.failed_icons[name] = True
|
|
if len(self.failed_icons) > 256:
|
|
self.failed_icons.popitem(False)
|
|
else:
|
|
return icd.add_icon(icd_key, data)
|
|
return ''
|
|
|
|
def category_for_notification(self, nc: NotificationCommand) -> MacOSNotificationCategory:
|
|
key = nc.buttons
|
|
if not key:
|
|
return self.default_category
|
|
if ans := self.category_cache.get(key):
|
|
self.category_cache.pop(key)
|
|
self.category_cache[key] = ans
|
|
return ans
|
|
ans = self.category_cache[key] = MacOSNotificationCategory(
|
|
str(next(self.category_id_counter)), nc.buttons, tuple(str(next(self.buttons_id_counter)) for x in nc.buttons)
|
|
)
|
|
if len(self.category_cache) > 32:
|
|
self.category_cache.popitem(False)
|
|
return ans
|
|
|
|
def notify(self, nc: NotificationCommand, existing_desktop_notification_id: Optional[int]) -> int:
|
|
desktop_notification_id = existing_desktop_notification_id or next(self.id_counter)
|
|
from .fast_data_types import cocoa_send_notification
|
|
# If the body is not set macos makes the title the body and uses
|
|
# "kitty" as the title. So use a single space for the body in this
|
|
# case. Although https://developer.apple.com/documentation/usernotifications/unnotificationcontent/body?language=objc
|
|
# says printf style strings are stripped this does not actually happen, so dont double % for %% escaping.
|
|
body = (nc.body or ' ')
|
|
assert nc.urgency is not None
|
|
image_path = ''
|
|
if nc.icon_names:
|
|
for name in nc.icon_names:
|
|
if image_path := self.get_icon_for_name(name):
|
|
break
|
|
image_path = image_path or nc.icon_path
|
|
if not image_path and nc.application_name:
|
|
image_path = self.get_icon_for_name(nc.application_name)
|
|
category = self.category_for_notification(nc)
|
|
categories = tuple(self.category_cache.values())
|
|
sc = frozenset(categories)
|
|
if sc == self.current_categories:
|
|
categories = ()
|
|
else:
|
|
self.current_categories = sc
|
|
|
|
cocoa_send_notification(
|
|
nc.application_name or 'kitty', str(desktop_notification_id), nc.title, body,
|
|
category=category, categories=categories, image_path=image_path, urgency=nc.urgency.value,
|
|
muted=nc.sound_name == 'silent' or nc.sound_name in standard_sound_names,
|
|
)
|
|
return desktop_notification_id
|
|
|
|
def notification_activated(self, event: str, ident: str, button_id: str) -> None:
|
|
if event == 'live':
|
|
live_ids = tuple(int(x) for x in ident.split(',') if x)
|
|
self.notification_manager.purge_dead_notifications(live_ids)
|
|
self.live_notification_queries, queries = [], self.live_notification_queries
|
|
for channel_id, req_id in queries:
|
|
self.notification_manager.send_live_response(channel_id, req_id, live_ids)
|
|
return
|
|
if debug_desktop_integration:
|
|
log_error(f'Notification {ident=} {event=} {button_id=}')
|
|
try:
|
|
desktop_notification_id = int(ident)
|
|
except Exception:
|
|
log_error(f'Got unexpected notification activated event with id: {ident!r} from cocoa')
|
|
return
|
|
if event == 'created':
|
|
n = self.notification_manager.notification_created(desktop_notification_id)
|
|
from .fast_data_types import cocoa_live_delivered_notifications
|
|
cocoa_live_delivered_notifications() # so that we purge dead notifications
|
|
if n and n.sound_name in standard_sound_names:
|
|
from .fast_data_types import cocoa_play_system_sound_by_id_async
|
|
cocoa_play_system_sound_by_id_async(standard_sound_names[n.sound_name][1])
|
|
elif event == 'activated':
|
|
self.notification_manager.notification_activated(desktop_notification_id, 0)
|
|
elif event == 'creation_failed':
|
|
self.notification_manager.notification_closed(desktop_notification_id)
|
|
elif event == 'closed': # sadly Crapple never delivers these events
|
|
self.notification_manager.notification_closed(desktop_notification_id)
|
|
elif event == 'button':
|
|
if n := self.notification_manager.in_progress_notification_commands.get(desktop_notification_id):
|
|
if debug_desktop_integration:
|
|
log_error('Button matches notification:', n)
|
|
for c in self.current_categories:
|
|
if c.buttons == n.buttons and button_id in c.button_ids:
|
|
if debug_desktop_integration:
|
|
log_error('Button number:', c.button_ids.index(button_id) + 1)
|
|
self.notification_manager.notification_activated(desktop_notification_id, c.button_ids.index(button_id) + 1)
|
|
break
|
|
else:
|
|
if debug_desktop_integration:
|
|
log_error('No category found with buttons:', n.buttons)
|
|
log_error('Current categories:', self.current_categories)
|
|
|
|
|
|
class FreeDesktopIntegration(DesktopIntegration):
|
|
|
|
supports_body_markup: bool = True
|
|
|
|
def initialize(self) -> None:
|
|
from .fast_data_types import dbus_set_notification_callback
|
|
dbus_set_notification_callback(self.dispatch_event_from_desktop)
|
|
# map the id returned by the notification daemon to the
|
|
# desktop_notification_id we use for the notification
|
|
self.dbus_to_desktop: 'OrderedDict[int, int]' = OrderedDict()
|
|
self.desktop_to_dbus: dict[int, int] = {}
|
|
|
|
def query_live_notifications(self, channel_id: int, identifier: str) -> None:
|
|
self.notification_manager.send_live_response(channel_id, identifier, tuple(self.desktop_to_dbus))
|
|
|
|
def close_notification(self, desktop_notification_id: int) -> bool:
|
|
from .fast_data_types import dbus_close_notification
|
|
close_succeeded = False
|
|
if dbus_id := self.get_dbus_notification_id(desktop_notification_id, 'close_request'):
|
|
close_succeeded = dbus_close_notification(dbus_id)
|
|
if debug_desktop_integration:
|
|
log_error(f'Close request for {desktop_notification_id=} {"succeeded" if close_succeeded else "failed"}')
|
|
return close_succeeded
|
|
|
|
def get_desktop_notification_id(self, dbus_notification_id: int, event: str) -> Optional[int]:
|
|
q = self.dbus_to_desktop.get(dbus_notification_id)
|
|
if q is None:
|
|
if debug_desktop_integration:
|
|
log_error(f'Could not find desktop_notification_id for {dbus_notification_id=} for event {event}')
|
|
return q
|
|
|
|
def get_dbus_notification_id(self, desktop_notification_id: int, event: str) ->Optional[int]:
|
|
q = self.desktop_to_dbus.get(desktop_notification_id)
|
|
if q is None:
|
|
if debug_desktop_integration:
|
|
log_error(f'Could not find dbus_notification_id for {desktop_notification_id=} for event {event}')
|
|
return q
|
|
|
|
def created(self, dbus_notification_id: int, desktop_notification_id: int) -> None:
|
|
self.dbus_to_desktop[desktop_notification_id] = dbus_notification_id
|
|
self.desktop_to_dbus[dbus_notification_id] = desktop_notification_id
|
|
if len(self.dbus_to_desktop) > 128:
|
|
k, v = self.dbus_to_desktop.popitem(False)
|
|
self.desktop_to_dbus.pop(v, None)
|
|
if n := self.notification_manager.notification_created(dbus_notification_id):
|
|
# self.supports_sound does not tell us if the notification server
|
|
# supports named sounds or not so we play the named sound
|
|
# ourselves and tell the server to mute any sound it might play.
|
|
if n.sound_name not in ('system', 'silent'):
|
|
sn = standard_sound_names[n.sound_name][0] if n.sound_name in standard_sound_names else n.sound_name
|
|
from .fast_data_types import play_desktop_sound_async
|
|
play_desktop_sound_async(sn, event_id='desktop notification')
|
|
|
|
def dispatch_event_from_desktop(self, event_type: str, dbus_notification_id: int, extra: Union[int, str]) -> None:
|
|
if event_type == 'capabilities':
|
|
capabilities = frozenset(str(extra).splitlines())
|
|
self.supports_body = 'body' in capabilities
|
|
self.supports_buttons = 'actions' in capabilities
|
|
self.supports_body_markup = 'body-markup' in capabilities
|
|
self.supports_sound = 'sound' in capabilities
|
|
if debug_desktop_integration:
|
|
log_error('Got notification server capabilities:', capabilities)
|
|
return
|
|
if debug_desktop_integration:
|
|
log_error(f'Got notification event from desktop: {event_type=} {dbus_notification_id=} {extra=}')
|
|
if event_type == 'created':
|
|
self.created(dbus_notification_id, int(extra))
|
|
return
|
|
if desktop_notification_id := self.get_desktop_notification_id(dbus_notification_id, event_type):
|
|
if event_type == 'activation_token':
|
|
self.notification_manager.notification_activation_token_received(desktop_notification_id, str(extra))
|
|
elif event_type == 'activated':
|
|
button = 0 if extra == 'default' else int(extra)
|
|
self.notification_manager.notification_activated(desktop_notification_id, button)
|
|
elif event_type == 'closed':
|
|
self.notification_manager.notification_closed(desktop_notification_id)
|
|
|
|
def notify(self, nc: NotificationCommand, existing_desktop_notification_id: Optional[int]) -> int:
|
|
from .fast_data_types import dbus_send_notification
|
|
from .xdg import icon_exists, icon_for_appname
|
|
app_icon = ''
|
|
if nc.icon_names:
|
|
for name in nc.icon_names:
|
|
if sn := standard_icon_names.get(name):
|
|
app_icon = sn[0]
|
|
break
|
|
if icon_exists(name):
|
|
app_icon = name
|
|
break
|
|
if not app_icon:
|
|
app_icon = nc.icon_path or nc.icon_names[0]
|
|
else:
|
|
app_icon = nc.icon_path or icon_for_appname(nc.application_name)
|
|
if not app_icon:
|
|
app_icon = get_custom_window_icon()[1] or logo_png_file
|
|
|
|
body = nc.body
|
|
if self.supports_body_markup:
|
|
body = body.replace('<', '<\u200c').replace('&', '&\u200c') # prevent HTML markup from being recognized
|
|
assert nc.urgency is not None
|
|
replaces_dbus_id = 0
|
|
if existing_desktop_notification_id:
|
|
replaces_dbus_id = self.get_dbus_notification_id(existing_desktop_notification_id, 'notify') or 0
|
|
actions = {'default': ' '} # dbus requires string to not be empty
|
|
for i, b in enumerate(nc.buttons):
|
|
actions[str(i+1)] = b
|
|
desktop_notification_id = dbus_send_notification(
|
|
app_name=nc.application_name or 'kitty', app_icon=app_icon, title=nc.title, body=body, actions=actions,
|
|
timeout=nc.timeout, urgency=nc.urgency.value, replaces=replaces_dbus_id,
|
|
category=(nc.notification_types or ('',))[0], muted=nc.sound_name == 'silent' or nc.sound_name != 'system',
|
|
)
|
|
if debug_desktop_integration:
|
|
log_error(f'Requested creation of notification with {desktop_notification_id=}')
|
|
if existing_desktop_notification_id and replaces_dbus_id:
|
|
self.dbus_to_desktop.pop(replaces_dbus_id, None)
|
|
self.desktop_to_dbus.pop(existing_desktop_notification_id, None)
|
|
return desktop_notification_id
|
|
|
|
|
|
class UIState(NamedTuple):
|
|
has_keyboard_focus: bool
|
|
is_visible: bool
|
|
|
|
|
|
class Channel:
|
|
|
|
def window_for_id(self, channel_id: int) -> Optional[WindowType]:
|
|
boss = get_boss()
|
|
if channel_id:
|
|
return boss.window_id_map.get(channel_id)
|
|
return boss.active_window
|
|
|
|
def ui_state(self, channel_id: int) -> UIState:
|
|
has_focus = is_visible = False
|
|
boss = get_boss()
|
|
if w := self.window_for_id(channel_id):
|
|
has_focus = w.is_active and w.os_window_id == current_focused_os_window_id()
|
|
# window is in the active OS window and the active tab and is visible in the tab layout
|
|
is_visible = w.os_window_id == current_focused_os_window_id() and w.tabref() is boss.active_tab and w.is_visible_in_layout
|
|
return UIState(has_focus, is_visible)
|
|
|
|
def send(self, channel_id: int, osc_escape_code: str) -> bool:
|
|
if w := self.window_for_id(channel_id):
|
|
if not w.destroyed:
|
|
w.screen.send_escape_code_to_child(ESC_OSC, osc_escape_code)
|
|
return True
|
|
return False
|
|
|
|
def focus(self, channel_id: int, activation_token: str) -> None:
|
|
if debug_desktop_integration:
|
|
log_error(f'Focusing window: {channel_id} with activation_token: {activation_token}')
|
|
boss = get_boss()
|
|
if w := self.window_for_id(channel_id):
|
|
boss.set_active_window(w, switch_os_window_if_needed=True, activation_token=activation_token)
|
|
|
|
|
|
sanitize_text = sanitize_control_codes
|
|
|
|
@run_once
|
|
def sanitize_identifier_pat() -> 're.Pattern[str]':
|
|
return re.compile(r'[^a-zA-Z0-9-_+.]+')
|
|
|
|
|
|
def sanitize_id(v: str) -> str:
|
|
return sanitize_identifier_pat().sub('', v)[:512]
|
|
|
|
|
|
class Log:
|
|
def __call__(self, *a: Any, **kw: str) -> None:
|
|
log_error(*a, **kw)
|
|
|
|
|
|
class NotificationManager:
|
|
|
|
def __init__(
|
|
self,
|
|
desktop_integration: Union[MacOSIntegration, FreeDesktopIntegration, None] = None,
|
|
channel: Channel = Channel(),
|
|
log: Log = Log(),
|
|
debug: bool = False,
|
|
base_cache_dir: str = '',
|
|
cleanup_at_exit: bool = True,
|
|
):
|
|
global debug_desktop_integration
|
|
debug_desktop_integration = debug
|
|
if desktop_integration is None:
|
|
self.desktop_integration = MacOSIntegration(self) if is_macos else FreeDesktopIntegration(self)
|
|
else:
|
|
self.desktop_integration = desktop_integration
|
|
self.channel = channel
|
|
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()
|
|
if cleanup_at_exit:
|
|
import atexit
|
|
atexit.register(self.cleanup)
|
|
|
|
def reset(self) -> None:
|
|
self.icon_data_cache.clear()
|
|
self.in_progress_notification_commands: 'OrderedDict[int, NotificationCommand]' = OrderedDict()
|
|
self.in_progress_notification_commands_by_client_id: dict[str, NotificationCommand] = {}
|
|
self.pending_commands: dict[int, NotificationCommand] = {}
|
|
|
|
def notification_created(self, desktop_notification_id: int) -> Optional[NotificationCommand]:
|
|
if n := self.in_progress_notification_commands.get(desktop_notification_id):
|
|
n.created_by_desktop = True
|
|
if n.timeout > 0 and not self.desktop_integration.supports_timeout_natively:
|
|
add_timer(partial(self.expire_notification, desktop_notification_id, id(n)), n.timeout / 1000, False)
|
|
return n
|
|
return None
|
|
|
|
def notification_activation_token_received(self, desktop_notification_id: int, token: str) -> None:
|
|
if n := self.in_progress_notification_commands.get(desktop_notification_id):
|
|
n.activation_token = token
|
|
|
|
def notification_activated(self, desktop_notification_id: int, button: int) -> None:
|
|
if n := self.in_progress_notification_commands.get(desktop_notification_id):
|
|
if not n.close_response_requested:
|
|
self.purge_notification(n)
|
|
if n.focus_requested:
|
|
self.channel.focus(n.channel_id, n.activation_token)
|
|
if n.report_requested:
|
|
self.channel.send(n.channel_id, f'99;i={n.identifier or "0"};{button or ""}')
|
|
if n.on_activation:
|
|
try:
|
|
n.on_activation(n, button)
|
|
except Exception as e:
|
|
self.log('Notification on_activation handler failed with error:', e)
|
|
|
|
def notification_replaced(self, old_cmd: NotificationCommand, new_cmd: NotificationCommand) -> None:
|
|
if old_cmd.desktop_notification_id != new_cmd.desktop_notification_id:
|
|
self.in_progress_notification_commands.pop(old_cmd.desktop_notification_id, None)
|
|
if old_cmd.on_update is not None:
|
|
try:
|
|
old_cmd.on_update(old_cmd, new_cmd)
|
|
except Exception as e:
|
|
self.log('Notification on_update 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 and self.desktop_integration.supports_close_events:
|
|
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)
|
|
|
|
def send_test_notification(self) -> None:
|
|
boss = get_boss()
|
|
if w := boss.active_window:
|
|
from time import monotonic
|
|
cmd = self.create_notification_cmd()
|
|
now = monotonic()
|
|
cmd.title = f'Test {now}'
|
|
cmd.body = f'At: {now}'
|
|
cmd.on_activation = print
|
|
self.notify_with_command(cmd, w.id)
|
|
|
|
def send_new_version_notification(self, version: str) -> None:
|
|
cmd = self.create_notification_cmd()
|
|
cmd.title = 'kitty update available!'
|
|
cmd.body = f'kitty version {version} released'
|
|
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, 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:
|
|
return False
|
|
if cmd.only_when is OnlyWhen.invisible and ui_state.is_visible:
|
|
return False
|
|
return True
|
|
|
|
@property
|
|
def filter_rules(self) -> Iterator[str]:
|
|
return iter(get_options().filter_notification.keys())
|
|
|
|
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 self.filter_rules:
|
|
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) or self.is_notification_filtered(cmd):
|
|
return None
|
|
existing_desktop_notification_id: Optional[int] = None
|
|
existing_cmd = self.in_progress_notification_commands_by_client_id.get(cmd.identifier) if cmd.identifier else None
|
|
if existing_cmd:
|
|
existing_desktop_notification_id = existing_cmd.desktop_notification_id
|
|
desktop_notification_id = self.desktop_integration.notify(cmd, existing_desktop_notification_id)
|
|
self.register_in_progress_notification(cmd, desktop_notification_id)
|
|
if existing_cmd:
|
|
self.notification_replaced(existing_cmd, cmd)
|
|
if not self.desktop_integration.supports_close_events and cmd.close_response_requested:
|
|
self.send_closed_response(channel_id, cmd.identifier, untracked=True)
|
|
return desktop_notification_id
|
|
|
|
def expire_notification(self, desktop_notification_id: int, command_id: int, timer_id: int) -> None:
|
|
if n := self.in_progress_notification_commands.get(desktop_notification_id):
|
|
if id(n) == command_id:
|
|
self.desktop_integration.close_notification(desktop_notification_id)
|
|
|
|
def register_in_progress_notification(self, cmd: NotificationCommand, desktop_notification_id: int) -> None:
|
|
cmd.desktop_notification_id = desktop_notification_id
|
|
self.in_progress_notification_commands[desktop_notification_id] = cmd
|
|
if cmd.identifier:
|
|
self.in_progress_notification_commands_by_client_id[cmd.identifier] = cmd
|
|
if len(self.in_progress_notification_commands) > 128:
|
|
_, cmd = self.in_progress_notification_commands.popitem(False)
|
|
self.in_progress_notification_commands_by_client_id.pop(cmd.identifier, None)
|
|
|
|
def parse_notification_cmd(
|
|
self, prev_cmd: NotificationCommand, channel_id: int, raw: str
|
|
) -> Optional[NotificationCommand]:
|
|
metadata, payload = raw.partition(';')[::2]
|
|
cmd = self.create_notification_cmd()
|
|
try:
|
|
payload_type, payload_is_encoded = cmd.parse_metadata(metadata, prev_cmd)
|
|
except Exception:
|
|
self.log('Malformed metadata section in OSC 99: ' + metadata)
|
|
return None
|
|
if payload_type is PayloadType.query:
|
|
self.channel.send(channel_id, self.desktop_integration.query_response(cmd.identifier))
|
|
return None
|
|
if payload_type is PayloadType.alive:
|
|
if cmd.identifier:
|
|
self.desktop_integration.query_live_notifications(channel_id, cmd.identifier)
|
|
return None
|
|
if payload_type is PayloadType.close:
|
|
if cmd.identifier:
|
|
to_close = self.in_progress_notification_commands_by_client_id.get(cmd.identifier)
|
|
if to_close:
|
|
if not self.desktop_integration.close_notification(to_close.desktop_notification_id):
|
|
if to_close.close_response_requested:
|
|
self.send_closed_response(to_close.channel_id, to_close.identifier)
|
|
self.purge_notification(to_close)
|
|
return None
|
|
|
|
if payload_type is PayloadType.unknown:
|
|
self.log(f'OSC 99: unknown payload type: {payload_type}, ignoring payload')
|
|
payload = ''
|
|
|
|
cmd.set_payload(payload_type, payload_is_encoded, payload, prev_cmd)
|
|
return cmd
|
|
|
|
def send_closed_response(self, channel_id: int, client_id: str, untracked: bool = False) -> None:
|
|
payload = 'untracked' if untracked else ''
|
|
self.channel.send(channel_id, f'99;i={client_id}:p={PayloadType.close.value};{payload}')
|
|
|
|
def send_live_response(self, channel_id: int, client_id: str, live_desktop_ids: Sequence[int]) -> None:
|
|
ids = []
|
|
for desktop_notification_id in live_desktop_ids:
|
|
if n := self.in_progress_notification_commands.get(desktop_notification_id):
|
|
if n.identifier and n.channel_id == channel_id:
|
|
ids.append(n.identifier)
|
|
self.channel.send(channel_id, f'99;i={client_id}:p={PayloadType.alive.value};{",".join(ids)}')
|
|
|
|
def purge_dead_notifications(self, live_desktop_ids: Sequence[int]) -> None:
|
|
for d in set(self.in_progress_notification_commands) - set(live_desktop_ids):
|
|
if debug_desktop_integration:
|
|
log_error(f'Purging dead notification {d} from list of live notifications:', live_desktop_ids)
|
|
self.purge_notification(self.in_progress_notification_commands[d])
|
|
|
|
def purge_notification(self, cmd: NotificationCommand) -> None:
|
|
self.in_progress_notification_commands_by_client_id.pop(cmd.identifier, None)
|
|
self.in_progress_notification_commands.pop(cmd.desktop_notification_id, None)
|
|
|
|
def handle_notification_cmd(self, channel_id: int, osc_code: int, raw: str) -> None:
|
|
if osc_code == 99:
|
|
cmd = self.pending_commands.pop(channel_id, None) or self.create_notification_cmd()
|
|
q = self.parse_notification_cmd(cmd, channel_id, raw)
|
|
if q is not None:
|
|
if q.done:
|
|
self.notify_with_command(q, channel_id)
|
|
else:
|
|
self.pending_commands[channel_id] = q
|
|
elif osc_code == 9:
|
|
n = self.create_notification_cmd()
|
|
n.title = raw
|
|
self.notify_with_command(n, channel_id)
|
|
elif osc_code == 777:
|
|
n = self.create_notification_cmd()
|
|
parts = raw.split(';', 1)
|
|
n.title, n.body = parts[0], (parts[1] if len(parts) > 1 else '')
|
|
self.notify_with_command(n, channel_id)
|
|
|
|
def close_notification(self, desktop_notification_id: int) -> None:
|
|
self.desktop_integration.close_notification(desktop_notification_id)
|
|
|
|
def cleanup(self) -> None:
|
|
del self.icon_data_cache
|