docker-infrastructure/uptime-kuma/manage_monitors.py

#!/usr/bin/env python3
"""
Uptime Kuma Monitor Management Script
Usage examples:
  ./manage_monitors.py list
  ./manage_monitors.py add --name "My Service" --url "https://example.com" --type http
  ./manage_monitors.py delete --id 123
  ./manage_monitors.py update --id 123 --name "Updated Name"
"""

import argparse
import sys
import os
from uptime_kuma_api import UptimeKumaApi, MonitorType

# Configuration
UPTIME_KUMA_URL = "https://uptime.kolpacksoftware.com"
# Set one of these (or use environment variables):
API_KEY = os.getenv('UPTIME_KUMA_API_KEY')  # Set via UPTIME_KUMA_API_KEY env var
USERNAME = os.getenv('UPTIME_KUMA_USERNAME')  # Or use UPTIME_KUMA_USERNAME
PASSWORD = os.getenv('UPTIME_KUMA_PASSWORD')  # And UPTIME_KUMA_PASSWORD

def connect():
    """Connect to Uptime Kuma API"""
    # Disable SSL verification for self-signed certificates
    api = UptimeKumaApi(UPTIME_KUMA_URL, ssl_verify=False)

    if USERNAME and PASSWORD:
        api.login(USERNAME, PASSWORD)
    elif API_KEY:
        api.login_by_token(API_KEY)
    else:
        print("Error: No credentials provided. Set UPTIME_KUMA_API_KEY or UPTIME_KUMA_USERNAME/PASSWORD environment variables.")
        sys.exit(1)

    return api

def list_monitors(api):
    """List all monitors"""
    monitors = api.get_monitors()
    if not monitors:
        print("No monitors found")
        return

    print(f"{'ID':<6} {'Name':<30} {'Type':<10} {'URL/Hostname':<50}")
    print("-" * 96)
    for monitor in monitors:
        url = monitor.get('url') or monitor.get('hostname', 'N/A')
        if url is None:
            url = 'N/A'
        print(f"{monitor['id']:<6} {monitor['name']:<30} {str(monitor['type']):<10} {url:<50}")

def add_monitor(api, name, url, monitor_type='http', interval=60, ignore_tls=False):
    """Add a new monitor"""
    type_map = {
        'http': MonitorType.HTTP,
        'https': MonitorType.HTTP,
        'tcp': MonitorType.PORT,
        'ping': MonitorType.PING,
        'dns': MonitorType.DNS,
        'docker': MonitorType.DOCKER,
    }

    monitor_type_value = type_map.get(monitor_type.lower(), MonitorType.HTTP)

    # Different monitor types use different parameters
    if monitor_type.lower() == 'ping':
        result = api.add_monitor(
            type=monitor_type_value,
            name=name,
            hostname=url,
            interval=interval
        )
    elif monitor_type.lower() == 'tcp':
        hostname, port = url.rsplit(':', 1)
        result = api.add_monitor(
            type=monitor_type_value,
            name=name,
            hostname=hostname,
            port=int(port),
            interval=interval
        )
    else:
        kwargs = {
            'type': monitor_type_value,
            'name': name,
            'url': url,
            'interval': interval
        }
        # Add SSL ignore for HTTPS URLs
        if ignore_tls or (url.startswith('https://') and monitor_type.lower() in ['http', 'https']):
            kwargs['ignoreTls'] = True
        result = api.add_monitor(**kwargs)
    print(f"Monitor added successfully: ID {result['monitorID']}")

def delete_monitor(api, monitor_id):
    """Delete a monitor"""
    api.delete_monitor(monitor_id)
    print(f"Monitor {monitor_id} deleted successfully")

def update_monitor(api, monitor_id, **kwargs):
    """Update a monitor"""
    # Get existing monitor
    monitor = api.get_monitor(monitor_id)

    # Update fields
    for key, value in kwargs.items():
        if value is not None:
            monitor[key] = value

    api.edit_monitor(monitor_id, **monitor)
    print(f"Monitor {monitor_id} updated successfully")

def main():
    parser = argparse.ArgumentParser(description='Manage Uptime Kuma monitors')
    subparsers = parser.add_subparsers(dest='command', help='Commands')

    # List command
    subparsers.add_parser('list', help='List all monitors')

    # Add command
    add_parser = subparsers.add_parser('add', help='Add a monitor')
    add_parser.add_argument('--name', required=True, help='Monitor name')
    add_parser.add_argument('--url', required=True, help='URL to monitor')
    add_parser.add_argument('--type', default='http', help='Monitor type (http, tcp, ping, dns, docker)')
    add_parser.add_argument('--interval', type=int, default=60, help='Check interval in seconds')
    add_parser.add_argument('--ignore-tls', action='store_true', help='Ignore TLS/SSL certificate errors')

    # Delete command
    delete_parser = subparsers.add_parser('delete', help='Delete a monitor')
    delete_parser.add_argument('--id', type=int, required=True, help='Monitor ID')

    # Update command
    update_parser = subparsers.add_parser('update', help='Update a monitor')
    update_parser.add_argument('--id', type=int, required=True, help='Monitor ID')
    update_parser.add_argument('--name', help='New name')
    update_parser.add_argument('--url', help='New URL')
    update_parser.add_argument('--interval', type=int, help='New interval')

    args = parser.parse_args()

    if not args.command:
        parser.print_help()
        sys.exit(1)

    # Connect to API
    api = connect()

    try:
        if args.command == 'list':
            list_monitors(api)
        elif args.command == 'add':
            ignore_tls = getattr(args, 'ignore_tls', False)
            add_monitor(api, args.name, args.url, args.type, args.interval, ignore_tls)
        elif args.command == 'delete':
            delete_monitor(api, args.id)
        elif args.command == 'update':
            update_monitor(api, args.id, name=args.name, url=args.url, interval=args.interval)
    finally:
        api.disconnect()

if __name__ == '__main__':
    main()