First commit

.gitignore

@@ -129,3 +129,5 @@ dmypy.json
 # Pyre type checker
 .pyre/
 
+# Ignore config
+conf/kimsufi.yaml

config/kimsufi.sample.yaml

@@ -0,0 +1,9 @@
+config:
+  - main:
+    - api_url: https://www.ovh.com/engine/api/dedicated/server/availabilities?country=FR
+    - id_server:
+      - 160sk2
+      - 160sk3
+  - notifications:
+      - email:
+        - host: domain.tld

servertracker.py

@@ -0,0 +1,9 @@
+#!/usr/bin/env python3
+# coding: utf-8
+# -*- coding: utf-8 -*-
+
+from src.core import OVHTracker
+
+config="config/kimsufi.yaml"
+
+OVHTracker = OVHTracker(config)

src/core.py

@@ -0,0 +1,52 @@
+#!/usr/bin/env python3
+# coding: utf-8
+# -*- coding: utf-8 -*-
+
+import requests
+import pprint
+from src.logs import Logger
+from src.notifications import Notifications
+from src.settings import Settings
+
+
+class OVHTracker:
+    def __init__(self, config):
+        self.settings = Settings().get(config)
+        self.logger = Logger(
+            self.settings["log"]["level"], self.settings["log"]["path"]
+        )
+        self.logger.info("App loaded")
+        self.main()
+
+    def main(self):
+        self.getStock()
+        self.findServer()
+
+    def getStock(self):
+        url = self.settings["main"]["api_url"]
+        self.logger.info("Getting {}".format(url))
+        try:
+            resp = requests.get(url)
+        except requests.exceptions.RequestException as e:
+            raise SystemExit(e)
+        if resp.json is not None:
+            self.stock = resp.json()
+
+    def findServer(self):
+        for server in self.stock:
+            for wantedServer in self.settings["main"]["id_server"]:
+                if wantedServer in server["hardware"]:
+                    for dc in server["datacenters"]:
+                        for wantedDatacenter in self.settings["main"]["datacenter"]:
+                            if wantedDatacenter in dc["datacenter"]:
+                                if dc["availability"] not in ("unavailable", "unknown"):
+                                    self.logger.info(
+                                        "{} is available in {}".format(
+                                            wantedServer, wantedDatacenter
+                                        )
+                                    )
+                                    self.notify(wantedServer)
+
+    def notify(self, server):
+        self.notification = Notifications(self.settings)
+        return True

src/logs.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python3
+# coding: utf-8
+# -*- coding: utf-8 -*-
+
+from logging.handlers import RotatingFileHandler
+
+import logging
+
+
+class Logger:
+    def __new__(self, level, log_path):
+        log = logging.getLogger()
+        log.setLevel(level)
+        formatter = logging.Formatter("%(asctime)s :: %(levelname)s :: %(message)s")
+
+        file = RotatingFileHandler(log_path, "a", 1000000, 1)
+        file.setLevel(level)
+        file.setFormatter(formatter)
+        log.addHandler(file)
+
+        return log

src/notifications.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+# coding: utf-8
+# -*- coding: utf-8 -*-
+
+import apprise
+import pprint
+
+from src.logs import Logger
+
+
+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):
+        pprint.pprint(self)
+        for key, val in self.config.items():
+            if key not in self.allowedNotifyChannel:
+                self.logger.error(
+                    "{} is not possible".format(self.allowedNotifyChannel)
+                )
+            else:
+                pprint.pprint(key)
+                pprint.pprint(self._notifyChannel)
+                self._notifyChannel[key]()
+        return True
+
+    def _addEmail(self):
+        self.logger.info("debug")
+        if self.config["email"]["secure"] is True:
+            mailto = "mailtos"
+        else:
+            mailto = "mailto"
+        self.apprise.add(
+            "{}://{}:{}@{}:{}?from={}&name=OVHTracker".format(
+                mailto,
+                self.config["email"]["user"],
+                self.config["email"]["password"],
+                self.config["email"]["host"],
+                self.config["email"]["port"],
+                self.config["email"]["from"]
+            )
+        )
+
+    def _addDiscord(self):
+        return True
+
+    def _addXmpp(self):
+        return True
+
+    def _addMatrix(self):
+        return True

src/settings.py

@@ -0,0 +1,15 @@
+#!/usr/bin/env python3
+# coding: utf-8
+# -*- coding: utf-8 -*-
+
+import yaml
+
+
+class Settings:
+    def get(self, config):
+        try:
+            with open(config, "r") as file:
+                settings = yaml.load(file, Loader=yaml.FullLoader)
+                return settings
+        except yaml.YAMLError as exc:
+            print("Error in configuration file: {}".format(exc))