#!/usr/bin/env python3 """Automated Backup Script — pg_dump + file backup with retention policy. Usage: python scripts/backup.py python scripts/backup.py --destination local python scripts/backup.py --destination s3 --bucket my-backups python scripts/backup.py --retention-days 30 --dry-run Creates timestamped backups in /backups/ directory: /backups/leocrm_backup_YYYYMMDD_HHMMSS/ ├── database.sql (pg_dump output) ├── files/ (copied storage files) └── manifest.json (backup metadata) Environment variables: DATABASE_URL — PostgreSQL connection string (asyncpg format, converted to psycopg2) STORAGE_PATH — Path to files to backup (default: /data/storage) BACKUP_DIR — Base backup directory (default: /backups) BACKUP_RETENTION_DAYS — Days to keep backups (default: 7) BACKUP_DESTINATION — local | s3 | nextcloud (default: local) BACKUP_S3_BUCKET — S3 bucket name (if destination=s3) BACKUP_NEXTCLOUD_URL — Nextcloud WebDAV URL (if destination=nextcloud) Exit codes: 0 — backup successful 1 — backup failed 2 — configuration error """ from __future__ import annotations import argparse import json import os import shutil import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path from typing import Any # Add project root to path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # ─── Backup Configuration ──────────────────────────────────────────── DEFAULT_BACKUP_DIR = os.environ.get("BACKUP_DIR", "/backups") DEFAULT_STORAGE_PATH = os.environ.get("STORAGE_PATH", "/data/storage") DEFAULT_RETENTION_DAYS = int(os.environ.get("BACKUP_RETENTION_DAYS", "7")) def get_db_connection_params() -> dict[str, str]: """Extract PostgreSQL connection params from DATABASE_URL env var.""" db_url = os.environ.get("DATABASE_URL", "") if not db_url: raise ValueError("DATABASE_URL environment variable not set") # Convert asyncpg URL to standard postgresql URL for pg_dump if db_url.startswith("postgresql+asyncpg://"): db_url = db_url.replace("postgresql+asyncpg://", "postgresql://", 1) # Parse the URL from urllib.parse import urlparse parsed = urlparse(db_url) return { "url": db_url, "host": parsed.hostname or "localhost", "port": str(parsed.port or 5432), "database": parsed.path.lstrip("/"), "username": parsed.username or "postgres", "password": parsed.password or "", } def create_backup_manifest( backup_dir: Path, db_success: bool, files_success: bool, db_size: int = 0, files_count: int = 0, files_size: int = 0, errors: list[str] | None = None, ) -> dict[str, Any]: """Create a manifest.json for the backup.""" return { "backup_id": backup_dir.name, "created_at": datetime.now(timezone.utc).isoformat(), "database": { "success": db_success, "size_bytes": db_size, }, "files": { "success": files_success, "count": files_count, "size_bytes": files_size, }, "errors": errors or [], "version": "1.0.0", } # ─── Database Backup ───────────────────────────────────────────────── def backup_database(backup_dir: Path, db_params: dict[str, str]) -> tuple[bool, int, str]: """Run pg_dump and save to backup directory. Returns (success, size_bytes, error_message). """ db_file = backup_dir / "database.sql" # Build pg_dump command cmd = [ "pg_dump", "--host", db_params["host"], "--port", db_params["port"], "--username", db_params["username"], "--format", "plain", "--no-owner", "--no-privileges", db_params["database"], ] # Set PGPASSWORD environment for pg_dump env = os.environ.copy() if db_params["password"]: env["PGPASSWORD"] = db_params["password"] try: with open(db_file, "w") as f: result = subprocess.run( cmd, stdout=f, stderr=subprocess.PIPE, env=env, timeout=600, ) if result.returncode != 0: error_msg = result.stderr.decode() if result.stderr else "pg_dump failed" return False, 0, error_msg size = db_file.stat().st_size return True, size, "" except subprocess.TimeoutExpired: return False, 0, "pg_dump timed out after 600s" except FileNotFoundError: return False, 0, "pg_dump command not found" except Exception as exc: return False, 0, str(exc) # ─── File Backup ───────────────────────────────────────────────────── def backup_files(backup_dir: Path, storage_path: str) -> tuple[bool, int, int, str]: """Copy storage files to backup directory. Returns (success, file_count, total_size_bytes, error_message). """ files_dir = backup_dir / "files" if not os.path.isdir(storage_path): # Storage path doesn't exist — not necessarily an error return True, 0, 0, "" try: files_dir.mkdir(parents=True, exist_ok=True) file_count = 0 total_size = 0 for root, dirs, files in os.walk(storage_path): rel_root = os.path.relpath(root, storage_path) dest_root = files_dir / rel_root if rel_root != "." else files_dir dest_root.mkdir(parents=True, exist_ok=True) for fname in files: src_file = os.path.join(root, fname) dest_file = dest_root / fname shutil.copy2(src_file, dest_file) file_count += 1 total_size += os.path.getsize(src_file) return True, file_count, total_size, "" except Exception as exc: return False, 0, 0, str(exc) # ─── Retention Policy ──────────────────────────────────────────────── def apply_retention_policy(backup_base_dir: Path, retention_days: int) -> list[str]: """Delete backups older than retention_days. Returns list of deleted backup directory names. """ if retention_days <= 0: return [] deleted = [] cutoff_time = time.time() - (retention_days * 86400) if not backup_base_dir.is_dir(): return deleted for entry in backup_base_dir.iterdir(): if not entry.is_dir(): continue if not entry.name.startswith("leocrm_backup_"): continue try: entry_mtime = entry.stat().st_mtime if entry_mtime < cutoff_time: shutil.rmtree(entry) deleted.append(entry.name) except Exception: pass # Don't fail on retention errors return deleted # ─── Notification ──────────────────────────────────────────────────── def send_backup_notification( success: bool, backup_dir: Path, errors: list[str], dry_run: bool = False, ) -> None: """Send backup status notification via system_notif plugin event bus. This publishes a 'notification.created' event that the system_notif plugin picks up and converts to a chat message. """ if dry_run: return try: import asyncio from app.core.event_bus import get_event_bus event_type = "backup.completed" if success else "backup.failed" title = "Backup erfolgreich" if success else "Backup fehlgeschlagen" body = f"Backup: {backup_dir.name}" if errors: body += f"\nFehler: {'; '.join(errors[:3])}" async def _publish(): bus = get_event_bus() await bus.publish("notification.created", { "title": title, "body": body, "event_type": event_type, "severity": "info" if success else "error", "tenant_id": os.environ.get("BACKUP_TENANT_ID", ""), "user_id": os.environ.get("BACKUP_USER_ID", ""), }) asyncio.run(_publish()) except Exception: # Notification is best-effort — don't fail backup on notification error pass # ─── Main Backup Pipeline ──────────────────────────────────────────── def run_backup( destination: str = "local", retention_days: int = DEFAULT_RETENTION_DAYS, storage_path: str = DEFAULT_STORAGE_PATH, backup_base_dir: str = DEFAULT_BACKUP_DIR, dry_run: bool = False, ) -> dict[str, Any]: """Run the full backup pipeline and return a report.""" timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") backup_name = f"leocrm_backup_{timestamp}" backup_dir = Path(backup_base_dir) / backup_name report: dict[str, Any] = { "backup_id": backup_name, "backup_path": str(backup_dir), "destination": destination, "dry_run": dry_run, "started_at": datetime.now(timezone.utc).isoformat(), "database": {"success": False, "size_bytes": 0, "error": ""}, "files": {"success": False, "count": 0, "size_bytes": 0, "error": ""}, "retention": {"deleted": []}, "overall_success": False, "errors": [], } if dry_run: report["database"]["error"] = "[DRY-RUN] Would run pg_dump" report["files"]["error"] = "[DRY-RUN] Would copy files" report["overall_success"] = True report["finished_at"] = datetime.now(timezone.utc).isoformat() return report # Create backup directory backup_dir.mkdir(parents=True, exist_ok=True) # Step 1: Database backup print(f"▶ Step 1: Database backup (pg_dump)") try: db_params = get_db_connection_params() db_success, db_size, db_error = backup_database(backup_dir, db_params) report["database"] = {"success": db_success, "size_bytes": db_size, "error": db_error} if db_success: print(f" ✓ Database backup: {db_size} bytes") else: print(f" ✗ Database backup failed: {db_error}") report["errors"].append(f"Database: {db_error}") except Exception as exc: report["database"]["error"] = str(exc) report["errors"].append(f"Database config: {exc}") print(f" ✗ Database config error: {exc}") # Step 2: File backup print(f"▶ Step 2: File backup ({storage_path})") files_success, files_count, files_size, files_error = backup_files(backup_dir, storage_path) report["files"] = { "success": files_success, "count": files_count, "size_bytes": files_size, "error": files_error, } if files_success: print(f" ✓ File backup: {files_count} files, {files_size} bytes") else: print(f" ✗ File backup failed: {files_error}") report["errors"].append(f"Files: {files_error}") # Step 3: Write manifest manifest = create_backup_manifest( backup_dir, db_success=report["database"]["success"], files_success=report["files"]["success"], db_size=report["database"]["size_bytes"], files_count=report["files"]["count"], files_size=report["files"]["size_bytes"], errors=report["errors"], ) manifest_file = backup_dir / "manifest.json" with open(manifest_file, "w") as f: json.dump(manifest, f, indent=2) print(f" ✓ Manifest written: {manifest_file}") # Step 4: Apply retention policy print(f"▶ Step 3: Retention policy ({retention_days} days)") deleted = apply_retention_policy(Path(backup_base_dir), retention_days) report["retention"]["deleted"] = deleted if deleted: print(f" ✓ Deleted {len(deleted)} old backup(s)") else: print(f" ✓ No old backups to delete") # Step 5: Determine overall success report["overall_success"] = report["database"]["success"] and report["files"]["success"] report["finished_at"] = datetime.now(timezone.utc).isoformat() # Step 6: Send notification send_backup_notification( success=report["overall_success"], backup_dir=backup_dir, errors=report["errors"], dry_run=dry_run, ) return report # ─── CLI Entry Point ───────────────────────────────────────────────── def main() -> int: parser = argparse.ArgumentParser( description="LeoCRM Automated Backup — pg_dump + file backup with retention." ) parser.add_argument( "--destination", choices=["local", "s3", "nextcloud"], default=os.environ.get("BACKUP_DESTINATION", "local"), help="Backup destination (default: local)", ) parser.add_argument( "--retention-days", type=int, default=DEFAULT_RETENTION_DAYS, help=f"Days to keep backups (default: {DEFAULT_RETENTION_DAYS})", ) parser.add_argument( "--storage-path", default=DEFAULT_STORAGE_PATH, help=f"Path to storage files (default: {DEFAULT_STORAGE_PATH})", ) parser.add_argument( "--backup-dir", default=DEFAULT_BACKUP_DIR, help=f"Base backup directory (default: {DEFAULT_BACKUP_DIR})", ) parser.add_argument( "--dry-run", action="store_true", help="Simulate backup without executing", ) parser.add_argument( "--output", default="-", help="Write JSON report to file (default: stdout)", ) args = parser.parse_args() print(f"\n{'='*60}") print(f" LeoCRM Backup Pipeline") print(f" {'[DRY-RUN]' if args.dry_run else '[LIVE]'}") print(f"{'='*60}\n") report = run_backup( destination=args.destination, retention_days=args.retention_days, storage_path=args.storage_path, backup_base_dir=args.backup_dir, dry_run=args.dry_run, ) report_json = json.dumps(report, indent=2, ensure_ascii=False) if args.output == "-": print(f"\n{'='*60}") print(" Backup Report") print(f"{'='*60}") print(report_json) else: with open(args.output, "w", encoding="utf-8") as f: f.write(report_json) print(f"\nReport written to {args.output}", file=sys.stderr) return 0 if report["overall_success"] else 1 if __name__ == "__main__": sys.exit(main())