backups: use apprise for notifications
This commit is contained in:
+15
-21
@@ -158,38 +158,32 @@ class Notifier(ABC):
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
class TelegramNotifier(Notifier):
|
||||
TYPE_NAME = "telegram"
|
||||
class AppriseNotifier(Notifier):
|
||||
TYPE_NAME = "apprise"
|
||||
|
||||
def __init__(self, name: str, params: Dict[str, Any]):
|
||||
self.name = name
|
||||
self.telegram_bot_token = str(params.get("telegram_bot_token", ""))
|
||||
self.telegram_chat_id = str(params.get("telegram_chat_id", ""))
|
||||
if not all(
|
||||
[
|
||||
self.telegram_bot_token,
|
||||
self.telegram_chat_id,
|
||||
]
|
||||
):
|
||||
self.api_url = str(params.get("api_url", "")).rstrip("/")
|
||||
self.tag = str(params.get("tag", ""))
|
||||
if not self.api_url or not self.tag:
|
||||
raise ValueError(
|
||||
f"Missing notification configuration values for backend {name}"
|
||||
)
|
||||
|
||||
def send(self, html_message: str) -> None:
|
||||
url = f"https://api.telegram.org/bot{self.telegram_bot_token}/sendMessage"
|
||||
data = {
|
||||
"chat_id": self.telegram_chat_id,
|
||||
"parse_mode": "HTML",
|
||||
"text": html_message,
|
||||
url = f"{self.api_url}/notify/{self.tag}/"
|
||||
payload = {
|
||||
"body": html_message,
|
||||
"format": "html",
|
||||
}
|
||||
|
||||
response = requests.post(url, data=data, timeout=30)
|
||||
response = requests.post(url, json=payload, timeout=30)
|
||||
|
||||
if response.status_code == 200:
|
||||
logger.info("Telegram notification sent successfully")
|
||||
if response.ok:
|
||||
logger.info("Apprise notification sent successfully")
|
||||
else:
|
||||
logger.error(
|
||||
f"Failed to send Telegram notification: {response.status_code} - {response.text}"
|
||||
f"Failed to send Apprise notification: {response.status_code} - {response.text}"
|
||||
)
|
||||
|
||||
|
||||
@@ -459,8 +453,8 @@ def initialize(config_path: Path) -> BackupManager:
|
||||
if not isinstance(params, dict):
|
||||
raise ValueError(f"Notificator config for {name} must be a table")
|
||||
notifier_type = params.get("type", "")
|
||||
if notifier_type == TelegramNotifier.TYPE_NAME:
|
||||
notifiers.append(TelegramNotifier(name, params))
|
||||
if notifier_type == AppriseNotifier.TYPE_NAME:
|
||||
notifiers.append(AppriseNotifier(name, params))
|
||||
if not notifiers:
|
||||
raise ValueError("At least one notification backend must be configured")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user