# -*- coding: utf-8 -*- # # Project: weechat-notify-send # Homepage: https://github.com/s3rvac/weechat-notify-send # Description: Sends highlight and message notifications through notify-send. # Requires libnotify. # License: MIT (see below) # # Copyright (c) 2015 by Petr Zemek and contributors # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in # all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. # from __future__ import print_function import os import re import subprocess import sys import time # Ensure that we are running under WeeChat. try: import weechat except ImportError: sys.exit('This script has to run under WeeChat (https://weechat.org/).') # Name of the script. SCRIPT_NAME = 'notify_send' # Author of the script. SCRIPT_AUTHOR = 's3rvac' # Version of the script. SCRIPT_VERSION = '0.8' # License under which the script is distributed. SCRIPT_LICENSE = 'MIT' # Description of the script. SCRIPT_DESC = 'Sends highlight and message notifications through notify-send.' # Name of a function to be called when the script is unloaded. SCRIPT_SHUTDOWN_FUNC = '' # Used character set (utf-8 by default). SCRIPT_CHARSET = '' # Script options. OPTIONS = { 'notify_on_highlights': ( 'on', 'Send notifications on highlights.' ), 'notify_on_privmsgs': ( 'on', 'Send notifications on private messages.' ), 'notify_on_filtered_messages': ( 'off', 'Send notifications also on filtered (hidden) messages.' ), 'notify_when_away': ( 'on', 'Send also notifications when away.' ), 'notify_for_current_buffer': ( 'on', 'Send also notifications for the currently active buffer.' ), 'notify_on_all_messages_in_buffers': ( '', 'A comma-separated list of buffers for which you want to receive ' 'notifications on all messages that appear in them.' ), 'notify_on_all_messages_in_buffers_that_match': ( '', 'A comma-separated list of regex patterns of buffers for which you ' 'want to receive notifications on all messages that appear in them.' ), 'notify_on_messages_that_match': ( '', 'A comma-separated list of regex patterns that you want to receive ' 'notifications on when message body matches.' ), 'min_notification_delay': ( '500', 'A minimal delay between successive notifications from the same ' 'buffer (in milliseconds; set to 0 to show all notifications).' ), 'ignore_messages_tagged_with': ( ','.join([ 'notify_none', # Buffer with line is not added to hotlist 'irc_join', # Joined IRC 'irc_quit', # Quit IRC 'irc_part', # Parted a channel 'irc_status', # Status messages 'irc_nick_back', # A nick is back on server 'irc_401', # No such nick/channel 'irc_402', # No such server ]), 'A comma-separated list of message tags for which no notifications ' 'should be shown.' ), 'ignore_buffers': ( '', 'A comma-separated list of buffers from which no notifications should ' 'be shown.' ), 'ignore_buffers_starting_with': ( '', 'A comma-separated list of buffer prefixes from which no ' 'notifications should be shown.' ), 'ignore_nicks': ( '', 'A comma-separated list of nicks from which no notifications should ' 'be shown.' ), 'ignore_nicks_starting_with': ( '', 'A comma-separated list of nick prefixes from which no ' 'notifications should be shown.' ), 'nick_separator': ( ': ', 'A separator between a nick and a message.' ), 'escape_html': ( 'on', "Escapes the '<', '>', and '&' characters in notification messages." ), 'max_length': ( '72', 'Maximal length of a notification (0 means no limit).' ), 'ellipsis': ( '[..]', 'Ellipsis to be used for notifications that are too long.' ), 'icon': ( '/usr/share/icons/hicolor/32x32/apps/weechat.png', 'Path to an icon to be shown in notifications.' ), 'timeout': ( '5000', 'Time after which the notification disappears (in milliseconds; ' 'set to 0 to disable).' ), 'transient': ( 'on', 'When a notification expires or is dismissed, remove it from the ' 'notification bar.' ), 'urgency': ( 'normal', 'Urgency (low, normal, critical).' ) } class Notification(object): """A representation of a notification.""" def __init__(self, source, message, icon, timeout, transient, urgency): self.source = source self.message = message self.icon = icon self.timeout = timeout self.transient = transient self.urgency = urgency def default_value_of(option): """Returns the default value of the given option.""" return OPTIONS[option][0] def add_default_value_to(description, default_value): """Adds the given default value to the given option description.""" # All descriptions end with a period, so do not add another period. return '{} Default: {}.'.format( description, default_value if default_value else '""' ) def nick_that_sent_message(tags, prefix): """Returns a nick that sent the message based on the given data passed to the callback. """ # 'tags' is a comma-separated list of tags that WeeChat passed to the # callback. It should contain a tag of the following form: nick_XYZ, where # XYZ is the nick that sent the message. for tag in tags: if tag.startswith('nick_'): return tag[5:] # There is no nick in the tags, so check the prefix as a fallback. # 'prefix' (str) is the prefix of the printed line with the message. # Usually (but not always), it is a nick with an optional mode (e.g. on # IRC, @ denotes an operator and + denotes a user with voice). We have to # remove the mode (if any) before returning the nick. # Strip also a space as some protocols (e.g. Matrix) may start prefixes # with a space. It probably means that the nick has no mode set. if prefix.startswith(('~', '&', '@', '%', '+', '-', ' ')): return prefix[1:] return prefix def parse_tags(tags): """Parses the given "list" of tags (str) from WeeChat into a list.""" return tags.split(',') def message_printed_callback(data, buffer, date, tags, is_displayed, is_highlight, prefix, message): """A callback when a message is printed.""" is_displayed = int(is_displayed) is_highlight = int(is_highlight) tags = parse_tags(tags) nick = nick_that_sent_message(tags, prefix) if notification_should_be_sent(buffer, tags, nick, is_displayed, is_highlight, message): notification = prepare_notification(buffer, nick, message) send_notification(notification) return weechat.WEECHAT_RC_OK def notification_should_be_sent(buffer, tags, nick, is_displayed, is_highlight, message): """Should a notification be sent?""" if notification_should_be_sent_disregarding_time(buffer, tags, nick, is_displayed, is_highlight, message): # The following function should be called only when the notification # should be sent (it updates the last notification time). if not is_below_min_notification_delay(buffer): return True return False def notification_should_be_sent_disregarding_time(buffer, tags, nick, is_displayed, is_highlight, message): """Should a notification be sent when not considering time?""" if not nick: # A nick is required to form a correct notification source/message. return False if i_am_author_of_message(buffer, nick): return False if not is_displayed: if not notify_on_filtered_messages(): return False if buffer == weechat.current_buffer(): if not notify_for_current_buffer(): return False if is_away(buffer): if not notify_when_away(): return False if ignore_notifications_from_messages_tagged_with(tags): return False if ignore_notifications_from_nick(nick): return False if ignore_notifications_from_buffer(buffer): return False if is_private_message(buffer): return notify_on_private_messages() if is_highlight: return notify_on_highlights() if notify_on_messages_that_match(message): return True if notify_on_all_messages_in_buffer(buffer): return True return False def is_below_min_notification_delay(buffer): """Is a notification in the given buffer below the minimal delay between successive notifications from the same buffer? When called, this function updates the time of the last notification. """ # We store the time of the last notification in a buffer-local variable to # make it persistent over the lifetime of this script. LAST_NOTIFICATION_TIME_VAR = 'notify_send_last_notification_time' last_notification_time = buffer_get_float( buffer, 'localvar_' + LAST_NOTIFICATION_TIME_VAR ) min_notification_delay = weechat.config_get_plugin('min_notification_delay') # min_notification_delay is in milliseconds (str). To compare it with # last_notification_time (float in seconds), we have to convert it to # seconds (float). min_notification_delay = float(min_notification_delay) / 1000 current_time = time.time() # We have to update the last notification time before returning the result. buffer_set_float( buffer, 'localvar_set_' + LAST_NOTIFICATION_TIME_VAR, current_time ) return (min_notification_delay > 0 and current_time - last_notification_time < min_notification_delay) def buffer_get_float(buffer, property): """A variant of weechat.buffer_get_x() for floats. This variant is needed because WeeChat supports only buffer_get_string() and buffer_get_int(). """ value = weechat.buffer_get_string(buffer, property) return float(value) if value else 0.0 def buffer_set_float(buffer, property, value): """A variant of weechat.buffer_set() for floats. This variant is needed because WeeChat supports only integers and strings. """ weechat.buffer_set(buffer, property, str(value)) def names_for_buffer(buffer): """Returns a list of all names for the given buffer.""" # The 'buffer' parameter passed to our callback is actually the buffer's ID # (e.g. '0x2719cf0'). We have to check its name (e.g. 'freenode.#weechat') # and short name (e.g. '#weechat') because these are what users specify in # their configs. buffer_names = [] full_name = weechat.buffer_get_string(buffer, 'name') if full_name: buffer_names.append(full_name) short_name = weechat.buffer_get_string(buffer, 'short_name') if short_name: buffer_names.append(short_name) # Consider >channel and #channel to be equal buffer names. The reason # is that the https://github.com/rawdigits/wee-slack script replaces # '#' with '>' to indicate that someone in the buffer is typing. This # fixes the behavior of several configuration options (e.g. # 'notify_on_all_messages_in_buffers') when weechat_notify_send is used # together with the wee_slack script. # # Note that this is only needed to be done for the short name. Indeed, # the full name always stays unchanged. if short_name.startswith('>'): buffer_names.append('#' + short_name[1:]) return buffer_names def notify_for_current_buffer(): """Should we also send notifications for the current buffer?""" return weechat.config_get_plugin('notify_for_current_buffer') == 'on' def notify_on_highlights(): """Should we send notifications on highlights?""" return weechat.config_get_plugin('notify_on_highlights') == 'on' def notify_on_private_messages(): """Should we send notifications on private messages?""" return weechat.config_get_plugin('notify_on_privmsgs') == 'on' def notify_on_filtered_messages(): """Should we also send notifications for filtered (hidden) messages?""" return weechat.config_get_plugin('notify_on_filtered_messages') == 'on' def notify_when_away(): """Should we also send notifications when away?""" return weechat.config_get_plugin('notify_when_away') == 'on' def is_away(buffer): """Is the user away?""" return weechat.buffer_get_string(buffer, 'localvar_away') != '' def is_private_message(buffer): """Has a private message been sent?""" return weechat.buffer_get_string(buffer, 'localvar_type') == 'private' def i_am_author_of_message(buffer, nick): """Am I (the current WeeChat user) the author of the message?""" return weechat.buffer_get_string(buffer, 'localvar_nick') == nick def split_option_value(option, separator=','): """Splits the value of the given plugin option by the given separator and returns the result in a list. """ values = weechat.config_get_plugin(option) if not values: # When there are no values, return the empty list instead of ['']. return [] return [value.strip() for value in values.split(separator)] def ignore_notifications_from_messages_tagged_with(tags): """Should notifications be ignored for a message tagged with the given tags? """ ignored_tags = split_option_value('ignore_messages_tagged_with') for ignored_tag in ignored_tags: for tag in tags: if tag == ignored_tag: return True return False def ignore_notifications_from_buffer(buffer): """Should notifications from the given buffer be ignored?""" buffer_names = names_for_buffer(buffer) for buffer_name in buffer_names: if buffer_name and buffer_name in ignored_buffers(): return True for buffer_name in buffer_names: for prefix in ignored_buffer_prefixes(): if prefix and buffer_name and buffer_name.startswith(prefix): return True return False def ignored_buffers(): """A generator of buffers from which notifications should be ignored.""" for buffer in split_option_value('ignore_buffers'): yield buffer def ignored_buffer_prefixes(): """A generator of buffer prefixes from which notifications should be ignored. """ for prefix in split_option_value('ignore_buffers_starting_with'): yield prefix def ignore_notifications_from_nick(nick): """Should notifications from the given nick be ignored?""" if nick in ignored_nicks(): return True for prefix in ignored_nick_prefixes(): if prefix and nick.startswith(prefix): return True return False def ignored_nicks(): """A generator of nicks from which notifications should be ignored.""" for nick in split_option_value('ignore_nicks'): yield nick def ignored_nick_prefixes(): """A generator of nick prefixes from which notifications should be ignored. """ for prefix in split_option_value('ignore_nicks_starting_with'): yield prefix def notify_on_messages_that_match(message): """Should we send a notification for the given message, provided it matches any of the requested patterns? """ message_patterns = split_option_value('notify_on_messages_that_match') for pattern in message_patterns: if re.search(pattern, message): return True return False def buffers_to_notify_on_all_messages(): """A generator of buffer names in which the user wants to be notified for all messages. """ for buffer in split_option_value('notify_on_all_messages_in_buffers'): yield buffer def buffer_patterns_to_notify_on_all_messages(): """A generator of buffer-name patterns in which the user wants to be notifier for all messages. """ for pattern in split_option_value('notify_on_all_messages_in_buffers_that_match'): yield pattern def notify_on_all_messages_in_buffer(buffer): """Does the user want to be notified for all messages in the given buffer? """ buffer_names = names_for_buffer(buffer) # Option notify_on_all_messages_in_buffers: for buf in buffers_to_notify_on_all_messages(): if buf in buffer_names: return True # Option notify_on_all_messages_in_buffers_that_match: for pattern in buffer_patterns_to_notify_on_all_messages(): for buf in buffer_names: if re.search(pattern, buf): return True return False def prepare_notification(buffer, nick, message): """Prepares a notification from the given data.""" if is_private_message(buffer): source = nick else: source = (weechat.buffer_get_string(buffer, 'short_name') or weechat.buffer_get_string(buffer, 'name')) message = nick + nick_separator() + message max_length = int(weechat.config_get_plugin('max_length')) if max_length > 0: ellipsis = weechat.config_get_plugin('ellipsis') message = shorten_message(message, max_length, ellipsis) if weechat.config_get_plugin('escape_html') == 'on': message = escape_html(message) message = escape_slashes(message) icon = weechat.config_get_plugin('icon') timeout = weechat.config_get_plugin('timeout') transient = should_notifications_be_transient() urgency = weechat.config_get_plugin('urgency') return Notification(source, message, icon, timeout, transient, urgency) def should_notifications_be_transient(): """Should the sent notifications be transient, i.e. should they be removed from the notification bar once they expire or are dismissed? """ return weechat.config_get_plugin('transient') == 'on' def nick_separator(): """Returns a nick separator to be used.""" separator = weechat.config_get_plugin('nick_separator') return separator if separator else default_value_of('nick_separator') def shorten_message(message, max_length, ellipsis): """Shortens the message to at most max_length characters by using the given ellipsis. """ # In Python 2, we need to decode the message and ellipsis into Unicode to # correctly (1) detect their length and (2) shorten the message. Failing to # do that could make the shortened message invalid and cause notify-send to # fail. For example, when we have bytes, we cannot guarantee that we do not # split the message inside of a multibyte character. if sys.version_info.major == 2: try: message = message.decode('utf-8') ellipsis = ellipsis.decode('utf-8') except UnicodeDecodeError: # Either (or both) of the two cannot be decoded. Continue in a # best-effort manner. pass message = shorten_unicode_message(message, max_length, ellipsis) if sys.version_info.major == 2: if not isinstance(message, str): message = message.encode('utf-8') return message def shorten_unicode_message(message, max_length, ellipsis): """An internal specialized version of shorten_message() when the both the message and ellipsis are str (in Python 3) or unicode (in Python 2). """ if max_length <= 0 or len(message) <= max_length: # Nothing to shorten. return message if len(ellipsis) >= max_length: # We cannot include any part of the message. return ellipsis[:max_length] return message[:max_length - len(ellipsis)] + ellipsis def escape_html(message): """Escapes HTML characters in the given message.""" # Only the following characters need to be escaped # (https://wiki.ubuntu.com/NotificationDevelopmentGuidelines). message = message.replace('&', '&') message = message.replace('<', '<') message = message.replace('>', '>') return message def escape_slashes(message): """Escapes slashes in the given message.""" # We need to escape backslashes to prevent notify-send from interpreting # them, e.g. we do not want to print a newline when the message contains # '\n'. return message.replace('\\', r'\\') def send_notification(notification): """Sends the given notification to the user.""" notify_cmd = ['notify-send', '--app-name', 'weechat'] if notification.icon: notify_cmd += ['--icon', notification.icon] if notification.timeout: notify_cmd += ['--expire-time', str(notification.timeout)] if notification.transient: notify_cmd += ['--hint', 'int:transient:1'] if notification.urgency: notify_cmd += ['--urgency', notification.urgency] # We need to add '--' before the source and message to ensure that # notify-send considers the remaining parameters as the source and the # message. This prevents errors when a source or message starts with '--'. notify_cmd += [ '--', # notify-send fails with "No summary specified." when no source is # specified, so ensure that there is always a non-empty source. notification.source or '-', notification.message ] # Prevent notify-send from messing up the WeeChat screen when occasionally # emitting assertion messages by redirecting the output to /dev/null (users # would need to run /redraw to fix the screen). # In Python < 3.3, there is no subprocess.DEVNULL, so we have to use a # workaround. with open(os.devnull, 'wb') as devnull: try: subprocess.check_call( notify_cmd, stderr=subprocess.STDOUT, stdout=devnull, ) except Exception as ex: error_message = '{} (reason: {!r}). {}'.format( 'Failed to send the notification via notify-send', '{}: {}'.format(ex.__class__.__name__, ex), 'Ensure that you have notify-send installed in your system.', ) print(error_message, file=sys.stderr) if __name__ == '__main__': # Registration. weechat.register( SCRIPT_NAME, SCRIPT_AUTHOR, SCRIPT_VERSION, SCRIPT_LICENSE, SCRIPT_DESC, SCRIPT_SHUTDOWN_FUNC, SCRIPT_CHARSET ) # Initialization. for option, (default_value, description) in OPTIONS.items(): description = add_default_value_to(description, default_value) weechat.config_set_desc_plugin(option, description) if not weechat.config_is_set_plugin(option): weechat.config_set_plugin(option, default_value) # Catch all messages on all buffers and strip colors from them before # passing them into the callback. weechat.hook_print('', '', '', 1, 'message_printed_callback', '')