leocrm/scripts/backup.py

#!/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())