stateless-uptime-kuma/stateless_uptime_kuma/tree_gen.py

"""
Classes to generate the item tree from json spec
"""

import logging
import sys

from .uptime_kuma import Monitor, Notification, Settings, StatusPage, Tag  # type: ignore

logger = logging.getLogger(__name__)


def die_tag_format_error():
    logger.error(
        "Fatal: You must provide tags in monitors in the format [{name:str, value:str}] (value is optional)"
    )
    sys.exit(1)


def from_dict(api, tree, autocreate_tags=True):
    notif = tree.get("notifications", [])
    indexed_notifications = {
        name: Notification(api, name, **kwargs) for name, kwargs in notif.items()
    }
    tags = tree.get("tags", [])
    indexed_tags = {name: Tag(api, name, **kwargs) for name, kwargs in tags.items()}
    monitors = tree.get("monitors", [])
    indexed_monitors = {}
    for monitor_name, monitor_kwargs in monitors.items():
        associated_tags = []
        for tag in monitor_kwargs.get("tags", []):
            if not isinstance(tag, dict) or "name" not in tag:
                die_tag_format_error()
            try:
                if autocreate_tags and tag["name"] not in indexed_tags:
                    indexed_tags[tag["name"]] = Tag(api, name=tag["name"])
                associated_tags.append((indexed_tags[tag["name"]], tag.get("value")))
            except IndexError:
                die_tag_format_error()
        monitor_kwargs["tags"] = associated_tags
        associated_notifications = [
            indexed_notifications[notif]
            for notif in monitor_kwargs.get("notifications", [])
        ]
        monitor_kwargs["notifications"] = associated_notifications
        indexed_monitors[monitor_name] = Monitor(api, monitor_name, **monitor_kwargs)
    status_pages = tree.get("status_pages", [])
    indexed_status_pages = {}
    for slug, kwargs in status_pages.items():
        for group in kwargs.get("publicGroupList", []):
            if "monitorList" in group:
                monitorList = []
                for monitor in group["monitorList"]:
                    if monitor not in indexed_monitors:
                        logger.error(
                            "Fatal: status page is referencing a monitor that doesn't exist"
                        )
                        sys.exit(1)
                    monitorList.append(indexed_monitors[monitor])
                group["monitorList"] = monitorList
        indexed_status_pages[slug] = StatusPage(api, slug, **kwargs)
    settings = []
    if "settings" in tree:
        parsed_settings = {}
        for k, v in tree["settings"].items():
            if k.endswith("__FILE"):
                with open(v, "rt") as f:
                    parsed_settings[k[:-len("__FILE")]] = f.read().strip()
            else:
                parsed_settings[k] = v
        settings = [Settings(api, **parsed_settings)]


    return {
        "monitors": indexed_monitors.values(),
        "tags": indexed_tags.values(),
        "notifications": indexed_notifications.values(),
        "status_pages": indexed_status_pages.values(),
        "settings": settings,
    }