# -*- coding: utf-8 -*- """ Management command for generating sample channel data for testing and development. """ import random from datetime import datetime, timedelta from django.core.management.base import BaseCommand from django.utils import timezone from django.core.files.base import ContentFile from ...models import ( Channel, ChannelZone, ChannelCodec, ChannelZoneRelation, EPGProgram, Jingle, JingleDetection, ChannelSchedule, IPSecConfiguration, OpenVPNConfiguration, WireGuardConfiguration ) class Command(BaseCommand): """ Management command to generate sample channel data. """ help = 'Generate sample channel data for testing and development' def add_arguments(self, parser): parser.add_argument( '--channels', type=int, default=50, help='Number of channels to create (default: 50)' ) parser.add_argument( '--zones', type=int, default=10, help='Number of zones to create (default: 10)' ) parser.add_argument( '--codecs', type=int, default=5, help='Number of codecs to create (default: 5)' ) parser.add_argument( '--epg-programs', type=int, default=500, help='Number of EPG programs to create (default: 500)' ) parser.add_argument( '--jingles', type=int, default=100, help='Number of jingles to create (default: 100)' ) parser.add_argument( '--schedules', type=int, default=200, help='Number of schedules to create (default: 200)' ) parser.add_argument( '--vpn-configs', type=int, default=15, help='Number of VPN configurations to create (default: 15)' ) parser.add_argument( '--clear-existing', action='store_true', help='Clear existing data before generating new data' ) parser.add_argument( '--seed', type=int, help='Random seed for reproducible data generation' ) def handle(self, *args, **options): num_channels = options['channels'] num_zones = options['zones'] num_codecs = options['codecs'] num_epg_programs = options['epg_programs'] num_jingles = options['jingles'] num_schedules = options['schedules'] num_vpn_configs = options['vpn_configs'] clear_existing = options['clear_existing'] seed = options.get('seed') if seed: random.seed(seed) self.stdout.write(f'Using random seed: {seed}') self.stdout.write( self.style.SUCCESS('Starting sample data generation...') ) if clear_existing: self.clear_existing_data() # Generate data in order of dependencies codecs = self.generate_codecs(num_codecs) zones = self.generate_zones(num_zones) vpn_configs = self.generate_vpn_configs(num_vpn_configs) channels = self.generate_channels(num_channels, codecs, zones) self.generate_channel_zone_relations(channels, zones, vpn_configs) self.generate_epg_programs(num_epg_programs, channels) self.generate_jingles(num_jingles, channels) self.generate_schedules(num_schedules, channels) self.generate_jingle_detections(channels) self.stdout.write( self.style.SUCCESS('Sample data generation completed!') ) def clear_existing_data(self): """ Clear existing data. """ self.stdout.write('Clearing existing data...') models_to_clear = [ JingleDetection, ChannelSchedule, Jingle, EPGProgram, ChannelZoneRelation, Channel, ChannelZone, ChannelCodec, IPSecConfiguration, OpenVPNConfiguration, WireGuardConfiguration, ] for model in models_to_clear: count = model.objects.count() if count > 0: model.objects.all().delete() self.stdout.write(f' Deleted {count} {model.__name__} records') def generate_codecs(self, count): """ Generate sample codecs. """ self.stdout.write(f'Generating {count} codecs...') codec_types = ['h264', 'h265', 'mpeg2', 'vp9', 'av1'] codec_names = { 'h264': ['H.264 Main', 'H.264 High', 'H.264 Baseline'], 'h265': ['H.265 Main', 'H.265 Main10', 'H.265 Main12'], 'mpeg2': ['MPEG-2 Main', 'MPEG-2 High'], 'vp9': ['VP9 Profile 0', 'VP9 Profile 2'], 'av1': ['AV1 Main', 'AV1 High'] } codecs = [] for i in range(count): codec_type = random.choice(codec_types) name = random.choice(codec_names[codec_type]) codec = ChannelCodec.objects.create( name=f'{name} {i+1}', codec_type=codec_type, description=f'Sample {codec_type.upper()} codec configuration', bitrate_min=random.randint(500, 2000), bitrate_max=random.randint(5000, 20000), resolution_width=random.choice([1280, 1920, 3840]), resolution_height=random.choice([720, 1080, 2160]), frame_rate=random.choice([25, 30, 50, 60]), is_active=random.choice([True, True, True, False]) # 75% active ) codecs.append(codec) self.stdout.write(f' Created {len(codecs)} codecs') return codecs def generate_zones(self, count): """ Generate sample zones. """ self.stdout.write(f'Generating {count} zones...') countries = ['US', 'UK', 'DE', 'FR', 'ES', 'IT', 'CA', 'AU', 'JP', 'BR'] regions = { 'US': ['East Coast', 'West Coast', 'Midwest', 'South'], 'UK': ['England', 'Scotland', 'Wales', 'Northern Ireland'], 'DE': ['North', 'South', 'East', 'West'], 'FR': ['North', 'South', 'East', 'West'], 'ES': ['North', 'South', 'East', 'Central'], 'IT': ['North', 'South', 'Central'], 'CA': ['Eastern', 'Western', 'Central'], 'AU': ['Eastern', 'Western', 'Central'], 'JP': ['Kanto', 'Kansai', 'Kyushu', 'Tohoku'], 'BR': ['Southeast', 'Northeast', 'South', 'North'] } zones = [] for i in range(count): country = random.choice(countries) region = random.choice(regions[country]) zone = ChannelZone.objects.create( name=f'{country} {region} Zone {i+1}', description=f'Broadcasting zone for {region}, {country}', country=country, region=region, timezone=random.choice([ 'UTC', 'US/Eastern', 'US/Pacific', 'Europe/London', 'Europe/Berlin', 'Asia/Tokyo', 'Australia/Sydney' ]), is_active=random.choice([True, True, True, False]) # 75% active ) zones.append(zone) self.stdout.write(f' Created {len(zones)} zones') return zones def generate_vpn_configs(self, count): """ Generate sample VPN configurations. """ self.stdout.write(f'Generating {count} VPN configurations...') vpn_configs = [] # Generate IPSec configurations for i in range(count // 3): config = IPSecConfiguration.objects.create( name=f'IPSec Config {i+1}', description=f'Sample IPSec VPN configuration {i+1}', server_address=f'ipsec{i+1}.example.com', username=f'ipsec_user_{i+1}', password=f'ipsec_pass_{i+1}', pre_shared_key=f'psk_{random.randint(1000, 9999)}', encryption_algorithm=random.choice(['aes128', 'aes256', '3des']), authentication_algorithm=random.choice(['sha1', 'sha256', 'md5']), is_active=True ) vpn_configs.append(config) # Generate OpenVPN configurations for i in range(count // 3): config = OpenVPNConfiguration.objects.create( name=f'OpenVPN Config {i+1}', description=f'Sample OpenVPN configuration {i+1}', server_address=f'openvpn{i+1}.example.com', username=f'ovpn_user_{i+1}', password=f'ovpn_pass_{i+1}', config_file=ContentFile( f'# OpenVPN Config {i+1}\nremote openvpn{i+1}.example.com 1194\n', name=f'openvpn_config_{i+1}.ovpn' ), protocol=random.choice(['udp', 'tcp']), port=random.choice([1194, 443, 8080]), is_active=True ) vpn_configs.append(config) # Generate WireGuard configurations for i in range(count - (2 * (count // 3))): config = WireGuardConfiguration.objects.create( name=f'WireGuard Config {i+1}', description=f'Sample WireGuard VPN configuration {i+1}', server_address=f'wireguard{i+1}.example.com', public_key=f'wg_public_key_{random.randint(1000, 9999)}', private_key=f'wg_private_key_{random.randint(1000, 9999)}', endpoint=f'wireguard{i+1}.example.com:51820', allowed_ips='0.0.0.0/0', listen_port=51820 + i, is_active=True ) vpn_configs.append(config) self.stdout.write(f' Created {len(vpn_configs)} VPN configurations') return vpn_configs def generate_channels(self, count, codecs, zones): """ Generate sample channels. """ self.stdout.write(f'Generating {count} channels...') categories = ['news', 'sports', 'entertainment', 'documentary', 'kids', 'music'] languages = ['en', 'es', 'fr', 'de', 'it', 'pt', 'ja', 'zh'] countries = ['US', 'UK', 'DE', 'FR', 'ES', 'IT', 'CA', 'AU', 'JP', 'BR'] channel_names = [ 'Global News', 'Sports Central', 'Entertainment Plus', 'Discovery World', 'Kids Zone', 'Music TV', 'Movie Channel', 'Documentary Network', 'Comedy Central', 'Drama Series', 'Action Movies', 'Sci-Fi Channel', 'History Channel', 'Nature World', 'Travel Guide', 'Food Network', 'Fashion TV', 'Tech News', 'Business Today', 'Weather Channel' ] channels = [] for i in range(count): base_name = random.choice(channel_names) channel = Channel.objects.create( name=f'{base_name} {i+1}', channel_number=str(100 + i), description=f'Sample {base_name.lower()} channel for testing', category=random.choice(categories), status=random.choice(['active', 'active', 'active', 'inactive']), # 75% active stream_url=f'http://stream{i+1}.example.com/live', backup_stream_url=f'http://backup{i+1}.example.com/live', language=random.choice(languages), country=random.choice(countries), is_hd=random.choice([True, False]), is_encrypted=random.choice([True, False]), codec=random.choice(codecs) if codecs else None, health_status=random.choice(['healthy', 'unhealthy', None]), last_health_check=timezone.now() - timedelta( minutes=random.randint(0, 1440) ) if random.choice([True, False]) else None ) # Assign random zones channel_zones = random.sample(zones, random.randint(1, min(3, len(zones)))) channel.zones.set(channel_zones) channels.append(channel) self.stdout.write(f' Created {len(channels)} channels') return channels def generate_channel_zone_relations(self, channels, zones, vpn_configs): """ Generate channel-zone relations with VPN configurations. """ self.stdout.write('Generating channel-zone relations...') relations = [] for channel in channels: for zone in channel.zones.all(): # Check if relation already exists relation, created = ChannelZoneRelation.objects.get_or_create( channel=channel, zone=zone, defaults={ 'stream_url': f'http://zone-{zone.id}-stream.example.com/{channel.channel_number}', 'backup_stream_url': f'http://zone-{zone.id}-backup.example.com/{channel.channel_number}', 'priority': random.randint(1, 10), 'is_active': random.choice([True, True, True, False]), # 75% active 'requires_vpn': random.choice([True, False]), } ) if created and relation.requires_vpn and vpn_configs: # Assign random VPN configuration vpn_config = random.choice(vpn_configs) if hasattr(vpn_config, 'ipsecconfiguration'): relation.ipsec_config = vpn_config elif hasattr(vpn_config, 'openvpnconfiguration'): relation.openvpn_config = vpn_config elif hasattr(vpn_config, 'wireguardconfiguration'): relation.wireguard_config = vpn_config relation.save() relations.append(relation) self.stdout.write(f' Created/updated {len(relations)} channel-zone relations') def generate_epg_programs(self, count, channels): """ Generate sample EPG programs. """ self.stdout.write(f'Generating {count} EPG programs...') program_titles = [ 'Morning News', 'Sports Update', 'Movie Night', 'Documentary Hour', 'Kids Show', 'Music Videos', 'Talk Show', 'Comedy Special', 'Drama Series', 'Reality TV', 'Game Show', 'Weather Report', 'Business News', 'Travel Guide', 'Cooking Show', 'Tech Review' ] genres = ['news', 'sports', 'movie', 'documentary', 'kids', 'music', 'talk', 'comedy'] categories = ['live', 'recorded', 'rerun'] ratings = ['G', 'PG', 'PG-13', 'R', 'TV-Y', 'TV-G', 'TV-PG', 'TV-14', 'TV-MA'] programs = [] for i in range(count): channel = random.choice(channels) title = random.choice(program_titles) # Generate random start time within the next 30 days start_time = timezone.now() + timedelta( days=random.randint(0, 30), hours=random.randint(0, 23), minutes=random.choice([0, 15, 30, 45]) ) # Generate duration between 30 minutes and 3 hours duration = timedelta(minutes=random.choice([30, 60, 90, 120, 180])) end_time = start_time + duration program = EPGProgram.objects.create( channel=channel, title=f'{title} {i+1}', description=f'Sample {title.lower()} program for testing', start_time=start_time, end_time=end_time, duration=duration, genre=random.choice(genres), category=random.choice(categories), rating=random.choice(ratings), language=channel.language, is_live=random.choice([True, False]), is_premiere=random.choice([True, False, False, False]), # 25% premiere episode_number=random.randint(1, 50) if random.choice([True, False]) else None, season_number=random.randint(1, 10) if random.choice([True, False]) else None ) programs.append(program) self.stdout.write(f' Created {len(programs)} EPG programs') return programs def generate_jingles(self, count, channels): """ Generate sample jingles. """ self.stdout.write(f'Generating {count} jingles...') jingle_types = ['intro', 'outro', 'commercial_break', 'station_id', 'weather', 'news'] jingle_names = { 'intro': ['Opening Theme', 'Show Intro', 'Program Start'], 'outro': ['Closing Theme', 'Show Outro', 'Program End'], 'commercial_break': ['Ad Break', 'Commercial Transition', 'Break Time'], 'station_id': ['Station ID', 'Channel Branding', 'Logo Sound'], 'weather': ['Weather Intro', 'Weather Update', 'Weather Alert'], 'news': ['News Flash', 'Breaking News', 'News Update'] } jingles = [] for i in range(count): channel = random.choice(channels) jingle_type = random.choice(jingle_types) name = random.choice(jingle_names[jingle_type]) jingle = Jingle.objects.create( channel=channel, name=f'{name} {i+1}', description=f'Sample {jingle_type} jingle for {channel.name}', jingle_type=jingle_type, duration=timedelta(seconds=random.randint(5, 30)), file_size=random.randint(100000, 1000000), # 100KB to 1MB audio_format=random.choice(['mp3', 'wav', 'aac']), sample_rate=random.choice([44100, 48000]), bit_rate=random.choice([128, 192, 256, 320]), is_active=random.choice([True, True, True, False]), # 75% active play_count=random.randint(0, 1000), last_played=timezone.now() - timedelta( hours=random.randint(0, 168) ) if random.choice([True, False]) else None ) jingles.append(jingle) self.stdout.write(f' Created {len(jingles)} jingles') return jingles def generate_schedules(self, count, channels): """ Generate sample schedules. """ self.stdout.write(f'Generating {count} schedules...') schedule_types = ['program', 'commercial', 'jingle', 'maintenance'] schedules = [] for i in range(count): channel = random.choice(channels) schedule_type = random.choice(schedule_types) # Generate random start time within the next 7 days start_time = timezone.now() + timedelta( days=random.randint(0, 7), hours=random.randint(0, 23), minutes=random.choice([0, 15, 30, 45]) ) # Generate duration based on schedule type if schedule_type == 'program': duration = timedelta(minutes=random.choice([30, 60, 90, 120])) elif schedule_type == 'commercial': duration = timedelta(minutes=random.choice([2, 3, 5])) elif schedule_type == 'jingle': duration = timedelta(seconds=random.randint(5, 30)) else: # maintenance duration = timedelta(hours=random.choice([1, 2, 4])) end_time = start_time + duration schedule = ChannelSchedule.objects.create( channel=channel, title=f'{schedule_type.title()} Schedule {i+1}', description=f'Sample {schedule_type} schedule for {channel.name}', schedule_type=schedule_type, start_time=start_time, end_time=end_time, duration=duration, priority=random.randint(1, 10), is_active=random.choice([True, True, True, False]), # 75% active is_recurring=random.choice([True, False]), recurrence_pattern=random.choice(['daily', 'weekly', 'monthly']) if random.choice([True, False]) else None ) schedules.append(schedule) self.stdout.write(f' Created {len(schedules)} schedules') return schedules def generate_jingle_detections(self, channels): """ Generate sample jingle detections. """ # Get all jingles jingles = list(Jingle.objects.all()) if not jingles: return detection_count = len(jingles) * random.randint(5, 20) # 5-20 detections per jingle self.stdout.write(f'Generating {detection_count} jingle detections...') detections = [] for i in range(detection_count): jingle = random.choice(jingles) # Generate detection time within the last 30 days detected_at = timezone.now() - timedelta( days=random.randint(0, 30), hours=random.randint(0, 23), minutes=random.randint(0, 59) ) detection = JingleDetection.objects.create( jingle=jingle, detected_at=detected_at, confidence_score=random.uniform(0.7, 1.0), # High confidence scores audio_fingerprint=f'fingerprint_{random.randint(100000, 999999)}', detection_method=random.choice(['audio_fingerprint', 'pattern_matching', 'ml_detection']), stream_position=timedelta(seconds=random.randint(0, 7200)), # Within 2 hours metadata={ 'detection_algorithm': random.choice(['v1.0', 'v1.1', 'v2.0']), 'processing_time_ms': random.randint(10, 500), 'audio_quality': random.choice(['high', 'medium', 'low']) } ) detections.append(detection) self.stdout.write(f' Created {len(detections)} jingle detections') return detections