feat(ai): catalogue d'outils mutations + confirmations two-step (#79 phase D)

backend/services/backups.py

@@ -12,6 +12,8 @@ from __future__ import annotations
 import difflib
 import logging
 import os
+import shutil
+import time
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
@@ -22,6 +24,19 @@ from backend.services.vaults import get_vault_root
 
 logger = logging.getLogger("obsigate.services.backups")
 
+# Default number of backups kept per file (matches ``max_backups_per_file``).
+DEFAULT_MAX_BACKUPS = 10
+
+
+def _default_max_backups() -> int:
+    """Read ``max_backups_per_file`` from app config (lazy, best-effort)."""
+    try:
+        from backend.main import _load_config
+
+        return int(_load_config().get("max_backups_per_file", DEFAULT_MAX_BACKUPS))
+    except Exception:  # pragma: no cover - config unavailable
+        return DEFAULT_MAX_BACKUPS
+
 
 def get_backup_dir(vault_name: str, relative_path: str) -> Path:
     """Return the directory where backups for a specific file are stored.
@@ -39,6 +54,59 @@ def get_backup_dir(vault_name: str, relative_path: str) -> Path:
     return backup_root / vault_name / Path(relative_path).parent
 
 
+def create_backup(
+    file_path: Path,
+    vault_name: str,
+    relative_path: str,
+    *,
+    max_backups: int | None = None,
+) -> Path | None:
+    """Create a timestamped backup of a file before it is modified.
+
+    Backups are stored as ``{backup_root}/{vault}/{relative_path}.{ts}.bak``.
+    Missing or unreadable files are skipped silently (returns ``None``) so a
+    backup failure never blocks the caller's mutation.
+
+    Args:
+        file_path: Absolute path of the file to back up.
+        vault_name: Name of the vault the file belongs to.
+        relative_path: Vault-relative path (used to mirror the tree).
+        max_backups: Number of backups to keep per file. ``None`` reads
+            ``max_backups_per_file`` from the app config.
+
+    Returns:
+        The created backup path, or ``None`` when skipped.
+    """
+    try:
+        if not file_path.exists() or not file_path.is_file():
+            logger.debug(f"Backup skipped: file not found {file_path}")
+            return None
+
+        backup_dir = get_backup_dir(vault_name, relative_path)
+        backup_dir.mkdir(parents=True, exist_ok=True)
+        backup_path = backup_dir / f"{file_path.name}.{int(time.time())}.bak"
+        shutil.copy2(file_path, backup_path)
+        logger.info(f"Backup created: {relative_path} -> {backup_path}")
+
+        keep = max_backups if max_backups is not None else _default_max_backups()
+        all_backups = sorted(
+            [
+                f
+                for f in backup_dir.iterdir()
+                if f.is_file() and f.name.startswith(file_path.name + ".") and f.name.endswith(".bak")
+            ],
+            key=lambda f: f.stat().st_mtime,
+            reverse=True,
+        )
+        for old in all_backups[keep:]:
+            old.unlink()
+            logger.debug(f"Auto-cleanup: removed old backup {old.name}")
+        return backup_path
+    except Exception as e:
+        logger.warning(f"Failed to backup {relative_path} (vault={vault_name}): {e}", exc_info=True)
+        return None
+
+
 def list_backup_files(vault_name: str, relative_path: str) -> list[dict[str, Any]]:
     """List all backup files for a vault file, sorted newest first.