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