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 validate_monitor_data(self, monitor_data): required_keys = ["type", "name", "url"] for key in required_keys: if key not in monitor_data: raise ValueError(f"Missing required monitor field: {key}") if not isinstance(monitor_data.get("notificationIDList", []), list): raise ValueError("notificationIDList must be a list") 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()