iframe-proxy | Sunbelt Computer Software

615 lines (559 loc) · 22.1 KB
# Blackmore's Enhanced IRC-Notification Collection (BEINC)
# Copyright (C) 2013-2024 Simeon Simeonov
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
"""BEINC client for Weechat"""
import datetime
import json
import urllib.parse
import urllib.request
import weechat
__author__ = 'Simeon Simeonov'
__version__ = '4.4'
__license__ = 'GPL3'
enabled = True
global_values = {}
# few constants #
BEINC_POLICY_NONE = 0
BEINC_POLICY_ALL = 1
BEINC_POLICY_LIST_ONLY = 2
BEINC_CURRENT_CONFIG_VERSION = 2
class WeechatTarget:
    The target (destination) class
    Each remote destination is represented as a WeechatTarget object
    def __init__(self, target_dict):
        """
        :param target_dict: The config-dict node that represents this instance
        :type target_dict: dict
        """
        self._name = target_dict.get('name', '')
        if self._name == '':
            raise Exception('"name" not defined for target')
        self._url = target_dict.get('target_url', '')
        if self._url == '':
            raise Exception('"target_url" not defined for target')
        self._password = target_dict.get('target_password', '')
        self._pm_title_template = target_dict.get(
            'pm_title_template', '%s @ %S'
        self._pm_message_template = target_dict.get(
            'pm_message_template', '%m'
        self._cm_title_template = target_dict.get(
            'cm_title_template', '%c @ %S'
        self._cm_message_template = target_dict.get(
            'cm_message_template', '%s -> %m'
        self._nm_title_template = target_dict.get(
            'nm_title_template', '%c @ %S'
        self._nm_message_template = target_dict.get(
            'nm_message_template', '%s -> %m'
        self._chans = set(target_dict.get('channel_list', []))
        self._nicks = set(target_dict.get('nick_list', []))
        self._chan_messages_policy = int(
            target_dict.get('channel_messages_policy', BEINC_POLICY_LIST_ONLY)
        self._priv_messages_policy = int(
            target_dict.get('private_messages_policy', BEINC_POLICY_ALL)
        self._notifications_policy = int(
            target_dict.get('notifications_policy', BEINC_POLICY_ALL)
        self._cert_file = target_dict.get('target_cert_file')
        self._timestamp_format = target_dict.get(
            'target_timestamp_format', '%H:%M:%S'
        self._debug = bool(target_dict.get('debug', False))
        self._enabled = bool(target_dict.get('enabled', True))
        self._socket_timeout = int(target_dict.get('socket_timeout', 3))
        self._ssl_ciphers = target_dict.get('ssl_ciphers', '')
        self._disable_hostname_check = bool(
            target_dict.get('disable-hostname-check', False)
        self._ssl_version = target_dict.get('ssl_version', 'auto')
        self._last_message = None  # datetime.datetime instance
        self._context = None
        self._context_setup()
    @property
    def name(self):
        """Target name (read-only property)"""
        return self._name
    @property
    def chans(self):
        """Target channel list (read-only property)"""
        return self._chans
    @property
    def nicks(self):
        """Target nick list (read-only property)"""
        return self._nicks
    @property
    def channel_messages_policy(self):
        """The target's channel messages policy (read-only property)"""
        return self._chan_messages_policy
    @property
    def private_messages_policy(self):
        """The target's private messages policy (read-only property)"""
        return self._priv_messages_policy
    @property
    def notifications_policy(self):
        """The target's notifications policy (read-only property)"""
        return self._notifications_policy
    @property
    def enabled(self):
        """The target's enabled status (bool property)"""
        return self._enabled
    @enabled.setter
    def enabled(self, value):
        """The target's enabled status (bool property)"""
        self._enabled = value
    def __repr__(self):
        """repr() implementation"""
        last_message = 'never'
        if self._last_message is not None:
            last_message = self._last_message.strftime('%Y-%m-%d %H:%M:%S')
        return (
            f'name: {self._name}\nurl: {self._url}\n'
            f"enabled: {'yes' if self._enabled else 'no'}\n"
            f"channel_list: {', '.join(self._chans)}\n"
            f"nick_list: {', '.join(self._nicks)}\n"
            f'channel_messages_policy: {self._chan_messages_policy}\n'
            f'private_messages_policy: {self._priv_messages_policy}\n'
            f'notifications_policy: {self._notifications_policy}\n'
            f'last message: {last_message}\n'
            f'socket timeout: {self._socket_timeout}\n'
            f'ssl-version: {self._ssl_version}\n'
            f"ciphers: {self._ssl_ciphers or 'auto'}\n"
            "disable hostname check: "
            f"{'yes' if self._disable_hostname_check else 'no'}\n"
            f"debug: {'yes' if self._debug else 'no'}\n\n"
    def send_private_message_notification(self, values):
        """
        Sends a private message notification to the represented target
        :param value: Dict pupulated by the irc msg-handler
        :type value: dict
        """
        try:
            title = self._fetch_formatted_str(self._pm_title_template, values)
            message = self._fetch_formatted_str(
                self._pm_message_template, values
            if not self._send_beinc_message(title, message) and self._debug:
                beinc_prnt(
                    f'BEINC DEBUG: send_private_message_notification-ERROR '
                    f'for "{self._name}": _send_beinc_message -> False'
        except Exception as exp:
            if self._debug:
                beinc_prnt(
                    f'BEINC DEBUG: send_private_message_notification-ERROR '
                    f'for "{self._name}": {exp}'
    def send_channel_message_notification(self, values):
        """
        Sends a channel message notification to the represented target
        :param value: Dict pupulated by the irc msg-handler
        :type value: dict
        """
        try:
            title = self._fetch_formatted_str(self._cm_title_template, values)
            message = self._fetch_formatted_str(
                self._cm_message_template, values
            if not self._send_beinc_message(title, message) and self._debug:
                beinc_prnt(
                    f'BEINC DEBUG: send_channel_message_notification-ERROR '
                    f'for "{self._name}": _send_beinc_message -> False'
        except Exception as exp:
            if self._debug:
                beinc_prnt(
                    f'BEINC DEBUG: send_channel_message_notification-ERROR '
                    f'for "{self._name}": {exp}'
    def send_notify_message_notification(self, values):
        """
        Sends a notify message notification to the represented target
        :param value: Dict pupulated by the irc msg-handler
        :type value: dict
        """
        try:
            title = self._fetch_formatted_str(self._nm_title_template, values)
            message = self._fetch_formatted_str(
                self._nm_message_template, values
            if not self._send_beinc_message(title, message) and self._debug:
                beinc_prnt(
                    f'BEINC DEBUG: send_notify_message_notification-ERROR '
                    f'for "{self._name}": _send_beinc_message -> False'
        except Exception as exp:
            if self._debug:
                beinc_prnt(
                    f'BEINC DEBUG: send_notify_message_notification-ERROR '
                    f'for "{self._name}": {exp}'
    def send_broadcast_notification(self, message):
        """
        Sends a 'pure' broadcast / test message notification
        to the represented target
        :param message: A single message string
        :type message: str
        """
        try:
            title = 'BEINC broadcast'
            if not self._send_beinc_message(title, message) and self._debug:
                beinc_prnt(
                    f'BEINC DEBUG: send_broadcast_notification-ERROR '
                    f'for "{self._name}": _send_beinc_message -> False'
        except Exception as exp:
            if self._debug:
                beinc_prnt(
                    f'BEINC DEBUG: send_broadcast_notification-ERROR '
                    f'for "{self._name}": {exp}'
    def _context_setup(self):
        """Sets up the SSL context"""
        if self._context is not None:
            return True
        try:
            context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
            if self._cert_file:
                context.verify_mode = ssl.CERT_REQUIRED
                context.load_verify_locations(
                    cafile=os.path.expanduser(self._cert_file)
                context.check_hostname = bool(not self._disable_hostname_check)
            else:
                context.check_hostname = False
                context.verify_mode = ssl.CERT_NONE
            if self._ssl_ciphers and self._ssl_ciphers != 'auto':
                context.set_ciphers(self._ssl_ciphers)
            self._context = context
            return True
        except ssl.SSLError as err:
            if self._debug:
                beinc_prnt(f'BEINC DEBUG: SSL/TLS error: {err}\n')
        except Exception as exp:
            if self._debug:
                beinc_prnt(f'BEINC DEBUG: Generic context error: {exp}\n')
        self._context = None
        return False
    def _fetch_formatted_str(self, template, values):
        """
        Returns a formatted string by replacing the defined
        macros in 'template' the the corresponding values from 'values'
        :param template: The template to use
        :type template: str
        :param values: The values dict
        :type values: dict
        :return: The formatted string
        :rtype: str
        """
        timestamp = datetime.datetime.now().strftime(self._timestamp_format)
        replacements = {
            '%S': values['server'],
            '%s': values['source_nick'],
            '%c': values['channel'],
            '%m': values['message'],
            '%t': timestamp,
            '%p': 'BEINC',
            '%n': values['own_nick'],
        for key, value in replacements.items():
            template = template.replace(key, value)
        return template
    def _send_beinc_message(self, title, message):
        """
        The method implements the BEINC "protocol" by generating a simple
        POST request
        :param title: The title
        :type title: str
        :param message: The message
        :type message: str
        :return: The status
        :rtype: bool
        """
        try:
            if self._context is None and not self._context_setup():
                return False
            response = urllib.request.urlopen(
                self._url,
                data=urllib.parse.urlencode(
                        ('resource_name', self._name),
                        ('password', self._password),
                        ('title', title),
                        ('message', message),
                ).encode('utf-8'),
                timeout=self._socket_timeout,
                context=self._context,
            response_dict = json.loads(response.read().decode('utf-8'))
            if response.code != 200:
                raise OSError(response_dict.get('message', ''))
            if self._debug:
                beinc_prnt(
                    "BEINC DEBUG: Server responded: "
                    f"{response_dict.get('message')}"
            self._last_message = datetime.datetime.now()
            return True
        except ssl.SSLError as err:
            if self._debug:
                beinc_prnt(f'BEINC DEBUG: SSL/TLS error: {err}\n')
        except OSError as err:
            if self._debug:
                beinc_prnt(f'BEINC DEBUG: Connection error: {err}\n')
        except Exception as exp:
            if self._debug:
                beinc_prnt(f'BEINC DEBUG: Unable to send message: {exp}\n')
        return False
def beinc_prnt(message_str):
    """wrapper around weechat.prnt"""
    if global_values['use_current_buffer']:
        weechat.prnt(weechat.current_buffer(), message_str)
        weechat.prnt('', message_str)
def beinc_cmd_broadcast_handler(cmd_tokens):
    """handles: '/beinc broadcast' command actions"""
    if not cmd_tokens:
        beinc_prnt('beinc broadcast <message>')
        return weechat.WEECHAT_RC_OK
    for target in target_list:
        if target.enabled:
            target.send_broadcast_notification(' '.join(cmd_tokens))
    return weechat.WEECHAT_RC_OK
def beinc_cmd_target_handler(cmd_tokens):
    """handles: '/beinc target' command actions"""
    if not cmd_tokens or cmd_tokens[0] not in ['list', 'enable', 'disable']:
        beinc_prnt('beinc target [ list | enable <name> | disable <name> ]')
        return weechat.WEECHAT_RC_OK
    if cmd_tokens[0] == 'list':
        beinc_prnt('--- Globals ---')
        for key, value in global_values.items():
            beinc_prnt(f'{key} -> {value}')
        beinc_prnt('--- Targets ---')
        for target in target_list:
            beinc_prnt(str(target))
        beinc_prnt('---------------')
    elif cmd_tokens[0] == 'enable':
        if not cmd_tokens[1:]:
            beinc_prnt('missing a name-argument')
            return weechat.WEECHAT_RC_OK
        name = ' '.join(cmd_tokens[1:])
        for target in target_list:
            if target.name == name:
                target.enabled = True
                beinc_prnt(f'target "{name}" enabled')
                break
        else:
            beinc_prnt(f'no matching target for "{name}"')
    elif cmd_tokens[0] == 'disable':
        if not cmd_tokens[1:]:
            beinc_prnt('missing a name-argument')
            return weechat.WEECHAT_RC_OK
        name = ' '.join(cmd_tokens[1:])
        for target in target_list:
            if target.name == name:
                target.enabled = False
                beinc_prnt(f'target "{name}" disabled')
                break
        else:
            beinc_prnt(f'no matching target for "{name}"')
    return weechat.WEECHAT_RC_OK
def beinc_command(data, buffer_obj, args):
    """Callback function handling the Weechat's /beinc command"""
    global enabled
    cmd_tokens = args.split()
    if not cmd_tokens:
        return weechat.WEECHAT_RC_OK
    if args == 'on':
        enabled = True
        beinc_prnt('BEINC on')
    elif args == 'off':
        enabled = False
        beinc_prnt('BEINC off')
    elif args == 'reload':
        beinc_prnt('Reloading BEINC...')
        beinc_init()
    elif cmd_tokens[0] in ('broadcast', 'test'):
        return beinc_cmd_broadcast_handler(cmd_tokens[1:])
    elif cmd_tokens[0] == 'target':
        return beinc_cmd_target_handler(cmd_tokens[1:])
        beinc_prnt(
            'syntax: /beinc < on | off | reload |'
            ' broadcast <text> | target <action> >'
    return weechat.WEECHAT_RC_OK
def beinc_privmsg_handler(data, signal, signal_data):
    """Callback function the *PRIVMSG* IRC messages hooked by Weechat"""
    if not enabled:
        return weechat.WEECHAT_RC_OK
    prvmsg_dict = weechat.info_get_hashtable(
        'irc_message_parse', {'message': signal_data}
    # packing the privmsg handler values
    ph_values = {}
    ph_values['server'] = signal.split(',')[0]
    ph_values['own_nick'] = weechat.info_get('irc_nick', ph_values['server'])
    ph_values['channel'] = prvmsg_dict['arguments'].split(':')[0].strip()
    ph_values['source_nick'] = prvmsg_dict['nick']
    ph_values['message'] = ':'.join(
        prvmsg_dict['arguments'].split(':')[1:]
    ).strip()
    if ph_values['channel'] == ph_values['own_nick']:
        # priv messages are handled here
        if not global_values['global_private_messages_policy']:
            return weechat.WEECHAT_RC_OK
        for target in target_list:
            if not target.enabled:
                continue
            p_messages_policy = target.private_messages_policy
            if p_messages_policy == BEINC_POLICY_ALL or (
                p_messages_policy == BEINC_POLICY_LIST_ONLY
                and f"{ph_values['server']}.{ph_values['source_nick'].lower()}"
                in target.nicks
                target.send_private_message_notification(ph_values)
    elif ph_values['own_nick'].lower() in ph_values['message'].lower():
        # notify messages are handled here
        if not global_values['global_notifications_policy']:
            return weechat.WEECHAT_RC_OK
        for target in target_list:
            if not target.enabled:
                continue
            if target.notifications_policy == BEINC_POLICY_ALL or (
                target.notifications_policy == BEINC_POLICY_LIST_ONLY
                and f"{ph_values['server']}.{ph_values['channel'].lower()}"
                in target.chans
                target.send_notify_message_notification(ph_values)
    elif global_values['global_channel_messages_policy']:
        # chan messages are handled here
        if not global_values['global_notifications_policy']:
            return weechat.WEECHAT_RC_OK
        for target in target_list:
            if not target.enabled:
                continue
            c_messages_policy = target.channel_messages_policy
            if c_messages_policy == BEINC_POLICY_ALL or (
                c_messages_policy == BEINC_POLICY_LIST_ONLY
                and f"{ph_values['server']}.{ph_values['channel'].lower()}"
                in target.chans
                target.send_channel_message_notification(ph_values)
    return weechat.WEECHAT_RC_OK
def beinc_init():
    Ran every time the script is (re)loaded
    It loads the config (.json) file and (re)loads its contents into memory
    beinc_init() will disable all notifications on failure
    global enabled
    global target_list
    global global_values
    # global chans/nicks sets are used to speed up the filtering
    global_values = {}
    target_list = []
    custom_error = ''
    global_values['global_channel_messages_policy'] = False
    global_values['global_private_messages_policy'] = False
    global_values['global_notifications_policy'] = False
    global_values['use_current_buffer'] = False
        beinc_config_file_str = os.path.join(
            weechat.info_get('weechat_dir', ''), 'beinc_weechat.json'
        beinc_prnt(f'Parsing {beinc_config_file_str}...')
        custom_error = 'load error'
        with io.open(beinc_config_file_str, 'r', encoding='utf-8') as fp:
            config_dict = json.load(fp)
        custom_error = 'target parse error'
        global_values['use_current_buffer'] = bool(
            config_dict['irc_client'].get('use_current_buffer', False)
        if (
            config_dict.get('config_version', 0)
            != BEINC_CURRENT_CONFIG_VERSION
            beinc_prnt(
                "WARNING: The version of the config-file: "
                f"{beinc_config_file_str} "
                f"({config_dict.get('config_version', 0)}) "
                "does not correspond to the latest version supported "
                f"by this program ({BEINC_CURRENT_CONFIG_VERSION})\n"
                "Check beinc_config_sample.json for the newest features!"
        for target in config_dict['irc_client']['targets']:
            try:
                new_target = WeechatTarget(target)
            except Exception as exp:
                beinc_prnt(f'Unable to add target: {exp}')
                continue
            if new_target.channel_messages_policy:
                global_values['global_channel_messages_policy'] = True
            if new_target.private_messages_policy:
                global_values['global_private_messages_policy'] = True
            if new_target.notifications_policy:
                global_values['global_notifications_policy'] = True
            target_list.append(new_target)
            beinc_prnt(f'BEINC target "{new_target.name}" added')
        beinc_prnt('Done!')
    except Exception as exp:
        beinc_prnt(
            f'ERROR: unable to parse {beinc_config_file_str}: '
            f'{custom_error} - {exp}\nBEINC is now disabled'
        enabled = False
        # do not return error / exit the script
        # in order to give a smoother opportunity to fix a 'broken' config
        return weechat.WEECHAT_RC_OK
    return weechat.WEECHAT_RC_OK
weechat.register(
    'beinc_weechat',
    __author__,
    __version__,
    __license__,
    "Blackmore's Extended IRC Notification Collection (Weechat Client)",
version = weechat.info_get('version_number', '') or 0
if int(version) < 0x00040000:
    weechat.prnt('', 'WeeChat version >= 0.4.0 is required to run beinc')
    weechat.hook_command(
        'beinc',
        'BEINC command',
        '< broadcast <message> | on | off | reload | target <action> >',
            'Available target actions:\n'
            'disable <target name>\nenable <target name>\nlist'
        'None',
        'beinc_command',
        '',
    weechat.hook_signal('*,irc_in2_privmsg', 'beinc_privmsg_handler', '')
    beinc_init()
    weechat.prnt('', 'beinc initiated!')
Sunbelt Computer Software

PL/B Language Development and Support

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Sunbelt Computer Software

PL/B Language Development and Support

FilesExpand file tree

beinc.py

Latest commit

History

beinc.py

File metadata and controls
