151 lines
4.8 KiB
Python
Executable File
151 lines
4.8 KiB
Python
Executable File
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Iterable
|
|
|
|
import requests
|
|
|
|
from .config import GotifyConfig, PushoverConfig
|
|
|
|
|
|
log = logging.getLogger("inspector.notify")
|
|
|
|
|
|
class GotifyNotifier:
|
|
def __init__(self, config: GotifyConfig, timeout: float = 5.0):
|
|
self._config = config
|
|
self._timeout = timeout
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return self._config.enabled and bool(self._config.url) and bool(self._config.token)
|
|
|
|
def notify(self, title: str, message: str, priority: int | None = None) -> None:
|
|
if not self.enabled:
|
|
return
|
|
url = f"{self._config.url}/message"
|
|
payload = {
|
|
"title": title,
|
|
"message": message,
|
|
"priority": self._config.priority if priority is None else priority,
|
|
"extras": {
|
|
"client::display": {"contentType": "text/markdown"},
|
|
},
|
|
}
|
|
try:
|
|
# Header, not ?token=, so the token stays out of access logs.
|
|
resp = requests.post(
|
|
url,
|
|
headers={"X-Gotify-Key": self._config.token},
|
|
json=payload,
|
|
timeout=self._timeout,
|
|
)
|
|
if resp.status_code >= 400:
|
|
log.error(
|
|
"Gotify notification failed: status=%s body=%s",
|
|
resp.status_code,
|
|
resp.text[:200],
|
|
)
|
|
except requests.RequestException as exc:
|
|
log.error("Gotify notification failed: %s", exc)
|
|
except Exception as exc:
|
|
log.exception("Gotify notifier crashed: %s", exc)
|
|
|
|
|
|
# Pushover's priority range is -2..2. Inspector uses a Gotify-style scale
|
|
# (roughly 1..10). Mapping chosen 2026-10-26 (see plan file).
|
|
# priority 2 ("emergency") requires retry+expire; defaults below.
|
|
_PUSHOVER_EMERGENCY_RETRY_SECONDS = 60
|
|
_PUSHOVER_EMERGENCY_EXPIRE_SECONDS = 3600
|
|
|
|
|
|
def _map_priority_to_pushover(inspector_priority: int) -> int:
|
|
if inspector_priority <= 2:
|
|
return -1
|
|
if inspector_priority <= 4:
|
|
return 0
|
|
if inspector_priority <= 6:
|
|
return 1
|
|
return 2
|
|
|
|
|
|
class PushoverNotifier:
|
|
API_URL = "https://api.pushover.net/1/messages.json"
|
|
|
|
def __init__(self, config: PushoverConfig, timeout: float = 5.0):
|
|
self._config = config
|
|
self._timeout = timeout
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return (
|
|
self._config.enabled
|
|
and bool(self._config.token)
|
|
and bool(self._config.user_key)
|
|
)
|
|
|
|
def notify(self, title: str, message: str, priority: int | None = None) -> None:
|
|
if not self.enabled:
|
|
return
|
|
|
|
inspector_priority = self._config.priority if priority is None else priority
|
|
pushover_priority = _map_priority_to_pushover(inspector_priority)
|
|
|
|
data = {
|
|
"token": self._config.token,
|
|
"user": self._config.user_key,
|
|
"title": title,
|
|
"message": message,
|
|
"priority": pushover_priority,
|
|
}
|
|
if self._config.device:
|
|
data["device"] = self._config.device
|
|
if self._config.sound:
|
|
data["sound"] = self._config.sound
|
|
if pushover_priority == 2:
|
|
data["retry"] = _PUSHOVER_EMERGENCY_RETRY_SECONDS
|
|
data["expire"] = _PUSHOVER_EMERGENCY_EXPIRE_SECONDS
|
|
|
|
try:
|
|
resp = requests.post(self.API_URL, data=data, timeout=self._timeout)
|
|
if resp.status_code >= 400:
|
|
log.error(
|
|
"Pushover notification failed: status=%s body=%s",
|
|
resp.status_code,
|
|
resp.text[:200],
|
|
)
|
|
except requests.RequestException as exc:
|
|
log.error("Pushover notification failed: %s", exc)
|
|
except Exception as exc:
|
|
log.exception("Pushover notifier crashed: %s", exc)
|
|
|
|
|
|
class MultiNotifier:
|
|
"""Fan out notify() to every enabled child notifier.
|
|
|
|
Enabled is true if any child is enabled. A failure in one child never
|
|
prevents the others from running.
|
|
"""
|
|
|
|
def __init__(self, notifiers: Iterable[object]):
|
|
self._notifiers = list(notifiers)
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return any(getattr(n, "enabled", False) for n in self._notifiers)
|
|
|
|
def notify(self, title: str, message: str, priority: int | None = None) -> None:
|
|
for n in self._notifiers:
|
|
try:
|
|
n.notify(title=title, message=message, priority=priority)
|
|
except Exception as exc:
|
|
log.exception("notifier %r crashed: %s", type(n).__name__, exc)
|
|
|
|
|
|
class NullNotifier:
|
|
"""No-op notifier: used to silence a muted container's alerts for one
|
|
_check_one call. Mirrors the notify() signature of the real notifiers."""
|
|
|
|
def notify(self, title: str, message: str, priority: int | None = None) -> None:
|
|
pass
|