"""Maintenance Mode Management Command This command enables or disables maintenance mode for the Adtlas platform: - Enable maintenance mode to block user access during updates - Disable maintenance mode to restore normal operation - Check current maintenance mode status - Schedule maintenance windows - Send notifications about maintenance - Graceful handling of active sessions Usage: python manage.py maintenance_mode [options] Options: --enable: Enable maintenance mode --disable: Disable maintenance mode --status: Check current maintenance mode status --message: Custom maintenance message --duration: Maintenance duration in minutes --schedule: Schedule maintenance for future time --notify: Send notifications to users --force: Force maintenance mode without confirmation --template: Custom maintenance template Examples: python manage.py maintenance_mode --enable python manage.py maintenance_mode --disable python manage.py maintenance_mode --status python manage.py maintenance_mode --enable --message "System upgrade in progress" python manage.py maintenance_mode --enable --duration 60 --notify python manage.py maintenance_mode --schedule "2024-01-15 02:00" --duration 120 """ import json import os from datetime import datetime, timedelta from typing import Dict, Optional from django.core.management.base import BaseCommand, CommandError from django.conf import settings from django.utils import timezone from django.core.cache import cache from django.core.mail import send_mass_mail from django.template.loader import render_to_string from django.contrib.auth import get_user_model from apps.core.models import SystemConfiguration, AuditLog from apps.core.logging import get_logger from apps.accounts.models import UserSession User = get_user_model() class Command(BaseCommand): """Maintenance Mode Command""" help = 'Enable, disable, or check maintenance mode status' def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.logger = get_logger('core.management.maintenance_mode') self.maintenance_file = os.path.join( settings.BASE_DIR, '.maintenance' ) self.config_key = 'maintenance_mode' def add_arguments(self, parser): """Add command arguments""" parser.add_argument( '--enable', action='store_true', help='Enable maintenance mode' ) parser.add_argument( '--disable', action='store_true', help='Disable maintenance mode' ) parser.add_argument( '--status', action='store_true', help='Check current maintenance mode status' ) parser.add_argument( '--message', type=str, default='System maintenance in progress. Please try again later.', help='Custom maintenance message' ) parser.add_argument( '--duration', type=int, help='Maintenance duration in minutes' ) parser.add_argument( '--schedule', type=str, help='Schedule maintenance for future time (YYYY-MM-DD HH:MM)' ) parser.add_argument( '--notify', action='store_true', help='Send notifications to users' ) parser.add_argument( '--force', action='store_true', help='Force maintenance mode without confirmation' ) parser.add_argument( '--template', type=str, help='Custom maintenance template' ) parser.add_argument( '--allowed-ips', type=str, help='Comma-separated list of allowed IP addresses' ) parser.add_argument( '--allowed-users', type=str, help='Comma-separated list of allowed usernames' ) def handle(self, *args, **options): """Main command handler""" try: enable = options.get('enable', False) disable = options.get('disable', False) status = options.get('status', False) schedule = options.get('schedule') # Validate arguments if sum([enable, disable, status, bool(schedule)]) != 1: raise CommandError( "Specify exactly one action: --enable, --disable, --status, or --schedule" ) if status: self.check_status() elif enable: self.enable_maintenance_mode(options) elif disable: self.disable_maintenance_mode(options) elif schedule: self.schedule_maintenance(options) except Exception as e: self.logger.error(f"Maintenance mode command failed: {str(e)}") raise CommandError(f"Command failed: {str(e)}") def check_status(self): """Check current maintenance mode status""" try: is_enabled = self.is_maintenance_enabled() if is_enabled: config = self.get_maintenance_config() self.stdout.write("Maintenance mode: ENABLED") if config: self.stdout.write(f"Message: {config.get('message', 'N/A')}") self.stdout.write(f"Started: {config.get('start_time', 'N/A')}") if config.get('end_time'): self.stdout.write(f"Scheduled end: {config.get('end_time')}") if config.get('allowed_ips'): self.stdout.write(f"Allowed IPs: {', '.join(config['allowed_ips'])}") if config.get('allowed_users'): self.stdout.write(f"Allowed users: {', '.join(config['allowed_users'])}") else: self.stdout.write("Maintenance mode: DISABLED") # Check for scheduled maintenance scheduled = self.get_scheduled_maintenance() if scheduled: self.stdout.write(f"Scheduled maintenance: {scheduled['start_time']}") self.stdout.write(f"Duration: {scheduled['duration']} minutes") except Exception as e: self.logger.error(f"Failed to check maintenance status: {str(e)}") raise CommandError(f"Status check failed: {str(e)}") def enable_maintenance_mode(self, options): """Enable maintenance mode""" try: if self.is_maintenance_enabled(): self.stdout.write("Maintenance mode is already enabled") return force = options.get('force', False) message = options.get('message') duration = options.get('duration') notify = options.get('notify', False) template = options.get('template') allowed_ips = self.parse_list(options.get('allowed_ips')) allowed_users = self.parse_list(options.get('allowed_users')) # Confirm if not forced if not force: if not self.confirm_enable(): self.stdout.write("Maintenance mode activation cancelled") return # Prepare maintenance configuration config = { 'enabled': True, 'message': message, 'start_time': timezone.now().isoformat(), 'template': template or 'core/maintenance.html', 'allowed_ips': allowed_ips or [], 'allowed_users': allowed_users or [], } if duration: end_time = timezone.now() + timedelta(minutes=duration) config['end_time'] = end_time.isoformat() config['duration'] = duration # Enable maintenance mode self.set_maintenance_config(config) self.create_maintenance_file(config) # Log the action self.log_maintenance_action('enabled', config) # Send notifications if requested if notify: self.send_maintenance_notifications(config, 'started') # Handle active sessions self.handle_active_sessions() self.stdout.write("Maintenance mode ENABLED") if duration: self.stdout.write(f"Scheduled to end in {duration} minutes") self.logger.info("Maintenance mode enabled") except Exception as e: self.logger.error(f"Failed to enable maintenance mode: {str(e)}") raise CommandError(f"Enable failed: {str(e)}") def disable_maintenance_mode(self, options): """Disable maintenance mode""" try: if not self.is_maintenance_enabled(): self.stdout.write("Maintenance mode is already disabled") return force = options.get('force', False) notify = options.get('notify', False) # Confirm if not forced if not force: if not self.confirm_disable(): self.stdout.write("Maintenance mode deactivation cancelled") return # Get current config for notifications config = self.get_maintenance_config() # Disable maintenance mode self.remove_maintenance_config() self.remove_maintenance_file() # Log the action self.log_maintenance_action('disabled', config) # Send notifications if requested if notify and config: self.send_maintenance_notifications(config, 'ended') self.stdout.write("Maintenance mode DISABLED") self.logger.info("Maintenance mode disabled") except Exception as e: self.logger.error(f"Failed to disable maintenance mode: {str(e)}") raise CommandError(f"Disable failed: {str(e)}") def schedule_maintenance(self, options): """Schedule future maintenance""" try: schedule_str = options.get('schedule') duration = options.get('duration', 60) message = options.get('message') notify = options.get('notify', False) # Parse schedule time try: schedule_time = datetime.strptime(schedule_str, '%Y-%m-%d %H:%M') schedule_time = timezone.make_aware(schedule_time) except ValueError: raise CommandError( "Invalid schedule format. Use: YYYY-MM-DD HH:MM" ) # Validate schedule time if schedule_time <= timezone.now(): raise CommandError("Schedule time must be in the future") # Prepare scheduled maintenance config scheduled_config = { 'start_time': schedule_time.isoformat(), 'duration': duration, 'message': message, 'notify': notify, } # Save scheduled maintenance self.save_scheduled_maintenance(scheduled_config) # Log the action self.log_maintenance_action('scheduled', scheduled_config) # Send advance notifications if requested if notify: self.send_maintenance_notifications(scheduled_config, 'scheduled') self.stdout.write(f"Maintenance scheduled for {schedule_time}") self.stdout.write(f"Duration: {duration} minutes") self.logger.info(f"Maintenance scheduled for {schedule_time}") except Exception as e: self.logger.error(f"Failed to schedule maintenance: {str(e)}") raise CommandError(f"Schedule failed: {str(e)}") def is_maintenance_enabled(self): """Check if maintenance mode is currently enabled""" # Check file-based maintenance mode if os.path.exists(self.maintenance_file): return True # Check database-based maintenance mode try: config = SystemConfiguration.objects.filter( key=self.config_key, is_active=True ).first() if config: config_data = config.value return config_data.get('enabled', False) except Exception: pass # Check cache-based maintenance mode return cache.get('maintenance_mode_enabled', False) def get_maintenance_config(self): """Get current maintenance configuration""" try: # Try database first config = SystemConfiguration.objects.filter( key=self.config_key, is_active=True ).first() if config: return config.value # Try cache cached_config = cache.get('maintenance_mode_config') if cached_config: return cached_config # Try file if os.path.exists(self.maintenance_file): with open(self.maintenance_file, 'r') as f: return json.load(f) except Exception as e: self.logger.error(f"Failed to get maintenance config: {str(e)}") return None def set_maintenance_config(self, config): """Set maintenance configuration""" try: # Save to database config_obj, created = SystemConfiguration.objects.get_or_create( key=self.config_key, defaults={'value': config, 'is_active': True} ) if not created: config_obj.value = config config_obj.is_active = True config_obj.save() # Save to cache cache.set('maintenance_mode_enabled', True, timeout=None) cache.set('maintenance_mode_config', config, timeout=None) except Exception as e: self.logger.error(f"Failed to set maintenance config: {str(e)}") raise def remove_maintenance_config(self): """Remove maintenance configuration""" try: # Remove from database SystemConfiguration.objects.filter( key=self.config_key ).update(is_active=False) # Remove from cache cache.delete('maintenance_mode_enabled') cache.delete('maintenance_mode_config') except Exception as e: self.logger.error(f"Failed to remove maintenance config: {str(e)}") def create_maintenance_file(self, config): """Create maintenance file""" try: with open(self.maintenance_file, 'w') as f: json.dump(config, f, indent=2) except Exception as e: self.logger.error(f"Failed to create maintenance file: {str(e)}") def remove_maintenance_file(self): """Remove maintenance file""" try: if os.path.exists(self.maintenance_file): os.remove(self.maintenance_file) except Exception as e: self.logger.error(f"Failed to remove maintenance file: {str(e)}") def handle_active_sessions(self): """Handle active user sessions during maintenance""" try: # Mark active sessions for maintenance notification active_sessions = UserSession.objects.filter( is_active=True, last_activity__gte=timezone.now() - timedelta(minutes=30) ) count = active_sessions.count() if count > 0: self.stdout.write(f"Found {count} active sessions") # You could implement graceful session handling here # For example, send WebSocket notifications to active users except Exception as e: self.logger.error(f"Failed to handle active sessions: {str(e)}") def send_maintenance_notifications(self, config, event_type): """Send maintenance notifications to users""" try: # Get users to notify (e.g., staff, superusers) users_to_notify = User.objects.filter( is_staff=True, is_active=True, email__isnull=False ).exclude(email='') if not users_to_notify.exists(): return # Prepare email content subject_map = { 'started': 'Maintenance Mode Activated', 'ended': 'Maintenance Mode Deactivated', 'scheduled': 'Scheduled Maintenance Notification' } subject = subject_map.get(event_type, 'Maintenance Notification') context = { 'config': config, 'event_type': event_type, 'site_name': getattr(settings, 'SITE_NAME', 'Adtlas'), } message = render_to_string( 'core/emails/maintenance_notification.txt', context ) html_message = render_to_string( 'core/emails/maintenance_notification.html', context ) # Prepare email list email_list = [] for user in users_to_notify: email_list.append(( subject, message, settings.DEFAULT_FROM_EMAIL, [user.email] )) # Send emails send_mass_mail(email_list, fail_silently=True) self.stdout.write(f"Sent notifications to {len(email_list)} users") except Exception as e: self.logger.error(f"Failed to send notifications: {str(e)}") def log_maintenance_action(self, action, config): """Log maintenance action to audit log""" try: AuditLog.objects.create( action=f'maintenance_mode_{action}', details={ 'action': action, 'config': config, 'timestamp': timezone.now().isoformat(), }, timestamp=timezone.now() ) except Exception as e: self.logger.error(f"Failed to log maintenance action: {str(e)}") def confirm_enable(self): """Confirm maintenance mode activation""" try: response = input( "This will enable maintenance mode and block user access. Continue? [y/N]: " ) return response.lower() in ['y', 'yes'] except (EOFError, KeyboardInterrupt): return False def confirm_disable(self): """Confirm maintenance mode deactivation""" try: response = input( "This will disable maintenance mode and restore user access. Continue? [y/N]: " ) return response.lower() in ['y', 'yes'] except (EOFError, KeyboardInterrupt): return False def parse_list(self, list_str): """Parse comma-separated list string""" if not list_str: return [] return [item.strip() for item in list_str.split(',')] def get_scheduled_maintenance(self): """Get scheduled maintenance information""" try: config = SystemConfiguration.objects.filter( key='scheduled_maintenance', is_active=True ).first() if config: return config.value except Exception: pass return None def save_scheduled_maintenance(self, config): """Save scheduled maintenance configuration""" try: config_obj, created = SystemConfiguration.objects.get_or_create( key='scheduled_maintenance', defaults={'value': config, 'is_active': True} ) if not created: config_obj.value = config config_obj.is_active = True config_obj.save() except Exception as e: self.logger.error(f"Failed to save scheduled maintenance: {str(e)}") raise