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_server.py | 255 +++++++++++++++++++++++++++----------------------------- 1 file changed, 122 insertions(+), 133 deletions(-) (limited to 'beinc_server.py') diff --git a/beinc_server.py b/beinc_server.py index ad81ce9..0f6c248 100755 --- a/beinc_server.py +++ b/beinc_server.py @@ -1,8 +1,8 @@ #!/usr/bin/env python # -*- coding: utf-8 -*- -# Blackmore's Enhanced IRC-Notification Collection (BEINC) v3.0 -# Copyright (C) 2013-2018 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 @@ -29,26 +29,16 @@ import sys from functools import wraps from logging.config import fileConfig - -PY2 = sys.version_info[0] == 2 -PY3 = sys.version_info[0] == 3 - -if PY3: - from http.server import BaseHTTPRequestHandler, HTTPServer -else: - from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer +from http.server import BaseHTTPRequestHandler, HTTPServer try: - if PY3: - import notify2 as pynotify - else: - import pynotify -except ImportError as e: + import notify2 as pynotify +except ImportError: pynotify = None __author__ = 'Simeon Simeonov' -__version__ = '3.0' +__version__ = '4.0' __license__ = 'GPL3' @@ -59,28 +49,30 @@ BEINC_CURRENT_CONFIG_VERSION = 3 class BEINCError401(Exception): - pass + """BEINCError401""" class BEINCError403(Exception): - pass + """BEINCError403""" class BEINCError404(Exception): - pass + """BEINCError404""" class BEINCError405(Exception): - pass + """BEINCError405""" -def beinc_login_required(method): - """ - Decorator for checking login credentials - """ +def eprint(*arg, **kwargs): + """stdderr print wrapper""" + print(*arg, file=sys.stderr, flush=True, **kwargs) + +def beinc_login_required(method): + """Decorator for checking login credentials""" @wraps(method) - def wrapper(self, data, *args, **kwargs): + def wrapper(self, data, *arg, **kwargs): if data.get('resource_name') is None: raise BEINCError403('Resource-name missing') if data.get('password') is None: @@ -91,122 +83,122 @@ def beinc_login_required(method): raise BEINCError401('Wrong instance or password') if not instance.password_match(data.get('password')): raise BEINCError401('Wrong instance or password') - return method(self, data, *args, **kwargs) + return method(self, data, *arg, **kwargs) return wrapper -class BEINCInstance(object): - """ - Represents a single server-instance - """ +class BEINCInstance: + """Represents a single server-instance""" def __init__(self, instance_dict): """ instance_dict: the config-dictionary node that represents this instance """ - self.__message_queue = list() - self.__osd_type = BEINC_OSD_TYPE_NONE - self.__osd_notification = None + self._message_queue = [] + self._osd_type = BEINC_OSD_TYPE_NONE + self._osd_notification = None - self.__name = instance_dict.get('name') - self.__password = instance_dict.get('password', '') - self.__queue_size = int(instance_dict.get('queue_size', 3)) + self._name = instance_dict.get('name') + self._password = instance_dict.get('password', '') + self._queue_size = int(instance_dict.get('queue_size', 3)) if instance_dict['osd_system'].lower() == 'pynotify': - self.__queue_size = 0 # disable queueing + self._queue_size = 0 # disable queueing if pynotify is None: - sys.stderr.write( - 'This server does not possess pynotify capability\n') - sys.stderr.write( - 'Remove the instance {0}'.format(self.__name)) - sys.stderr.write( - 'or define it with "osd_system": "none" ' - 'or other available backend\n') + eprint('This server does not possess pynotify capability') + eprint(f'Remove the instance {self._name} or define it with ' + f'"osd_system": "none" or other ' + f'available backend') sys.exit(errno.EPERM) try: - self.__osd_notification = pynotify.Notification(' ') - if PY3: - self.__osd_notification.timeout = 1000 * int( - instance_dict.get('osd_timeout', 5)) - self.__osd_notification.set_category('im.received') - else: - self.__osd_notification.set_timeout( - 1000 * int(instance_dict.get('osd_timeout', 5))) - self.__osd_notification.set_property( - 'app_name', - '{0} {1}'.format(sys.argv[0], __version__)) - self.__osd_type = BEINC_OSD_TYPE_PYNOTIFY + self._osd_notification = pynotify.Notification(' ') + self._osd_notification.timeout = 1000 * int( + instance_dict.get('osd_timeout', 5)) + self._osd_notification.set_category('im.received') + self._osd_type = BEINC_OSD_TYPE_PYNOTIFY except Exception as e: - sys.stderr.write( - 'Unable to set up a ' - 'pynotify notification object for "{0}" ({1})\n'.format( - self.__name, - e)) + eprint(f'Unable to set up a pynotify notification object ' + f'for "{self._name}" ({e})') sys.exit(errno.EPERM) @property def name(self): - """ - name-property for the server instance (read-only) - """ - return self.__name + """name-property for the server instance (read-only)""" + return self._name @property def queueable(self): - """ - True if this instance has a queueing capability (read-only) - """ - return bool(self.__queue_size) + """True if this instance has a queueing capability (read-only)""" + return bool(self._queue_size) def password_match(self, password): """ Returns True if 'passowrd' matches the instance-password, otherwise - False + + :param password: The password to compare + :type password: str + + :return: True if the password matches, False otherwise + :rtype: bool """ - return True if self.__password == password else False + return self._password == password def send_message(self, title, message): """ Displays or enqueues the message, depending on the instance's type in regard to the osd_system + + :param title: The title + :type title: str + + :param message: The message + :type message: str """ - if self.__osd_type == BEINC_OSD_TYPE_PYNOTIFY: - self.__send_pynotify_messaage(title, message) + if self._osd_type == BEINC_OSD_TYPE_PYNOTIFY: + self._send_pynotify_messaage(title, message) else: - self.__send_message_to_queue(title, message) + self._send_message_to_queue(title, message) def get_queue(self): - """ - Returns a list of dict representation of the message queue - """ - r_value = self.__message_queue - self.__message_queue = list() + """Returns a list of dict representation of the message queue""" + r_value = self._message_queue + self._message_queue = [] return r_value - def __send_pynotify_messaage(self, title, message): + def _send_pynotify_messaage(self, title, message): """ Displays pynotify message + + :param title: The title + :type title: str + + :param message: The message + :type message: str """ - self.__osd_notification.update(summary=title, message=message) - self.__osd_notification.show() + self._osd_notification.update(summary=title, message=message) + self._osd_notification.show() - def __send_message_to_queue(self, title, message): + def _send_message_to_queue(self, title, message): """ Enqueues the message + + :param title: The title + :type title: str + + :param message: The message + :type message: str """ - if len(self.__message_queue) >= self.__queue_size: - self.__message_queue.pop(0) - self.__message_queue.append({'title': title, 'message': message}) + if len(self._message_queue) >= self._queue_size: + self._message_queue.pop(0) + self._message_queue.append({'title': title, 'message': message}) class BEINCCustomHandler(BaseHTTPRequestHandler): - """ - """ + """Custom handler""" def do_POST(self): - """ - Handle POST requests - """ + """Handle POST requests""" if self.path.strip('/') not in ('beinc/push', 'beinc/pull'): - self.__generate_json_error(404, 'Invalid resource path') + self._generate_json_error(404, 'Invalid resource path') return form = cgi.FieldStorage( fp=self.rfile, @@ -222,43 +214,38 @@ class BEINCCustomHandler(BaseHTTPRequestHandler): try: result = {} if self.path.strip('/') == 'beinc/push': - result = self.__handle_push(POST_data) + result = self._handle_push(POST_data) elif self.path.strip('/') == 'beinc/pull': - result = self.__handle_pull(POST_data) - self.__render_to_JSON_response(result) + result = self._handle_pull(POST_data) + self._render_to_JSON_response(result) except BEINCError401 as e: - self.__generate_json_error(401, str(e)) + self._generate_json_error(401, str(e)) except BEINCError403 as e: - self.__generate_json_error(403, str(e)) + self._generate_json_error(403, str(e)) except BEINCError404 as e: - self.__generate_json_error(404, str(e)) + self._generate_json_error(404, str(e)) except BEINCError405 as e: - self.__generate_json_error(405, str(e)) + self._generate_json_error(405, str(e)) except Exception as e: - self.__generate_json_error(500, - 'Unexpected error: {}'.format(str(e))) + self._generate_json_error(500, f'Unexpected error: {e}') def do_GET(self): - """ - Handle GET Requests - """ - self.__generate_json_error(405, 'Unsupported method') + """Handle GET Requests""" + self._generate_json_error(405, 'Unsupported method') @beinc_login_required - def __handle_push(self, data): - """ - """ + def _handle_push(self, data): + """Handle push""" instance = self.server.instances[data.get('resource_name')] try: instance.send_message(data.get('title'), data.get('message')) return {'message': 'OK. Sent.'} except Exception as e: - self.__generate_json_error(500, str(e)) + self._generate_json_error(500, str(e)) @beinc_login_required - def __handle_pull(self, data): - """ - """ + def _handle_pull(self, data): + """Handle pull""" instance = self.server.instances[data.get('resource_name')] try: if not instance.queueable: @@ -267,13 +254,12 @@ class BEINCCustomHandler(BaseHTTPRequestHandler): return {'message': 'OK. Fetched.', 'data': {'messages': instance.get_queue()}} except Exception as e: - self.__generate_json_error(500, str(e)) + self._generate_json_error(500, str(e)) - def __generate_json_error(self, code, message): + def _generate_json_error(self, code, message): """ Generates response header and json content for errors - Keyword Arguments: :param code: the HTTP code :type code: int @@ -288,9 +274,8 @@ class BEINCCustomHandler(BaseHTTPRequestHandler): sort_keys=True, indent=4).encode('utf-8')) - def __render_to_JSON_response(self, context): + def _render_to_JSON_response(self, context): """ - Keyword Arguments: :param context: the context-dict to be converted to json :type context: dict """ @@ -305,31 +290,36 @@ class BEINCCustomHandler(BaseHTTPRequestHandler): class BEINCNotifyServer(HTTPServer): - """ - """ + """BEINCNotifyServer class""" + def __init__(self, *arg, **kwargs): + """Default constructor""" + super().__init__(*arg, **kwargs) + self._config = None + self._instances = {} + def set_config(self, config): """ Sets the configuration dict for the server, instantiates the BEINC instances and initiates the defined OSD backends """ - self.__config = config - self.__instances = dict() + self._config = config # initialize pynotify if the module exists and if needed if pynotify: - for instance in self.__config['server']['instances']: + for instance in self._config['server']['instances']: # check if we have at least one instance that uses pynotify # before initializing it if instance.get('osd_system', '').lower() == 'pynotify': if not pynotify.init('BEINC Notify'): - sys.stderr.write('pynotify.init failed! Exiting...\n') + eprint('pynotify.init failed! Exiting...') sys.exit(1) break + instance = {'name': 'Invalid'} try: - for instance in self.__config['server']['instances']: - self.__instances[instance['name']] = BEINCInstance(instance) - logger.info('Instance "{0}" added'.format(instance['name'])) + for instance in self._config['server']['instances']: + self._instances[instance['name']] = BEINCInstance(instance) + logger.info('Instance %s added', instance['name']) except Exception as e: - sys.stderr.write('Unable to create instance "{0}": {1}\n'.format( + eprint('Unable to create instance "{0}": {1}'.format( instance['name'], e)) sys.exit(1) @@ -339,7 +329,7 @@ class BEINCNotifyServer(HTTPServer): """ a property that returns the instance list (read-only) """ - return self.__instances + return self._instances if __name__ == '__main__': @@ -390,15 +380,14 @@ if __name__ == '__main__': parser.add_argument( '-v', '--version', action='version', - version='%(prog)s {0}'.format(__version__), + version=f'%(prog)s {__version__}', help='Display program-version and exit') args = parser.parse_args() try: with open(args.config_file, 'r') as fp: config_dict = json.load(fp) except Exception as e: - sys.stderr.write('Unable to parse {0}: {1}\n'.format(args.config_file, - e)) + eprint(f'Unable to parse {args.config_file}: {e}') sys.exit(errno.EIO) try: if os.path.isfile(args.logger_config): @@ -411,11 +400,11 @@ if __name__ == '__main__': logger = logging.getLogger('beinc') logger.info('BEINC starting. Loading config...') if config_dict.get('config_version') != BEINC_CURRENT_CONFIG_VERSION: - sys.stderr.write( + eprint( 'WARNING: The version of the config-file: {0} ({1}) ' 'does not correspond to the latest version supported ' 'by this program ({2})\nCheck beinc_config_sample.json ' - 'for the newest features!\n'.format( + 'for the newest features!'.format( args.config_file, config_dict.get('config_version', 'Not set'), BEINC_CURRENT_CONFIG_VERSION)) @@ -441,6 +430,6 @@ if __name__ == '__main__': except KeyboardInterrupt: print('\n\nTerminating...') except Exception as e: - sys.stderr.write('BEINCServer critical error: {0}\n'.format(e)) + eprint(f'BEINCServer critical error: {e}') sys.exit(1) sys.exit(0) -- cgit v1.3