#!/usr/bin/env python3 """ Backup and Analysis Infrastructure for Codebase Cleanup This script provides: 1. Timestamped backup creation of entire codebase 2. File scanning utility to categorize files by type 3. Dependency tracking system to map import relationships Requirements: 1.1, 1.2 """ import ast from datetime import datetime from enum import Enum import json from pathlib import Path import re import shutil from typing import Dict, List, Optional, Set, Tuple from dataclasses import asdict, dataclass class FileType(Enum): PYTHON_SOURCE = "python_source" PYTHON_TEST = "python_test" PYTHON_CACHE = "python_cache" DOCUMENTATION = "documentation" CONFIGURATION = "configuration" MIGRATION = "migration" BACKUP = "backup" DEPLOYMENT = "deployment" DATA = "data" UNKNOWN = "unknown" class CleanupAction(Enum): KEEP = "keep" MOVE = "move" ARCHIVE = "archive" DELETE = "delete" CONSOLIDATE = "consolidate" RENAME = "rename" @dataclass class FileMetadata: path: str size: int last_modified: str file_type: FileType dependencies: List[str] is_redundant: bool suggested_action: CleanupAction def to_dict(self): return { 'path': self.path, 'size': self.size, 'last_modified': self.last_modified, 'file_type': self.file_type.value, 'dependencies': self.dependencies, 'is_redundant': self.is_redundant, 'suggested_action': self.suggested_action.value } class CodebaseAnalyzer: """Analyzes codebase structure and dependencies""" def __init__(self, root_path: str = "."): self.root_path = Path(root_path).resolve() self.file_metadata: Dict[str, FileMetadata] = {} self.import_graph: Dict[str, Set[str]] = {} def create_timestamped_backup(self) -> str: """Create a timestamped backup of the entire codebase""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_name = f"codebase_backup_{timestamp}" backup_path = self.root_path / "backups" / backup_name # Create backups directory if it doesn't exist backup_path.parent.mkdir(exist_ok=True) print(f"Creating backup: {backup_path}") # Copy entire codebase excluding certain directories exclude_patterns = { '__pycache__', '.git', 'node_modules', '.venv', 'venv', 'atlas_env', 'backups' # Don't backup previous backups } def ignore_patterns(dir_path, names): return [name for name in names if name in exclude_patterns] shutil.copytree( self.root_path, backup_path, ignore=ignore_patterns ) # Create backup manifest manifest = { 'timestamp': timestamp, 'backup_path': str(backup_path), 'source_path': str(self.root_path), 'excluded_patterns': list(exclude_patterns) } manifest_path = backup_path / "backup_manifest.json" with open(manifest_path, 'w') as f: json.dump(manifest, f, indent=2) print(f"Backup created successfully: {backup_path}") return str(backup_path) def categorize_file_type(self, file_path: Path) -> FileType: """Categorize file by type based on path and content""" path_str = str(file_path).lower() name = file_path.name.lower() # Python cache files if '__pycache__' in path_str or name.endswith('.pyc'): return FileType.PYTHON_CACHE # Python test files if (name.startswith('test_') or name.endswith('_test.py') or 'test' in path_str and name.endswith('.py')): return FileType.PYTHON_TEST # Python source files if name.endswith('.py'): # Check for migration patterns if any(keyword in name for keyword in ['migrate', 'migration', 'rollback']): return FileType.MIGRATION return FileType.PYTHON_SOURCE # Documentation files if (name.endswith(('.md', '.rst', '.txt')) or name in ['readme', 'changelog', 'license']): return FileType.DOCUMENTATION # Configuration files if (name.endswith(('.json', '.yaml', '.yml', '.toml', '.ini', '.cfg')) or name in ['.gitignore', '.gitattributes', 'dockerfile', '.env']): return FileType.CONFIGURATION # Backup files if 'backup' in path_str or name.endswith('.bak'): return FileType.BACKUP # Deployment files if (name in ['start.sh', 'requirements.txt', 'dockerfile'] or 'deploy' in path_str): return FileType.DEPLOYMENT # Data files if name.endswith(('.db', '.sqlite', '.json')) and 'data' in path_str: return FileType.DATA return FileType.UNKNOWN def extract_python_imports(self, file_path: Path) -> List[str]: """Extract import statements from Python files""" imports = [] try: with open(file_path, 'r', encoding='utf-8') as f: content = f.read() # Parse AST to extract imports try: tree = ast.parse(content) for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: imports.append(alias.name) elif isinstance(node, ast.ImportFrom): if node.module: imports.append(node.module) except SyntaxError: # Fallback to regex if AST parsing fails import_patterns = [ r'^\s*import\s+([^\s#]+)', r'^\s*from\s+([^\s#]+)\s+import' ] for line in content.split('\n'): for pattern in import_patterns: match = re.match(pattern, line) if match: imports.append(match.group(1)) except Exception as e: print(f"Error reading {file_path}: {e}") return imports def suggest_cleanup_action(self, metadata: FileMetadata) -> CleanupAction: """Suggest cleanup action based on file metadata""" file_type = metadata.file_type path = metadata.path # Cache files should be deleted if file_type == FileType.PYTHON_CACHE: return CleanupAction.DELETE # Migration files should be archived if file_type == FileType.MIGRATION: return CleanupAction.ARCHIVE # Backup files can be deleted if old if file_type == FileType.BACKUP: return CleanupAction.DELETE # Test files might need consolidation if file_type == FileType.PYTHON_TEST: # Check for potential duplicates based on naming if any(keyword in path.lower() for keyword in ['duplicate', 'old', 'backup']): return CleanupAction.CONSOLIDATE return CleanupAction.KEEP # Keep most other files by default return CleanupAction.KEEP def scan_codebase(self) -> Dict[str, FileMetadata]: """Scan entire codebase and categorize all files""" print("Scanning codebase...") for file_path in self.root_path.rglob('*'): if file_path.is_file(): try: # Get file stats stat = file_path.stat() relative_path = str(file_path.relative_to(self.root_path)) # Categorize file type file_type = self.categorize_file_type(file_path) # Extract dependencies for Python files dependencies = [] if file_type in [FileType.PYTHON_SOURCE, FileType.PYTHON_TEST]: dependencies = self.extract_python_imports(file_path) # Create metadata metadata = FileMetadata( path=relative_path, size=stat.st_size, last_modified=datetime.fromtimestamp(stat.st_mtime).isoformat(), file_type=file_type, dependencies=dependencies, is_redundant=False, # Will be determined later suggested_action=CleanupAction.KEEP # Will be determined later ) # Suggest cleanup action metadata.suggested_action = self.suggest_cleanup_action(metadata) self.file_metadata[relative_path] = metadata except Exception as e: print(f"Error processing {file_path}: {e}") print(f"Scanned {len(self.file_metadata)} files") return self.file_metadata def build_dependency_graph(self) -> Dict[str, Set[str]]: """Build dependency graph from import relationships""" print("Building dependency graph...") # Map module names to file paths module_to_file = {} for file_path, metadata in self.file_metadata.items(): if metadata.file_type in [FileType.PYTHON_SOURCE, FileType.PYTHON_TEST]: # Convert file path to module name module_name = file_path.replace('/', '.').replace('.py', '') module_to_file[module_name] = file_path # Build dependency graph for file_path, metadata in self.file_metadata.items(): if metadata.dependencies: self.import_graph[file_path] = set() for import_name in metadata.dependencies: # Check if it's a local import if import_name in module_to_file: self.import_graph[file_path].add(module_to_file[import_name]) else: # Check for partial matches (submodules) for module_name, module_file in module_to_file.items(): if import_name.startswith(module_name): self.import_graph[file_path].add(module_file) print(f"Built dependency graph with {len(self.import_graph)} nodes") return self.import_graph def generate_analysis_report(self, output_path: str = "analysis_report.json"): """Generate comprehensive analysis report""" print("Generating analysis report...") # File type summary type_counts = {} for metadata in self.file_metadata.values(): file_type = metadata.file_type.value type_counts[file_type] = type_counts.get(file_type, 0) + 1 # Action summary action_counts = {} for metadata in self.file_metadata.values(): action = metadata.suggested_action.value action_counts[action] = action_counts.get(action, 0) + 1 # Dependency analysis dependency_stats = { 'total_files_with_dependencies': len(self.import_graph), 'average_dependencies_per_file': ( sum(len(deps) for deps in self.import_graph.values()) / len(self.import_graph) if self.import_graph else 0 ) } report = { 'timestamp': datetime.now().isoformat(), 'total_files': len(self.file_metadata), 'file_type_counts': type_counts, 'suggested_action_counts': action_counts, 'dependency_stats': dependency_stats, 'files': [metadata.to_dict() for metadata in self.file_metadata.values()], 'dependency_graph': { file_path: list(deps) for file_path, deps in self.import_graph.items() } } with open(output_path, 'w') as f: json.dump(report, f, indent=2) print(f"Analysis report saved to: {output_path}") return report def main(): """Main function to run backup and analysis""" print("Starting codebase backup and analysis...") analyzer = CodebaseAnalyzer() # Create backup backup_path = analyzer.create_timestamped_backup() # Scan codebase file_metadata = analyzer.scan_codebase() # Build dependency graph dependency_graph = analyzer.build_dependency_graph() # Generate report report = analyzer.generate_analysis_report() print("\n=== Analysis Summary ===") print(f"Total files analyzed: {len(file_metadata)}") print(f"Files with dependencies: {len(dependency_graph)}") print(f"Backup created at: {backup_path}") print("Analysis report saved to: analysis_report.json") # Print file type breakdown print("\n=== File Type Breakdown ===") type_counts = {} for metadata in file_metadata.values(): file_type = metadata.file_type.value type_counts[file_type] = type_counts.get(file_type, 0) + 1 for file_type, count in sorted(type_counts.items()): print(f"{file_type}: {count}") # Print suggested actions print("\n=== Suggested Actions ===") action_counts = {} for metadata in file_metadata.values(): action = metadata.suggested_action.value action_counts[action] = action_counts.get(action, 0) + 1 for action, count in sorted(action_counts.items()): print(f"{action}: {count}") if __name__ == "__main__": main()