dc831b430b
Move modules from fluxer_admin_alerts/ to the repository root, switch to absolute imports, fix PROJECT_ROOT resolution, and update README commands and notifier paths.
42 lines
1.5 KiB
Python
42 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
from .base import Message, Notifier, NotifierError
|
|
|
|
EMBEDS_PER_MESSAGE = 10
|
|
COLOR_URGENT = 0xE74C3C
|
|
COLOR_NORMAL = 0x3498DB
|
|
USERNAME = "Admin Alerts"
|
|
|
|
|
|
class FluxerWebhookNotifier(Notifier):
|
|
def __init__(self, webhook_url: str, timeout: float) -> None:
|
|
self._url = webhook_url
|
|
self._timeout = timeout
|
|
|
|
def send(self, messages: Sequence[Message]) -> None:
|
|
for start in range(0, len(messages), EMBEDS_PER_MESSAGE):
|
|
batch = messages[start : start + EMBEDS_PER_MESSAGE]
|
|
payload = {"username": USERNAME, "embeds": [_embed(m) for m in batch]}
|
|
try:
|
|
resp = requests.post(self._url, json=payload, timeout=self._timeout)
|
|
except requests.RequestException as exc:
|
|
# The exception text contains the webhook URL, which holds the token, so only the type is kept.
|
|
raise NotifierError(f"Fluxer webhook request failed ({type(exc).__name__})") from None
|
|
if not resp.ok:
|
|
raise NotifierError(f"Fluxer webhook returned {resp.status_code}")
|
|
|
|
|
|
def _embed(message: Message) -> dict[str, Any]:
|
|
embed: dict[str, Any] = {
|
|
"title": message.title[:256],
|
|
"description": message.body[:4096],
|
|
"color": COLOR_URGENT if message.urgent else COLOR_NORMAL,
|
|
}
|
|
if message.link:
|
|
embed["url"] = message.link
|
|
return embed
|