""" Django management command to migrate existing permissions to the enhanced granular system """ from django.core.management.base import BaseCommand, CommandError from django.db import transaction, connection from django.utils import timezone from users.models import CustomUser, RolePermission, DefaultRolePermission, UserPermission from users.enhanced_permissions_models import PagePermission, RolePermissionTemplate, UserPagePermission import json import logging logger = logging.getLogger(__name__) class Command(BaseCommand): help = 'Migrate existing permissions to enhanced granular permission system' def add_arguments(self, parser): parser.add_argument( '--dry-run', action='store_true', help='Run migration in dry-run mode without making changes', ) parser.add_argument( '--backup', action='store_true', help='Create backup of existing permissions before migration', ) parser.add_argument( '--force', action='store_true', help='Force migration even if granular permissions already exist', ) parser.add_argument( '--validate-only', action='store_true', help='Only validate data integrity without migrating', ) def handle(self, *args, **options): self.dry_run = options['dry_run'] self.create_backup = options['backup'] self.force = options['force'] self.validate_only = options['validate_only'] self.stdout.write( self.style.SUCCESS('Starting permission migration to granular system...') ) try: if self.validate_only: self.validate_data_integrity() return if self.create_backup: self.create_permission_backup() if not self.force and self.check_existing_granular_permissions(): self.stdout.write( self.style.WARNING( 'Granular permissions already exist. Use --force to override.' ) ) return with transaction.atomic(): # Step 1: Create page permissions from existing module/action combinations self.create_page_permissions() # Step 2: Migrate role permissions to role permission templates self.migrate_role_permissions() # Step 3: Migrate user permissions to user page permissions self.migrate_user_permissions() # Step 4: Validate migration self.validate_migration() if self.dry_run: self.stdout.write( self.style.WARNING('DRY RUN: Rolling back transaction...') ) raise transaction.TransactionManagementError("Dry run - rolling back") self.stdout.write( self.style.SUCCESS('Permission migration completed successfully!') ) except Exception as e: self.stdout.write( self.style.ERROR(f'Migration failed: {str(e)}') ) if not self.dry_run: logger.error(f'Permission migration failed: {str(e)}', exc_info=True) raise CommandError(f'Migration failed: {str(e)}') def check_existing_granular_permissions(self): """Check if granular permissions already exist""" return PagePermission.objects.exists() def create_permission_backup(self): """Create backup of existing permissions""" backup_data = { 'timestamp': timezone.now().isoformat(), 'role_permissions': [], 'default_role_permissions': [], 'user_permissions': [] } # Backup role permissions for perm in RolePermission.objects.all(): backup_data['role_permissions'].append({ 'role': perm.role, 'module': perm.module, 'action': perm.action, 'is_allowed': perm.is_allowed, 'created_at': perm.created_at.isoformat(), 'updated_at': perm.updated_at.isoformat(), }) # Backup default role permissions for perm in DefaultRolePermission.objects.all(): backup_data['default_role_permissions'].append({ 'role': perm.role, 'module': perm.module, 'action': perm.action, 'is_allowed': perm.is_allowed, 'description': perm.description, 'created_at': perm.created_at.isoformat(), 'updated_at': perm.updated_at.isoformat(), }) # Backup user permissions for perm in UserPermission.objects.all(): backup_data['user_permissions'].append({ 'user_id': str(perm.user.id), 'user_username': perm.user.username, 'module': perm.module, 'action': perm.action, 'is_allowed': perm.is_allowed, 'granted_by_id': str(perm.granted_by.id) if perm.granted_by else None, 'reason': perm.reason, 'expires_at': perm.expires_at.isoformat() if perm.expires_at else None, 'created_at': perm.created_at.isoformat(), 'updated_at': perm.updated_at.isoformat(), }) # Save backup to file backup_filename = f'permission_backup_{timezone.now().strftime("%Y%m%d_%H%M%S")}.json' with open(backup_filename, 'w') as f: json.dump(backup_data, f, indent=2) self.stdout.write( self.style.SUCCESS(f'Permission backup created: {backup_filename}') ) return backup_filename def create_page_permissions(self): """Create PagePermission objects from existing module/action combinations""" self.stdout.write('Creating page permissions...') # Get unique module/action combinations from existing permissions existing_combinations = set() for perm in RolePermission.objects.all(): existing_combinations.add((perm.module, perm.action)) for perm in DefaultRolePermission.objects.all(): existing_combinations.add((perm.module, perm.action)) for perm in UserPermission.objects.all(): existing_combinations.add((perm.module, perm.action)) # Create PagePermission objects created_count = 0 for module, action in existing_combinations: page_name = self.extract_page_name(module) action_code = f"{module}_{action}" action_name = self.generate_action_name(module, action) description = self.generate_description(module, action) category = self.determine_category(action) page_permission, created = PagePermission.objects.get_or_create( page_name=page_name, action_code=action_code, defaults={ 'action_name': action_name, 'description': description, 'category': category, 'is_critical': self.is_critical_permission(module, action), 'is_active': True, } ) if created: created_count += 1 if not self.dry_run: self.stdout.write(f' Created: {page_permission}') self.stdout.write( self.style.SUCCESS(f'Created {created_count} page permissions') ) def migrate_role_permissions(self): """Migrate RolePermission and DefaultRolePermission to RolePermissionTemplate""" self.stdout.write('Migrating role permissions...') # Migrate DefaultRolePermission first (these are the templates) migrated_count = 0 for perm in DefaultRolePermission.objects.all(): try: page_permission = PagePermission.objects.get( action_code=f"{perm.module}_{perm.action}" ) template, created = RolePermissionTemplate.objects.get_or_create( role=perm.role, page_permission=page_permission, defaults={ 'is_allowed': perm.is_allowed, 'can_override': True, # Default to allowing overrides 'priority': 0, 'created_by': None, } ) if created: migrated_count += 1 if not self.dry_run: self.stdout.write(f' Migrated template: {template}') except PagePermission.DoesNotExist: self.stdout.write( self.style.WARNING( f'PagePermission not found for {perm.module}_{perm.action}' ) ) # If no DefaultRolePermission exists, migrate from RolePermission if DefaultRolePermission.objects.count() == 0: for perm in RolePermission.objects.all(): try: page_permission = PagePermission.objects.get( action_code=f"{perm.module}_{perm.action}" ) template, created = RolePermissionTemplate.objects.get_or_create( role=perm.role, page_permission=page_permission, defaults={ 'is_allowed': perm.is_allowed, 'can_override': True, 'priority': 0, 'created_by': None, } ) if created: migrated_count += 1 if not self.dry_run: self.stdout.write(f' Migrated from role perm: {template}') except PagePermission.DoesNotExist: self.stdout.write( self.style.WARNING( f'PagePermission not found for {perm.module}_{perm.action}' ) ) self.stdout.write( self.style.SUCCESS(f'Migrated {migrated_count} role permission templates') ) def migrate_user_permissions(self): """Migrate UserPermission to UserPagePermission""" self.stdout.write('Migrating user permissions...') migrated_count = 0 for perm in UserPermission.objects.all(): try: page_permission = PagePermission.objects.get( action_code=f"{perm.module}_{perm.action}" ) user_perm, created = UserPagePermission.objects.get_or_create( user=perm.user, page_permission=page_permission, defaults={ 'is_allowed': perm.is_allowed, 'granted_by': perm.granted_by, 'reason': perm.reason or 'Migrated from legacy system', 'expires_at': perm.expires_at, } ) if created: migrated_count += 1 if not self.dry_run: self.stdout.write(f' Migrated user perm: {user_perm}') except PagePermission.DoesNotExist: self.stdout.write( self.style.WARNING( f'PagePermission not found for {perm.module}_{perm.action}' ) ) self.stdout.write( self.style.SUCCESS(f'Migrated {migrated_count} user permissions') ) def validate_migration(self): """Validate the migration was successful""" self.stdout.write('Validating migration...') # Count original permissions original_role_perms = RolePermission.objects.count() original_default_perms = DefaultRolePermission.objects.count() original_user_perms = UserPermission.objects.count() # Count migrated permissions migrated_page_perms = PagePermission.objects.count() migrated_role_templates = RolePermissionTemplate.objects.count() migrated_user_perms = UserPagePermission.objects.count() self.stdout.write(f'Original permissions:') self.stdout.write(f' Role permissions: {original_role_perms}') self.stdout.write(f' Default role permissions: {original_default_perms}') self.stdout.write(f' User permissions: {original_user_perms}') self.stdout.write(f'Migrated permissions:') self.stdout.write(f' Page permissions: {migrated_page_perms}') self.stdout.write(f' Role templates: {migrated_role_templates}') self.stdout.write(f' User page permissions: {migrated_user_perms}') # Validate data integrity self.validate_data_integrity() def validate_data_integrity(self): """Validate data integrity of the migration""" self.stdout.write('Validating data integrity...') issues = [] # Check for orphaned permissions for template in RolePermissionTemplate.objects.all(): if not template.page_permission: issues.append(f'Role template {template.id} has no page permission') for user_perm in UserPagePermission.objects.all(): if not user_perm.page_permission: issues.append(f'User permission {user_perm.id} has no page permission') if not user_perm.user: issues.append(f'User permission {user_perm.id} has no user') # Check for duplicate permissions duplicates = PagePermission.objects.values('page_name', 'action_code').annotate( count=models.Count('id') ).filter(count__gt=1) for dup in duplicates: issues.append( f'Duplicate page permission: {dup["page_name"]} - {dup["action_code"]}' ) if issues: self.stdout.write(self.style.WARNING('Data integrity issues found:')) for issue in issues: self.stdout.write(f' - {issue}') else: self.stdout.write(self.style.SUCCESS('Data integrity validation passed')) def extract_page_name(self, module): """Extract page name from module""" # Map modules to page names page_mapping = { 'dashboard': 'dashboard', 'dashboard_overview': 'dashboard', 'dashboard_metrics': 'dashboard', 'dashboard_charts': 'dashboard', 'dashboard_recent_activities': 'dashboard', 'dashboard_quick_actions': 'dashboard', 'dashboard_loan_summary': 'dashboard', 'dashboard_collection_summary': 'dashboard', 'dashboard_branch_performance': 'dashboard', 'dashboard_alerts': 'dashboard', 'clients': 'clients', 'clients_view_list': 'clients', 'clients_search_filter': 'clients', 'clients_create_new': 'clients', 'clients_edit_profile': 'clients', 'clients_delete_client': 'clients', 'clients_view_details': 'clients', 'clients_loan_history': 'clients', 'clients_payment_history': 'clients', 'clients_kyc_documents': 'clients', 'clients_assign_portfolio': 'clients', 'clients_registration_fee': 'clients', 'clients_status_change': 'clients', 'clients_export_data': 'clients', 'clients_bulk_actions': 'clients', 'loans': 'loans', 'loans_view_list': 'loans', 'loans_search_filter': 'loans', 'loans_create_application': 'loans', 'loans_edit_application': 'loans', 'loans_delete_loan': 'loans', 'loans_approve_application': 'loans', 'loans_reject_application': 'loans', 'loans_disburse_funds': 'loans', 'loans_view_details': 'loans', 'loans_amortization_schedule': 'loans', 'loans_rollover_loan': 'loans', 'loans_calculate_interest': 'loans', 'loans_generate_receipt': 'loans', 'loans_modify_terms': 'loans', 'loans_close_loan': 'loans', 'loans_export_data': 'loans', 'repayments': 'repayments', 'repayments_view_list': 'repayments', 'repayments_search_filter': 'repayments', 'repayments_record_payment': 'repayments', 'repayments_edit_payment': 'repayments', 'repayments_delete_payment': 'repayments', 'repayments_verify_payment': 'repayments', 'repayments_reconcile_mpesa': 'repayments', 'repayments_generate_receipt': 'repayments', 'repayments_bulk_import': 'repayments', 'repayments_export_data': 'repayments', 'repayments_view_analytics': 'repayments', 'portfolio': 'portfolio', 'portfolio_view_overview': 'portfolio', 'portfolio_manager_stats': 'portfolio', 'portfolio_performance_metrics': 'portfolio', 'portfolio_client_assignments': 'portfolio', 'portfolio_reassign_clients': 'portfolio', 'portfolio_view_analytics': 'portfolio', 'portfolio_generate_reports': 'portfolio', 'portfolio_target_tracking': 'portfolio', 'reports_statements': 'reports', 'reports_view_dashboard': 'reports', 'reports_loans_due': 'reports', 'reports_delinquent_loans': 'reports', 'reports_arrears_analysis': 'reports', 'reports_processing_fees': 'reports', 'reports_interest_income': 'reports', 'reports_registration_fees': 'reports', 'reports_customer_requests': 'reports', 'reports_collection_summary': 'reports', 'reports_branch_performance': 'reports', 'reports_export_all': 'reports', 'reports_schedule_automated': 'reports', 'statements_generate_loan': 'reports', 'statements_generate_client': 'reports', 'statements_download_pdf': 'reports', 'statements_email_client': 'reports', 'documents': 'documents', 'documents_view_library': 'documents', 'documents_upload_files': 'documents', 'documents_download_files': 'documents', 'documents_delete_files': 'documents', 'documents_organize_folders': 'documents', 'documents_search_content': 'documents', 'documents_share_access': 'documents', 'documents_version_control': 'documents', 'customer_documents': 'documents', 'customer_docs_view_all': 'documents', 'customer_docs_upload_kyc': 'documents', 'customer_docs_verify_identity': 'documents', 'customer_docs_approve_kyc': 'documents', 'customer_docs_reject_kyc': 'documents', 'customer_docs_request_additional': 'documents', 'customer_docs_download_files': 'documents', 'customer_docs_compliance_check': 'documents', 'payment_receipts': 'receipts', 'receipts_view_all': 'receipts', 'receipts_generate_new': 'receipts', 'receipts_edit_existing': 'receipts', 'receipts_delete_receipt': 'receipts', 'receipts_print_receipt': 'receipts', 'receipts_download_pdf': 'receipts', 'receipts_email_client': 'receipts', 'receipts_bulk_generate': 'receipts', 'receipts_template_manage': 'receipts', 'notifications': 'notifications', 'notifications_view_inbox': 'notifications', 'notifications_send_individual': 'notifications', 'notifications_send_bulk': 'notifications', 'notifications_manage_templates': 'notifications', 'notifications_configure_settings': 'notifications', 'notifications_view_history': 'notifications', 'notifications_manage_channels': 'notifications', 'settings': 'settings', 'settings_view_general': 'settings', 'settings_edit_system': 'settings', 'settings_manage_users': 'settings', 'settings_configure_permissions': 'settings', 'settings_backup_restore': 'settings', 'settings_integration_config': 'settings', 'settings_audit_logs': 'settings', 'branch_settings': 'settings', 'branch_view_info': 'settings', 'branch_edit_details': 'settings', 'branch_manage_staff': 'settings', 'branch_configure_mpesa': 'settings', 'branch_view_performance': 'settings', 'branch_manage_targets': 'settings', 'system_settings': 'settings', 'system_database_management': 'settings', 'system_server_monitoring': 'settings', 'system_security_settings': 'settings', 'system_maintenance_mode': 'settings', 'system_api_management': 'settings', 'system_integration_logs': 'settings', } return page_mapping.get(module, module.split('_')[0] if '_' in module else module) def generate_action_name(self, module, action): """Generate human-readable action name""" # Clean up module name module_clean = module.replace('_', ' ').title() action_clean = action.replace('_', ' ').title() # Create descriptive name if action == 'access': return f"Access {module_clean}" elif action == 'create': return f"Create {module_clean}" elif action == 'edit': return f"Edit {module_clean}" elif action == 'delete': return f"Delete {module_clean}" elif action == 'export': return f"Export {module_clean}" else: return f"{action_clean} {module_clean}" def generate_description(self, module, action): """Generate description for the permission""" return f"Allows user to {action} {module.replace('_', ' ')}" def determine_category(self, action): """Determine category based on action""" category_mapping = { 'access': 'view', 'view': 'view', 'create': 'create', 'add': 'create', 'edit': 'edit', 'modify': 'edit', 'update': 'edit', 'delete': 'delete', 'remove': 'delete', 'approve': 'approve', 'reject': 'approve', 'verify': 'approve', 'export': 'export', 'download': 'export', 'print': 'export', 'manage': 'manage', 'configure': 'manage', 'assign': 'manage', 'process': 'process', 'calculate': 'process', 'generate': 'process', } return category_mapping.get(action, 'manage') def is_critical_permission(self, module, action): """Determine if this is a critical permission""" critical_combinations = [ ('settings', 'delete'), ('system_settings', 'edit'), ('users', 'delete'), ('permissions', 'edit'), ('database', 'delete'), ('backup', 'restore'), ] return (module, action) in critical_combinations