4da7f899c6
All checks were successful
Build and Push Container / build (push) Successful in 31s
Uptime Kuma v2 requires a non-null conditions field when creating monitors. Adding empty array to satisfy the constraint. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
287 lines
9.1 KiB
Python
287 lines
9.1 KiB
Python
import os
|
|
import json
|
|
from typing import Optional
|
|
from dataclasses import dataclass
|
|
from uptime_kuma_api import UptimeKumaApi, MonitorType
|
|
|
|
from config import get_config
|
|
|
|
|
|
# Map our monitor types to Uptime Kuma types
|
|
TYPE_MAP = {
|
|
"http": MonitorType.HTTP,
|
|
"tcp": MonitorType.PORT,
|
|
"ping": MonitorType.PING,
|
|
"docker": MonitorType.DOCKER,
|
|
"keyword": MonitorType.KEYWORD,
|
|
"push": MonitorType.PUSH,
|
|
}
|
|
|
|
# Token storage path
|
|
TOKEN_FILE = os.environ.get("KUMA_TOKEN_FILE", "/tmp/kuma_token.json")
|
|
|
|
|
|
@dataclass
|
|
class Monitor:
|
|
"""Uptime Kuma monitor configuration."""
|
|
|
|
type: str # http, tcp, ping, docker, keyword, push
|
|
name: str
|
|
url: Optional[str] = None # For HTTP monitors
|
|
hostname: Optional[str] = None # For TCP/Ping monitors
|
|
port: Optional[int] = None # For TCP monitors
|
|
interval: int = 60
|
|
keyword: Optional[str] = None # For keyword monitors
|
|
docker_container: Optional[str] = None # For Docker monitors
|
|
docker_host: Optional[str] = None # For Docker monitors
|
|
push_token: Optional[str] = None # For Push monitors
|
|
retries: int = 3
|
|
retry_interval: int = 60
|
|
max_redirects: int = 10
|
|
accepted_statuscodes: list[str] = None
|
|
|
|
def __post_init__(self):
|
|
if self.accepted_statuscodes is None:
|
|
self.accepted_statuscodes = ["200-299"]
|
|
|
|
|
|
def load_saved_token() -> Optional[str]:
|
|
"""Load saved token from file."""
|
|
try:
|
|
if os.path.exists(TOKEN_FILE):
|
|
with open(TOKEN_FILE, "r") as f:
|
|
data = json.load(f)
|
|
return data.get("token")
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def save_token(token: str) -> None:
|
|
"""Save token to file."""
|
|
try:
|
|
with open(TOKEN_FILE, "w") as f:
|
|
json.dump({"token": token}, f)
|
|
except Exception as e:
|
|
print(f"Warning: Could not save token: {e}")
|
|
|
|
|
|
def clear_saved_token() -> None:
|
|
"""Clear saved token."""
|
|
try:
|
|
if os.path.exists(TOKEN_FILE):
|
|
os.remove(TOKEN_FILE)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
class UptimeKumaClient:
|
|
"""Client for Uptime Kuma using Socket.io API."""
|
|
|
|
def __init__(self):
|
|
config = get_config()
|
|
self.base_url = config.uptime_kuma_url.rstrip("/")
|
|
self.api_key = config.uptime_kuma_api_key # Can be empty if using saved token
|
|
self._api = None
|
|
self._token = None
|
|
|
|
def _get_token(self) -> Optional[str]:
|
|
"""Get token from config or saved file."""
|
|
if self._token:
|
|
return self._token
|
|
if self.api_key:
|
|
return self.api_key
|
|
return load_saved_token()
|
|
|
|
def _get_api(self) -> UptimeKumaApi:
|
|
"""Get connected API instance."""
|
|
if self._api is None:
|
|
token = self._get_token()
|
|
if not token:
|
|
raise Exception("Not authenticated. Please login to Uptime Kuma first.")
|
|
|
|
self._api = UptimeKumaApi(self.base_url)
|
|
try:
|
|
self._api.login_by_token(token)
|
|
except Exception as e:
|
|
self._api = None
|
|
clear_saved_token()
|
|
raise Exception(f"Authentication failed. Please login again: {e}")
|
|
return self._api
|
|
|
|
def _disconnect(self):
|
|
"""Disconnect API."""
|
|
if self._api:
|
|
try:
|
|
self._api.disconnect()
|
|
except Exception:
|
|
pass
|
|
self._api = None
|
|
|
|
def login(self, username: str, password: str, totp: Optional[str] = None) -> dict:
|
|
"""Login with username/password and optional TOTP."""
|
|
self._disconnect()
|
|
try:
|
|
api = UptimeKumaApi(self.base_url)
|
|
if totp:
|
|
result = api.login(username, password, totp)
|
|
else:
|
|
result = api.login(username, password)
|
|
|
|
# Check if 2FA is required
|
|
if result.get("tokenRequired"):
|
|
api.disconnect()
|
|
raise Exception("2FA token required. Please enter your TOTP code.")
|
|
|
|
token = result.get("token")
|
|
if not token:
|
|
api.disconnect()
|
|
raise Exception("Login failed: No authentication token received")
|
|
|
|
save_token(token)
|
|
self._token = token
|
|
|
|
api.disconnect()
|
|
return {"success": True, "message": "Login successful"}
|
|
except Exception as e:
|
|
error_msg = str(e)
|
|
# Re-raise with clear message for 2FA requirement
|
|
if "tokenRequired" in error_msg or "2fa" in error_msg.lower():
|
|
raise Exception("2FA token required. Please enter your TOTP code.")
|
|
raise Exception(f"Login failed: {error_msg}")
|
|
|
|
def is_authenticated(self) -> bool:
|
|
"""Check if we have valid authentication."""
|
|
token = self._get_token()
|
|
if not token:
|
|
return False
|
|
try:
|
|
api = UptimeKumaApi(self.base_url)
|
|
api.login_by_token(token)
|
|
api.disconnect()
|
|
return True
|
|
except Exception:
|
|
clear_saved_token()
|
|
return False
|
|
|
|
def logout(self) -> None:
|
|
"""Clear saved authentication."""
|
|
self._disconnect()
|
|
self._token = None
|
|
clear_saved_token()
|
|
|
|
def get_monitors(self) -> list[dict]:
|
|
"""Get all monitors."""
|
|
try:
|
|
api = self._get_api()
|
|
return api.get_monitors()
|
|
except Exception as e:
|
|
self._disconnect()
|
|
raise Exception(f"Failed to get monitors: {str(e)}")
|
|
|
|
def get_monitor(self, monitor_id: int) -> dict:
|
|
"""Get a specific monitor."""
|
|
try:
|
|
api = self._get_api()
|
|
return api.get_monitor(monitor_id)
|
|
except Exception as e:
|
|
self._disconnect()
|
|
raise Exception(f"Failed to get monitor {monitor_id}: {str(e)}")
|
|
|
|
def create_monitor(self, monitor: Monitor) -> dict:
|
|
"""Create a new monitor."""
|
|
try:
|
|
api = self._get_api()
|
|
|
|
monitor_type = TYPE_MAP.get(monitor.type)
|
|
if not monitor_type:
|
|
raise ValueError(f"Unknown monitor type: {monitor.type}")
|
|
|
|
kwargs = {
|
|
"type": monitor_type,
|
|
"name": monitor.name,
|
|
"interval": monitor.interval,
|
|
"maxretries": monitor.retries,
|
|
"retryInterval": monitor.retry_interval,
|
|
"conditions": [], # Required for Uptime Kuma v2
|
|
}
|
|
|
|
# Add type-specific fields
|
|
if monitor.type == "http":
|
|
kwargs["url"] = monitor.url
|
|
kwargs["maxredirects"] = monitor.max_redirects
|
|
kwargs["accepted_statuscodes"] = monitor.accepted_statuscodes
|
|
elif monitor.type == "keyword":
|
|
kwargs["url"] = monitor.url
|
|
kwargs["keyword"] = monitor.keyword
|
|
elif monitor.type == "tcp":
|
|
kwargs["hostname"] = monitor.hostname
|
|
kwargs["port"] = monitor.port
|
|
elif monitor.type == "ping":
|
|
kwargs["hostname"] = monitor.hostname
|
|
elif monitor.type == "docker":
|
|
kwargs["docker_container"] = monitor.docker_container
|
|
if monitor.docker_host:
|
|
kwargs["docker_host"] = monitor.docker_host
|
|
elif monitor.type == "push":
|
|
# Push monitors are created and get a token back
|
|
pass
|
|
|
|
result = api.add_monitor(**kwargs)
|
|
return result
|
|
except Exception as e:
|
|
self._disconnect()
|
|
raise Exception(f"Failed to create monitor: {str(e)}")
|
|
|
|
def delete_monitor(self, monitor_id: int) -> dict:
|
|
"""Delete a monitor."""
|
|
try:
|
|
api = self._get_api()
|
|
result = api.delete_monitor(monitor_id)
|
|
return result
|
|
except Exception as e:
|
|
self._disconnect()
|
|
raise Exception(f"Failed to delete monitor {monitor_id}: {str(e)}")
|
|
|
|
def pause_monitor(self, monitor_id: int) -> dict:
|
|
"""Pause a monitor."""
|
|
try:
|
|
api = self._get_api()
|
|
result = api.pause_monitor(monitor_id)
|
|
return result
|
|
except Exception as e:
|
|
self._disconnect()
|
|
raise Exception(f"Failed to pause monitor {monitor_id}: {str(e)}")
|
|
|
|
def resume_monitor(self, monitor_id: int) -> dict:
|
|
"""Resume a paused monitor."""
|
|
try:
|
|
api = self._get_api()
|
|
result = api.resume_monitor(monitor_id)
|
|
return result
|
|
except Exception as e:
|
|
self._disconnect()
|
|
raise Exception(f"Failed to resume monitor {monitor_id}: {str(e)}")
|
|
|
|
def test_connection(self) -> bool:
|
|
"""Test connection to Uptime Kuma."""
|
|
try:
|
|
api = self._get_api()
|
|
api.get_monitors()
|
|
return True
|
|
except Exception:
|
|
self._disconnect()
|
|
return False
|
|
|
|
|
|
# Global client instance
|
|
_kuma_client: Optional[UptimeKumaClient] = None
|
|
|
|
|
|
def get_kuma_client() -> UptimeKumaClient:
|
|
"""Get the global Uptime Kuma client instance."""
|
|
global _kuma_client
|
|
if _kuma_client is None:
|
|
_kuma_client = UptimeKumaClient()
|
|
return _kuma_client
|