diff --git a/AGENTS.md b/AGENTS.md index 5405d46..3a5f898 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -59,6 +59,7 @@ Ansible-based server automation for personal services. Playbooks provision Docke - Ansible lint: `ansible-lint .` (CI default). - Authelia config validation: `task authelia-validate-config` (renders with secrets then validates via docker). - Black formatting for Python helpers: `task format-py-files`. +- Python types validation with mypy: `mypy `. ## Operational Notes - Deployments rely on `production.yml` inventory and per-app playbooks; run with `--diff` for visibility. diff --git a/files/backups/backup-all.py b/files/backups/backup-all.py index 122cbf2..0ad787f 100644 --- a/files/backups/backup-all.py +++ b/files/backups/backup-all.py @@ -12,7 +12,7 @@ import logging import pwd from dataclasses import dataclass from pathlib import Path -from typing import List, Optional +from typing import Dict, List, Optional import requests import tomllib from collections.abc import Iterable @@ -29,43 +29,107 @@ logging.basicConfig( ) logger = logging.getLogger(__name__) -try: - with open("/etc/backup/config.toml", "rb") as config_file: - config = tomllib.load(config_file) -except OSError as e: - logger.error(f"Failed to read config file: {e}") - raise -ROOTS = config.get("roots") or [] +@dataclass +class StorageConfig: + type: str + restic_repository: str + restic_password: str + aws_access_key_id: str + aws_secret_access_key: str + aws_default_region: str -storage_cfg = config.get("storage", {}).get("yandex", {}) -RESTIC_REPOSITORY = storage_cfg.get("restic_repository") -RESTIC_PASSWORD = storage_cfg.get("restic_password") -AWS_ACCESS_KEY_ID = storage_cfg.get("aws_access_key_id") -AWS_SECRET_ACCESS_KEY = storage_cfg.get("aws_secret_access_key") -AWS_DEFAULT_REGION = storage_cfg.get("aws_default_region") -notifications_cfg = config.get("notifications", {}).get("telegram", {}) -TELEGRAM_BOT_TOKEN = notifications_cfg.get("telegram_bot_token") -TELEGRAM_CHAT_ID = notifications_cfg.get("telegram_chat_id") -NOTIFICATIONS_NAME = notifications_cfg.get("notifications_name") +@dataclass +class TelegramConfig: + type: str + telegram_bot_token: str + telegram_chat_id: str + notifications_name: str -if not isinstance(ROOTS, Iterable) or not ROOTS: - raise ValueError("roots must be a non-empty list of paths in config.toml") -if not all( - [ - RESTIC_REPOSITORY, - RESTIC_PASSWORD, - AWS_ACCESS_KEY_ID, - AWS_SECRET_ACCESS_KEY, - AWS_DEFAULT_REGION, - TELEGRAM_BOT_TOKEN, - TELEGRAM_CHAT_ID, - NOTIFICATIONS_NAME, - ] -): - raise ValueError("Missing required configuration values in config.toml") +@dataclass +class Config: + roots: List[Path] + storage: Dict[str, StorageConfig] + notifications: Dict[str, TelegramConfig] + + +def read_config(config_path: Path) -> Config: + try: + with config_path.open("rb") as config_file: + raw_config = tomllib.load(config_file) + except OSError as e: + logger.error(f"Failed to read config file {config_path}: {e}") + raise + + 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") + roots = [Path(root) for root in roots_raw] + + storage_raw = raw_config.get("storage") or {} + storage: Dict[str, StorageConfig] = {} + for name, cfg in storage_raw.items(): + if not isinstance(cfg, dict): + raise ValueError(f"Storage config for {name} must be a table") + storage[name] = StorageConfig( + type=cfg.get("type", ""), + restic_repository=cfg.get("restic_repository", ""), + restic_password=cfg.get("restic_password", ""), + aws_access_key_id=cfg.get("aws_access_key_id", ""), + aws_secret_access_key=cfg.get("aws_secret_access_key", ""), + aws_default_region=cfg.get("aws_default_region", ""), + ) + + if not storage: + 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: + raise ValueError("At least one notification backend must be configured") + + for name, cfg in storage.items(): + if not all( + [ + cfg.type, + cfg.restic_repository, + cfg.restic_password, + cfg.aws_access_key_id, + cfg.aws_secret_access_key, + cfg.aws_default_region, + ] + ): + raise ValueError(f"Missing storage configuration values for backend {name}") + + 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}" + ) + + return Config(roots=roots, storage=storage, notifications=notifications) + + +CONFIG_PATH = Path("/etc/backup/config.toml") # File name to store directories and files to back up BACKUP_TARGETS_FILE = "backup-targets" @@ -86,11 +150,22 @@ class BackupManager: self.errors: List[str] = [] self.warnings: List[str] = [] self.successful_backups: List[str] = [] + self.config = read_config(CONFIG_PATH) + + def _select_storage(self) -> StorageConfig: + if "yandex" in self.config.storage: + return self.config.storage["yandex"] + return next(iter(self.config.storage.values())) + + 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) def find_applications(self) -> List[Application]: """Get all application directories and their owners.""" applications: List[Application] = [] - source_dirs = itertools.chain(*(Path(root).iterdir() for root in ROOTS)) + source_dirs = itertools.chain(*(root.iterdir() for root in self.config.roots)) for app_dir in source_dirs: if "lost+found" in str(app_dir): @@ -227,19 +302,21 @@ class BackupManager: logger.warning("No backup directories found") return True + storage_cfg = self._select_storage() + try: logger.info("Starting restic backup") - logger.info("Destination: %s", RESTIC_REPOSITORY) + logger.info("Destination: %s", storage_cfg.restic_repository) # Set environment variables for restic env = os.environ.copy() env.update( { - "RESTIC_REPOSITORY": RESTIC_REPOSITORY, - "RESTIC_PASSWORD": RESTIC_PASSWORD, - "AWS_ACCESS_KEY_ID": AWS_ACCESS_KEY_ID, - "AWS_SECRET_ACCESS_KEY": AWS_SECRET_ACCESS_KEY, - "AWS_DEFAULT_REGION": AWS_DEFAULT_REGION, + "RESTIC_REPOSITORY": storage_cfg.restic_repository, + "RESTIC_PASSWORD": storage_cfg.restic_password, + "AWS_ACCESS_KEY_ID": storage_cfg.aws_access_key_id, + "AWS_SECRET_ACCESS_KEY": storage_cfg.aws_secret_access_key, + "AWS_DEFAULT_REGION": storage_cfg.aws_default_region, } ) @@ -308,15 +385,22 @@ class BackupManager: 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 + try: if success and not self.errors: - message = f"{NOTIFICATIONS_NAME}: бекап успешно завершен!" + message = ( + f"{telegram_cfg.notifications_name}: бекап успешно завершен!" + ) if self.successful_backups: message += ( f"\n\nУспешные бекапы: {', '.join(self.successful_backups)}" ) else: - message = f"{NOTIFICATIONS_NAME}: бекап завершен с ошибками!" + message = f"{telegram_cfg.notifications_name}: бекап завершен с ошибками!" if self.successful_backups: message += ( @@ -329,8 +413,12 @@ class BackupManager: if self.errors: message += f"\n\n❌ Ошибки:\n" + "\n".join(self.errors) - url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage" - data = {"chat_id": TELEGRAM_CHAT_ID, "parse_mode": "HTML", "text": message} + 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) diff --git a/playbook-backups.yml b/playbook-backups.yml index 07b404e..3f53e35 100644 --- a/playbook-backups.yml +++ b/playbook-backups.yml @@ -7,7 +7,7 @@ vars: backup_config_dir: "/etc/backup" - backup_config_file: "{{ (backup_config_dir, 'config.ini') | path_join }}" + backup_config_file: "{{ (backup_config_dir, 'config.toml') | path_join }}" restic_shell_script: "{{ (bin_prefix, 'restic-shell.sh') | path_join }}" backup_all_script: "{{ (bin_prefix, 'backup-all.py') | path_join }}" @@ -23,7 +23,7 @@ - name: "Create backup config file" ansible.builtin.template: - src: "files/backups/config.template.ini" + src: "files/backups/config.template.toml" dest: "{{ backup_config_file }}" owner: root group: root