docker-mods-uptime-kuma-timeout-fix/root/app/auto_uptime_kuma/uptime_kuma_service.py

import requests
from uptime_kuma_api.api import UptimeKumaApi, MonitorType
from auto_uptime_kuma.log import Log
from auto_uptime_kuma.config_service import ConfigService


class UptimeKumaService:

    api: UptimeKumaApi
    swag_tag_name = "swag"
    swag_tag_color = "#ff4f97"
    swag_tag = None
    monitors: list
    groups: list
    default_notifications: list

    config_service: ConfigService

    default_monitor_data = {
        "type": MonitorType.HTTP,
        "description": "Automatically generated by SWAG auto-uptime-kuma",
    }

    def __init__(self, config_service):
        self.config_service = config_service

    def connect(self, url, username, password):
        response = requests.get(url, allow_redirects=True, timeout=5)
        if response.status_code != 200:
            Log.info(
                f"Unable to connect to UptimeKuma at '{url}' (Status code: {response.status_code})."
                " Please check if the host is running."
            )
            return False

        self.api = UptimeKumaApi(url)
        self.api.login(username, password)

        return True

    def disconnect(self):
        """
        API has to be disconnected at the end as the connection is blocking
        """
        self.api.disconnect()

    def load_data(self):
        monitors = self.api.get_monitors()
        self.get_swag_tag()

        self.default_notifications = self.get_default_notifications()

        self.monitors = [
            monitor
            for monitor in monitors
            if monitor["type"] != MonitorType.GROUP
            and self.get_monitor_swag_tag_value(monitor) is not None
        ]
        self.groups = [
            group
            for group in monitors
            if group["type"] == MonitorType.GROUP
            and self.get_monitor_swag_tag_value(group) is not None
        ]

    def build_monitor_data(self, container_name, configured_monitor_data):
        """
        Some of the container label values might have to be converted before sending to API.
        Additionally merge default config with label config.
        """
        result_data = dict(self.default_monitor_data)
        result_data.update(
            {
                "name": container_name.title(),
                "url": f"https://{container_name}.{self.config_service.domain_name}",
            }
        )
        result_data.update(configured_monitor_data)

        # Convert strings that are lists in API
        for key in ["accepted_statuscodes", "notificationIDList"]:
            if key in result_data and isinstance(result_data[key], str):
                result_data[key] = result_data[key].split(",")

        # If Monitor Groups are used then create them if needed and switch into ID as value
        if "parent" in result_data and not str(result_data["parent"]).isdigit():
            if self.group_exists(result_data["parent"]):
                group_data = self.get_group(result_data["parent"])
            else:
                group_data = self.create_group(result_data["parent"])
            result_data["parent"] = group_data["id"]

        if self.default_notifications:
            default_notification_ids = [
                notification["id"] for notification in self.default_notifications
            ]
            if "notificationIDList" in result_data:
                result_data["notificationIDList"] += default_notification_ids
                result_data["notificationIDList"] = list(
                    set(result_data["notificationIDList"])
                )
            else:
                result_data["notificationIDList"] = default_notification_ids

        # All numeric values sent to API have to be of type int
        for key, value in result_data.items():
            if str(value).isdigit():
                result_data[key] = int(value)

        return result_data

    def get_monitor(self, container_name):
        for monitor in self.monitors:
            swag_tag = self.get_monitor_swag_tag_value(monitor)
            if swag_tag is not None and swag_tag == container_name.lower():
                return monitor
        return None

    def monitor_exists(self, container_name):
        return self.get_monitor(container_name) is not None

    def create_monitor(self, container_name, monitor_data):
        monitor_data = self.build_monitor_data(container_name, monitor_data)
        self.validate_monitor_data(monitor_data)

        if self.monitor_exists(container_name):
            Log.info(
                f"Uptime Kuma already contains Monitor '{monitor_data['name']}'"
                f" for container '{container_name}', skipping..."
            )
            return None

        Log.info(f"Adding Monitor '{monitor_data['name']}' for container '{container_name}'")
        Log.debug(f"Sending monitor data: {monitor_data}")

        try:
            monitor = self.api.add_monitor(**monitor_data)
        except socketio.exceptions.TimeoutError:
            Log.error("Timeout while trying to add monitor to Uptime Kuma. Is the server responsive?")
            return None
        except Exception as e:
            Log.error(f"Failed to create monitor due to unexpected error: {e}")
            return None

        self.api.add_monitor_tag(
            tag_id=self.get_swag_tag()["id"],
            monitor_id=monitor["monitorID"],
            value=container_name.lower(),
        )

        self.config_service.create_config(container_name, monitor_data)
        monitor = self.api.get_monitor(monitor["monitorID"])
        self.monitors.append(monitor)

        return monitor

    def edit_monitor(self, container_name, monitor_data):
        """
        Please note that due to API limitations the "update" action
        is actually "delete" followed by "add"
        so that in the end the monitors are actually recreated
        """
        new_monitor_data = self.build_monitor_data(container_name, monitor_data)
        existing_monitor_data = self.get_monitor(container_name)
        old_content = self.config_service.read_config_content(container_name)
        new_content = self.config_service.build_config_content(new_monitor_data)

        if not old_content == new_content:
            Log.info(
                "Updating (Delete and Add) Monitor"
                f" {existing_monitor_data['id']}:{existing_monitor_data['name']}"
            )
            self.delete_monitor(container_name)
            self.create_monitor(container_name, new_monitor_data)
        else:
            Log.info(
                f"Monitor {existing_monitor_data['id']}:{existing_monitor_data['name']}"
                " has no changes, skipping..."
            )

    def delete_monitor(self, container_name: str):
        monitor_data = self.get_monitor(container_name)
        if monitor_data is not None:
            Log.info(f"Deleting Monitor {monitor_data['id']}:{monitor_data['name']}")
            self.api.delete_monitor(monitor_data["id"])

            for i, monitor in enumerate(self.monitors):
                if monitor["id"] == monitor_data["id"]:
                    del self.monitors[i]
                    break

    def delete_monitors(self, container_names: list[str]):
        if container_names:
            Log.info(
                f"Deleting Monitors for the following containers: {container_names}"
            )
            for container_name in container_names:
                self.delete_monitor(container_name)
        else:
            Log.info("No Monitors to delete")

    def get_group(self, group_name):
        for group in self.groups:
            swag_tag = self.get_monitor_swag_tag_value(group)
            if swag_tag is not None and swag_tag == group_name.lower():
                return group
        return None

    def group_exists(self, container_name):
        return self.get_group(container_name) is not None

    def create_group(self, group_name):
        monitor_data = {
            "type": MonitorType.GROUP,
            "name": group_name,
            "description": "Automatically generated by SWAG auto-uptime-kuma",
        }
        if self.group_exists(group_name):
            Log.info(
                f"Uptime Kuma already contains Group '{monitor_data['name']}', skipping..."
            )
            return

        Log.info(f"Adding Group '{monitor_data['name']}'")

        group = self.api.add_monitor(**monitor_data)

        self.api.add_monitor_tag(
            tag_id=self.get_swag_tag()["id"],
            monitor_id=group["monitorID"],
            value=group_name.lower(),
        )

        group = self.api.get_monitor(group["monitorID"])
        self.groups.append(group)

        return group

    ## This intentionally does not exist. Groups management by this mod is made
    ## simple so that they can be only added or deleted.
    # def edit_group(self, group_name):

    def delete_group(self, group_name):
        group_data = self.get_group(group_name)
        if group_data is not None:
            Log.info(f"Deleting Group {group_data['id']}:{group_data['name']}")
            self.api.delete_monitor(group_data["id"])

            for i, group in enumerate(self.groups):
                if group["id"] == group_data["id"]:
                    del self.groups[i]
                    break

    def delete_groups(self, group_names: list[str]):
        if group_names:
            Log.info(f"Deleting Groups with following names: {group_names}")
            for group_name in group_names:
                self.delete_group(group_name)
        else:
            Log.info("No Groups to delete")

    def get_swag_tag(self):
        """
        The "swag" tag is used to detect in API which monitors were created using this script.
        """
        # If the tag was not fetched yet
        if self.swag_tag is None:
            for tag in self.api.get_tags():
                if tag["name"] == self.swag_tag_name:
                    self.swag_tag = tag
                    break

        # If the tag was not in API then it has to be created
        if self.swag_tag is None:
            self.swag_tag = self.create_swag_tag()

        return self.swag_tag

    def swag_tag_exists(self):
        return self.get_swag_tag() is not None

    def create_swag_tag(self):
        self.swag_tag = self.api.add_tag(
            name=self.swag_tag_name, color=self.swag_tag_color
        )
        return self.swag_tag

    def delete_swag_tag(self):
        swag_tag = self.get_swag_tag()
        if swag_tag is not None:
            self.api.delete_tag(self.swag_tag["id"])
        swag_tag = None

    def get_monitor_swag_tag_value(self, monitor_data):
        """
        This value is container name itself. Used to link containers with monitors.
        """
        for tag in monitor_data.get("tags"):
            if "name" in tag and tag["name"] == self.swag_tag_name:
                return tag["value"]
        return None

    def get_default_notifications(self):
        default_notifications = [
            notification
            for notification in self.api.get_notifications()
            if notification["isDefault"] is True
        ]

        return default_notifications

    def purge_data(self):
        """
        Deletes all of the monitors created with this script
        """
        Log.info("Purging all Uptime Kuma Monitors added by this mod")

        monitor_names = [
            self.get_monitor_swag_tag_value(monitor) for monitor in self.monitors
        ]
        self.delete_monitors(monitor_names)
        Log.info("Monitor purging finished")
        group_names = [self.get_monitor_swag_tag_value(group) for group in self.groups]
        self.delete_groups(group_names)
        Log.info("Group purging finished")
        Log.info("Deleting the Swag tag")
        self.delete_swag_tag()