""" 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, }