U k,hO@sdZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl mZmZmZmZmZddlmZmZddlmZddlmZddlmZdd lmZdd lmZmZm Z m!Z!m"Z"dd l#m$Z$m%Z%m&Z&m'Z'dd l(m)Z)e*d Z+e,dddZ-eee.e fdddZ/e"ee.dddZ0e!ee.dddZ1dQeee!ee"ee.dddZ2dRee3e,ddd Z4ed!d"dd#dSe.e.e.d$d%d&Z5ee.d'd(d)Z6dTeee!ee"edd*d+Z7dUeee!ee"eedd,d-Z8ee.d'd.d/Z9ee,d0d1d2Z:ee.d3d4d5Z;ed6d7Zed!d>d;d#e.ee.d?d@dAZ?ed!d>d;d#e.dBdCdDZ@e.e.eAdEdFdGZBee,ddHdIZCedJdKZDee.dLdMdNZEedOdPZFdS)Wz Celery Tasks for Stream Processing This module contains all stream processing functionality as Celery tasks for better task management. All FFmpeg operations, HLS processing, and stream management are handled as background tasks. N)Path)OptionalDictAnyListTuple)datetime timedelta) shared_task)settings)timezone)ObjectDoesNotExist) StreamSessionChannel HLSSegmentVideoConfigurationAudioConfiguration)JingleTemplateJingleDetectionAdBreakDetectionStatistics)NotificationServicezstream_processor.tasks)returncCszDtjddgdddd}|jdkr2tdWdStd|jWn8tk r`td Yntjk r|td YnXd S) z Validate that FFmpeg is available and accessible. Returns: bool: True if FFmpeg is available, raises exception otherwise Raises: RuntimeError: If FFmpeg is not found or not executable ffmpegz-versionT capture_outputtexttimeoutrzFFmpeg validation successfulzFFmpeg test failed: z0FFmpeg not found in PATH. Please install FFmpeg.zFFmpeg validation timed outN) subprocessrun returncodeloggerinfo RuntimeErrorstderrFileNotFoundErrorTimeoutExpired)resultr)7/var/www/html/StreamProcessor/src/apps/streams/tasks.py_validate_ffmpeg%s    r+)channelrc Csttjd|j}|d}|d}|d}||||d}|D]v\}}z,|jddd|dtd |WqBt k r}z"t d |d |d |W5d }~XYqBXqB|S)z Create the necessary output directories for a channel. Args: channel (Channel): Channel model instance Returns: Dict[str, Path]: Dictionary containing paths to created directories OUTPUT_DIRhlslogsiframes)baser.r/r0Tparentsexist_okizCreated directory: zFailed to create z directory : N) rr STREAM_CONFIGslugitemsmkdirchmodr"r#OSErrorerror) r, base_pathhls_pathZ logs_path iframes_path directoriesnamepather)r)r*_create_output_directoriesBs$  rD) audio_configrc CsFg}|jr|ddg|d|jdt|jdt|jd|jg|S)z Build audio encoding options from configuration. Args: audio_config (AudioConfiguration): Audio configuration object Returns: List[str]: Audio encoding options for FFmpeg -afloudnorm=I=-16:LRA=11:TP=-1-c:a-ar-ac-b:a) normalizeextendcodecstr sample_ratechannelsbitrate)rEoptionsr)r)r*_build_audio_optionsgs rT) video_configrcCsld|jd|jd|jd|jd|jdt|jd|jd|jd |jg}|jd krh| d |j d d ddddddg |S)z Build video encoding options from configuration. Args: video_config (VideoConfiguration): Video configuration object Returns: List[str]: Video encoding options for FFmpeg -c:v -profile:v-level-s-aspectz-r-b:v-maxrate-bufsizeh264z-preset-crf20 -sc_threshold0-g48 -keyint_min) rNprofilelevel resolution aspect_ratiorO frame_rate min_bitrate max_bitraterMpreset)rUrSr)r)r*_build_video_optionssB  rn)r,rUrErcCs6t|}|d}dddddddd d |jd d d d g}|rF|t|n8|dddddttjddttjddtjdg |r|t|nX|dddddddtjdd tjd!d"tjd#d$tjd%d&tjd#d'd(d)d*d+d,d-d,g|d.dd/t|j d0t|j d1d2d3d4d5d6d7d2d8t|d9t|d:g|S);a] Build the FFmpeg command for stream capture. Args: channel (Channel): Channel configuration video_config (VideoConfiguration, optional): Video encoding settings audio_config (AudioConfiguration, optional): Audio encoding settings Returns: List[str]: FFmpeg command as list of arguments r.rz -hide_bannerz -loglevelr#z -user_agentzcmdr)r)r*_build_ffmpeg_commands     r<)sessionrrc Cslt}t||kr:z|jr0ztd|j}|sptd|jdd|_d|_|WWdS|d}|rt |dT}| }d | krtd |jd d|_d |_|W5QRWWdSW5QRXWnHt t fk r.}z$td |jd|WYWdSd}~XYnXt|jj|jj}|d} | d} | rt| d} t| dkrdd| D} | rtdt| dd|_|tt|jWdStdtdWqtk r6}z8td|d|_dt||_|WYdSd}~XYqXqtd|dd|_d |d|_|dS)!a Monitor a capture session for initial success. Args: session (StreamSession): Session to monitor timeout (int): Timeout in seconds for initial success check Returns: bool: True if capture appears successful, False otherwise /proc/zFFmpeg process z has terminatedfailed&FFmpeg process terminated unexpectedlyFcmdlinerrProcess z is not FFmpegzInvalid process IDzError checking process r5Nr.rz*.tsrcSsg|]}|jdkr|qS)r)statst_size).0sr)r)r* 7sz$_monitor_capture..z$Stream capture appears successful - z valid segments foundactiveTz!Found segments but they are emptyzError monitoring capture: zMonitoring error: z#Capture monitoring timed out after z secondszMonitoring timeout after )time process_idrexistsr"r<status last_errorsaveopenreadlowerr;IOErrorr,output_directoryr7listgloblenr#monitor_segmentsdelayrOidwarningsleep Exception) rr start_time proc_pathZ cmdline_pathfrrCZ output_dirr>Z playlist_filesegmentsZvalid_segmentsr)r)r*_monitor_capturesf      rT)bind max_retriesdefault_retry_delay) channel_idvideo_config_idaudio_config_idc Csztd|tjj|d}|r4tjj|dnd}|rJtjj|dnd}t|jj dd}| st d|j dt }|jdd |j d |j d td d t|||d} | rptd|j t }|jd|j t| j| j| r|nddd | r>td|j d|dntd|j ddt| j|j | jdWSd|j d} t| t }|jd|j | tdd t| Wntjk rd|d} t| t| Yntk r} ztd| |jj|jkrHtd |jjd!d"|j| d#n>z.t }|jd|t| tdd$d Wn YnXW5d} ~ XYnXdS)%a2 Start stream capture for a specific channel. This task initiates the FFmpeg stream capture process and creates the necessary directory structure and configuration files. Args: channel_id (str): UUID of the channel to capture video_config_id (str, optional): UUID of video configuration to use audio_config_id (str, optional): UUID of audio configuration to use Returns: dict: Task result with session information Raises: Exception: If stream capture fails to start z*Starting stream capture task for channel: rNT is_activezChannel zC has no active jingle templates. Jingle detection will be disabled. system_alertZmissing_jingle_templatesz Channel "zT" has no jingle templates. Please add jingle templates to enable ad break detection.)Z alert_type channel_namemessage timestamp template_typecontext)r,rUrEz0Stream capture started successfully for channel stream_startedr)r session_idrZjingle_templates_countz%Jingle detection enabled for channel z with z templatesz&Jingle detection disabled for channel z - no templates found)successrr started_atz"Stream capture failed for channel z after all retry attempts stream_error)r error_messagerChannel with ID not foundzStream capture task failed: z&Retrying stream capture task (attempt ))exc)rrrZretries_exhausted)r"r#robjectsgetrrr+jingle_templatesfilterrrrArsend_notificationr now isoformat_capture_with_retryrOrrcountr<r DoesNotExistrequestretriesrretry) selfrrrr,rUrErnotification_servicer error_msgrCr)r)r*start_stream_captureSs             r)rc Csz^tjj|d}|j}td|jd}ttjj|dj ddd}t | d}d }|D]}|j }||krrq\d } d } d } | | kr| d krzt |j} d d l} d d lm}d d l}|}t|d d}|| d|d dd}tjj|||t|| dd} WqWq~tk r}zdt|kr| d7} | | krb| d| WYPq~n&td|d| d|WY*qn td|d|WYqW5d }~XYq~Xq~| r\|d7}td||jd7_|jdgdq\|d kr6td|d||jdkrRt j!|gd dn|jdkrRt j!|gd d|||jd WStj"k rtd!|d"d#d$iYStk r}z*td%|d|d#t|iWYSd }~XYnXd S)&z Monitor and create database records for new HLS segments. Args: session_id (str): UUID of the stream session to monitor rz/app/media/streams/z/hls)rfilenameT)flatrrrN)ri@Biii)rsequence_numberr file_pathduration is_availablez.duplicate key value violates unique constraintrg{Gz?z$Failed to create segment record for z after z retries: r5zCreated HLS segment record: segments_processed) update_fieldszCreated z! new segment records for session )r processing)args countdown)rZ new_segmentstotal_segmentsSession z! not found for segment monitoringr<Zsession_not_foundz&Error monitoring segments for session )#rrrr,rr7setrr values_listrrrAfloatr~rrrandomrintrrandintcreaterOrrr"r<r#rrrr apply_asyncr)rrr,r>Zexisting_segmentsZ segment_filesZnew_segments_createdZ segment_filerr retry_countZ hls_segmentrrrrrZ timestamp_msrrCr)r)r*rs           rc Cstd|jtjj|ddgd}|rd}|jrtzd|j}tj |rzft |ddL}| }d | krd }td |jd |j|W5QRWWSW5QRXWnttfk rYnX|std |jd|jdd|_t|_d|_|WnXtk rp}z8td|d|_t|_d||_|W5d}~XYnXn2td |jdd|_t|_d|_|tjj|||dtd} znt|||} tdd| tj| tjtjd dd d} t | j!| _d| _| td| j!| WStk r}zDd| _t || _t| _| td|t"d|W5d}~XYnXdS) a Start stream capture for a channel. Args: channel (Channel): Channel to capture video_config (VideoConfiguration, optional): Video encoding settings audio_config (AudioConfiguration, optional): Audio encoding settings Returns: StreamSession: Created session object Raises: RuntimeError: If capture fails to start z%Starting stream capture for channel: rr)r, status__inFrz/cmdlinerrTzFound existing active session z for channel zExisting session z has dead process (PID: z), marking as failedrrzError checking process status: zProcess check failed: Nz% has no process ID, marking as failedzNo process ID recordedpending)r,rUrErrzFFmpeg command:  r)stdoutr%rbufsizeuniversal_newlinesz!Stream capture started with PID: z Failed to start stream capture: zStream capture failed: )#r"r#rArrrfirstrosrBrrrrrrr;rrr rended_atrrrr<rrjoinrPopenPIPErOpidr$) r,rUrEZexisting_sessionZprocess_runningrrrrCrrprocessr)r)r*_start_captureDs    "         rc Csd}|j}|j}||krzdtd|dd|t|||}t|rN|WS|d7}||krztd|dt|Wqt k r}zJt d|dd||d7}||krtd |dt|W5d }~XYqXqt d |d |j d S) az Start stream capture with automatic retry logic. Args: channel (Channel): Channel to capture video_config (VideoConfiguration, optional): Video encoding settings audio_config (AudioConfiguration, optional): Audio encoding settings Returns: Optional[StreamSession]: Session if successful, None if all retries failed rzStream capture attempt r/zCapture failed, retrying in z seconds...zCapture attempt z failed: z Retrying in NzAll z capture attempts failed for ) retry_attemptsretry_intervalr"r#rrrrrrr<rA)r,rUrErrrrrCr)r)r*rs2   rc Cs0ztd|tjj|d}t|}|rtd|t}|jd|jj t |j | rjt | nd|j dddt |j |jj |jr|jnd d WSd |}t|t|Wnhtjk rd |d }t|t|Yn6tk r*}ztd|W5d }~XYnXd S)z Stop an active stream capture session. Args: session_id (str): UUID of the stream session to stop Returns: dict: Task result with session information z%Stopping stream capture for session: rz0Stream capture stopped successfully for session stream_stoppedUnknown)rrrrrTN)rrrrz*Failed to stop stream capture for session zStream session with ID rz!Stop stream capture task failed: )r"r#rrr _stop_capturerrr,rArOrrrrrr<rr)rrrrrrCr)r)r*stop_stream_captures<        r)rrc Cstd|jz<|jr.ztt|jtjtd|jt dzz'_cleanup_old_segments..)keyrrzDeleted segment: zFailed to delete segment r5 Cleaned up z old segments for zError in cleanup task: )rr get_hls_pathsortedrrunlinkr"rr;rr#rArr<)r,rr>r deleted_countZsegments_to_deleterrCr)r)r*_cleanup_old_segmentss*   *r$cCsPztjj|d}|jj}tj|jsFt d|jdddWStj |jdkrvt d|jdd dWS|j jd d }|st d |jd z.ttjd|j}|d}|jd d dWnbttfk r0}z>t d|ttjd|j}|d}|jd d dW5d}~XYnXt|jj}||d} ddd|jddddddddt| g } ztj| d d d d!} | jdkr*t d"| j||d#} ddd|jdd$dddd%t| g } tj| d d d d!}|jdkrt d&|jdd'|jdWWS| } t d(|Wn4tjk r`t d)|dd*dYWSXt|  d+d,}t!t""|}g}|D]}zttj |d-krddl#}|$|}|dk r|j%ddkr|j%d.dkr|&|n t'|n t'|WnBt|j fk rBzt'|Wntk r<YnXYnXqt d/t(|d0||r|rt)*||n|rt d/t(|d1d2t(|||jd3WStj+k rt d4|dd5dYSt,k rJ}zJt d6|d7||j-j.|j/kr(|j0d8|d9dt|dWYSd}~XYnXdS):z Extract I-frames from an HLS segment for jingle detection. Args: segment_id (str): ID of the HLS segment to process Returns: dict: Task result with extracted frame paths rzSegment file does not exist: r<Zsegment_file_not_foundrr<rzSegment file is empty: rZempty_segment_fileTrz No jingle templates for channel z3, but extracting iframes for user to create jinglesr-r0r2z#Failed to create output directory: streamsNz_%d.pngrz-yroz-vfzselect=gt(scene\,0.10)z-vsyncZvfrz -frame_ptstruez-q:v2Zrz"Primary iframe extraction failed: z_iframe_%d.pngzselect=eq(pict_type\,I)3z Both extraction methods failed: zffmpeg_failed: z0Used alternative I-frame extraction for segment zFFmpeg timeout for segment Zffmpeg_timeoutz%d*irz Extracted z valid frames from segment z. iframes but no jingle templates for detectionr)rZframes_extracted frame_pathsZsegment_processedzSegment not found: Zsegment_not_foundz&Error extracting iframes from segment r5rrr)1rrrrr,rrBrrr"r<getsizerrrr#rArr r6r7r9KeyErrorr; MEDIA_ROOTstemrOrr r!r%r'replacer!rcv2imreadshaperremoverdetect_jingles_in_framesrrrrrrr)rr rr,rr=r?rCZ video_nameZoutput_patternrr(Zalt_output_patternZalt_cmdZ alt_resultZ frame_patternZ frame_filesZ valid_frames frame_pathr4imgr)r)r*r s          *  r r)r r-c Csz6tjj|d}|jj}|jjdd}g}|D]}|D]}|sFq8t|j |} |j p`t j d} | | kr8t jj|j||d| |d| d| dd } || td |jd | jd d t| jt} | jd|j|j| j| jddq8q0|D]*}zt|Wqtk r YqXqdt|t|dWStk r} zDt d| |j!j"|j#krz|j$d| ddt%| dWYSd} ~ XYnXdS)z Detect jingles by comparing extracted frames with templates. Args: segment_id (str): ID of the HLS segment frame_paths (List[str]): List of extracted frame file paths Returns: dict: Detection results rTrSIMILARITY_THRESHOLD?Zopencv_comparison)Zsimilarity_scoreZdetection_methodZthreshold_used)rrtemplateconfidence_scorer9frame_timestampmetadatazJingle detected: z (confidence: z.3frjingle_detected)rZ jingle_name confidencerrr)rZdetections_countZframes_processedz#Error detecting jingles in frames: rr.r<r&N)&rrrrr,rr image_exists_compare_images image_pathsimilarity_thresholdr JINGLE_CONFIGrrrr"r#rAr?process_ad_breakrrrrdetection_timerrr7r;rrr<rrrrrO)rr r-rr,r detectionsr9r>Z similarity threshold detectionrrCr)r)r*r8sd     r8) detection_idc Cszhtjj|d}|j}tjj|dd}|dkrftjj||jj d||j dd}t d|j n|j |j}tjd }tjd }||kr|krnnr||_|j |_t||_d |_|t d |j d |ddt} | jd|jj||j|jddnJ||_|j |_d|_d|_d|_d|_|t d|j d|dddddWStk r} zJt d|d| |jj |j!kr|j"d| ddt#| dWYSd} ~ XYnXdS)z Process a jingle detection to create or update ad breaks. Args: detection_id (str): ID of the jingle detection Returns: dict: Ad break processing result rT)rZend_time__isnullNGlobalr)rrregionstart_detectionrrzStarted new ad break: MIN_AD_BREAK_DURATIONMAX_AD_BREAK_DURATIONrzCompleted ad break: z (duration: z.1fzs)Zad_break_detected)rrrend_timerzReset ad break: z (invalid duration: r)rZad_break_processedz(Error processing ad break for detection r5rr.r<r&)$rrrrrrrrr,r7rJr"r#rrrr rH end_detectionrTrduration_secondsrrrrrArrQrr<rrrrrO) rrNrMrZactive_adbreakZad_breakrZ min_durationZ max_durationrrCr)r)r*rIsd       rI) image_1_path image_2_pathrc Csztj|rtj|s8td|d|WdSt|}t|}|dks\|dkrxtd|d|WdSt|jd|jdt|jd|jd}}|d ks|d krtd |d |WdSt |||f}t |||f}t |tj }t |tj } z"dd l m } | || } d| } Wntk rJd} YnXt|gdgdd gdd g} t| gdgdd gdd g}t| |tj}|jd| jdkr|jd| jdkrt| |tj}t|}nt|| tj}t|}d|}t|t| td}t|dd}| dk rV| d|d|d|d}n|d|d|d}tdtd|}|WStk r}z&td|d|d|WYdSd}~XYnXdS)aA Compare two images and return a similarity score using multiple methods. Args: image_1_path (str): Path to the first image (template) image_2_path (str): Path to the second image (frame) Returns: float: Similarity score (0.0 = identical, 1.0 = completely different) zImage files not found: z or r<Nz&Failed to load images for comparison: z, rr2z*Images too small for reliable comparison: r)structural_similarityrgo@g?g333333?g?g?r=zError comparing images z and r5)rrBrr"rr4r5minr6resizeZcvtColorZCOLOR_BGR2GRAYZskimage.metricsrZ ImportErrorZcalcHistZ compareHistZHISTCMP_BHATTACHARYYAZ matchTemplateZTM_CCOEFF_NORMEDnpmaxZmeanZastyperrr<)rWrXZimage_1Zimage_2heightwidthZimage_1_resizedZimage_2_resizedZgray_1Zgray_2ZssimZ ssim_scoreZ ssim_diffZhist_1Zhist_2Z hist_diffZtemplate_resultZtemplate_matchZ template_diffZmseZmse_normalizedZ combined_diffrCr)r)r*rEGsT   .    ,   "rEc Csz`t|}|d}ddddddg}t|d}|d |W5QRXtd |Wd Stk r}ztd |WYd Sd}~XYnXdS)z Create a master HLS playlist file for a channel. Args: channel (Channel): Channel to create playlist for Returns: bool: True if playlist was created successfully z master.m3u8z#EXTM3Uz#EXT-X-VERSION:3z#EXT-X-INDEPENDENT-SEGMENTSzz#EXT-X-MEDIA:TYPE=CLOSED-CAPTIONS,GROUP-ID="CC",LANGUAGE="eng",NAME="english",DEFAULT=YES,AUTOSELECT=YES,INSTREAM-ID="CC1"zy#EXT-X-STREAM-INF:BANDWIDTH=3405600,RESOLUTION=1280x720,CODECS="avc1.4d401f,mp4a.40.2",FRAME-RATE=25,CLOSED-CAPTIONS="CC"rzw zCreated master playlist: Tz"Failed to create master playlist: FN) rr rwriterr"r#rr<)r,r>Z master_pathcontentrrCr)r)r*_create_master_playlists"   rgc Cs@ztdtjjdd}d}|D]|}z4t|}||7}|dkrZtd|d|jWq$tk r}z$t d|jd|WYq$W5d }~XYq$Xq$dd l m }t |d d }tjj|d d}|}|td|d|dd||dWStk r:}zt d|W5d }~XYnXd S)z Periodic task to clean up old segment files and maintain disk space. This task removes old segment files beyond the configured retention limits for each channel. Returns: dict: Cleanup results zRunning segment cleanup taskTrrrz segments for channel z'Failed to cleanup segments for channel r5N)r )daysF)Zprocessed_at__ltrzSegment cleanup completed: z files, z DB records)rZ files_deletedZdb_records_deletedzSegment cleanup task failed: )r"rrrrr$r#rArr<rr r rrrdelete) rQZ total_deletedr,r#rCr Z cutoff_timeZ old_segmentsZdb_deleted_countr)r)r*cleanup_old_segmentss:   rk)rc CszLtd|tjj|d}t|}|rBdt|j|jdWSt dWnftj k rd|d}t |t |Yn4t k r}zt d|W5d }~XYnXd S) z Create HLS playlists for a channel. Args: channel_id (str): UUID of the channel Returns: dict: Task result z Creating playlists for channel: rT)rrrz Failed to create master playlistrrzPlaylist creation task failed: N) r"r#rrrrgrOrrArrr<)rr,rrrCr)r)r*create_playlist_for_channels"     rlc Csztdtjjdd}tjjtd}ddl m }|jjtd}ddl m }|jjtd }t }|j d d ||||td d d||||dWStk r}ztd|W5d}~XYnXdS)z Send periodic status updates about system operation. This task provides regular status reports about stream processing activity and system health. Returns: dict: Status report data z!Generating periodic status reportr)r)Zstarted_at__dater)r)Zdetection_time__date)r)Zstart_time__daterZperiodic_report)Z status_typersessions_todaydetections_todayad_breaks_todayrrT)rrrmrnrozPeriodic status task failed: N)r"rrrrrr rdateapps.jingles.modelsrrrrrrr<)rZtotal_sessions_todayrrnrrorrCr)r)r*send_periodic_statussD         rr)NN)r)NN)NN)NN)N)G__doc__rr4rrrrloggingrnumpyr_pathlibrtypingrrrrrrr celeryr django.confr django.utilsr django.core.exceptionsr Zapps.streams.modelsrrrrrrqrrrrZapps.notifications.servicesr getLoggerr"boolr+rOrDrTrnrrrrrrrrrrrr$r r8rIrrErgrkrlrrr)r)r)r*s       %% ^P p t 6771 c(  S QV# 8%