sdbs-infra/sdbs_infra/dashboard/views.py

import asyncio
import socket
import time
from collections import namedtuple

import aiohttp
import psutil
from bs4 import BeautifulSoup
from django.views.generic import TemplateView
from humanize import naturalsize

from sdbs_infra.dashboard.models import Service, ServiceStatus, Link


class IndexView(TemplateView):
    template_name = "index.html"

    def get_context_data(self, **kwargs):

        return {
            'links': asyncio.run(self.process_links(list(Link.objects.all()))),
            'services': asyncio.run(self.process_services(list(Service.objects.all()))),
            'vps_stats': self.vps_stats()
        }

    async def process_links(self, links):
        result = []

        session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=5, sock_connect=1))

        for link in links:
            index_text = None, None
            if not link.image:
                try:
                    async with session.get(link.url) as response:
                        index_status, index_text = response.status, await response.text()
                except asyncio.TimeoutError:
                    pass

            image = link.image.url if link.image else self.extract_favicon(link.url, index_text)

            result.append({
                'image_url': image,
                **vars(link)
            })

        await session.close()
        return result

    async def process_services(self, services):
        result = []

        session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=5, sock_connect=1))

        for service in services:
            index_status, index_text = None, None
            if not service.port or not service.image:
                try:
                    async with session.get(service.url) as response:
                        index_status, index_text = response.status, await response.text()
                except asyncio.TimeoutError:
                    pass

            if service.port:
                a_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                location = ("localhost", service.port)
                result_of_check = a_socket.connect_ex(location)
                if result_of_check == 0:
                    status = ServiceStatus.OK
                else:
                    status = ServiceStatus.DOWN
            elif index_status:
                status = ServiceStatus.OK if index_status == 200 else ServiceStatus.DOWN
            else:
                status = ServiceStatus.UNKNOWN

            image = service.image.url if service.image else self.extract_favicon(service.url, index_text)

            result.append({
                'status': status.value,
                'image_url': image,
                **vars(service)
            })

        await session.close()
        return result

    @staticmethod
    def extract_favicon(url, index_text):
        if not index_text:
            return None

        parsed_html = BeautifulSoup(index_text, features="html.parser")
        link_tags = parsed_html.find_all('link')
        for rel in ['apple-touch-icon', 'shortcut', 'icon']:
            for link_tag in link_tags:
                if rel in link_tag.attrs['rel']:
                    href = link_tag.attrs['href']
                    if url not in href:
                        image = url + (href if href.startswith("/") else f"/{href}")
                    else:
                        image = href
                    return image

    # noinspection PyListCreation
    @staticmethod
    def vps_stats():
        stats = []

        stats.append(f"<em>LOAD AVG:</em> {', '.join(map(str, psutil.getloadavg()))}")

        memory = psutil.virtual_memory()
        stats.append(
            f"<em>MEM:</em> {naturalsize(memory.used)}/{naturalsize(memory.total)} ({memory.percent}% USED)"
        )

        disk = psutil.disk_usage('/')
        stats.append(
            f"<em>DISK:</em> {naturalsize(disk.used)}/{naturalsize(disk.total)} ({disk.percent}% USED)"
        )

        uptime = normalize_seconds(time.time() - psutil.boot_time())

        stats.append(
            f"<em>UPTIME:</em> {int(uptime.days)} days, {int(uptime.hours)} hours, {int(uptime.minutes)} minutes"
        )

        return " / ".join(map(lambda stat: stat.replace(" ", "&nbsp;"), stats))


def normalize_seconds(seconds: int):
    (days, remainder) = divmod(seconds, 86400)
    (hours, remainder) = divmod(remainder, 3600)
    (minutes, seconds) = divmod(remainder, 60)

    return namedtuple("_", ("days", "hours", "minutes", "seconds"))(days, hours, minutes, seconds)