# Adtlas TV Advertising Platform - Channel Maintenance Command # Management command for maintaining and optimizing TV channel data from django.core.management.base import BaseCommand, CommandError from django.db import transaction, connection from django.db.models import Count, Q, Avg, Max, Min from django.utils import timezone from django.core.cache import cache from channels.models import ( TVChannel, BroadcastNetwork, GeographicZone, ChannelCoverage, ContentSchedule, AudienceDemographics ) import logging from datetime import timedelta, datetime from typing import Dict, List, Any, Optional # Configure logging for maintenance operations logger = logging.getLogger('adtlas.channels.maintenance') class Command(BaseCommand): """ Django management command for maintaining TV channel data. Performs various maintenance tasks: - Clean up orphaned records - Update statistics and analytics - Optimize database performance - Validate data integrity - Archive old schedule data - Update coverage calculations Usage: python manage.py maintain_channels --all python manage.py maintain_channels --cleanup --stats python manage.py maintain_channels --archive-days 30 """ help = 'Maintain and optimize TV channel data' def add_arguments(self, parser): """ Define command-line arguments for the maintenance command. Args: parser: Django's argument parser instance """ parser.add_argument( '--all', action='store_true', help='Run all maintenance tasks' ) parser.add_argument( '--cleanup', action='store_true', help='Clean up orphaned and invalid records' ) parser.add_argument( '--stats', action='store_true', help='Update statistics and analytics' ) parser.add_argument( '--optimize', action='store_true', help='Optimize database performance' ) parser.add_argument( '--validate', action='store_true', help='Validate data integrity' ) parser.add_argument( '--archive', action='store_true', help='Archive old schedule data' ) parser.add_argument( '--archive-days', type=int, default=30, help='Number of days to keep schedule data (default: 30)' ) parser.add_argument( '--update-coverage', action='store_true', help='Update coverage calculations' ) parser.add_argument( '--clear-cache', action='store_true', help='Clear application cache' ) parser.add_argument( '--dry-run', action='store_true', help='Perform a dry run without making changes' ) parser.add_argument( '--verbose', action='store_true', help='Enable verbose output' ) def handle(self, *args, **options): """ Main command handler for maintenance operations. Args: *args: Positional arguments **options: Command options from argument parser """ # Extract options run_all = options['all'] cleanup = options['cleanup'] or run_all stats = options['stats'] or run_all optimize = options['optimize'] or run_all validate = options['validate'] or run_all archive = options['archive'] or run_all archive_days = options['archive_days'] update_coverage = options['update_coverage'] or run_all clear_cache = options['clear_cache'] or run_all dry_run = options['dry_run'] verbose = options['verbose'] # Set up logging level if verbose: logging.getLogger('adtlas.channels.maintenance').setLevel(logging.DEBUG) # Log maintenance operation start logger.info("Starting channel maintenance operations") if dry_run: self.stdout.write( self.style.WARNING('DRY RUN MODE: No database changes will be made') ) try: maintenance_results = { 'start_time': timezone.now(), 'tasks_completed': [], 'errors': [], 'statistics': {} } # Run maintenance tasks if cleanup: self._run_cleanup(dry_run, verbose, maintenance_results) if validate: self._run_validation(verbose, maintenance_results) if stats: self._update_statistics(dry_run, verbose, maintenance_results) if update_coverage: self._update_coverage_calculations(dry_run, verbose, maintenance_results) if archive: self._archive_old_data(archive_days, dry_run, verbose, maintenance_results) if optimize: self._optimize_database(dry_run, verbose, maintenance_results) if clear_cache: self._clear_cache(dry_run, verbose, maintenance_results) # Final summary maintenance_results['end_time'] = timezone.now() maintenance_results['duration'] = ( maintenance_results['end_time'] - maintenance_results['start_time'] ).total_seconds() self._print_summary(maintenance_results) # Log completion logger.info( f"Channel maintenance completed in {maintenance_results['duration']:.2f} seconds" ) except Exception as e: error_msg = f"Maintenance failed: {str(e)}" logger.error(error_msg, exc_info=True) raise CommandError(error_msg) def _run_cleanup(self, dry_run: bool, verbose: bool, results: Dict[str, Any]): """ Clean up orphaned and invalid records. Args: dry_run: Whether to perform a dry run verbose: Whether to enable verbose output results: Results dictionary to update """ self.stdout.write("Running cleanup tasks...") cleanup_stats = { 'orphaned_coverage': 0, 'invalid_schedules': 0, 'inactive_networks': 0, 'duplicate_channels': 0 } try: # Clean up orphaned coverage records orphaned_coverage = ChannelCoverage.objects.filter( Q(channel__isnull=True) | Q(zone__isnull=True) ) cleanup_stats['orphaned_coverage'] = orphaned_coverage.count() if verbose: self.stdout.write(f"Found {cleanup_stats['orphaned_coverage']} orphaned coverage records") if not dry_run and cleanup_stats['orphaned_coverage'] > 0: orphaned_coverage.delete() # Clean up invalid schedule entries (past entries with no end time) cutoff_date = timezone.now() - timedelta(days=7) invalid_schedules = ContentSchedule.objects.filter( start_time__lt=cutoff_date, end_time__isnull=True ) cleanup_stats['invalid_schedules'] = invalid_schedules.count() if verbose: self.stdout.write(f"Found {cleanup_stats['invalid_schedules']} invalid schedule entries") if not dry_run and cleanup_stats['invalid_schedules'] > 0: invalid_schedules.delete() # Identify networks with no active channels inactive_networks = BroadcastNetwork.objects.annotate( active_channel_count=Count('channels', filter=Q(channels__is_active=True)) ).filter(active_channel_count=0, is_active=True) cleanup_stats['inactive_networks'] = inactive_networks.count() if verbose: self.stdout.write(f"Found {cleanup_stats['inactive_networks']} networks with no active channels") if not dry_run and cleanup_stats['inactive_networks'] > 0: inactive_networks.update(is_active=False) # Find potential duplicate channels (same call sign) duplicate_channels = TVChannel.objects.values('call_sign').annotate( count=Count('id') ).filter(count__gt=1) cleanup_stats['duplicate_channels'] = duplicate_channels.count() if verbose and cleanup_stats['duplicate_channels'] > 0: self.stdout.write( self.style.WARNING( f"Found {cleanup_stats['duplicate_channels']} potential duplicate call signs" ) ) for dup in duplicate_channels: self.stdout.write(f" - Call sign '{dup['call_sign']}' appears {dup['count']} times") results['tasks_completed'].append('cleanup') results['statistics']['cleanup'] = cleanup_stats self.stdout.write( self.style.SUCCESS( f"Cleanup completed: {sum(cleanup_stats.values())} issues addressed" ) ) except Exception as e: error_msg = f"Cleanup task failed: {str(e)}" results['errors'].append(error_msg) logger.error(error_msg) self.stdout.write(self.style.ERROR(error_msg)) def _run_validation(self, verbose: bool, results: Dict[str, Any]): """ Validate data integrity. Args: verbose: Whether to enable verbose output results: Results dictionary to update """ self.stdout.write("Running validation checks...") validation_stats = { 'channels_without_network': 0, 'channels_without_coverage': 0, 'invalid_frequencies': 0, 'missing_call_signs': 0, 'overlapping_schedules': 0 } try: # Check channels without networks channels_without_network = TVChannel.objects.filter(network__isnull=True) validation_stats['channels_without_network'] = channels_without_network.count() # Check channels without coverage areas channels_without_coverage = TVChannel.objects.annotate( coverage_count=Count('coverage_areas') ).filter(coverage_count=0) validation_stats['channels_without_coverage'] = channels_without_coverage.count() # Check for invalid frequencies (should be positive numbers) invalid_frequencies = TVChannel.objects.filter( Q(frequency__lte=0) | Q(frequency__isnull=True) ) validation_stats['invalid_frequencies'] = invalid_frequencies.count() # Check for missing call signs missing_call_signs = TVChannel.objects.filter( Q(call_sign__isnull=True) | Q(call_sign__exact='') ) validation_stats['missing_call_signs'] = missing_call_signs.count() # Check for overlapping schedules (same channel, overlapping times) overlapping_schedules = 0 for channel in TVChannel.objects.prefetch_related('schedule_entries'): schedules = list(channel.schedule_entries.order_by('start_time')) for i in range(len(schedules) - 1): current = schedules[i] next_schedule = schedules[i + 1] if (current.end_time and current.end_time > next_schedule.start_time): overlapping_schedules += 1 validation_stats['overlapping_schedules'] = overlapping_schedules # Report validation results total_issues = sum(validation_stats.values()) if verbose or total_issues > 0: for issue_type, count in validation_stats.items(): if count > 0: self.stdout.write( self.style.WARNING( f" - {issue_type.replace('_', ' ').title()}: {count}" ) ) results['tasks_completed'].append('validation') results['statistics']['validation'] = validation_stats if total_issues == 0: self.stdout.write( self.style.SUCCESS("Validation completed: No issues found") ) else: self.stdout.write( self.style.WARNING( f"Validation completed: {total_issues} issues found" ) ) except Exception as e: error_msg = f"Validation task failed: {str(e)}" results['errors'].append(error_msg) logger.error(error_msg) self.stdout.write(self.style.ERROR(error_msg)) def _update_statistics(self, dry_run: bool, verbose: bool, results: Dict[str, Any]): """ Update statistics and analytics. Args: dry_run: Whether to perform a dry run verbose: Whether to enable verbose output results: Results dictionary to update """ self.stdout.write("Updating statistics...") try: # Calculate channel statistics channel_stats = { 'total_channels': TVChannel.objects.count(), 'active_channels': TVChannel.objects.filter(is_active=True).count(), 'total_networks': BroadcastNetwork.objects.count(), 'active_networks': BroadcastNetwork.objects.filter(is_active=True).count(), 'total_zones': GeographicZone.objects.count(), 'coverage_relationships': ChannelCoverage.objects.count(), } # Calculate coverage statistics coverage_stats = ChannelCoverage.objects.aggregate( avg_coverage=Avg('coverage_percentage'), max_coverage=Max('coverage_percentage'), min_coverage=Min('coverage_percentage') ) # Calculate network distribution network_distribution = list( BroadcastNetwork.objects.annotate( channel_count=Count('channels') ).values('name', 'channel_count').order_by('-channel_count')[:10] ) # Calculate zone distribution zone_distribution = list( GeographicZone.objects.annotate( channel_count=Count('coverage_areas') ).values('name', 'zone_type', 'channel_count').order_by('-channel_count')[:10] ) statistics = { 'channels': channel_stats, 'coverage': coverage_stats, 'top_networks': network_distribution, 'top_zones': zone_distribution, 'last_updated': timezone.now().isoformat() } if verbose: self.stdout.write("Channel Statistics:") for key, value in channel_stats.items(): self.stdout.write(f" - {key.replace('_', ' ').title()}: {value}") if coverage_stats['avg_coverage']: self.stdout.write( f" - Average Coverage: {coverage_stats['avg_coverage']:.2f}%" ) # Cache statistics for application use if not dry_run: cache.set('channel_statistics', statistics, timeout=3600) # 1 hour results['tasks_completed'].append('statistics') results['statistics']['channel_stats'] = statistics self.stdout.write( self.style.SUCCESS("Statistics updated successfully") ) except Exception as e: error_msg = f"Statistics update failed: {str(e)}" results['errors'].append(error_msg) logger.error(error_msg) self.stdout.write(self.style.ERROR(error_msg)) def _update_coverage_calculations(self, dry_run: bool, verbose: bool, results: Dict[str, Any]): """ Update coverage calculations and relationships. Args: dry_run: Whether to perform a dry run verbose: Whether to enable verbose output results: Results dictionary to update """ self.stdout.write("Updating coverage calculations...") try: updated_count = 0 # Update coverage percentages based on zone population and signal strength for coverage in ChannelCoverage.objects.select_related('channel', 'zone'): old_percentage = coverage.coverage_percentage # Simple calculation based on signal strength # In a real implementation, this would use more sophisticated algorithms signal_multipliers = { 'excellent': 1.0, 'strong': 0.9, 'good': 0.8, 'fair': 0.6, 'poor': 0.4, 'weak': 0.2 } base_coverage = 100.0 # Base coverage percentage multiplier = signal_multipliers.get(coverage.signal_strength, 0.8) new_percentage = base_coverage * multiplier if abs(new_percentage - old_percentage) > 1.0: # Only update if significant change coverage.coverage_percentage = new_percentage if not dry_run: coverage.save(update_fields=['coverage_percentage', 'updated_at']) updated_count += 1 if verbose: self.stdout.write( f" Updated {coverage.channel.name} in {coverage.zone.name}: " f"{old_percentage:.1f}% -> {new_percentage:.1f}%" ) results['tasks_completed'].append('coverage_update') results['statistics']['coverage_updates'] = {'updated_count': updated_count} self.stdout.write( self.style.SUCCESS( f"Coverage calculations updated: {updated_count} records modified" ) ) except Exception as e: error_msg = f"Coverage update failed: {str(e)}" results['errors'].append(error_msg) logger.error(error_msg) self.stdout.write(self.style.ERROR(error_msg)) def _archive_old_data(self, archive_days: int, dry_run: bool, verbose: bool, results: Dict[str, Any]): """ Archive old schedule data. Args: archive_days: Number of days to keep data dry_run: Whether to perform a dry run verbose: Whether to enable verbose output results: Results dictionary to update """ self.stdout.write(f"Archiving data older than {archive_days} days...") try: cutoff_date = timezone.now() - timedelta(days=archive_days) # Archive old schedule entries old_schedules = ContentSchedule.objects.filter(end_time__lt=cutoff_date) schedule_count = old_schedules.count() if verbose: self.stdout.write(f"Found {schedule_count} old schedule entries to archive") if not dry_run and schedule_count > 0: # In a real implementation, you might move these to an archive table # For now, we'll just delete them old_schedules.delete() # Archive old audience demographics old_demographics = AudienceDemographics.objects.filter( measurement_date__lt=cutoff_date ) demographics_count = old_demographics.count() if verbose: self.stdout.write(f"Found {demographics_count} old demographic records to archive") if not dry_run and demographics_count > 0: old_demographics.delete() total_archived = schedule_count + demographics_count results['tasks_completed'].append('archive') results['statistics']['archive'] = { 'schedules_archived': schedule_count, 'demographics_archived': demographics_count, 'total_archived': total_archived } self.stdout.write( self.style.SUCCESS( f"Archiving completed: {total_archived} records processed" ) ) except Exception as e: error_msg = f"Archiving failed: {str(e)}" results['errors'].append(error_msg) logger.error(error_msg) self.stdout.write(self.style.ERROR(error_msg)) def _optimize_database(self, dry_run: bool, verbose: bool, results: Dict[str, Any]): """ Optimize database performance. Args: dry_run: Whether to perform a dry run verbose: Whether to enable verbose output results: Results dictionary to update """ self.stdout.write("Optimizing database performance...") try: optimization_stats = { 'tables_analyzed': 0, 'indexes_checked': 0 } if not dry_run: # Run database-specific optimization commands with connection.cursor() as cursor: # PostgreSQL specific optimizations if connection.vendor == 'postgresql': # Analyze tables for better query planning tables = [ 'channels_tvchannel', 'channels_broadcastnetwork', 'channels_geographiczone', 'channels_channelcoverage', 'channels_contentschedule', 'channels_audiencedemographics' ] for table in tables: try: cursor.execute(f'ANALYZE {table};') optimization_stats['tables_analyzed'] += 1 if verbose: self.stdout.write(f" Analyzed table: {table}") except Exception as e: if verbose: self.stdout.write( self.style.WARNING( f" Could not analyze {table}: {str(e)}" ) ) # MySQL specific optimizations elif connection.vendor == 'mysql': # Optimize tables tables = [ 'channels_tvchannel', 'channels_broadcastnetwork', 'channels_geographiczone', 'channels_channelcoverage' ] for table in tables: try: cursor.execute(f'OPTIMIZE TABLE {table};') optimization_stats['tables_analyzed'] += 1 if verbose: self.stdout.write(f" Optimized table: {table}") except Exception as e: if verbose: self.stdout.write( self.style.WARNING( f" Could not optimize {table}: {str(e)}" ) ) results['tasks_completed'].append('optimization') results['statistics']['optimization'] = optimization_stats self.stdout.write( self.style.SUCCESS( f"Database optimization completed: {optimization_stats['tables_analyzed']} tables processed" ) ) except Exception as e: error_msg = f"Database optimization failed: {str(e)}" results['errors'].append(error_msg) logger.error(error_msg) self.stdout.write(self.style.ERROR(error_msg)) def _clear_cache(self, dry_run: bool, verbose: bool, results: Dict[str, Any]): """ Clear application cache. Args: dry_run: Whether to perform a dry run verbose: Whether to enable verbose output results: Results dictionary to update """ self.stdout.write("Clearing application cache...") try: if not dry_run: # Clear specific cache keys cache_keys = [ 'channel_statistics', 'network_list', 'zone_list', 'coverage_map_data' ] cleared_count = 0 for key in cache_keys: if cache.delete(key): cleared_count += 1 if verbose: self.stdout.write(f" Cleared cache key: {key}") # Clear all channel-related cache (pattern-based) try: cache.clear() if verbose: self.stdout.write(" Cleared entire cache") except Exception: # Fallback if clear() is not supported pass results['tasks_completed'].append('cache_clear') results['statistics']['cache_clear'] = {'keys_cleared': len(cache_keys) if not dry_run else 0} self.stdout.write( self.style.SUCCESS("Cache cleared successfully") ) except Exception as e: error_msg = f"Cache clearing failed: {str(e)}" results['errors'].append(error_msg) logger.error(error_msg) self.stdout.write(self.style.ERROR(error_msg)) def _print_summary(self, results: Dict[str, Any]): """ Print maintenance summary. Args: results: Results dictionary with maintenance information """ self.stdout.write("\n" + "=" * 50) self.stdout.write("MAINTENANCE SUMMARY") self.stdout.write("=" * 50) self.stdout.write(f"Duration: {results['duration']:.2f} seconds") self.stdout.write(f"Tasks completed: {', '.join(results['tasks_completed'])}") if results['errors']: self.stdout.write("\nErrors encountered:") for error in results['errors']: self.stdout.write(self.style.ERROR(f" - {error}")) else: self.stdout.write("\nNo errors encountered.") # Print detailed statistics if available if 'channel_stats' in results['statistics']: stats = results['statistics']['channel_stats']['channels'] self.stdout.write("\nChannel Statistics:") self.stdout.write(f" Total Channels: {stats['total_channels']}") self.stdout.write(f" Active Channels: {stats['active_channels']}") self.stdout.write(f" Total Networks: {stats['total_networks']}") self.stdout.write(f" Coverage Relationships: {stats['coverage_relationships']}") self.stdout.write("=" * 50)