feat: uptime command: add app domains to uptime kuma

alakazam.py

@@ -19,6 +19,7 @@ from ruamel.yaml import YAML
 from ruamel.yaml.constructor import SafeConstructor
 from ruamel.yaml.nodes import ScalarNode
 from packaging import version
+from uptime_kuma_api import UptimeKumaApi, MonitorType
 
 COMBINE_PATH = os.path.dirname(os.path.realpath(__file__)) + "/combine.yml"
 # INSTANCE_CONFIGS: dict: contains all app organized by recipe names and instance domains
@@ -1362,5 +1363,76 @@ def diff(apps: List[str]) -> None:
             print(f'\t {file}')
 
 
+@cli.command()
+@click.option('-a', '--apps', multiple=True)
+@click.option('-u', '--update', is_flag=True)
+@click.option('missing', '-m', '--get-missing', is_flag=True)
+def uptime(apps: List[str], update: bool, missing: bool) -> None:
+    """
+    Add the app domains to an uptime kuma instance. 
+
+    Args:
+        apps (list): List of applications.
+        update (bool): Update the uptime kuma monitors
+        missing (bool): Only print missing uptime kuma monitors
+    """
+    if not (uptime_config:= (SETTINGS.get('uptime_kuma'))):
+        logging.error(f"No uptime kuma settings provided in {SETTINGS_PATH}")
+        exit(1)
+    if not uptime_config.get('url'):
+        logging.error(f"No uptime kuma 'url' provided in {SETTINGS_PATH}")
+        exit(1)
+    if not uptime_config.get('user'):
+        logging.error(f"No uptime kuma 'user' provided in {SETTINGS_PATH}")
+        exit(1)
+    if not uptime_config.get('password'):
+        logging.error(f"No uptime kuma 'password' provided in {SETTINGS_PATH}")
+        exit(1)
+    uptime_kuma_url = uptime_config.get('url')
+    api = UptimeKumaApi(uptime_kuma_url)
+    api.login(uptime_config.get('user'), uptime_config.get('password'))
+
+    monitor_parameter = uptime_config.get('parameter')
+    if not monitor_parameter:
+        monitor_parameter = {}
+
+    instance_apps = get_apps(apps)
+    if not missing:
+        print_all_apps(instance_apps)
+    else:
+        print(f"The following domains are missing in {uptime_kuma_url}")
+    if not instance_apps or (not missing and input(f"Do you really want to add these apps to {uptime_kuma_url}? Type YES: ") != "YES"):
+        return 
+    domains = list(zip(*sum(instance_apps.values(), [])))[1]
+
+    monitors = api.get_monitors()
+    added_monitors = {m['name']:m['id'] for m in monitors if m['type'] == MonitorType.HTTP}
+
+    excluded_subdomains = ['traefik', 'monitoring', 'backup']
+    for domain in domains:
+        if any(domain.startswith(s) for s in excluded_subdomains):
+            logging.debug(f"Skip {domain}")
+            continue
+        if 'matrix' in domain:
+            url = f"https://{domain}/.well-known/matrix/server"
+        elif 'collab' in domain:
+            url = f"https://{domain}/hosting/discovery"
+        else:
+            url = f"https://{domain}"
+        if (monitor:= added_monitors.get(domain)):
+            if not missing and not update:
+                print(f"{domain} already exists.")
+            if not not missing or not update:
+                continue
+            response = api.edit_monitor(monitor, **monitor_parameter)
+        elif missing:
+            print(domain)
+            continue
+        else:
+            response = api.add_monitor(type=MonitorType.HTTP, name=domain, url=url, **monitor_parameter)
+        print(f"{domain}: {response.get('msg')}")
+
+
+
 if __name__ == '__main__':
     cli()

requirements.txt

@@ -7,3 +7,4 @@ python-dotenv==1.0.0
 icecream==2.1.3
 packaging==24.0 
 GitPython==3.1.43
+uptime_kuma_api==1.2.1