uniswap_auto_clp/tools/cleanup_manager.py

#!/usr/bin/env python3
"""
Cleanup Manager for Git Agent
Manages backup branch rotation (keep last 100)
"""

import os
import subprocess
import logging
from typing import Dict, Any, List

class CleanupManager:
    """Manages backup branch cleanup and rotation"""
    
    def __init__(self, config: Dict[str, Any], logger: logging.Logger):
        self.config = config
        self.logger = logger
        self.backup_config = config.get('backup', {})
        self.prefix = self.backup_config.get('branch_prefix', 'backup-')
        self.max_backups = self.backup_config.get('keep_max_count', 100)
        self.project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    
    def cleanup_old_backups(self) -> bool:
        """Clean up old backup branches to keep only the last N"""
        try:
            # Get all backup branches
            backup_branches = self._get_backup_branches()
            
            if len(backup_branches) <= self.max_backups:
                self.logger.info(f"✅ Backup count ({len(backup_branches)}) within limit ({self.max_backups})")
                return False  # No cleanup needed
            
            # Branches to delete (oldest ones)
            branches_to_delete = backup_branches[self.max_backups:]
            
            if not branches_to_delete:
                return False
            
            self.logger.info(f"🧹 Cleaning up {len(branches_to_delete)} old backup branches")
            
            deleted_count = 0
            for branch in branches_to_delete:
                # Delete local branch
                if self._delete_local_branch(branch):
                    # Delete remote branch
                    if self._delete_remote_branch(branch):
                        deleted_count += 1
                        self.logger.debug(f"  ✅ Deleted: {branch}")
                    else:
                        self.logger.warning(f"  ⚠️ Local deleted, remote failed: {branch}")
                else:
                    self.logger.warning(f"  ❌ Failed to delete: {branch}")
            
            if deleted_count > 0:
                self.logger.info(f"✅ Cleanup completed: deleted {deleted_count} old backup branches")
                return True
            else:
                self.logger.warning("⚠️ No branches were successfully deleted")
                return False
                
        except Exception as e:
            self.logger.error(f"❌ Cleanup failed: {e}")
            return False
    
    def _get_backup_branches(self) -> List[str]:
        """Get all backup branches sorted by timestamp (newest first)"""
        try:
            result = subprocess.run(
                ['git', 'branch', '-a'],
                cwd=self.project_root,
                capture_output=True,
                text=True,
                check=False
            )
            
            if result.returncode != 0:
                return []
            
            branches = []
            for line in result.stdout.strip().split('\n'):
                if line.strip():
                    # Clean up branch name
                    branch = line.strip().replace('* ', '').replace('remotes/origin/', '')
                    if branch.startswith(self.prefix):
                        branches.append(branch)
            
            # Sort by timestamp (extract from branch name)
            # Format: backup-YYYY-MM-DD-HH
            branches.sort(key=lambda x: x.replace(self.prefix, ''), reverse=True)
            return branches
            
        except Exception as e:
            self.logger.error(f"Error getting backup branches: {e}")
            return []
    
    def _delete_local_branch(self, branch_name: str) -> bool:
        """Delete local branch"""
        try:
            result = subprocess.run(
                ['git', 'branch', '-D', branch_name],
                cwd=self.project_root,
                capture_output=True,
                text=True,
                check=False
            )
            
            if result.returncode == 0:
                return True
            else:
                self.logger.debug(f"Local delete failed for {branch_name}: {result.stderr}")
                return False
                
        except Exception as e:
            self.logger.error(f"Exception deleting local branch {branch_name}: {e}")
            return False
    
    def _delete_remote_branch(self, branch_name: str) -> bool:
        """Delete remote branch"""
        try:
            result = subprocess.run(
                ['git', 'push', 'origin', '--delete', branch_name],
                cwd=self.project_root,
                capture_output=True,
                text=True,
                check=False
            )
            
            if result.returncode == 0:
                return True
            else:
                # Might already be deleted remotely, that's ok
                if "not found" in result.stderr.lower() or "does not exist" in result.stderr.lower():
                    return True
                self.logger.debug(f"Remote delete failed for {branch_name}: {result.stderr}")
                return False
                
        except Exception as e:
            self.logger.error(f"Exception deleting remote branch {branch_name}: {e}")
            return False
    
    def get_cleanup_stats(self) -> Dict[str, Any]:
        """Get statistics about backup cleanup"""
        backup_branches = self._get_backup_branches()
        current_count = len(backup_branches)
        
        return {
            'current_backup_count': current_count,
            'max_allowed': self.max_backups,
            'cleanup_needed': current_count > self.max_backups,
            'branches_to_delete': max(0, current_count - self.max_backups),
            'newest_backup': backup_branches[0] if backup_branches else None,
            'oldest_backup': backup_branches[-1] if backup_branches else None
        }