OVH-server-tracker/src/notifications.py

#!/usr/bin/env python3
# coding: utf-8
# -*- coding: utf-8 -*-

import apprise

from src.logs import Logger
from src.settings import Settings


class Notifications:
    def __init__(self, config):
        self.logger = Logger(config["log"]["level"], config["log"]["path"])
        self.config = config["notifications"]
        self.apprise = apprise.Apprise()
        self.allowedNotifyChannel = ["email", "discord", "matrix", "xmpp"]
        self._notifyChannel = {
            "email": self._addEmail,
            "discord": self._addDiscord,
            "xmpp": self._addXmpp,
            "matrix": self._addMatrix,
        }
        self.setNotificationsChannel()

    def setNotificationsChannel(self):
        for key, val in self.config.items():
            if key not in self.allowedNotifyChannel:
                self.logger.error(
                    "{} is not possible".format(self.allowedNotifyChannel)
                )
            else:
                self._notifyChannel[key]()
        return True

    def run(self, message):
        refOfServer = Settings().get("config/refToName.yml")
        self.apprise.notify(
            body="",
            title="{}".format(message),
            notify_type=apprise.NotifyType.INFO,
        )

    def _addEmail(self):
        if self.config["email"]["secure"] is True:
            mailto = "mailtos"
            mode = self.config["email"]["mode"]
        else:
            mailto = "mailto"
            mode = ""
        self.apprise.add(
            "{}://{}:{}@{}:{}?from={}&name=OVHTracker&to={}?mode={}".format(
                mailto,
                self.config["email"]["user"],
                self.config["email"]["password"],
                self.config["email"]["host"],
                self.config["email"]["port"],
                self.config["email"]["from"],
                self.config["email"]["to"],
                mode,
            )
        )

    def _addXmpp(self):
        if self.config["xmpp"]["secure"] is True:
            xmpp = "xmpps"
        else:
            xmpp = "xmpp"
        self.apprise.add(
            "{}://{}:{}@{}:{}?to={}".format(
                xmpp,
                self.config["xmpp"]["user"],
                self.config["xmpp"]["password"],
                self.config["xmpp"]["host"],
                self.config["xmpp"]["port"],
                self.config["xmpp"]["to"],
            )
        )

    def _addDiscord(self):
        self.apprise.add(
            "discord://{}/{}/".format(
                self.config["discord"]["webhookid"],
                self.config["discord"]["webhooktoken"],
            )
        )

    def _addMatrix(self):
        self.apprise.add(
            "matrixs://{}:{}@{}/!{}?webhook=matrix".format(
                self.config["matrix"]["user"],
                self.config["matrix"]["token"],
                self.config["matrix"]["host"],
                self.config["matrix"]["room_id"],
            )
        )