линтеры python перенастроены: mypy заменён на pyrefly

- набор правил ruff расширен по канону rp-local-env (ANN, PTH, C90, G и др.), в lefthook добавлен job ruff check --fix - код приведён под новые правила: логирование через %s вместо f-строк, os.path заменён на pathlib, run_backup_process и initialize разбиты на функции по порогу цикломатической сложности
2026-08-22 11:32:03 +03:00
parent 321f6e7b6b
commit bdc319df64
7 changed files with 266 additions and 253 deletions
@@ -160,7 +160,7 @@ class Storage(ABC):
 class ResticStorage(Storage):
    TYPE_NAME = "restic"

-    def __init__(self, name: str, params: Dict[str, Any]):
+    def __init__(self, name: str, params: Dict[str, Any]) -> None:
        self.name = name
        self.restic_repository = str(params.get("restic_repository", ""))
        self.restic_password = str(params.get("restic_password", ""))
@@ -306,7 +306,7 @@ class Notifier(ABC):
 class AppriseNotifier(Notifier):
    TYPE_NAME = "apprise"

-    def __init__(self, name: str, params: Dict[str, Any]):
+    def __init__(self, name: str, params: Dict[str, Any]) -> None:
        self.name = name
        self.api_url = str(params.get("api_url", "")).rstrip("/")
        self.tag = str(params.get("tag", ""))
@@ -329,12 +329,14 @@ class AppriseNotifier(Notifier):
            logger.info("Apprise notification sent successfully")
        else:
            logger.error(
-                f"Failed to send Apprise notification: {response.status_code} - {response.text}"
+                "Failed to send Apprise notification: %s - %s",
+                response.status_code,
+                response.text,
            )


 class ApplicationFinder:
-    def __init__(self, roots: List[Path]):
+    def __init__(self, roots: List[Path]) -> None:
        self.roots = roots
        self.warnings: List[str] = []

@@ -361,7 +363,7 @@ class ApplicationFinder:
                        )
                    )
                except (KeyError, OSError) as e:
-                    logger.warning(f"Could not get owner for {app_dir}: {e}")
+                    logger.warning("Could not get owner for %s: %s", app_dir, e)

        applications.sort(key=lambda app: app.path.name)
        return applications
@@ -373,10 +375,9 @@ class ApplicationFinder:
            if script_path.exists():
                if os.access(script_path, os.X_OK):
                    return script_path
-                else:
-                    logger.warning(
-                        f"Backup script {script_path} exists but is not executable"
-                    )
+                logger.warning(
+                    "Backup script %s exists but is not executable", script_path
+                )
        return None

    def _find_backup_targets(self, app_dir: Path) -> List[Path]:
@@ -435,7 +436,7 @@ class BackupManager:
        schedule: Schedule,
        maintenance: MaintenanceOptions,
        forced_phases: Optional[List[str]] = None,
-    ):
+    ) -> None:
        self.errors: List[str] = []
        self.warnings: List[str] = []
        self.backed_up_apps: List[str] = []
@@ -452,69 +453,107 @@ class BackupManager:
    def run_backup_process(self, applications: List[Application]) -> bool:
        """Main backup process"""
        logger.info("Starting backup process")
-        logger.info(f"Found {len(applications)} application directories")
+        logger.info("Found %d application directories", len(applications))

-        # Какие фазы выполняем в этот прогон: либо принудительно из CLI, либо по расписанию.
+        self.active_phases = self._resolve_phases()
+        self._run_archive_phase(applications)
+        backup_dirs = self._collect_backup_dirs(applications)
+        overall_success = self._run_storages(backup_dirs)
+
+        self._send_notification(overall_success)
+
+        logger.info("Backup process completed")
+
+        if self.errors:
+            logger.error("Backup completed with %d errors", len(self.errors))
+            return False
+        if self.warnings:
+            logger.warning("Backup completed with %d warnings", len(self.warnings))
+            return True
+        logger.info("Backup completed successfully")
+        return True
+
+    def _resolve_phases(self) -> List[str]:
+        """Какие фазы выполняем в этот прогон: принудительно из CLI или по расписанию."""
        if self.forced_phases is not None:
-            self.active_phases = self.forced_phases
-            logger.info("Phases (forced): %s", ", ".join(self.active_phases))
-        else:
-            self.active_phases = self.schedule.due_phases(datetime.now())
-            logger.info("Phases (scheduled): %s", ", ".join(self.active_phases))
+            logger.info("Phases (forced): %s", ", ".join(self.forced_phases))
+            return self.forced_phases

+        phases = self.schedule.due_phases(datetime.now())
+        logger.info("Phases (scheduled): %s", ", ".join(phases))
+        return phases
+
+    def _run_archive_phase(self, applications: List[Application]) -> None:
+        """Прогнать скрипты дампов приложений и собрать список того, что уедет в restic.
+
+        Фаза нужна только вместе с restic backup: без неё дампы делать некому и незачем.
+        """
        archive_start = time.monotonic()
-        # Archive phase (per-app backup scripts) нужна только если будем делать restic backup.
+
        if PHASE_BACKUP in self.active_phases:
            for app in applications:
-                app_dir = str(app.path)
-                username = app.owner
-                app_name = app.path.name
-
-                if app.backup_script is None:
-                    if app.backup_targets:
-                        # Приложение без дампа: restic забирает его данные как есть,
-                        # отдельный шаг архивации ему не нужен.
-                        logger.info(
-                            f"No backup script for app: {app_dir} (user {username}), "
-                            f"data directories go to restic as is"
-                        )
-                        self.backed_up_apps.append(app_name)
-                    else:
-                        warning_msg = (
-                            f"Nothing to back up for app: {app_dir} (user {username}): "
-                            f"no backup script and no backup targets"
-                        )
-                        logger.warning(warning_msg)
-                        self.warnings.append(warning_msg)
-                    continue
-
-                logger.info(f"Processing backup for app: {app_dir} (user {username})")
-                if not self._run_app_backup(str(app.backup_script), app_dir, username):
-                    continue
-                # Дамп сделан, но в restic он попадёт только если есть цели бекапа;
-                # об их отсутствии уже предупредил ApplicationFinder.
-                if app.backup_targets:
-                    self.backed_up_apps.append(app_name)
+                self._archive_app(app)
        else:
            logger.info("Backup phase not active, skipping per-app archive scripts")
+
        self.archive_duration = time.monotonic() - archive_start
        logger.info(
            "Archive phase finished in %s", format_duration(self.archive_duration)
        )

-        # Collect backup directories from applications
+    def _archive_app(self, app: Application) -> None:
+        """Обработать одно приложение: сделать дамп, если он предусмотрен."""
+        app_dir = str(app.path)
+        username = app.owner
+        app_name = app.path.name
+
+        if app.backup_script is None:
+            if app.backup_targets:
+                # Приложение без дампа: restic забирает его данные как есть,
+                # отдельный шаг архивации ему не нужен.
+                logger.info(
+                    "No backup script for app: %s (user %s), "
+                    "data directories go to restic as is",
+                    app_dir,
+                    username,
+                )
+                self.backed_up_apps.append(app_name)
+            else:
+                warning_msg = (
+                    f"Nothing to back up for app: {app_dir} (user {username}): "
+                    f"no backup script and no backup targets"
+                )
+                logger.warning(warning_msg)
+                self.warnings.append(warning_msg)
+            return
+
+        logger.info("Processing backup for app: %s (user %s)", app_dir, username)
+        if not self._run_app_backup(str(app.backup_script), app_dir, username):
+            return
+        # Дамп сделан, но в restic он попадёт только если есть цели бекапа;
+        # об их отсутствии уже предупредил ApplicationFinder.
+        if app.backup_targets:
+            self.backed_up_apps.append(app_name)
+
+    @staticmethod
+    def _collect_backup_dirs(applications: List[Application]) -> List[str]:
+        """Собрать цели бекапа всех приложений, сохраняя порядок и убирая дубли."""
        backup_dirs: List[str] = []
        for app in applications:
            for target in app.backup_targets:
                target_str = str(target)
                if target_str not in backup_dirs:
                    backup_dirs.append(target_str)
-        logger.info(f"Found backup directories: {backup_dirs}")
+        logger.info("Found backup directories: %s", backup_dirs)
+        return backup_dirs

+    def _run_storages(self, backup_dirs: List[str]) -> bool:
+        """Прогнать активные фазы по всем хранилищам.
+
+        Хранилища независимы: падение одного не отменяет попытку для остальных.
+        """
        overall_success = True

-        # Each storage is processed independently: a failure in one storage
-        # must not prevent the others from being attempted.
        for storage in self.storages:
            storage_start = time.monotonic()
            try:
@@ -547,28 +586,14 @@ class BackupManager:
                    error_msg += f": {backup_result.error}"
                self.errors.append(error_msg)

-            # Determine overall success
            overall_success = overall_success and backup_result.success

-        # Send notification
-        self._send_notification(overall_success)
-
-        logger.info("Backup process completed")
-
-        if self.errors:
-            logger.error(f"Backup completed with {len(self.errors)} errors")
-            return False
-        elif self.warnings:
-            logger.warning(f"Backup completed with {len(self.warnings)} warnings")
-            return True
-        else:
-            logger.info("Backup completed successfully")
-            return True
+        return overall_success

    def _run_app_backup(self, script_path: str, app_dir: str, username: str) -> bool:
        """Run backup script as the specified user"""
        try:
-            logger.info(f"Running backup script {script_path} (user {username})")
+            logger.info("Running backup script %s (user %s)", script_path, username)

            # Use su to run the script as the user
            cmd = ["su", "--login", username, "--command", script_path]
@@ -582,15 +607,14 @@ class BackupManager:
            )

            if result.returncode == 0:
-                logger.info(f"Backup script for {username} completed successfully")
+                logger.info("Backup script for %s completed successfully", username)
                return True
-            else:
-                error_msg = f"Backup script {script_path} failed with return code {result.returncode}"
-                if result.stderr:
-                    error_msg += f": {result.stderr}"
-                logger.error(error_msg)
-                self.errors.append(f"App {username}: {error_msg}")
-                return False
+            error_msg = f"Backup script {script_path} failed with return code {result.returncode}"
+            if result.stderr:
+                error_msg += f": {result.stderr}"
+            logger.error(error_msg)
+            self.errors.append(f"App {username}: {error_msg}")
+            return False

        except subprocess.TimeoutExpired:
            error_msg = f"Backup script {script_path} timed out"
@@ -644,7 +668,7 @@ class BackupManager:
            try:
                notificator.send(title, message)
            except Exception as e:
-                logger.error(f"Failed to send notification: {str(e)}")
+                logger.error("Failed to send notification: %s", e)


 def parse_phases(raw: str) -> List[str]:
@@ -659,6 +683,65 @@ def parse_phases(raw: str) -> List[str]:
    return [p for p in PHASE_ORDER if p in requested]


+def build_storages(raw_config: Dict[str, Any]) -> List[Storage]:
+    """Собрать хранилища из секции [storage] конфига."""
+    storage_raw = raw_config.get("storage") or {}
+    storages: List[Storage] = []
+    for name, params in storage_raw.items():
+        if not isinstance(params, dict):
+            raise ValueError(f"Storage config for {name} must be a table")
+        if params.get("type", "") == ResticStorage.TYPE_NAME:
+            storages.append(ResticStorage(name, params))
+    if not storages:
+        raise ValueError("At least one storage backend must be configured")
+    return storages
+
+
+def build_notifiers(raw_config: Dict[str, Any]) -> List[Notifier]:
+    """Собрать нотификаторы из секции [notifier] конфига."""
+    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")
+        if params.get("type", "") == AppriseNotifier.TYPE_NAME:
+            notifiers.append(AppriseNotifier(name, params))
+    if not notifiers:
+        raise ValueError("At least one notification backend must be configured")
+    return notifiers
+
+
+def build_schedule(raw_config: Dict[str, Any]) -> Schedule:
+    """Собрать расписание обслуживающих фаз из секции [schedule] конфига."""
+    schedule_raw = raw_config.get("schedule") or {}
+    if not isinstance(schedule_raw, dict):
+        raise ValueError("'schedule' must be a table in config.toml")
+    return Schedule(
+        cron={
+            phase: str(schedule_raw[phase])
+            for phase in SCHEDULED_PHASES
+            if phase in schedule_raw
+        }
+    )
+
+
+def build_maintenance(raw_config: Dict[str, Any]) -> MaintenanceOptions:
+    """Собрать параметры обслуживания из секции [maintenance] конфига."""
+    maintenance_raw = raw_config.get("maintenance") or {}
+    if not isinstance(maintenance_raw, dict):
+        raise ValueError("'maintenance' must be a table in config.toml")
+    defaults = MaintenanceOptions()
+    return MaintenanceOptions(
+        verify_subset=str(maintenance_raw.get("verify_subset", defaults.verify_subset)),
+        prune_max_unused=str(
+            maintenance_raw.get("prune_max_unused", defaults.prune_max_unused)
+        ),
+        prune_max_repack=str(
+            maintenance_raw.get("prune_max_repack", defaults.prune_max_repack)
+        ),
+    )
+
+
 def initialize(
    config_path: Path,
    forced_phases: Optional[List[str]] = None,
@@ -667,7 +750,7 @@ def initialize(
        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}")
+        logger.error("Failed to read config file %s: %s", config_path, e)
        raise

    host_name = str(raw_config.get("host_name", "unknown"))
@@ -677,52 +760,10 @@ def initialize(
        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 {}
-    storages: List[Storage] = []
-    for name, params in storage_raw.items():
-        if not isinstance(params, dict):
-            raise ValueError(f"Storage config for {name} must be a table")
-        storage_type = params.get("type", "")
-        if storage_type == ResticStorage.TYPE_NAME:
-            storages.append(ResticStorage(name, params))
-    if not storages:
-        raise ValueError("At least one storage backend must be configured")
-
-    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 == AppriseNotifier.TYPE_NAME:
-            notifiers.append(AppriseNotifier(name, params))
-    if not notifiers:
-        raise ValueError("At least one notification backend must be configured")
-
-    schedule_raw = raw_config.get("schedule") or {}
-    if not isinstance(schedule_raw, dict):
-        raise ValueError("'schedule' must be a table in config.toml")
-    schedule = Schedule(
-        cron={
-            phase: str(schedule_raw[phase])
-            for phase in SCHEDULED_PHASES
-            if phase in schedule_raw
-        }
-    )
-
-    maintenance_raw = raw_config.get("maintenance") or {}
-    if not isinstance(maintenance_raw, dict):
-        raise ValueError("'maintenance' must be a table in config.toml")
-    defaults = MaintenanceOptions()
-    maintenance = MaintenanceOptions(
-        verify_subset=str(maintenance_raw.get("verify_subset", defaults.verify_subset)),
-        prune_max_unused=str(
-            maintenance_raw.get("prune_max_unused", defaults.prune_max_unused)
-        ),
-        prune_max_repack=str(
-            maintenance_raw.get("prune_max_repack", defaults.prune_max_repack)
-        ),
-    )
+    storages = build_storages(raw_config)
+    notifiers = build_notifiers(raw_config)
+    schedule = build_schedule(raw_config)
+    maintenance = build_maintenance(raw_config)

    config = Config(host_name=host_name)
    app_finder = ApplicationFinder(roots)
@@ -767,7 +808,7 @@ def main() -> None:
        logger.info("Backup process interrupted by user")
        sys.exit(130)
    except Exception as e:
-        logger.error(f"Unexpected error in backup process: {str(e)}")
+        logger.error("Unexpected error in backup process: %s", e)
        sys.exit(1)


@@ -1,7 +1,8 @@
 #!/usr/bin/env python3

-import os
 import argparse
+import os
+from pathlib import Path


 def main() -> None:
@@ -18,12 +19,13 @@ def main() -> None:
    if args.keep < 0:
        parser.error("--keep value cannot be negative")

-    if not os.path.isdir(args.directory):
+    directory = Path(args.directory)
+    if not directory.is_dir():
        parser.error(f"Directory not found: {args.directory}")

    # Get list of files (exclude subdirectories)
    files = []
-    with os.scandir(args.directory) as entries:
+    with os.scandir(directory) as entries:
        for entry in entries:
            if entry.is_file():
                files.append(entry.name)
@@ -36,9 +38,9 @@ def main() -> None:

    # Delete files and print results
    for filename in to_delete:
-        filepath = os.path.join(args.directory, filename)
+        filepath = directory / filename
        try:
-            os.remove(filepath)
+            filepath.unlink()
            print(f"Deleted: {filename}")
        except Exception as e:
            print(f"Error deleting {filename}: {str(e)}")