# -*- coding: utf-8 -*- """ Management command for performing health checks on channels. """ import time from django.core.management.base import BaseCommand, CommandError from django.db.models import Q from django.utils import timezone from datetime import timedelta from ...models import Channel from ...tasks import perform_channel_health_check, perform_bulk_health_checks class Command(BaseCommand): """ Management command to perform health checks on channels. """ help = 'Perform health checks on channels' def add_arguments(self, parser): parser.add_argument( '--channel-id', type=int, help='Specific channel ID to check' ) parser.add_argument( '--channel-number', type=str, help='Specific channel number to check' ) parser.add_argument( '--zone', type=str, help='Check all channels in a specific zone' ) parser.add_argument( '--status', type=str, choices=['active', 'inactive', 'maintenance'], help='Check channels with specific status' ) parser.add_argument( '--unhealthy-only', action='store_true', help='Check only channels that are currently unhealthy' ) parser.add_argument( '--stale-only', action='store_true', help='Check only channels with stale health data (older than 1 hour)' ) parser.add_argument( '--async', action='store_true', help='Run health checks asynchronously using Celery' ) parser.add_argument( '--batch-size', type=int, default=10, help='Number of channels to process in each batch (default: 10)' ) parser.add_argument( '--delay', type=float, default=1.0, help='Delay between health checks in seconds (default: 1.0)' ) parser.add_argument( '--timeout', type=int, default=30, help='Timeout for each health check in seconds (default: 30)' ) parser.add_argument( '--dry-run', action='store_true', help='Show which channels would be checked without actually checking them' ) def handle(self, *args, **options): channel_id = options.get('channel_id') channel_number = options.get('channel_number') zone = options.get('zone') status = options.get('status') unhealthy_only = options['unhealthy_only'] stale_only = options['stale_only'] async_mode = options['async'] batch_size = options['batch_size'] delay = options['delay'] timeout = options['timeout'] dry_run = options['dry_run'] # Build queryset based on filters queryset = Channel.objects.all() if channel_id: queryset = queryset.filter(id=channel_id) elif channel_number: queryset = queryset.filter(channel_number=channel_number) else: if zone: queryset = queryset.filter(zones__name__icontains=zone) if status: queryset = queryset.filter(status=status) if unhealthy_only: queryset = queryset.filter( Q(health_status='unhealthy') | Q(health_status__isnull=True) ) if stale_only: stale_threshold = timezone.now() - timedelta(hours=1) queryset = queryset.filter( Q(last_health_check__lt=stale_threshold) | Q(last_health_check__isnull=True) ) queryset = queryset.distinct().order_by('id') total_channels = queryset.count() if total_channels == 0: self.stdout.write( self.style.WARNING('No channels found matching the criteria.') ) return self.stdout.write( self.style.SUCCESS( f'Found {total_channels} channel(s) to check' ) ) if dry_run: self.stdout.write( self.style.WARNING('DRY RUN MODE - No health checks will be performed') ) for channel in queryset: self.stdout.write( f' - {channel.name} ({channel.channel_number}) - ' f'Status: {channel.status}, Health: {channel.health_status or "Unknown"}' ) return if async_mode: self.run_async_health_checks(queryset, batch_size) else: self.run_sync_health_checks(queryset, delay, timeout) def run_async_health_checks(self, queryset, batch_size): """ Run health checks asynchronously using Celery. """ self.stdout.write( self.style.SUCCESS('Starting asynchronous health checks...') ) channel_ids = list(queryset.values_list('id', flat=True)) # Process in batches for i in range(0, len(channel_ids), batch_size): batch = channel_ids[i:i + batch_size] self.stdout.write( f'Queuing batch {i//batch_size + 1}: {len(batch)} channels' ) # Queue bulk health check task task = perform_bulk_health_checks.delay(batch) self.stdout.write( f' Task ID: {task.id}' ) self.stdout.write( self.style.SUCCESS( f'Queued health checks for {len(channel_ids)} channels in ' f'{(len(channel_ids) + batch_size - 1) // batch_size} batches' ) ) def run_sync_health_checks(self, queryset, delay, timeout): """ Run health checks synchronously. """ self.stdout.write( self.style.SUCCESS('Starting synchronous health checks...') ) success_count = 0 error_count = 0 for i, channel in enumerate(queryset, 1): self.stdout.write( f'[{i}/{queryset.count()}] Checking {channel.name} ' f'({channel.channel_number})...' ) try: # Perform health check result = self.perform_single_health_check(channel, timeout) if result['success']: success_count += 1 self.stdout.write( self.style.SUCCESS( f' ✓ Health check passed - Status: {result["status"]}' ) ) else: error_count += 1 self.stdout.write( self.style.ERROR( f' ✗ Health check failed - Error: {result["error"]}' ) ) except Exception as e: error_count += 1 self.stdout.write( self.style.ERROR( f' ✗ Exception during health check: {str(e)}' ) ) # Add delay between checks if delay > 0 and i < queryset.count(): time.sleep(delay) # Print summary self.stdout.write('\n' + '='*50) self.stdout.write(self.style.SUCCESS('HEALTH CHECK SUMMARY')) self.stdout.write('='*50) self.stdout.write( self.style.SUCCESS(f'Successful checks: {success_count}') ) self.stdout.write( self.style.ERROR(f'Failed checks: {error_count}') ) self.stdout.write( f'Total channels checked: {success_count + error_count}' ) self.stdout.write('='*50) def perform_single_health_check(self, channel, timeout): """ Perform a single health check on a channel. """ try: # This is a simplified health check implementation # In a real implementation, you would check the stream URL, # verify connectivity, check stream quality, etc. import requests from urllib.parse import urlparse if not channel.stream_url: return { 'success': False, 'error': 'No stream URL configured' } # Parse URL to check if it's valid parsed_url = urlparse(channel.stream_url) if not parsed_url.scheme or not parsed_url.netloc: return { 'success': False, 'error': 'Invalid stream URL format' } # Perform HTTP HEAD request to check if stream is accessible try: response = requests.head( channel.stream_url, timeout=timeout, allow_redirects=True ) if response.status_code == 200: # Update channel health status channel.health_status = 'healthy' channel.last_health_check = timezone.now() channel.save(update_fields=['health_status', 'last_health_check']) return { 'success': True, 'status': 'healthy', 'response_code': response.status_code } else: # Update channel health status channel.health_status = 'unhealthy' channel.last_health_check = timezone.now() channel.save(update_fields=['health_status', 'last_health_check']) return { 'success': False, 'error': f'HTTP {response.status_code}' } except requests.exceptions.RequestException as e: # Update channel health status channel.health_status = 'unhealthy' channel.last_health_check = timezone.now() channel.save(update_fields=['health_status', 'last_health_check']) return { 'success': False, 'error': f'Request failed: {str(e)}' } except Exception as e: return { 'success': False, 'error': f'Unexpected error: {str(e)}' }