diff --git a/files/backups/backup-all.py b/files/backups/backup-all.py
index b585778..42235ab 100644
--- a/files/backups/backup-all.py
+++ b/files/backups/backup-all.py
@@ -39,18 +39,10 @@ logging.basicConfig(
logger = logging.getLogger(__name__)
-@dataclass
-class TelegramConfig:
- type: str
- telegram_bot_token: str
- telegram_chat_id: str
- notifications_name: str
-
-
@dataclass
class Config:
+ host_name: str
roots: List[Path]
- notifications: Dict[str, TelegramConfig]
@dataclass
@@ -160,19 +152,61 @@ class ResticStorage(Storage):
return True
+class Notifier(ABC):
+ def send(self, html_message: str):
+ raise NotImplementedError()
+
+
+class TelegramNotifier(Notifier):
+ TYPE_NAME = "telegram"
+
+ 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,
+ ]
+ ):
+ raise ValueError(
+ f"Missing notification configuration values for backend {name}"
+ )
+
+ def send(self, html_message: str):
+ 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,
+ }
+
+ response = requests.post(url, data=data, timeout=30)
+
+ if response.status_code == 200:
+ logger.info("Telegram notification sent successfully")
+ else:
+ logger.error(
+ f"Failed to send Telegram notification: {response.status_code} - {response.text}"
+ )
+
+
class BackupManager:
- def __init__(self, config: Config, roots: List[Path], storages: List[Storage]):
+ def __init__(
+ self,
+ config: Config,
+ roots: List[Path],
+ storages: List[Storage],
+ notifiers: List[Notifier],
+ ):
self.errors: List[str] = []
self.warnings: List[str] = []
self.successful_backups: List[str] = []
self.config = config
self.roots: List[Path] = roots
self.storages = storages
-
- def _select_telegram(self) -> Optional[TelegramConfig]:
- if "telegram" in self.config.notifications:
- return self.config.notifications["telegram"]
- return next(iter(self.config.notifications.values()), None)
+ self.notifiers = notifiers
def find_applications(self) -> List[Application]:
"""Get all application directories and their owners."""
@@ -308,54 +342,32 @@ class BackupManager:
return backup_dirs
- def send_telegram_notification(self, success: bool) -> None:
- """Send notification to Telegram"""
- telegram_cfg = self._select_telegram()
- if telegram_cfg is None:
- logger.warning("No telegram notification backend configured")
- return
+ def send_notification(self, success: bool) -> None:
+ """Send notification to Notifiers"""
- try:
- if success and not self.errors:
- message = (
- f"{telegram_cfg.notifications_name}: бекап успешно завершен!"
- )
- if self.successful_backups:
- message += (
- f"\n\nУспешные бекапы: {', '.join(self.successful_backups)}"
- )
- else:
- message = f"{telegram_cfg.notifications_name}: бекап завершен с ошибками!"
+ if success and not self.errors:
+ message = f"{self.config.host_name}: бекап успешно завершен!"
+ if self.successful_backups:
+ message += f"\n\nУспешные бекапы: {', '.join(self.successful_backups)}"
+ else:
+ message = f"{self.config.host_name}: бекап завершен с ошибками!"
- if self.successful_backups:
- message += (
- f"\n\n✅ Успешные бекапы: {', '.join(self.successful_backups)}"
- )
-
- if self.warnings:
- message += f"\n\n⚠️ Предупреждения:\n" + "\n".join(self.warnings)
-
- if self.errors:
- message += f"\n\n❌ Ошибки:\n" + "\n".join(self.errors)
-
- url = f"https://api.telegram.org/bot{telegram_cfg.telegram_bot_token}/sendMessage"
- data = {
- "chat_id": telegram_cfg.telegram_chat_id,
- "parse_mode": "HTML",
- "text": message,
- }
-
- response = requests.post(url, data=data, timeout=30)
-
- if response.status_code == 200:
- logger.info("Telegram notification sent successfully")
- else:
- logger.error(
- f"Failed to send Telegram notification: {response.status_code} - {response.text}"
+ if self.successful_backups:
+ message += (
+ f"\n\n✅ Успешные бекапы: {', '.join(self.successful_backups)}"
)
- except Exception as e:
- logger.error(f"Failed to send Telegram notification: {str(e)}")
+ if self.warnings:
+ message += f"\n\n⚠️ Предупреждения:\n" + "\n".join(self.warnings)
+
+ if self.errors:
+ message += f"\n\n❌ Ошибки:\n" + "\n".join(self.errors)
+
+ for notificator in self.notifiers:
+ try:
+ notificator.send(message)
+ except Exception as e:
+ logger.error(f"Failed to send notification: {str(e)}")
def run_backup_process(self) -> bool:
"""Main backup process"""
@@ -399,7 +411,7 @@ class BackupManager:
overall_success = overall_success and backup_result
# Send notification
- self.send_telegram_notification(overall_success)
+ self.send_notification(overall_success)
logger.info("Backup process completed")
@@ -422,6 +434,8 @@ def initialize(config_path: Path) -> BackupManager:
logger.error(f"Failed to read config file {config_path}: {e}")
raise
+ host_name = str(raw_config.get("host_name", "unknown"))
+
roots_raw = raw_config.get("roots") or []
if not isinstance(roots_raw, list) or not roots_raw:
raise ValueError("roots must be a non-empty list of paths in config.toml")
@@ -438,37 +452,22 @@ def initialize(config_path: Path) -> BackupManager:
if not storages:
raise ValueError("At least one storage backend must be configured")
- notifications_raw = raw_config.get("notifications") or {}
- notifications: Dict[str, TelegramConfig] = {}
- for name, cfg in notifications_raw.items():
- if not isinstance(cfg, dict):
- raise ValueError(f"Notification config for {name} must be a table")
- notifications[name] = TelegramConfig(
- type=cfg.get("type", ""),
- telegram_bot_token=cfg.get("telegram_bot_token", ""),
- telegram_chat_id=cfg.get("telegram_chat_id", ""),
- notifications_name=cfg.get("notifications_name", ""),
- )
-
- if not notifications:
+ notifications_raw = raw_config.get("notifier") or {}
+ notifiers: List[Notifier] = []
+ for name, params in notifications_raw.items():
+ 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 not notifiers:
raise ValueError("At least one notification backend must be configured")
- for name, cfg in notifications.items():
- if not all(
- [
- cfg.type,
- cfg.telegram_bot_token,
- cfg.telegram_chat_id,
- cfg.notifications_name,
- ]
- ):
- raise ValueError(
- f"Missing notification configuration values for backend {name}"
- )
+ config = Config(host_name=host_name, roots=roots)
- config = Config(roots=roots, notifications=notifications)
-
- return BackupManager(config=config, roots=roots, storages=storages)
+ return BackupManager(
+ config=config, roots=roots, storages=storages, notifiers=notifiers
+ )
def main():
diff --git a/files/backups/config.template.toml b/files/backups/config.template.toml
index b8ba4d5..c4e4ead 100644
--- a/files/backups/config.template.toml
+++ b/files/backups/config.template.toml
@@ -1,8 +1,10 @@
+host_name = "{{ notifications_name }}"
+
roots = [
"{{ application_dir }}"
]
-[storage.yandex]
+[storage.yandex_cloud_s3]
type = "restic"
restic_repository = "{{ restic_repository }}"
restic_password = "{{ restic_password }}"
@@ -10,8 +12,7 @@ aws_access_key_id = "{{ restic_s3_access_key }}"
aws_secret_access_key = "{{ restic_s3_access_secret }}"
aws_default_region = "{{ restic_s3_region }}"
-[notifications.telegram]
+[notifier.server_notifications_channel]
type = "telegram"
telegram_bot_token = "{{ notifications_tg_bot_token }}"
telegram_chat_id = "{{ notifications_tg_chat_id }}"
-notifications_name = "{{ notifications_name }}"