# Adtlas TV Advertising Platform - Channel Data Export Command # Management command for exporting TV channel data to external formats from django.core.management.base import BaseCommand, CommandError from django.db.models import Q, Prefetch from django.utils import timezone from channels.models import TVChannel, BroadcastNetwork, GeographicZone, ChannelCoverage import json import csv import xml.etree.ElementTree as ET from xml.dom import minidom import logging from typing import Dict, List, Any, Optional import os # Configure logging for export operations logger = logging.getLogger('adtlas.channels.export') class Command(BaseCommand): """ Django management command for exporting TV channel data. Supports multiple output formats: - JSON: Structured channel data with relationships - CSV: Tabular channel data for spreadsheet applications - XML: Industry-standard EPG formats Usage: python manage.py export_channels --output channels.json --format json python manage.py export_channels --output channels.csv --format csv --network "ABC" python manage.py export_channels --output channels.xml --format xml --active-only """ help = 'Export TV channel data to external formats (JSON, CSV, XML)' def add_arguments(self, parser): """ Define command-line arguments for the export command. Args: parser: Django's argument parser instance """ parser.add_argument( '--output', type=str, required=True, help='Path to the output file' ) parser.add_argument( '--format', type=str, choices=['json', 'csv', 'xml'], default='json', help='Format of the output file (default: json)' ) parser.add_argument( '--network', type=str, help='Filter channels by network name' ) parser.add_argument( '--zone', type=str, help='Filter channels by geographic zone' ) parser.add_argument( '--active-only', action='store_true', help='Export only active channels' ) parser.add_argument( '--include-coverage', action='store_true', help='Include coverage information in export' ) parser.add_argument( '--include-schedule', action='store_true', help='Include schedule information in export' ) parser.add_argument( '--pretty', action='store_true', help='Pretty-print JSON/XML output (larger file size)' ) def handle(self, *args, **options): """ Main command handler for exporting channel data. Args: *args: Positional arguments **options: Command options from argument parser """ output_path = options['output'] output_format = options['format'] network_filter = options.get('network') zone_filter = options.get('zone') active_only = options['active_only'] include_coverage = options['include_coverage'] include_schedule = options['include_schedule'] pretty_print = options['pretty'] # Log export operation start logger.info(f"Starting channel export to {output_path} (format: {output_format})") try: # Build queryset with filters queryset = self._build_queryset( network_filter, zone_filter, active_only, include_coverage, include_schedule ) # Get channel count channel_count = queryset.count() self.stdout.write(f"Exporting {channel_count} channels...") if channel_count == 0: self.stdout.write( self.style.WARNING('No channels found matching the specified criteria') ) return # Create output directory if it doesn't exist output_dir = os.path.dirname(output_path) if output_dir and not os.path.exists(output_dir): os.makedirs(output_dir) # Export data based on format if output_format == 'json': self._export_json(queryset, output_path, include_coverage, include_schedule, pretty_print) elif output_format == 'csv': self._export_csv(queryset, output_path, include_coverage) elif output_format == 'xml': self._export_xml(queryset, output_path, include_coverage, include_schedule, pretty_print) # Success message file_size = os.path.getsize(output_path) self.stdout.write( self.style.SUCCESS( f"Export complete: {channel_count} channels exported to {output_path} " f"({file_size:,} bytes)" ) ) # Log completion logger.info(f"Channel export completed: {channel_count} channels, {file_size} bytes") except Exception as e: error_msg = f"Export failed: {str(e)}" logger.error(error_msg, exc_info=True) raise CommandError(error_msg) def _build_queryset(self, network_filter: Optional[str], zone_filter: Optional[str], active_only: bool, include_coverage: bool, include_schedule: bool): """ Build the queryset for channels based on filters. Args: network_filter: Network name filter zone_filter: Zone name filter active_only: Whether to include only active channels include_coverage: Whether to prefetch coverage data include_schedule: Whether to prefetch schedule data Returns: Filtered and optimized queryset """ # Start with base queryset queryset = TVChannel.objects.select_related('network') # Apply filters if network_filter: queryset = queryset.filter(network__name__icontains=network_filter) if zone_filter: queryset = queryset.filter( coverage_areas__zone__name__icontains=zone_filter ).distinct() if active_only: queryset = queryset.filter(is_active=True) # Add prefetch for related data if requested prefetch_list = [] if include_coverage: prefetch_list.append( Prefetch( 'coverage_areas', queryset=ChannelCoverage.objects.select_related('zone') ) ) if include_schedule: prefetch_list.append('schedule_entries') if prefetch_list: queryset = queryset.prefetch_related(*prefetch_list) # Order by network and name for consistent output queryset = queryset.order_by('network__name', 'name') return queryset def _export_json(self, queryset, output_path: str, include_coverage: bool, include_schedule: bool, pretty_print: bool): """ Export channels to JSON format. Args: queryset: Channel queryset to export output_path: Path to output file include_coverage: Whether to include coverage data include_schedule: Whether to include schedule data pretty_print: Whether to format JSON for readability """ channels_data = [] for channel in queryset: channel_dict = { 'id': channel.id, 'name': channel.name, 'call_sign': channel.call_sign, 'frequency': str(channel.frequency), 'is_active': channel.is_active, 'description': channel.description, 'website_url': channel.website_url, 'created_at': channel.created_at.isoformat(), 'updated_at': channel.updated_at.isoformat(), 'network': { 'id': channel.network.id if channel.network else None, 'name': channel.network.name if channel.network else None, 'description': channel.network.description if channel.network else None, } if channel.network else None } # Add coverage information if requested if include_coverage: coverage_list = [] for coverage in channel.coverage_areas.all(): coverage_list.append({ 'zone': { 'id': coverage.zone.id, 'name': coverage.zone.name, 'zone_type': coverage.zone.zone_type, 'description': coverage.zone.description, }, 'coverage_percentage': float(coverage.coverage_percentage), 'signal_strength': coverage.signal_strength, 'is_primary': coverage.is_primary, }) channel_dict['coverage_areas'] = coverage_list # Add schedule information if requested if include_schedule: schedule_list = [] for schedule in channel.schedule_entries.all(): schedule_list.append({ 'program_title': schedule.program_title, 'start_time': schedule.start_time.isoformat(), 'end_time': schedule.end_time.isoformat(), 'description': schedule.description, 'genre': schedule.genre, 'rating': schedule.rating, }) channel_dict['schedule'] = schedule_list channels_data.append(channel_dict) # Create export metadata export_data = { 'metadata': { 'export_timestamp': timezone.now().isoformat(), 'total_channels': len(channels_data), 'format_version': '1.0', 'source': 'Adtlas TV Advertising Platform', 'includes_coverage': include_coverage, 'includes_schedule': include_schedule, }, 'channels': channels_data } # Write to file with open(output_path, 'w', encoding='utf-8') as file: if pretty_print: json.dump(export_data, file, indent=2, ensure_ascii=False) else: json.dump(export_data, file, ensure_ascii=False) def _export_csv(self, queryset, output_path: str, include_coverage: bool): """ Export channels to CSV format. Args: queryset: Channel queryset to export output_path: Path to output file include_coverage: Whether to include coverage data """ # Define CSV fieldnames fieldnames = [ 'id', 'name', 'call_sign', 'frequency', 'is_active', 'description', 'website_url', 'network_name', 'network_description', 'created_at', 'updated_at' ] if include_coverage: fieldnames.extend([ 'primary_zone', 'coverage_zones', 'avg_coverage_percentage' ]) with open(output_path, 'w', newline='', encoding='utf-8') as file: writer = csv.DictWriter(file, fieldnames=fieldnames) writer.writeheader() for channel in queryset: row = { 'id': channel.id, 'name': channel.name, 'call_sign': channel.call_sign, 'frequency': str(channel.frequency), 'is_active': channel.is_active, 'description': channel.description, 'website_url': channel.website_url, 'network_name': channel.network.name if channel.network else '', 'network_description': channel.network.description if channel.network else '', 'created_at': channel.created_at.isoformat(), 'updated_at': channel.updated_at.isoformat(), } if include_coverage: coverage_areas = channel.coverage_areas.all() primary_zones = [c.zone.name for c in coverage_areas if c.is_primary] all_zones = [c.zone.name for c in coverage_areas] avg_coverage = sum(c.coverage_percentage for c in coverage_areas) / len(coverage_areas) if coverage_areas else 0 row.update({ 'primary_zone': '; '.join(primary_zones), 'coverage_zones': '; '.join(all_zones), 'avg_coverage_percentage': f"{avg_coverage:.2f}", }) writer.writerow(row) def _export_xml(self, queryset, output_path: str, include_coverage: bool, include_schedule: bool, pretty_print: bool): """ Export channels to XML format (EPG-style). Args: queryset: Channel queryset to export output_path: Path to output file include_coverage: Whether to include coverage data include_schedule: Whether to include schedule data pretty_print: Whether to format XML for readability """ # Create root element root = ET.Element('adtlas_channels') root.set('export_timestamp', timezone.now().isoformat()) root.set('total_channels', str(queryset.count())) root.set('format_version', '1.0') # Add metadata metadata = ET.SubElement(root, 'metadata') ET.SubElement(metadata, 'source').text = 'Adtlas TV Advertising Platform' ET.SubElement(metadata, 'includes_coverage').text = str(include_coverage).lower() ET.SubElement(metadata, 'includes_schedule').text = str(include_schedule).lower() # Add channels channels_elem = ET.SubElement(root, 'channels') for channel in queryset: channel_elem = ET.SubElement(channels_elem, 'channel') channel_elem.set('id', str(channel.id)) # Basic channel information ET.SubElement(channel_elem, 'name').text = channel.name ET.SubElement(channel_elem, 'call_sign').text = channel.call_sign ET.SubElement(channel_elem, 'frequency').text = str(channel.frequency) ET.SubElement(channel_elem, 'is_active').text = str(channel.is_active).lower() ET.SubElement(channel_elem, 'description').text = channel.description or '' ET.SubElement(channel_elem, 'website_url').text = channel.website_url or '' ET.SubElement(channel_elem, 'created_at').text = channel.created_at.isoformat() ET.SubElement(channel_elem, 'updated_at').text = channel.updated_at.isoformat() # Network information if channel.network: network_elem = ET.SubElement(channel_elem, 'network') network_elem.set('id', str(channel.network.id)) ET.SubElement(network_elem, 'name').text = channel.network.name ET.SubElement(network_elem, 'description').text = channel.network.description or '' # Coverage information if include_coverage: coverage_elem = ET.SubElement(channel_elem, 'coverage_areas') for coverage in channel.coverage_areas.all(): area_elem = ET.SubElement(coverage_elem, 'coverage_area') area_elem.set('is_primary', str(coverage.is_primary).lower()) zone_elem = ET.SubElement(area_elem, 'zone') zone_elem.set('id', str(coverage.zone.id)) ET.SubElement(zone_elem, 'name').text = coverage.zone.name ET.SubElement(zone_elem, 'zone_type').text = coverage.zone.zone_type ET.SubElement(zone_elem, 'description').text = coverage.zone.description or '' ET.SubElement(area_elem, 'coverage_percentage').text = str(coverage.coverage_percentage) ET.SubElement(area_elem, 'signal_strength').text = coverage.signal_strength # Schedule information if include_schedule: schedule_elem = ET.SubElement(channel_elem, 'schedule') for schedule in channel.schedule_entries.all(): program_elem = ET.SubElement(schedule_elem, 'program') ET.SubElement(program_elem, 'title').text = schedule.program_title ET.SubElement(program_elem, 'start_time').text = schedule.start_time.isoformat() ET.SubElement(program_elem, 'end_time').text = schedule.end_time.isoformat() ET.SubElement(program_elem, 'description').text = schedule.description or '' ET.SubElement(program_elem, 'genre').text = schedule.genre or '' ET.SubElement(program_elem, 'rating').text = schedule.rating or '' # Write to file if pretty_print: # Pretty print XML rough_string = ET.tostring(root, encoding='unicode') reparsed = minidom.parseString(rough_string) with open(output_path, 'w', encoding='utf-8') as file: file.write(reparsed.toprettyxml(indent=" ")) else: # Compact XML tree = ET.ElementTree(root) tree.write(output_path, encoding='utf-8', xml_declaration=True)