Modules

1 files changed, 119 insertions, 0 deletions

diff --git a/fortnite_status/__init__.py b/fortnite_status/__init__.py
new file mode 100644
index 0000000..244502b
--- /dev/null
+++ b/fortnite_status/__init__.py
@@ -0,0 +1,119 @@
+# Simple utility class which provides an access to Fortnite service status report.
+# Usage: ForniteStatus().getStatus()
+
+import requests as req
+import html5lib
+from bs4 import BeautifulSoup
+
+# Polling interval in seconds
+__polling_interval__ = 5 * 60 # 5 minutes
+
+class FortniteStatusObserver:
+    async def update(self, fortnite_status) -> None:
+        pass
+
+class FortniteStatus:
+    """
+    Instantiate a FortniteStatus class.
+    """
+
+    class Status:
+        serviceStatuses = []
+
+        def __init__(self, serviceStatuses):
+            self.serviceStatuses = serviceStatuses
+        
+        def __eq__(self, other): 
+            if not isinstance(other, FortniteStatus.Status):
+                return NotImplemented
+            return sorted(self.serviceStatuses) == sorted(other.serviceStatuses)
+
+        def prettify(self):
+            return 'Fortnite services status:\n' + '\n'.join([serviceStatus.prettify() for serviceStatus in self.serviceStatuses])
+
+    class ServiceStatus:
+        serviceName = ''
+        status = False
+
+        def __init__(self, serviceName, status):
+            self.serviceName = serviceName
+            self.status = status
+        
+        def __lt__(self, other):
+            if not isinstance(other, FortniteStatus.ServiceStatus):
+                return NotImplemented
+            return self.serviceName < other.serviceName    
+        
+        def __eq__(self, other): 
+            if not isinstance(other, FortniteStatus.ServiceStatus):
+                return NotImplemented
+            return self.serviceName == other.serviceName and self.status == other.status
+
+
+        def prettify(self):
+            return f'{self.serviceName}, {self.status}'
+
+    def __findFortniteStatusHtmlComponent(self, html):
+        for component in html.findAll('div', {'class': 'component-container'}):
+                innerContainers = component.findAll('div', {'class': 'component-inner-container'})
+                for innerContainer in innerContainers:
+                    for names in innerContainer.findAll('span', {'class': 'name'}):
+                        for name in names.findAll('span'):
+                            if 'class' not in name.attrs and 'Fortnite' in name.text:
+                                return component
+
+    def __parseFortniteStatus(self, html):
+        component = self.__findFortniteStatusHtmlComponent(html)
+        serviceStatuses = []
+        childContainer = component.find('div', {'class': 'child-components-container'})
+        for innerContainer in childContainer.findAll('div', {'class': 'component-inner-container'}):
+            name = innerContainer.find('span', {'class': 'name'}).text.strip()
+            statusString = innerContainer.find('span', {'class': 'component-status'}).text.strip()
+            if statusString == 'Operational':
+                statusCode = True
+            else:
+                statusCode = False
+            serviceStatuses.append(self.ServiceStatus(name, statusCode))
+        return self.Status(serviceStatuses)
+
+    def getStatus(self):
+        webContent = req.get("https://status.epicgames.com/")
+        parsedHtml = BeautifulSoup(webContent.text, 'html5lib')
+
+        return self.__parseFortniteStatus(parsedHtml)
+
+
+
+    def printStatus(self):
+        """
+        Prints a current Fortnite services status in stdout.
+        Example:
+
+        """
+        print(self.getStatus().prettify())
+
+class FortniteStatusNotifier:
+
+    __fortnite_status_observer: FortniteStatusObserver
+    __fortnite_status: FortniteStatus
+    __last_fortnite_status: any
+
+    def __init__(self, fortnite_status_observer: FortniteStatusObserver):
+        self.__fortnite_status_observer = fortnite_status_observer
+        self.__fortnite_status = FortniteStatus()
+    
+    async def run(self):
+        # Initialize status
+        self.__last_fortnite_status = self.__fortnite_status.getStatus()
+        while True:
+            await self.__read_status()
+            await asyncio.sleep(__polling_interval__)
+    
+    async def __read_status(self):
+        service_status_tmp = self.__fortnite_status.getStatus()
+        if service_status_tmp != self.__last_fortnite_status:
+            await self.__notify(service_status_tmp)
+        self.__last_fortnite_status = service_status_tmp
+
+    async def __notify(self, fortnite_status):
+        await self.__fortnite_status_observer.update(fortnite_status)
\ No newline at end of file