diff --git a/RequestLogger.py b/RequestLogger.py new file mode 100644 index 0000000..21825f2 --- /dev/null +++ b/RequestLogger.py @@ -0,0 +1,136 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import json +import os +import re +import sys +from json import JSONDecodeError +from types import SimpleNamespace + +class Request: + def __init__(self, id: int, method: str, params: dict = None) -> None: + self.id = id + self.method = method + self.params = params + + +class Response: + def __init__(self, id: int, error: str = None, result: object = None) -> None: + self.id = id + self.error = error + self.result = result + + def __str__(self) -> str: + try: + return json.dumps( + Response.__remove_empty_properties(self.__dict__.copy()), + default=lambda o: o.__dict__, + sort_keys=True, + indent=None + ) + except BaseException as _: + return self.__repr__() + + @staticmethod + def __remove_empty_properties(dictionary) -> dict: + for k, v in list(dictionary.items()): + if v is None: + del dictionary[k] + elif isinstance(v, dict): + Response.__remove_empty_properties(v) + return dictionary + +class FSCLogger: + + def __init__(self: FSCLogger) -> None: + self.token = None + self.chat_id = None + self.base_url = None + self.log("starting") + + @staticmethod + def log(message: str) -> None: + sys.stderr.write("[FSCLogger] {message}{linesep}".format(message=message, linesep=os.linesep)) + sys.stdout.flush() + + @staticmethod + def return_and_continue(response: Response) -> None: + if response.error: + # removing result in case of an error + response.result = None + FSCLogger.log("returning an error: {}".format(response.error)) + + FSCLogger.log("returning response: {response}{linesep}".format(response=response, linesep=os.linesep)) + sys.stdout.write("{response}{linesep}".format(response=response, linesep=os.linesep)) + sys.stdout.flush() + + @staticmethod + def return_and_exit(response: Response) -> None: + FSCLogger.return_and_continue(response) + + if response.error: + exit(1) + else: + exit(0) + + def run(self) -> None: + self.log("waiting for requests") + for line in sys.stdin: + parsed_id = None + raw = None + + # parse request identifier + id_match = re.match(r'.*"id":(\d+)', line) + if id_match: + parsed_id = id_match.group(1) + + # parse request + try: + raw = json.loads(line.strip()) + except JSONDecodeError as err: + self.return_and_exit(Response(id=parsed_id, error="failed decoding input: {}".format(err.msg))) + + # map request + if raw: + request = Request(**raw) + self.log("received request: {}".format(vars(request))) + self.handle_request(request) + else: + self.log("dropped input as it couldn't be mapped to a request") + + self.log("exiting") + + def handle_request(self, request: Request) -> None: + match request.method: + case 'GetInfo': + # return information to Icinga Notifications about this channel + info = SimpleNamespace() + info.name = 'FSCLogger' + info.version = '0.1.0' + info.author = 'FSC' + info.config_attrs = [ + { + "name": "logfile", + "type": "string", + "label": { + "de_DE": "Logfile Pfad", + "en_US": "Logfile path" + }, + "help": { + "de_DE": "Pfad zum Logfile", + "en_US": "Path to logfile." + }, + "required": True, + "min": None, + "max": None + }, + ] + + self.return_and_exit(Response(id=request.id, result=info)) + + +# run entrypoint +if __name__ == '__main__': + app = FSCLogger() + app.run()