Backup: remove old config

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
@@ -93,89 +85,133 @@ class ResticStorage(Storage):
         if not backup_dirs:
             logger.warning("No backup directories found")
             return True
-
         try:
-            logger.info("Starting restic backup")
-            logger.info("Destination: %s", self.restic_repository)
-
-            env = os.environ.copy()
-            env.update(
-                {
-                    "RESTIC_REPOSITORY": self.restic_repository,
-                    "RESTIC_PASSWORD": self.restic_password,
-                    "AWS_ACCESS_KEY_ID": self.aws_access_key_id,
-                    "AWS_SECRET_ACCESS_KEY": self.aws_secret_access_key,
-                    "AWS_DEFAULT_REGION": self.aws_default_region,
-                }
-            )
-
-            backup_cmd = ["restic", "backup", "--verbose"] + backup_dirs
-            result = subprocess.run(backup_cmd, env=env, capture_output=True, text=True)
-
-            if result.returncode != 0:
-                logger.error("Restic backup failed: %s", result.stderr)
-                return False
-
-            logger.info("Restic backup completed successfully")
-
-            check_cmd = ["restic", "check"]
-            result = subprocess.run(check_cmd, env=env, capture_output=True, text=True)
-
-            if result.returncode != 0:
-                logger.error("Restic check failed: %s", result.stderr)
-                return False
-
-            logger.info("Restic check completed successfully")
-
-            forget_cmd = [
-                "restic",
-                "forget",
-                "--compact",
-                "--prune",
-                "--keep-daily",
-                "90",
-                "--keep-monthly",
-                "36",
-            ]
-            result = subprocess.run(forget_cmd, env=env, capture_output=True, text=True)
-
-            if result.returncode != 0:
-                logger.error("Restic forget/prune failed: %s", result.stderr)
-                return False
-
-            logger.info("Restic forget/prune completed successfully")
-
-            result = subprocess.run(check_cmd, env=env, capture_output=True, text=True)
-
-            if result.returncode != 0:
-                logger.error("Final restic check failed: %s", result.stderr)
-                return False
-
-            logger.info("Final restic check completed successfully")
-            return True
-
+            return self.__backup_internal(backup_dirs)
         except Exception as exc:  # noqa: BLE001
             logger.error("Restic backup process failed: %s", exc)
             return False
 
+    def __backup_internal(self, backup_dirs: List[str]) -> bool:
+        logger.info("Starting restic backup")
+        logger.info("Destination: %s", self.restic_repository)
+
+        env = os.environ.copy()
+        env.update(
+            {
+                "RESTIC_REPOSITORY": self.restic_repository,
+                "RESTIC_PASSWORD": self.restic_password,
+                "AWS_ACCESS_KEY_ID": self.aws_access_key_id,
+                "AWS_SECRET_ACCESS_KEY": self.aws_secret_access_key,
+                "AWS_DEFAULT_REGION": self.aws_default_region,
+            }
+        )
+
+        backup_cmd = ["restic", "backup", "--verbose"] + backup_dirs
+        result = subprocess.run(backup_cmd, env=env, capture_output=True, text=True)
+
+        if result.returncode != 0:
+            logger.error("Restic backup failed: %s", result.stderr)
+            return False
+
+        logger.info("Restic backup completed successfully")
+
+        check_cmd = ["restic", "check"]
+        result = subprocess.run(check_cmd, env=env, capture_output=True, text=True)
+
+        if result.returncode != 0:
+            logger.error("Restic check failed: %s", result.stderr)
+            return False
+
+        logger.info("Restic check completed successfully")
+
+        forget_cmd = [
+            "restic",
+            "forget",
+            "--compact",
+            "--prune",
+            "--keep-daily",
+            "90",
+            "--keep-monthly",
+            "36",
+        ]
+        result = subprocess.run(forget_cmd, env=env, capture_output=True, text=True)
+
+        if result.returncode != 0:
+            logger.error("Restic forget/prune failed: %s", result.stderr)
+            return False
+
+        logger.info("Restic forget/prune completed successfully")
+
+        result = subprocess.run(check_cmd, env=env, capture_output=True, text=True)
+
+        if result.returncode != 0:
+            logger.error("Final restic check failed: %s", result.stderr)
+            return False
+
+        logger.info("Final restic check completed successfully")
+        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, 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."""
         applications: List[Application] = []
-        source_dirs = itertools.chain(*(root.iterdir() for root in self.config.roots))
+        source_dirs = itertools.chain(*(root.iterdir() for root in self.roots))
 
         for app_dir in source_dirs:
             if "lost+found" in str(app_dir):
@@ -306,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"<b>{telegram_cfg.notifications_name}</b>: бекап успешно завершен!"
-                )
-                if self.successful_backups:
-                    message += (
-                        f"\n\nУспешные бекапы: {', '.join(self.successful_backups)}"
-                    )
-            else:
-                message = f"<b>{telegram_cfg.notifications_name}</b>: бекап завершен с ошибками!"
+        if success and not self.errors:
+            message = f"<b>{self.config.host_name}</b>: бекап успешно завершен!"
+            if self.successful_backups:
+                message += f"\n\nУспешные бекапы: {', '.join(self.successful_backups)}"
+        else:
+            message = f"<b>{self.config.host_name}</b>: бекап завершен с ошибками!"
 
-                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"""
@@ -397,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")
 
@@ -420,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")
@@ -436,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, storages=storages)
+    return BackupManager(
+        config=config, roots=roots, storages=storages, notifiers=notifiers
+    )
 
 
 def main():

files/backups/config.template.ini

@@ -1,11 +0,0 @@
-[restic]
-RESTIC_REPOSITORY={{ restic_repository }}
-RESTIC_PASSWORD={{ restic_password }}
-AWS_ACCESS_KEY_ID={{ restic_s3_access_key }}
-AWS_SECRET_ACCESS_KEY={{ restic_s3_access_secret }}
-AWS_DEFAULT_REGION={{ restic_s3_region }}
-
-[telegram]
-TELEGRAM_BOT_TOKEN={{ notifications_tg_bot_token }}
-TELEGRAM_CHAT_ID={{ notifications_tg_chat_id }}
-NOTIFICATIONS_NAME={{ notifications_name }}

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 }}"