"""Celery tasks for Jingles app.""" import os import hashlib import logging from datetime import datetime, timedelta from typing import List, Dict, Any, Optional from celery import shared_task from django.conf import settings from django.core.files.storage import default_storage from django.core.mail import send_mail from django.db import transaction from django.utils import timezone from django.template.loader import render_to_string from django.contrib.auth import get_user_model from .models import ( Jingle, JinglePlaylist, JinglePlaylistItem, JingleSchedule, JingleCategory, JingleType, ) from apps.channels.models import Channel User = get_user_model() logger = logging.getLogger(__name__) @shared_task(bind=True, max_retries=3) def process_jingle_upload(self, jingle_id: int, file_path: str) -> Dict[str, Any]: """Process uploaded jingle file - extract metadata, validate, and update record.""" try: jingle = Jingle.objects.get(id=jingle_id) # Update status to processing jingle.status = 'processing' jingle.save(update_fields=['status', 'updated_at']) # Extract file metadata metadata = _extract_file_metadata(file_path) # Update jingle with metadata jingle.file_size = metadata.get('file_size', 0) jingle.duration_seconds = metadata.get('duration', 0) jingle.file_format = metadata.get('format', '').lower() jingle.md5_hash = metadata.get('md5_hash', '') # Set video properties if it's a video file if metadata.get('is_video', False): jingle.video_width = metadata.get('width') jingle.video_height = metadata.get('height') jingle.video_fps = metadata.get('fps') jingle.video_bitrate = metadata.get('video_bitrate') # Set audio properties jingle.audio_bitrate = metadata.get('audio_bitrate') jingle.audio_sample_rate = metadata.get('sample_rate') jingle.audio_channels = metadata.get('channels') # Validate file validation_result = _validate_jingle_file(jingle, metadata) if validation_result['is_valid']: jingle.status = 'pending' jingle.quality = _determine_quality(metadata) else: jingle.status = 'rejected' jingle.rejection_reason = validation_result['reason'] jingle.save() # Send notification if jingle.uploaded_by: _send_processing_notification(jingle) logger.info(f"Processed jingle upload: {jingle_id} - Status: {jingle.status}") return { 'success': True, 'jingle_id': jingle_id, 'status': jingle.status, 'metadata': metadata } except Jingle.DoesNotExist: logger.error(f"Jingle not found: {jingle_id}") return {'success': False, 'error': 'Jingle not found'} except Exception as exc: logger.error(f"Error processing jingle upload {jingle_id}: {str(exc)}") # Update jingle status to failed try: jingle = Jingle.objects.get(id=jingle_id) jingle.status = 'failed' jingle.rejection_reason = f"Processing failed: {str(exc)}" jingle.save() except: pass # Retry task if self.request.retries < self.max_retries: raise self.retry(countdown=60 * (2 ** self.request.retries)) return {'success': False, 'error': str(exc)} @shared_task def bulk_approve_jingles(jingle_ids: List[int], approved_by_id: int) -> Dict[str, Any]: """Bulk approve multiple jingles.""" try: approved_by = User.objects.get(id=approved_by_id) approved_count = 0 failed_ids = [] with transaction.atomic(): for jingle_id in jingle_ids: try: jingle = Jingle.objects.get( id=jingle_id, status='pending' ) jingle.status = 'approved' jingle.approved_by = approved_by jingle.approved_at = timezone.now() jingle.save() approved_count += 1 # Send approval notification if jingle.uploaded_by: _send_approval_notification(jingle, approved=True) except Jingle.DoesNotExist: failed_ids.append(jingle_id) except Exception as e: logger.error(f"Error approving jingle {jingle_id}: {str(e)}") failed_ids.append(jingle_id) logger.info(f"Bulk approved {approved_count} jingles by user {approved_by_id}") return { 'success': True, 'approved_count': approved_count, 'failed_ids': failed_ids } except User.DoesNotExist: return {'success': False, 'error': 'User not found'} except Exception as e: logger.error(f"Error in bulk approve: {str(e)}") return {'success': False, 'error': str(e)} @shared_task def bulk_reject_jingles(jingle_ids: List[int], rejected_by_id: int, reason: str = '') -> Dict[str, Any]: """Bulk reject multiple jingles.""" try: rejected_by = User.objects.get(id=rejected_by_id) rejected_count = 0 failed_ids = [] with transaction.atomic(): for jingle_id in jingle_ids: try: jingle = Jingle.objects.get( id=jingle_id, status='pending' ) jingle.status = 'rejected' jingle.rejection_reason = reason or 'Bulk rejection' jingle.save() rejected_count += 1 # Send rejection notification if jingle.uploaded_by: _send_approval_notification(jingle, approved=False) except Jingle.DoesNotExist: failed_ids.append(jingle_id) except Exception as e: logger.error(f"Error rejecting jingle {jingle_id}: {str(e)}") failed_ids.append(jingle_id) logger.info(f"Bulk rejected {rejected_count} jingles by user {rejected_by_id}") return { 'success': True, 'rejected_count': rejected_count, 'failed_ids': failed_ids } except User.DoesNotExist: return {'success': False, 'error': 'User not found'} except Exception as e: logger.error(f"Error in bulk reject: {str(e)}") return {'success': False, 'error': str(e)} @shared_task def update_jingle_play_count(jingle_id: int, playlist_item_id: Optional[int] = None) -> Dict[str, Any]: """Update play count for jingle and playlist item.""" try: with transaction.atomic(): # Update jingle play count jingle = Jingle.objects.select_for_update().get(id=jingle_id) jingle.play_count += 1 jingle.last_played = timezone.now() jingle.save(update_fields=['play_count', 'last_played', 'updated_at']) # Update playlist item play count if provided if playlist_item_id: try: playlist_item = JinglePlaylistItem.objects.select_for_update().get( id=playlist_item_id ) playlist_item.play_count += 1 playlist_item.last_played = timezone.now() playlist_item.save(update_fields=['play_count', 'last_played']) except JinglePlaylistItem.DoesNotExist: pass return { 'success': True, 'jingle_id': jingle_id, 'new_play_count': jingle.play_count } except Jingle.DoesNotExist: return {'success': False, 'error': 'Jingle not found'} except Exception as e: logger.error(f"Error updating play count for jingle {jingle_id}: {str(e)}") return {'success': False, 'error': str(e)} @shared_task def generate_playlist_from_schedule(schedule_id: int) -> Dict[str, Any]: """Generate playlist items based on schedule configuration.""" try: schedule = JingleSchedule.objects.get(id=schedule_id) if not schedule.is_active or not schedule.playlist: return {'success': False, 'error': 'Schedule or playlist not active'} # Get available jingles based on schedule criteria jingles = _get_jingles_for_schedule(schedule) if not jingles: return {'success': False, 'error': 'No available jingles found'} # Clear existing playlist items if configured if schedule.auto_clear_playlist: schedule.playlist.playlist_items.all().delete() # Generate playlist items created_items = [] for order, jingle in enumerate(jingles, 1): playlist_item, created = JinglePlaylistItem.objects.get_or_create( playlist=schedule.playlist, jingle=jingle, defaults={ 'order': order, 'is_active': True, 'start_date': schedule.start_date, 'end_date': schedule.end_date, } ) if created: created_items.append(playlist_item.id) logger.info(f"Generated {len(created_items)} playlist items for schedule {schedule_id}") return { 'success': True, 'schedule_id': schedule_id, 'created_items': len(created_items), 'item_ids': created_items } except JingleSchedule.DoesNotExist: return {'success': False, 'error': 'Schedule not found'} except Exception as e: logger.error(f"Error generating playlist from schedule {schedule_id}: {str(e)}") return {'success': False, 'error': str(e)} @shared_task def cleanup_expired_jingles() -> Dict[str, Any]: """Clean up expired jingles and playlist items.""" try: today = timezone.now().date() # Deactivate expired jingles expired_jingles = Jingle.objects.filter( end_date__lt=today, is_active=True ) expired_count = expired_jingles.count() expired_jingles.update(is_active=False) # Deactivate expired playlist items expired_items = JinglePlaylistItem.objects.filter( end_date__lt=today, is_active=True ) expired_items_count = expired_items.count() expired_items.update(is_active=False) # Deactivate expired schedules expired_schedules = JingleSchedule.objects.filter( end_date__lt=today, is_active=True ) expired_schedules_count = expired_schedules.count() expired_schedules.update(is_active=False) logger.info( f"Cleanup completed: {expired_count} jingles, " f"{expired_items_count} playlist items, " f"{expired_schedules_count} schedules expired" ) return { 'success': True, 'expired_jingles': expired_count, 'expired_playlist_items': expired_items_count, 'expired_schedules': expired_schedules_count } except Exception as e: logger.error(f"Error in cleanup task: {str(e)}") return {'success': False, 'error': str(e)} @shared_task def cleanup_old_jingle_files(retention_days: int = 30) -> Dict[str, Any]: """Clean up old jingle files that are no longer referenced.""" try: cutoff_date = timezone.now() - timedelta(days=retention_days) # Find jingles marked as deleted and older than retention period old_jingles = Jingle.objects.filter( is_deleted=True, updated_at__lt=cutoff_date ) deleted_files = [] deleted_count = 0 for jingle in old_jingles: try: # Delete file from storage if jingle.file and default_storage.exists(jingle.file.name): default_storage.delete(jingle.file.name) deleted_files.append(jingle.file.name) # Delete jingle record jingle.delete() deleted_count += 1 except Exception as e: logger.error(f"Error deleting jingle {jingle.id}: {str(e)}") logger.info(f"Cleaned up {deleted_count} old jingle files") return { 'success': True, 'deleted_count': deleted_count, 'deleted_files': deleted_files } except Exception as e: logger.error(f"Error in file cleanup task: {str(e)}") return {'success': False, 'error': str(e)} @shared_task def generate_jingle_usage_report(channel_id: Optional[int] = None, days: int = 30) -> Dict[str, Any]: """Generate jingle usage report for specified period.""" try: end_date = timezone.now() start_date = end_date - timedelta(days=days) # Base queryset jingles_qs = Jingle.objects.filter( last_played__gte=start_date, last_played__lte=end_date ) if channel_id: jingles_qs = jingles_qs.filter(channel_id=channel_id) # Generate statistics total_jingles = jingles_qs.count() total_plays = sum(jingles_qs.values_list('play_count', flat=True)) # Top played jingles top_jingles = list( jingles_qs.order_by('-play_count')[:10].values( 'id', 'name', 'play_count', 'channel__name' ) ) # Category breakdown category_stats = {} for jingle in jingles_qs.select_related('category'): category_name = jingle.category.name if jingle.category else 'Uncategorized' if category_name not in category_stats: category_stats[category_name] = {'count': 0, 'plays': 0} category_stats[category_name]['count'] += 1 category_stats[category_name]['plays'] += jingle.play_count # Channel breakdown (if not filtered by channel) channel_stats = {} if not channel_id: for jingle in jingles_qs.select_related('channel'): channel_name = jingle.channel.name if jingle.channel else 'No Channel' if channel_name not in channel_stats: channel_stats[channel_name] = {'count': 0, 'plays': 0} channel_stats[channel_name]['count'] += 1 channel_stats[channel_name]['plays'] += jingle.play_count report_data = { 'period': { 'start_date': start_date.isoformat(), 'end_date': end_date.isoformat(), 'days': days }, 'summary': { 'total_jingles': total_jingles, 'total_plays': total_plays, 'average_plays': total_plays / total_jingles if total_jingles > 0 else 0 }, 'top_jingles': top_jingles, 'category_breakdown': category_stats, 'channel_breakdown': channel_stats } logger.info(f"Generated jingle usage report for {days} days") return { 'success': True, 'report_data': report_data } except Exception as e: logger.error(f"Error generating usage report: {str(e)}") return {'success': False, 'error': str(e)} @shared_task def send_daily_jingle_summary() -> Dict[str, Any]: """Send daily summary of jingle activities to administrators.""" try: today = timezone.now().date() yesterday = today - timedelta(days=1) # Get statistics for yesterday new_jingles = Jingle.objects.filter( created_at__date=yesterday ).count() approved_jingles = Jingle.objects.filter( approved_at__date=yesterday ).count() pending_jingles = Jingle.objects.filter( status='pending' ).count() total_plays_yesterday = Jingle.objects.filter( last_played__date=yesterday ).aggregate( total=models.Sum('play_count') )['total'] or 0 # Get top played jingles yesterday top_jingles = list( Jingle.objects.filter( last_played__date=yesterday ).order_by('-play_count')[:5].values( 'name', 'play_count', 'channel__name' ) ) # Prepare email context context = { 'date': yesterday, 'new_jingles': new_jingles, 'approved_jingles': approved_jingles, 'pending_jingles': pending_jingles, 'total_plays': total_plays_yesterday, 'top_jingles': top_jingles, } # Render email content subject = f"Daily Jingle Summary - {yesterday.strftime('%Y-%m-%d')}" html_content = render_to_string('jingles/emails/daily_summary.html', context) text_content = render_to_string('jingles/emails/daily_summary.txt', context) # Get admin email addresses admin_emails = list( User.objects.filter( is_staff=True, is_active=True, email__isnull=False ).exclude(email='').values_list('email', flat=True) ) if admin_emails: send_mail( subject=subject, message=text_content, html_message=html_content, from_email=settings.DEFAULT_FROM_EMAIL, recipient_list=admin_emails, fail_silently=False ) logger.info(f"Sent daily jingle summary to {len(admin_emails)} administrators") return { 'success': True, 'recipients': len(admin_emails), 'summary': context } except Exception as e: logger.error(f"Error sending daily summary: {str(e)}") return {'success': False, 'error': str(e)} # Helper functions def _extract_file_metadata(file_path: str) -> Dict[str, Any]: """Extract metadata from media file.""" metadata = { 'file_size': 0, 'duration': 0, 'format': '', 'md5_hash': '', 'is_video': False, 'width': None, 'height': None, 'fps': None, 'video_bitrate': None, 'audio_bitrate': None, 'sample_rate': None, 'channels': None, } try: # Get file size if os.path.exists(file_path): metadata['file_size'] = os.path.getsize(file_path) # Calculate MD5 hash with open(file_path, 'rb') as f: metadata['md5_hash'] = hashlib.md5(f.read()).hexdigest() # Get file format from extension _, ext = os.path.splitext(file_path) metadata['format'] = ext.lstrip('.').lower() # Determine if it's a video file video_formats = ['mp4', 'mov', 'avi', 'mkv', 'webm'] metadata['is_video'] = metadata['format'] in video_formats # Try to extract media metadata using ffprobe (if available) try: import subprocess import json cmd = [ 'ffprobe', '-v', 'quiet', '-print_format', 'json', '-show_format', '-show_streams', file_path ] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode == 0: data = json.loads(result.stdout) # Extract format info if 'format' in data: metadata['duration'] = float(data['format'].get('duration', 0)) # Extract stream info for stream in data.get('streams', []): if stream.get('codec_type') == 'video': metadata['width'] = stream.get('width') metadata['height'] = stream.get('height') metadata['fps'] = eval(stream.get('r_frame_rate', '0/1')) metadata['video_bitrate'] = stream.get('bit_rate') elif stream.get('codec_type') == 'audio': metadata['audio_bitrate'] = stream.get('bit_rate') metadata['sample_rate'] = stream.get('sample_rate') metadata['channels'] = stream.get('channels') except (ImportError, subprocess.SubprocessError, json.JSONDecodeError): # Fallback to basic metadata if ffprobe is not available pass except Exception as e: logger.error(f"Error extracting metadata from {file_path}: {str(e)}") return metadata def _validate_jingle_file(jingle: Jingle, metadata: Dict[str, Any]) -> Dict[str, Any]: """Validate jingle file based on metadata and business rules.""" validation_result = {'is_valid': True, 'reason': ''} try: # Check file size limits max_size = getattr(settings, 'JINGLE_MAX_FILE_SIZE', 50 * 1024 * 1024) # 50MB default if metadata['file_size'] > max_size: validation_result['is_valid'] = False validation_result['reason'] = f"File size exceeds limit ({max_size} bytes)" return validation_result # Check duration limits max_duration = getattr(settings, 'JINGLE_MAX_DURATION', 300) # 5 minutes default if metadata['duration'] > max_duration: validation_result['is_valid'] = False validation_result['reason'] = f"Duration exceeds limit ({max_duration} seconds)" return validation_result # Check allowed formats allowed_formats = getattr(settings, 'JINGLE_ALLOWED_FORMATS', ['mp3', 'wav', 'aac', 'mp4', 'mov']) if metadata['format'] not in allowed_formats: validation_result['is_valid'] = False validation_result['reason'] = f"Format '{metadata['format']}' not allowed" return validation_result # Check for duplicate MD5 hash if metadata['md5_hash']: existing = Jingle.objects.filter( md5_hash=metadata['md5_hash'], is_deleted=False ).exclude(id=jingle.id) if existing.exists(): validation_result['is_valid'] = False validation_result['reason'] = "Duplicate file detected" return validation_result except Exception as e: logger.error(f"Error validating jingle file: {str(e)}") validation_result['is_valid'] = False validation_result['reason'] = f"Validation error: {str(e)}" return validation_result def _determine_quality(metadata: Dict[str, Any]) -> str: """Determine quality rating based on file metadata.""" try: # Basic quality determination based on bitrate and format audio_bitrate = int(metadata.get('audio_bitrate', 0)) file_format = metadata.get('format', '').lower() # High quality criteria if file_format in ['wav', 'flac'] or audio_bitrate >= 320000: return 'high' # Medium quality criteria elif file_format in ['aac', 'mp3'] and audio_bitrate >= 128000: return 'medium' # Low quality else: return 'low' except: return 'medium' # Default to medium if unable to determine def _get_jingles_for_schedule(schedule: JingleSchedule) -> List[Jingle]: """Get available jingles based on schedule criteria.""" today = timezone.now().date() # Base queryset for available jingles jingles = Jingle.objects.filter( channel=schedule.channel, is_active=True, status='approved', is_deleted=False ).filter( models.Q(start_date__isnull=True) | models.Q(start_date__lte=today) ).filter( models.Q(end_date__isnull=True) | models.Q(end_date__gte=today) ) # Apply schedule-specific filters if schedule.jingle_categories.exists(): jingles = jingles.filter(category__in=schedule.jingle_categories.all()) if schedule.jingle_types.exists(): jingles = jingles.filter(jingle_type__in=schedule.jingle_types.all()) # Order by priority and randomize within same priority jingles = jingles.order_by('-priority', '?') # Limit based on schedule configuration max_items = getattr(schedule, 'max_playlist_items', 50) return list(jingles[:max_items]) def _send_processing_notification(jingle: Jingle) -> None: """Send notification about jingle processing status.""" try: if not jingle.uploaded_by or not jingle.uploaded_by.email: return context = { 'jingle': jingle, 'user': jingle.uploaded_by, } subject = f"Jingle Processing Update: {jingle.name}" html_content = render_to_string('jingles/emails/processing_notification.html', context) text_content = render_to_string('jingles/emails/processing_notification.txt', context) send_mail( subject=subject, message=text_content, html_message=html_content, from_email=settings.DEFAULT_FROM_EMAIL, recipient_list=[jingle.uploaded_by.email], fail_silently=True ) except Exception as e: logger.error(f"Error sending processing notification: {str(e)}") def _send_approval_notification(jingle: Jingle, approved: bool) -> None: """Send notification about jingle approval/rejection.""" try: if not jingle.uploaded_by or not jingle.uploaded_by.email: return context = { 'jingle': jingle, 'user': jingle.uploaded_by, 'approved': approved, } status = 'Approved' if approved else 'Rejected' subject = f"Jingle {status}: {jingle.name}" template_name = 'approval_notification' if approved else 'rejection_notification' html_content = render_to_string(f'jingles/emails/{template_name}.html', context) text_content = render_to_string(f'jingles/emails/{template_name}.txt', context) send_mail( subject=subject, message=text_content, html_message=html_content, from_email=settings.DEFAULT_FROM_EMAIL, recipient_list=[jingle.uploaded_by.email], fail_silently=True ) except Exception as e: logger.error(f"Error sending approval notification: {str(e)}")