path: root/playbooks/roles/backup/templates/volumes/scripts/cleanup.py

#!/usr/bin/env python3
"""
Backup Cleanup Script
Manages backup retention by removing old Borg backup archives
"""

import os
import sys
import logging
import argparse
import subprocess
import json
from datetime import datetime, timedelta
from pathlib import Path
from typing import List
import urllib.request
import urllib.parse


def setup_logging(log_level: str = "INFO") -> logging.Logger:
    """Setup logging configuration"""
    logger = logging.getLogger("cleanup")
    logger.setLevel(getattr(logging, log_level.upper()))

    # Console handler
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    console_formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    console_handler.setFormatter(console_formatter)
    logger.addHandler(console_handler)

    # File handler
    log_dir = Path("/scripts/logs")
    log_dir.mkdir(exist_ok=True)
    log_file = log_dir / f"cleanup-{datetime.now().strftime('%Y%m%d')}.log"
    file_handler = logging.FileHandler(log_file)
    file_handler.setLevel(logging.DEBUG)
    file_formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    file_handler.setFormatter(file_formatter)
    logger.addHandler(file_handler)

    return logger


class BorgBackupCleanup:
    """Borg backup cleanup manager"""

    def __init__(self, logger: logging.Logger):
        self.logger = logger
        # Get Borg configuration from environment
        self.borg_repo = os.getenv('BORG_REPO', '/backups/borg-repo')
        self.borg_passphrase = os.getenv('BORG_PASSPHRASE', '')
        self.keep_daily = int(os.getenv('BORG_KEEP_DAILY', '7'))
        self.keep_weekly = int(os.getenv('BORG_KEEP_WEEKLY', '4'))
        self.keep_monthly = int(os.getenv('BORG_KEEP_MONTHLY', '6'))

    def _send_notification(self, success: bool, message: str):
        """Send notification via ntfy if configured"""
        ntfy_topic = os.getenv('NTFY_TOPIC')
        if not ntfy_topic:
            return

        try:
            # Use ASCII-safe status indicators
            status = "CLEANUP" if success else "CLEANUP FAILED"
            title = f"Homelab Backup {status}"

            url = f'https://ntfy.sh/{ntfy_topic}'
            data = message.encode('utf-8')

            req = urllib.request.Request(url, data=data, method='POST')
            req.add_header('Title', title.encode('ascii', 'ignore').decode('ascii'))
            req.add_header('Content-Type', 'text/plain; charset=utf-8')

            with urllib.request.urlopen(req, timeout=10) as response:
                if response.status == 200:
                    self.logger.debug("Notification sent successfully")
                else:
                    self.logger.warning(f"Notification returned status {response.status}")

        except Exception as e:
            self.logger.error(f"Failed to send notification: {e}")

    def _get_borg_env(self) -> dict:
        """Get environment for Borg commands"""
        borg_env = os.environ.copy()
        borg_env['BORG_REPO'] = self.borg_repo
        borg_env['BORG_PASSPHRASE'] = self.borg_passphrase
        borg_env['BORG_RELOCATED_REPO_ACCESS_IS_OK'] = 'yes'
        borg_env['BORG_UNKNOWN_UNENCRYPTED_REPO_ACCESS_IS_OK'] = 'yes'
        # SSH configuration to accept unknown hosts
        borg_env['BORG_RSH'] = 'ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null'
        return borg_env

    def cleanup_old_backups(self) -> bool:
        """Remove old Borg backup archives using retention policy"""
        self.logger.info(f"Starting Borg cleanup with retention: {self.keep_daily}d/{self.keep_weekly}w/{self.keep_monthly}m")

        try:
            borg_env = self._get_borg_env()

            # Use Borg's built-in pruning with retention policy
            prune_cmd = [
                'borg', 'prune',
                '--verbose',
                '--stats',
                '--show-rc',
                '--keep-daily', str(self.keep_daily),
                '--keep-weekly', str(self.keep_weekly),
                '--keep-monthly', str(self.keep_monthly),
                self.borg_repo
            ]

            self.logger.info(f"Running: {' '.join(prune_cmd)}")

            result = subprocess.run(
                prune_cmd,
                env=borg_env,
                capture_output=True,
                text=True,
                timeout=300  # 5 minutes timeout
            )

            if result.returncode != 0:
                self.logger.error(f"Borg prune failed with return code {result.returncode}")
                self.logger.error(f"STDERR: {result.stderr}")
                raise subprocess.CalledProcessError(result.returncode, prune_cmd)

            # Log pruning statistics
            if result.stdout:
                self.logger.info("Borg prune statistics:")
                for line in result.stdout.split('\n'):
                    if line.strip() and ('Deleted' in line or 'Kept' in line or 'Repository size' in line):
                        self.logger.info(f"  {line.strip()}")

            # Run compact to reclaim space
            self.logger.info("Running Borg compact to reclaim space...")
            compact_cmd = ['borg', 'compact', '--verbose', self.borg_repo]

            compact_result = subprocess.run(
                compact_cmd,
                env=borg_env,
                capture_output=True,
                text=True,
                timeout=600  # 10 minutes timeout
            )

            if compact_result.returncode == 0:
                self.logger.info("Borg compact completed successfully")
            else:
                self.logger.warning(f"Borg compact completed with warnings: {compact_result.stderr}")

            message = f"Borg cleanup complete: kept {self.keep_daily}d/{self.keep_weekly}w/{self.keep_monthly}m"
            self.logger.info(message)
            self._send_notification(True, message)

            return True

        except subprocess.TimeoutExpired:
            self.logger.error("Borg cleanup timed out")
            self._send_notification(False, "Borg cleanup timed out")
            return False
        except Exception as e:
            error_msg = f"Borg cleanup failed: {str(e)}"
            self.logger.error(error_msg)
            self._send_notification(False, error_msg)
            return False

    def get_backup_stats(self):
        """Get statistics about current Borg repository"""
        try:
            borg_env = self._get_borg_env()

            # Get repository info
            info_cmd = ['borg', 'info', '--json', self.borg_repo]
            info_result = subprocess.run(
                info_cmd,
                env=borg_env,
                capture_output=True,
                text=True,
                timeout=60
            )

            if info_result.returncode != 0:
                self.logger.warning(f"Could not get repository info: {info_result.stderr}")
                return

            repo_info = json.loads(info_result.stdout)

            # Get archive list
            list_cmd = ['borg', 'list', '--json', self.borg_repo]
            list_result = subprocess.run(
                list_cmd,
                env=borg_env,
                capture_output=True,
                text=True,
                timeout=60
            )

            if list_result.returncode != 0:
                self.logger.warning(f"Could not get archive list: {list_result.stderr}")
                return

            archive_info = json.loads(list_result.stdout)
            archives = archive_info.get('archives', [])

            # Display statistics
            self.logger.info(f"Borg repository statistics:")
            self.logger.info(f"  Repository: {self.borg_repo}")
            self.logger.info(f"  Total archives: {len(archives)}")

            if 'cache' in repo_info:
                cache_stats = repo_info['cache']['stats']
                total_size_mb = cache_stats.get('total_size', 0) / (1024 * 1024)
                total_csize_mb = cache_stats.get('total_csize', 0) / (1024 * 1024)
                unique_csize_mb = cache_stats.get('unique_csize', 0) / (1024 * 1024)

                self.logger.info(f"  Original size: {total_size_mb:.1f} MB")
                self.logger.info(f"  Compressed size: {total_csize_mb:.1f} MB")
                self.logger.info(f"  Deduplicated size: {unique_csize_mb:.1f} MB")

                if total_size_mb > 0:
                    compression_ratio = (1 - total_csize_mb / total_size_mb) * 100
                    dedup_ratio = (1 - unique_csize_mb / total_csize_mb) * 100 if total_csize_mb > 0 else 0
                    self.logger.info(f"  Compression ratio: {compression_ratio:.1f}%")
                    self.logger.info(f"  Deduplication ratio: {dedup_ratio:.1f}%")

            if archives:
                oldest_archive = min(archives, key=lambda a: a['time'])
                newest_archive = max(archives, key=lambda a: a['time'])

                oldest_time = datetime.fromisoformat(oldest_archive['time'].replace('Z', '+00:00'))
                newest_time = datetime.fromisoformat(newest_archive['time'].replace('Z', '+00:00'))

                self.logger.info(f"  Oldest archive: {oldest_archive['name']} ({oldest_time.strftime('%Y-%m-%d %H:%M')})")
                self.logger.info(f"  Newest archive: {newest_archive['name']} ({newest_time.strftime('%Y-%m-%d %H:%M')})")

        except Exception as e:
            self.logger.error(f"Failed to get backup statistics: {e}")


def main():
    """Main function"""
    parser = argparse.ArgumentParser(description='Borg Backup Cleanup Script')
    parser.add_argument('backup_dir', nargs='?', help='Ignored for compatibility - Borg repo from env')
    parser.add_argument('--retention-days', type=int,
                        help='Ignored for compatibility - uses Borg retention policy from env')
    parser.add_argument('--log-level', default='INFO',
                        choices=['DEBUG', 'INFO', 'WARNING', 'ERROR'])
    parser.add_argument('--stats-only', action='store_true',
                        help='Only show backup statistics, do not perform cleanup')
    parser.add_argument('--blocklist', nargs='*',
                        help='Ignored parameter for compatibility with backup script')

    args = parser.parse_args()

    # Setup logging
    logger = setup_logging(args.log_level)

    # Create cleanup manager
    cleanup_manager = BorgBackupCleanup(logger=logger)

    # Show current statistics
    cleanup_manager.get_backup_stats()

    if args.stats_only:
        logger.info("Stats-only mode, no cleanup performed")
        sys.exit(0)

    # Perform cleanup
    success = cleanup_manager.cleanup_old_backups()
    sys.exit(0 if success else 1)


if __name__ == '__main__':
    main()