"""Celery Tasks for Channels App This module contains Celery tasks for the channels application, including FFmpeg processing for iframe extraction and jingle detection for ad break monitoring. Tasks: - extract_iframes_from_stream: Extract iframes from live stream URLs - detect_jingles_in_frames: Detect jingles in extracted frames - process_ad_break_detection: Main task for ad break detection - monitor_channel_health: Health monitoring for channels Author: Senior Django Developer Version: 1.0.0 """ import os import subprocess import tempfile import logging from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple from celery import shared_task, group, chord from celery.exceptions import Retry from django.conf import settings from django.utils import timezone from django.core.files.base import ContentFile from django.db import transaction from apps.channels.models import ( Channel, ChannelZone, ChannelZoneRelation, Jingle, JingleDetection, Adbreak, StandaloneVPNConfiguration, StandaloneFTPConfiguration ) from apps.activities.models import Activity # Configure logging logger = logging.getLogger('celery') @shared_task(bind=True, max_retries=3) def extract_iframes_from_stream(self, channel_id: str, zone_id: str, duration: int = 30): """ Extract iframes from a live stream URL using FFmpeg. This task captures iframes from the live stream for the specified duration and saves them for jingle detection analysis. Args: channel_id (str): UUID of the channel zone_id (str): UUID of the zone duration (int): Duration in seconds to capture (default: 30) Returns: dict: Task result with extracted frame information Raises: Retry: If FFmpeg extraction fails and retries are available """ try: # Get channel and zone relation relation = ChannelZoneRelation.objects.select_related( 'channel', 'zone' ).get(channel_id=channel_id, zone_id=zone_id) if not relation.stream_url: logger.error(f"No stream URL configured for channel {channel_id} in zone {zone_id}") return { 'success': False, 'error': 'No stream URL configured', 'channel_id': channel_id, 'zone_id': zone_id } # Create temporary directory for frames with tempfile.TemporaryDirectory() as temp_dir: output_pattern = os.path.join(temp_dir, 'frame_%04d.jpg') # FFmpeg command to extract iframes ffmpeg_cmd = [ 'ffmpeg', '-i', relation.stream_url, '-t', str(duration), '-vf', 'select=eq(pict_type\,I)', '-vsync', 'vfr', '-q:v', '2', '-f', 'image2', output_pattern, '-y' # Overwrite output files ] logger.info(f"Starting iframe extraction for channel {relation.channel.name} in zone {relation.zone.name}") # Execute FFmpeg command process = subprocess.run( ffmpeg_cmd, capture_output=True, text=True, timeout=duration + 60 # Add buffer time ) if process.returncode != 0: error_msg = f"FFmpeg failed: {process.stderr}" logger.error(error_msg) # Retry if we have retries left if self.request.retries < self.max_retries: logger.info(f"Retrying iframe extraction (attempt {self.request.retries + 1})") raise self.retry(countdown=60, exc=Exception(error_msg)) return { 'success': False, 'error': error_msg, 'channel_id': channel_id, 'zone_id': zone_id } # Count extracted frames frame_files = [f for f in os.listdir(temp_dir) if f.startswith('frame_') and f.endswith('.jpg')] frame_count = len(frame_files) logger.info(f"Extracted {frame_count} iframes from stream") # Trigger jingle detection for extracted frames if frame_count > 0: detect_jingles_in_frames.delay( channel_id=channel_id, zone_id=zone_id, frame_directory=temp_dir, frame_count=frame_count ) return { 'success': True, 'frame_count': frame_count, 'channel_id': channel_id, 'zone_id': zone_id, 'duration': duration } except ChannelZoneRelation.DoesNotExist: error_msg = f"Channel-Zone relation not found: {channel_id} - {zone_id}" logger.error(error_msg) return { 'success': False, 'error': error_msg, 'channel_id': channel_id, 'zone_id': zone_id } except subprocess.TimeoutExpired: error_msg = f"FFmpeg timeout after {duration + 60} seconds" logger.error(error_msg) if self.request.retries < self.max_retries: raise self.retry(countdown=120, exc=Exception(error_msg)) return { 'success': False, 'error': error_msg, 'channel_id': channel_id, 'zone_id': zone_id } except Exception as exc: error_msg = f"Unexpected error in iframe extraction: {str(exc)}" logger.error(error_msg, exc_info=True) if self.request.retries < self.max_retries: raise self.retry(countdown=60, exc=exc) return { 'success': False, 'error': error_msg, 'channel_id': channel_id, 'zone_id': zone_id } @shared_task(bind=True, max_retries=3) def detect_jingles_in_frames(self, channel_id: str, zone_id: str, frame_directory: str, frame_count: int): """ Detect jingles in extracted video frames. This task analyzes the extracted iframes to detect known jingles and identify potential ad break boundaries. Args: channel_id (str): UUID of the channel zone_id (str): UUID of the zone frame_directory (str): Directory containing extracted frames frame_count (int): Number of frames to analyze Returns: dict: Detection results with jingle matches """ try: # Get channel and active jingles channel = Channel.objects.get(id=channel_id) zone = ChannelZone.objects.get(id=zone_id) # Filter only active jingles active_jingles = Jingle.objects.filter( channel=channel, is_active=True ) # Process detections using available fingerprints # Note: video_fingerprint field was removed; using frames_fingerprint instead where applicable for jingle in active_jingles: if jingle.audio_fingerprint or getattr(jingle, 'frames_fingerprint', None): # Perform detection logic (placeholder) pass if not active_jingles.exists(): logger.info(f"No active jingles found for channel {channel.name}") return { 'success': True, 'detections': [], 'channel_id': channel_id, 'zone_id': zone_id } detections = [] # Analyze frames for jingle detection # Note: This is a simplified implementation # In production, you would use audio/video fingerprinting libraries for jingle in active_jingles: if jingle.audio_fingerprint or getattr(jingle, 'frames_fingerprint', None): # Simulate jingle detection logic # In real implementation, use libraries like: # - dejavu for audio fingerprinting # - OpenCV for video analysis # - TensorFlow/PyTorch for ML-based detection confidence_score = self._simulate_jingle_detection( jingle, frame_directory, frame_count ) if confidence_score > 0.7: # Threshold for positive detection detection_time = timezone.now() # Create jingle detection record with transaction.atomic(): detection = JingleDetection.objects.create( channel=channel, jingle=jingle, start_timestamp=detection_time, end_timestamp=detection_time + timedelta(seconds=jingle.duration or 5), confidence_score=confidence_score, detection_method='video_analysis', stream_position=0, # Would be calculated from frame timing status='detected' ) detections.append({ 'jingle_id': str(jingle.id), 'jingle_name': jingle.name, 'jingle_type': jingle.jingle_type, 'placement_type': jingle.placement_type, 'confidence_score': float(confidence_score), 'detection_id': str(detection.id) }) # Log activity Activity.objects.create( user=None, # System-generated action='jingle_detected', content_object=detection, description=f"Jingle '{jingle.name}' detected in {channel.name} with {confidence_score:.2%} confidence" ) logger.info(f"Detected {len(detections)} jingles in {frame_count} frames") # Trigger ad break analysis if commercial jingles detected commercial_detections = [ d for d in detections if d.get('placement_type') in ['commercial_start', 'commercial_end'] ] if commercial_detections: analyze_ad_breaks.delay( channel_id=channel_id, zone_id=zone_id, detections=commercial_detections ) return { 'success': True, 'detections': detections, 'channel_id': channel_id, 'zone_id': zone_id, 'frame_count': frame_count } except (Channel.DoesNotExist, ChannelZone.DoesNotExist) as exc: error_msg = f"Channel or Zone not found: {str(exc)}" logger.error(error_msg) return { 'success': False, 'error': error_msg, 'channel_id': channel_id, 'zone_id': zone_id } except Exception as exc: error_msg = f"Error in jingle detection: {str(exc)}" logger.error(error_msg, exc_info=True) if self.request.retries < self.max_retries: raise self.retry(countdown=30, exc=exc) return { 'success': False, 'error': error_msg, 'channel_id': channel_id, 'zone_id': zone_id } @shared_task(bind=True, max_retries=2) def analyze_ad_breaks(self, channel_id: str, zone_id: str, detections: List[Dict]): """ Analyze detected jingles to identify ad break patterns. Args: channel_id (str): UUID of the channel zone_id (str): UUID of the zone detections (List[Dict]): List of jingle detections Returns: dict: Ad break analysis results """ try: channel = Channel.objects.get(id=channel_id) # Analyze ad break patterns ad_breaks = [] for detection in detections: if detection.get('placement_type') == 'commercial_start': # Look for corresponding end jingle end_detection = next( (d for d in detections if d.get('placement_type') == 'commercial_end'), None ) if end_detection: ad_break_duration = 30 # Estimated duration ad_breaks.append({ 'start_jingle': detection, 'end_jingle': end_detection, 'estimated_duration': ad_break_duration }) logger.info(f"Identified {len(ad_breaks)} ad breaks for channel {channel.name}") return { 'success': True, 'ad_breaks': ad_breaks, 'channel_id': channel_id, 'zone_id': zone_id } except Exception as exc: error_msg = f"Error in ad break analysis: {str(exc)}" logger.error(error_msg, exc_info=True) return { 'success': False, 'error': error_msg, 'channel_id': channel_id, 'zone_id': zone_id } @shared_task def process_ad_break_detection(channel_id: str, zone_id: str, duration: int = 30): """ Main task to orchestrate the ad break detection process. This task coordinates iframe extraction and jingle detection for a specific channel-zone combination. Args: channel_id (str): UUID of the channel zone_id (str): UUID of the zone duration (int): Duration in seconds to monitor Returns: dict: Overall process results """ logger.info(f"Starting ad break detection for channel {channel_id} in zone {zone_id}") # Start iframe extraction result = extract_iframes_from_stream.delay( channel_id=channel_id, zone_id=zone_id, duration=duration ) return { 'task_id': result.id, 'channel_id': channel_id, 'zone_id': zone_id, 'duration': duration, 'status': 'started' } @shared_task def monitor_all_channels(): """ Periodic task to monitor all active channels for ad break detection. This task should be scheduled to run periodically (e.g., every 5 minutes) to continuously monitor active channels. """ active_relations = ChannelZoneRelation.objects.filter( is_active=True, channel__status='active', zone__is_active=True ).select_related('channel', 'zone') tasks = [] for relation in active_relations: if relation.stream_url: # Only monitor relations with stream URLs task = process_ad_break_detection.delay( channel_id=str(relation.channel.id), zone_id=str(relation.zone.id), duration=30 ) tasks.append(task.id) logger.info(f"Started monitoring {len(tasks)} channel-zone combinations") return { 'monitored_relations': len(tasks), 'task_ids': tasks } @shared_task(bind=True, max_retries=3) def update_channel_health(self, channel_id: str): """ Update health status for a specific channel. Args: channel_id (str): UUID of the channel Returns: dict: Health check results """ try: channel = Channel.objects.get(id=channel_id) # Check stream URLs for all zones healthy_zones = 0 total_zones = 0 for relation in channel.zone_relations.filter(is_active=True): total_zones += 1 if relation.stream_url: # Simple health check (in production, use proper stream validation) try: import requests response = requests.head(relation.stream_url, timeout=10) if response.status_code == 200: healthy_zones += 1 except: pass # Stream is not healthy # Update channel health is_online = healthy_zones > 0 if total_zones > 0 else False channel.update_health_status(is_online=is_online) return { 'success': True, 'channel_id': channel_id, 'is_online': is_online, 'healthy_zones': healthy_zones, 'total_zones': total_zones } except Channel.DoesNotExist: error_msg = f"Channel not found: {channel_id}" logger.error(error_msg) return { 'success': False, 'error': error_msg, 'channel_id': channel_id } except Exception as exc: error_msg = f"Error updating channel health: {str(exc)}" logger.error(error_msg, exc_info=True) if self.request.retries < self.max_retries: raise self.retry(countdown=60, exc=exc) return { 'success': False, 'error': error_msg, 'channel_id': channel_id } @shared_task(bind=True, max_retries=2) def create_adbreak_from_detection(self, detection_data: Dict): """ Create an Adbreak record from jingle detection data. This task processes jingle detection results and creates corresponding Adbreak records for tracking and analysis. Args: detection_data (dict): Detection data including timestamps and metadata Returns: dict: Task result with created adbreak information """ try: channel_id = detection_data.get('channel_id') jingle_id = detection_data.get('jingle_id') start_timestamp = detection_data.get('start_timestamp') end_timestamp = detection_data.get('end_timestamp') confidence_score = detection_data.get('confidence_score', 0.0) # Get the channel and jingle channel = Channel.objects.get(id=channel_id) jingle = Jingle.objects.get(id=jingle_id) # Determine adbreak type based on jingle placement adbreak_type = 'Mid' # Default if jingle.placement_type in ['commercial_start', 'start']: adbreak_type = 'Pre' elif jingle.placement_type in ['commercial_end', 'end']: adbreak_type = 'Post' # Calculate duration start_dt = datetime.fromisoformat(start_timestamp) end_dt = datetime.fromisoformat(end_timestamp) duration = end_dt - start_dt # Create Adbreak record adbreak = Adbreak.objects.create( channel=channel, date=start_dt.date(), start_at=start_dt.time(), end_at=end_dt.time(), duration=duration, adbreak_type=adbreak_type, adbreak_category='RealTime', show_before=detection_data.get('show_before', ''), show_after=detection_data.get('show_after', ''), if_show_during=detection_data.get('if_show_during', True) ) logger.info(f"Created adbreak {adbreak.id} for channel {channel.name}") return { 'success': True, 'adbreak_id': str(adbreak.id), 'channel_id': channel_id, 'adbreak_type': adbreak_type, 'duration_seconds': duration.total_seconds(), 'confidence_score': confidence_score } except Exception as exc: logger.error(f"Error creating adbreak from detection: {exc}") if self.request.retries < self.max_retries: raise self.retry(countdown=30) return { 'success': False, 'error': str(exc), 'detection_data': detection_data } @shared_task(bind=True) def setup_channel_monitoring(self, channel_id: str, zone_id: str = None): """ Set up comprehensive monitoring for a channel including VPN/FTP configuration validation. This task validates all configurations for a channel and sets up monitoring for ad break detection, health checks, and configuration validation. Args: channel_id (str): UUID of the channel to monitor zone_id (str, optional): Specific zone to monitor Returns: dict: Setup result with configuration status """ try: channel = Channel.objects.get(id=channel_id) setup_results = { 'channel_id': channel_id, 'channel_name': channel.name, 'zones_configured': [], 'vpn_configs_validated': [], 'ftp_configs_validated': [], 'monitoring_enabled': False } # Get zones to monitor if zone_id: zones = [ChannelZone.objects.get(id=zone_id)] else: zones = channel.get_active_zones() for zone in zones: zone_config = { 'zone_id': str(zone.id), 'zone_name': zone.name, 'stream_configured': False, 'vpn_configured': False, 'ftp_configured': False } # Check channel-zone relation try: relation = ChannelZoneRelation.objects.get( channel=channel, zone=zone, is_active=True ) # Validate stream URL if relation.stream_url: zone_config['stream_configured'] = True zone_config['stream_url'] = relation.stream_url # Validate VPN configuration vpn_config = relation.get_active_vpn_configuration() if vpn_config: zone_config['vpn_configured'] = True if hasattr(vpn_config, 'name'): # Standalone config zone_config['vpn_config_name'] = vpn_config.name setup_results['vpn_configs_validated'].append({ 'config_name': vpn_config.name, 'vpn_type': vpn_config.vpn_type, 'is_active': vpn_config.is_active }) else: # Legacy inline config zone_config['vpn_type'] = relation.vpn_type # Validate FTP configuration ftp_config = relation.get_active_ftp_configuration() if ftp_config: zone_config['ftp_configured'] = True if relation.standalone_ftp_config: # Standalone config ftp_standalone = relation.standalone_ftp_config zone_config['ftp_config_name'] = ftp_standalone.name setup_results['ftp_configs_validated'].append({ 'config_name': ftp_standalone.name, 'host': ftp_standalone.host, 'is_active': ftp_standalone.is_active }) else: # Legacy inline config zone_config['ftp_host'] = ftp_config['host'] setup_results['zones_configured'].append(zone_config) except ChannelZoneRelation.DoesNotExist: logger.warning(f"No active relation found for channel {channel.name} in zone {zone.name}") setup_results['zones_configured'].append(zone_config) # Enable monitoring if at least one zone is properly configured if any(z['stream_configured'] for z in setup_results['zones_configured']): setup_results['monitoring_enabled'] = True logger.info(f"Monitoring enabled for channel {channel.name}") return setup_results except Exception as exc: logger.error(f"Error setting up channel monitoring: {exc}") return { 'success': False, 'error': str(exc), 'channel_id': channel_id } @shared_task(bind=True, max_retries=2) def validate_standalone_configurations(self): """ Validate all standalone VPN and FTP configurations. This task checks the health and connectivity of standalone configurations and updates their status accordingly. Returns: dict: Validation results for all configurations """ try: results = { 'vpn_configs': [], 'ftp_configs': [], 'validation_time': timezone.now().isoformat() } # Validate VPN configurations vpn_configs = StandaloneVPNConfiguration.objects.filter(is_active=True) for vpn_config in vpn_configs: validation_result = { 'config_id': str(vpn_config.id), 'name': vpn_config.name, 'vpn_type': vpn_config.vpn_type, 'is_valid': False, 'error_message': None } try: # Basic validation based on VPN type if vpn_config.vpn_type == 'ipsec': # Validate IPSec configuration if vpn_config.ipsec_config and vpn_config.ipsec_config.gateway: validation_result['is_valid'] = True elif vpn_config.vpn_type == 'openvpn': # Validate OpenVPN configuration if vpn_config.openvpn_config and vpn_config.openvpn_config.server_address: validation_result['is_valid'] = True elif vpn_config.vpn_type == 'wireguard': # Validate WireGuard configuration if vpn_config.wireguard_config and vpn_config.wireguard_config.endpoint: validation_result['is_valid'] = True except Exception as e: validation_result['error_message'] = str(e) results['vpn_configs'].append(validation_result) # Validate FTP configurations ftp_configs = StandaloneFTPConfiguration.objects.filter(is_active=True) for ftp_config in ftp_configs: validation_result = { 'config_id': str(ftp_config.id), 'name': ftp_config.name, 'host': ftp_config.host, 'is_valid': False, 'error_message': None } try: # Basic FTP connectivity test import socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(10) result = sock.connect_ex((ftp_config.host, ftp_config.port)) sock.close() if result == 0: validation_result['is_valid'] = True else: validation_result['error_message'] = f"Cannot connect to {ftp_config.host}:{ftp_config.port}" except Exception as e: validation_result['error_message'] = str(e) results['ftp_configs'].append(validation_result) logger.info(f"Validated {len(results['vpn_configs'])} VPN and {len(results['ftp_configs'])} FTP configurations") return { 'success': True, 'results': results } except Exception as exc: logger.error(f"Error validating standalone configurations: {exc}") if self.request.retries < self.max_retries: raise self.retry(countdown=60) return { 'success': False, 'error': str(exc) } def _simulate_jingle_detection(jingle, frame_directory: str, frame_count: int) -> float: """ Simulate jingle detection logic. In a real implementation, this would use audio/video fingerprinting libraries to match the jingle against the extracted frames. Args: jingle: Jingle object to detect frame_directory: Directory containing frames frame_count: Number of frames Returns: float: Confidence score (0.0 to 1.0) """ # Simulate detection based on jingle properties # This is a placeholder - implement actual detection logic import random # Higher chance for commercial jingles if jingle.placement_type in ['commercial_start', 'commercial_end']: return random.uniform(0.6, 0.95) else: return random.uniform(0.3, 0.8)