From 1a5e60fc1069d39db1b0935f4ac8e4cace83fb76 Mon Sep 17 00:00:00 2001 From: Simeon Simeonov Date: Fri, 10 Apr 2020 23:27:15 +0200 Subject: Remove Python2 support and cleanup the code and the documentation --- beinc_weechat.py | 416 ++++++++++++++++++++++++++----------------------------- 1 file changed, 196 insertions(+), 220 deletions(-) (limited to 'beinc_weechat.py') diff --git a/beinc_weechat.py b/beinc_weechat.py index 132a22a..c1a1e1e 100644 --- a/beinc_weechat.py +++ b/beinc_weechat.py @@ -1,7 +1,7 @@ # -*- coding: utf-8 -*- -# Blackmore's Enhanced IRC-Notification Collection (BEINC) v3.0 -# Copyright (C) 2013-2019 Simeon Simeonov +# Blackmore's Enhanced IRC-Notification Collection (BEINC) v4.0 +# Copyright (C) 2013-2020 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 @@ -15,35 +15,25 @@ # You should have received a copy of the GNU General Public License # along with this program. If not, see . - - +"""BEINC client for Weechat""" import datetime import json import os import socket import ssl -import sys +import urllib.parse +import urllib.request import weechat -PY2 = sys.version_info[0] == 2 -PY3 = sys.version_info[0] == 3 - -if PY3: - from urllib.parse import urlencode - from urllib.request import urlopen -else: - from urllib import urlencode - from urllib2 import urlopen - __author__ = 'Simeon Simeonov' -__version__ = '3.0' +__version__ = '4.0' __license__ = 'GPL3' enabled = True -global_values = dict() +global_values = {} # few constants # BEINC_POLICY_NONE = 0 @@ -52,322 +42,317 @@ BEINC_POLICY_LIST_ONLY = 2 BEINC_CURRENT_CONFIG_VERSION = 2 -class WeechatTarget(object): +class WeechatTarget: """ The target (destination) class + Each remote destination is represented as a WeechatTarget object """ def __init__(self, target_dict): """ - target_dict: the config-dictionary node that represents this instance + :param target_dict: The config-dict node that represents this instance + :type target_dict: dict """ - self.__name = target_dict.get('name', '') - if self.__name == '': + 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 == '': + 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', list())) - self.__nicks = set(target_dict.get('nick_list', list())) - self.__chan_messages_policy = int(target_dict.get( + 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( + self._priv_messages_policy = int(target_dict.get( 'private_messages_policy', BEINC_POLICY_ALL)) - self.__notifications_policy = int(target_dict.get( + 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( + 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() + 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 + """Target name (read-only property)""" + return self._name @property def chans(self): - """ - Target channel list (read-only property) - """ - return self.__chans + """Target channel list (read-only property)""" + return self._chans @property def nicks(self): - """ - Target nick list (read-only property) - """ - return self.__nicks + """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 + """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 + """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 + """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 + """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 + """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') + if self._last_message is not None: + last_message = self._last_message.strftime('%Y-%m-%d %H:%M:%S') return ('name: {0}\nurl: {1}\nenabled: {2}\nchannel_list: {3}\n' 'nick_list: {4}\nchannel_messages_policy: {5}\n' 'private_messages_policy: {6}\nnotifications_policy: {7}\n' 'last message: {8}\nsocket timeout: {9}\nssl-version: {10}\n' 'ciphers: {11}\ndisable hostname check: {12}\n' 'debug: {13}\n\n'.format( - self.__name, - self.__url, - 'yes' if self.__enabled else 'no', - ', '.join(self.__chans), - ', '.join(self.__nicks), - self.__chan_messages_policy, - self.__priv_messages_policy, - self.__notifications_policy, + self._name, + self._url, + 'yes' if self._enabled else 'no', + ', '.join(self._chans), + ', '.join(self._nicks), + self._chan_messages_policy, + self._priv_messages_policy, + self._notifications_policy, last_message, - self.__socket_timeout, - self.__ssl_version, - self.__ssl_ciphers or 'auto', - 'yes' if self.__disable_hostname_check else 'no', - 'yes' if self.__debug else 'no')) + self._socket_timeout, + self._ssl_version, + self._ssl_ciphers or 'auto', + 'yes' if self._disable_hostname_check else 'no', + 'yes' if self._debug else 'no')) def send_private_message_notification(self, values): """ - sends a private message notification to the represented target + Sends a private message notification to the represented target - values: dict pupulated by the irc msg-handler + :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, + 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: + if not self._send_beinc_message(title, message) and self._debug: beinc_prnt( - 'BEINC DEBUG: send_private_message_notification-ERROR ' - 'for "{0}": __send_beinc_message -> False'.format( - self.__name)) + f'BEINC DEBUG: send_private_message_notification-ERROR ' + f'for "{self._name}": _send_beinc_message -> False') except Exception as e: - if self.__debug: + if self._debug: beinc_prnt( - 'BEINC DEBUG: send_private_message_notification-ERROR ' - 'for "{0}": {1}'.format(self.__name, e)) + f'BEINC DEBUG: send_private_message_notification-ERROR ' + f'for "{self._name}": {e}') def send_channel_message_notification(self, values): """ - sends a channel message notification to the represented target + Sends a channel message notification to the represented target - values: dict pupulated by the irc msg-handler + :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: + 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( - 'BEINC DEBUG: send_channel_message_notification-ERROR ' - 'for "{0}": __send_beinc_message -> False'.format( - self.__name)) + f'BEINC DEBUG: send_channel_message_notification-ERROR ' + f'for "{self._name}": _send_beinc_message -> False') except Exception as e: - if self.__debug: + if self._debug: beinc_prnt( - 'BEINC DEBUG: send_channel_message_notification-ERROR ' - 'for "{0}": {1}'.format(self.__name, e)) + f'BEINC DEBUG: send_channel_message_notification-ERROR ' + f'for "{self._name}": {e}') def send_notify_message_notification(self, values): """ - sends a notify message notification to the represented target + Sends a notify message notification to the represented target - values: dict pupulated by the irc msg-handler + :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: + 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( - 'BEINC DEBUG: send_notify_message_notification-ERROR ' - 'for "{0}": __send_beinc_message -> False'.format( - self.__name)) + f'BEINC DEBUG: send_notify_message_notification-ERROR ' + f'for "{self._name}": _send_beinc_message -> False') except Exception as e: - if self.__debug: + if self._debug: beinc_prnt( - 'BEINC DEBUG: send_notify_message_notification-ERROR ' - 'for "{0}": {1}'.format(self.__name, e)) + f'BEINC DEBUG: send_notify_message_notification-ERROR ' + f'for "{self._name}": {e}') def send_broadcast_notification(self, message): """ - sends a 'pure' broadcast / test message notification + Sends a 'pure' broadcast / test message notification to the represented target - message: a single message string + :param message: A single message string + :type message: str """ try: title = 'BEINC broadcast' - if not self.__send_beinc_message(title, message) and self.__debug: + if not self._send_beinc_message(title, message) and self._debug: beinc_prnt( - 'BEINC DEBUG: send_broadcast_notification-ERROR ' - 'for "{0}": __send_beinc_message -> False'.format( - self.__name)) + f'BEINC DEBUG: send_broadcast_notification-ERROR ' + f'for "{self._name}": _send_beinc_message -> False') except Exception as e: - if self.__debug: + if self._debug: beinc_prnt( - 'BEINC DEBUG: send_broadcast_notification-ERROR ' - 'for "{0}": {1}'.format(self.__name, e)) + f'BEINC DEBUG: send_broadcast_notification-ERROR ' + f'for "{self._name}": {e}') - def __context_setup(self): - """ - """ - if self.__context is not None: + def _context_setup(self): + """Sets up the SSL context""" + if self._context is not None: return True try: context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2) context.verify_mode = ssl.CERT_NONE - if self.__cert_file: + if self._cert_file: context.verify_mode = ssl.CERT_REQUIRED context.load_verify_locations(cafile=os.path.expanduser( - self.__cert_file)) + self._cert_file)) context.check_hostname = bool( - not self.__disable_hostname_check) - if self.__ssl_ciphers and self.__ssl_ciphers != 'auto': - context.set_ciphers(self.__ssl_ciphers) - self.__context = context + not self._disable_hostname_check) + if self._ssl_ciphers and self._ssl_ciphers != 'auto': + context.set_ciphers(self._ssl_ciphers) + self._context = context return True except ssl.SSLError as e: - if self.__debug: - beinc_prnt('BEINC DEBUG: SSL/TLS error: {0}\n'.format(e)) + if self._debug: + beinc_prnt(f'BEINC DEBUG: SSL/TLS error: {e}\n') except Exception as e: - if self.__debug: - beinc_prnt('BEINC DEBUG: Generic context error: {0}\n'.format( - e)) - self.__context = None + if self._debug: + beinc_prnt(f'BEINC DEBUG: Generic context error: {e}\n') + self._context = None return False - def __fetch_formatted_str(self, template, values): + def _fetch_formatted_str(self, template, values): """ - returns a formatted string by replacing the defined + Returns a formatted string by replacing the defined macros in 'template' the the corresponding values from 'values' - values: dict - template: str + :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) + 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': u'BEINC', + '%p': 'BEINC', '%n': values['own_nick']} for key, value in replacements.items(): template = template.replace(key, value) - return template.encode('utf-8') + return template - def __send_beinc_message(self, title, message): + def _send_beinc_message(self, title, message): """ - the function implements the BEINC "protocol" by generating a simple + 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(): + if self._context is None and not self._context_setup(): return False - response = urlopen( - self.__url, - data=urlencode( + response = urllib.request.urlopen( + self._url, + data=urllib.parse.urlencode( ( - ('resource_name', self.__name), - ('password', self.__password), + ('resource_name', self._name), + ('password', self._password), ('title', title), ('message', message) )).encode('utf-8'), - timeout=self.__socket_timeout, - context=self.__context) + timeout=self._socket_timeout, + context=self._context) response_dict = json.loads(response.read().decode('utf-8')) if response.code != 200: raise socket.error(response_dict.get('message', '')) - if self.__debug: + if self._debug: beinc_prnt('BEINC DEBUG: Server responded: {0}'.format( response_dict.get('message'))) - self.__last_message = datetime.datetime.now() + self._last_message = datetime.datetime.now() return True except ssl.SSLError as e: - if self.__debug: - beinc_prnt('BEINC DEBUG: SSL/TLS error: {0}\n'.format(e)) + if self._debug: + beinc_prnt(f'BEINC DEBUG: SSL/TLS error: {e}\n') except socket.error as e: - if self.__debug: - beinc_prnt('BEINC DEBUG: Connection error: {0}\n'.format(e)) + if self._debug: + beinc_prnt(f'BEINC DEBUG: Connection error: {e}\n') except Exception as e: - if self.__debug: - beinc_prnt('BEINC DEBUG: Unable to send message: {0}\n'.format( - e)) + if self._debug: + beinc_prnt(f'BEINC DEBUG: Unable to send message: {e}\n') return False def beinc_prnt(message_str): - """ - wrapper around weechat.prnt - """ + """wrapper around weechat.prnt""" if global_values['use_current_buffer']: weechat.prnt(weechat.current_buffer(), message_str) else: @@ -375,9 +360,7 @@ def beinc_prnt(message_str): def beinc_cmd_broadcast_handler(cmd_tokens): - """ - handles: '/beinc broadcast' command actions - """ + """handles: '/beinc broadcast' command actions""" if not cmd_tokens: beinc_prnt('beinc broadcast ') return weechat.WEECHAT_RC_OK @@ -388,9 +371,7 @@ def beinc_cmd_broadcast_handler(cmd_tokens): def beinc_cmd_target_handler(cmd_tokens): - """ - handles: '/beinc target' command actions - """ + """handles: '/beinc target' command actions""" if not cmd_tokens or cmd_tokens[0] not in ['list', 'enable', 'disable']: beinc_prnt('beinc target [ list | enable | disable ]') return weechat.WEECHAT_RC_OK @@ -410,10 +391,10 @@ def beinc_cmd_target_handler(cmd_tokens): for target in target_list: if target.name == name: target.enabled = True - beinc_prnt('target "{0}" enabled'.format(name)) + beinc_prnt(f'target "{name}" enabled') break else: - beinc_prnt('no matching target for "{0}"'.format(name)) + beinc_prnt(f'no matching target for "{name}"') elif cmd_tokens[0] == 'disable': if not cmd_tokens[1:]: beinc_prnt('missing a name-argument') @@ -422,17 +403,15 @@ def beinc_cmd_target_handler(cmd_tokens): for target in target_list: if target.name == name: target.enabled = False - beinc_prnt('target "{0}" disabled'.format(name)) + beinc_prnt(f'target "{name}" disabled') break else: - beinc_prnt('no matching target for "{0}"'.format(name)) + 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 - """ + """Callback function handling the Weechat's /beinc command""" global enabled cmd_tokens = args.split() if not cmd_tokens: @@ -457,15 +436,13 @@ def beinc_command(data, buffer_obj, args): def beinc_privmsg_handler(data, signal, signal_data): - """ - Callback function the *PRIVMSG* IRC messages hooked by Weechat - """ + """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 = dict() + 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() @@ -529,8 +506,8 @@ def beinc_init(): global global_values # global chans/nicks sets are used to speed up the filtering - global_values = dict() - target_list = list() + global_values = {} + target_list = [] custom_error = '' global_values['global_channel_messages_policy'] = False global_values['global_private_messages_policy'] = False @@ -541,7 +518,7 @@ def beinc_init(): beinc_config_file_str = os.path.join( weechat.info_get('weechat_dir', ''), 'beinc_weechat.json') - beinc_prnt('Parsing {0}...'.format(beinc_config_file_str)) + beinc_prnt(f'Parsing {beinc_config_file_str}...') custom_error = 'load error' with open(beinc_config_file_str, 'r') as fp: config_dict = json.load(fp, encoding='utf-8') @@ -562,7 +539,7 @@ def beinc_init(): try: new_target = WeechatTarget(target) except Exception as e: - beinc_prnt('Unable to add target: {0}'.format(e)) + beinc_prnt(f'Unable to add target: {e}') continue if new_target.channel_messages_policy: global_values['global_channel_messages_policy'] = True @@ -571,12 +548,11 @@ def beinc_init(): if new_target.notifications_policy: global_values['global_notifications_policy'] = True target_list.append(new_target) - beinc_prnt('BEINC target "{0}" added'.format(new_target.name)) + beinc_prnt(f'BEINC target "{new_target.name}" added') beinc_prnt('Done!') except Exception as e: - beinc_prnt('ERROR: unable to parse {0}: {1} - {2}\n' - 'BEINC is now disabled'.format( - beinc_config_file_str, custom_error, e)) + beinc_prnt(f'ERROR: unable to parse {beinc_config_file_str}: ' + f'{custom_error} - {e}\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 -- cgit v1.3