import os import subprocess import re import json import redis import shutil import sys from app.config import settings class VideoProcessor: """Process video files to HLS format using FFmpeg""" def __init__(self): self.ffmpeg_cache_path = settings.FFMPEG_CACHE_PATH self.storage_path = settings.STORAGE_PATH self.upload_path = settings.UPLOAD_PATH from app.redis_client import redis_client self.redis_client = redis_client def process_video(self, stream_id: str) -> bool: """Convert video to HLS format""" progress_key = f"processing_progress:{stream_id}" state_key = f"processing_state:{stream_id}" try: # Paths video_path = f"{self.upload_path}/videos/{stream_id}.mp4" hls_dir = f"{self.storage_path}/hls/{stream_id}" hls_path = f"{hls_dir}/master.m3u8" hls_url_path = f"/storage/hls/{stream_id}/master.m3u8" # Check if there's existing state to resume from existing_state = self.redis_client.get(state_key) resume_from = 0 if existing_state: state = json.loads(existing_state) if state.get('status') == 'interrupted': resume_from = state.get('last_successful_segment', 0) sys.stderr.write(f"[PROCESSING] Resuming from segment {resume_from}\n") sys.stderr.flush() # Create HLS directory (only clean if not resuming) if resume_from == 0: if os.path.exists(hls_dir): shutil.rmtree(hls_dir) os.makedirs(hls_dir, exist_ok=True) else: os.makedirs(hls_dir, exist_ok=True) sys.stderr.write(f"[PROCESSING] Starting HLS processing for stream {stream_id}\n") sys.stderr.write(f"[PROCESSING] Input: {video_path}\n") sys.stderr.write(f"[PROCESSING] Output: {hls_path}\n") sys.stderr.flush() # Initialize progress in Redis self.redis_client.set(progress_key, json.dumps({'percentage': 0, 'status': 'starting'})) # FFmpeg command for HLS conversion with progress if resume_from > 0: start_time = resume_from * 2 cmd = [ 'ffmpeg', '-ss', str(start_time), '-i', video_path, '-c:v', 'libx264', '-c:a', 'aac', '-b:a', '128k', '-f', 'hls', '-hls_time', '2', '-hls_list_size', '0', '-hls_segment_filename', f'{hls_dir}/segment_%03d.ts', '-progress', 'pipe:1', hls_path ] sys.stderr.write(f"[PROCESSING] Seeking to {start_time} seconds\n") sys.stderr.flush() else: cmd = [ 'ffmpeg', '-i', video_path, '-c:v', 'libx264', '-c:a', 'aac', '-b:a', '128k', '-f', 'hls', '-hls_time', '2', '-hls_list_size', '0', '-hls_segment_filename', f'{hls_dir}/segment_%03d.ts', '-progress', 'pipe:1', hls_path ] # Run FFmpeg and track progress process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1 ) # Parse progress from FFmpeg output duration = None for line in process.stdout: # Parse duration if 'Duration:' in line: match = re.search(r'Duration: (\d+):(\d+):(\d+)', line) if match: hours, minutes, seconds = map(int, match.groups()) duration = hours * 3600 + minutes * 60 + seconds # Parse progress if 'out_time=' in line: match = re.search(r'out_time=(\d+):(\d+):(\d+)', line) if match and duration: hours, minutes, seconds = map(int, match.groups()) current_time = hours * 3600 + minutes * 60 + seconds # Add resume offset to current time for correct percentage total_current_time = current_time + (resume_from * 2) percentage = (total_current_time / duration) * 100 if duration > 0 else 0 # Parse speed speed_match = re.search(r'speed=([\d.]+)x', line) speed = float(speed_match.group(1)) if speed_match else 1.0 # Calculate ETA (remaining time / speed) remaining_time = duration - total_current_time eta_seconds = (remaining_time / speed) if speed > 0 else 0 # Count segments segment_files = [f for f in os.listdir(hls_dir) if f.endswith('.ts')] current_segments = len(segment_files) + resume_from total_segments = int(duration / 2) if duration > 0 else 0 # Update progress in Redis progress_data = { 'percentage': min(percentage, 100), 'status': 'processing', 'speed': speed, 'eta_seconds': eta_seconds, 'current_segments': current_segments, 'total_segments': total_segments } self.redis_client.set(progress_key, json.dumps(progress_data)) # Update state for resume capability state_data = { 'last_successful_segment': current_segments, 'total_segments': total_segments, 'last_position_seconds': total_current_time, 'video_duration': duration, 'status': 'processing' } self.redis_client.set(state_key, json.dumps(state_data)) # Wait for completion return_code = process.wait(timeout=7200) # 2 hour timeout if return_code != 0: # Update progress with error self.redis_client.set(progress_key, json.dumps({'percentage': 0, 'status': 'error'})) raise RuntimeError(f"FFmpeg failed with code {return_code}") # Verify HLS output if not os.path.exists(hls_path): self.redis_client.set(progress_key, json.dumps({'percentage': 0, 'status': 'error'})) raise RuntimeError("HLS master playlist not created") sys.stderr.write(f"[PROCESSING] HLS processing completed for stream {stream_id}\n") sys.stderr.flush() # Update progress to thumbnail generation self.redis_client.set(progress_key, json.dumps({'percentage': 95, 'status': 'generating_thumbnails', 'step_name': 'Generating thumbnails'})) # Generate thumbnails sys.stderr.write(f"[PROCESSING] Starting thumbnail generation for stream {stream_id}\n") sys.stderr.flush() try: # Get segment files segment_files = [f for f in os.listdir(hls_dir) if f.endswith('.ts')] segment_files.sort() if not segment_files: raise RuntimeError("No segment files found") total_segments = len(segment_files) generated_count = 0 # Create thumbnails directory thumbnails_dir = f"{self.storage_path}/thumbnails/{stream_id}" os.makedirs(thumbnails_dir, exist_ok=True) # Update database with thumbnail progress from app.database import SessionLocal from app.models.stream import Stream db = SessionLocal() try: stream = db.query(Stream).filter(Stream.id == stream_id).first() if stream: stream.thumbnail_progress_total = total_segments stream.thumbnail_progress_generated = 0 stream.thumbnail_progress_message = 'Starting thumbnail generation' stream.thumbnail_progress_error = None stream.thumbnail_progress_updated_at = None db.commit() finally: db.close() # Generate thumbnails for each segment for i, segment_file in enumerate(segment_files): segment_path = f"{hls_dir}/{segment_file}" thumbnail_path = f"{thumbnails_dir}/thumb_{i}.jpg" # Skip if thumbnail already exists if os.path.exists(thumbnail_path): generated_count += 1 # Update progress db = SessionLocal() try: stream = db.query(Stream).filter(Stream.id == stream_id).first() if stream: stream.thumbnail_progress_generated = generated_count stream.thumbnail_progress_message = f'Processing segment {i + 1} of {total_segments} (skipped, already exists)' stream.thumbnail_progress_updated_at = None db.commit() finally: db.close() continue # Generate thumbnail using ffmpeg cmd = [ 'ffmpeg', '-i', segment_path, '-vframes', '1', '-vf', 'scale=320:180', '-y', thumbnail_path ] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode == 0: generated_count += 1 # Update progress in Redis thumbnail_percentage = 95 + (generated_count / total_segments) * 5 self.redis_client.set(progress_key, json.dumps({ 'percentage': thumbnail_percentage, 'status': 'generating_thumbnails', 'step_name': 'Generating thumbnails', 'current_thumbnail': generated_count, 'total_thumbnails': total_segments })) # Update progress in database db = SessionLocal() try: stream = db.query(Stream).filter(Stream.id == stream_id).first() if stream: stream.thumbnail_progress_generated = generated_count stream.thumbnail_progress_message = f'Processing segment {i + 1} of {total_segments}' stream.thumbnail_progress_updated_at = None db.commit() finally: db.close() else: sys.stderr.write(f"[PROCESSING] Failed to generate thumbnail for segment {i}: {result.stderr}\n") sys.stderr.flush() # Clear thumbnail progress in database db = SessionLocal() try: stream = db.query(Stream).filter(Stream.id == stream_id).first() if stream: stream.thumbnail_progress_generated = 0 stream.thumbnail_progress_total = 0 stream.thumbnail_progress_message = None stream.thumbnail_progress_error = None stream.thumbnail_progress_updated_at = None db.commit() finally: db.close() sys.stderr.write(f"[PROCESSING] Thumbnail generation completed for stream {stream_id}\n") sys.stderr.flush() except Exception as e: sys.stderr.write(f"[PROCESSING ERROR] Thumbnail generation failed: {e}\n") sys.stderr.flush() import traceback traceback.print_exc() # Update progress with error self.redis_client.set(progress_key, json.dumps({'percentage': 0, 'status': 'error', 'error': str(e)})) raise # Update progress to complete self.redis_client.set(progress_key, json.dumps({'percentage': 100, 'status': 'completed'})) # Clear state on successful completion self.redis_client.delete(state_key) # Update database with HLS path from app.database import SessionLocal from app.models.stream import Stream db = SessionLocal() try: stream = db.query(Stream).filter(Stream.id == stream_id).first() if stream: stream.hls_path = hls_url_path stream.status = "ready" db.commit() sys.stderr.write(f"[PROCESSING] Updated stream {stream_id} status to ready\n") sys.stderr.flush() finally: db.close() # Regenerate master ffconcat from ..services.ffconcat_service import regenerate_master_ffconcat regenerate_master_ffconcat() # Clear progress after 5 minutes self.redis_client.expire(progress_key, 300) return True except subprocess.TimeoutExpired: self.redis_client.set(f"processing_progress:{stream_id}", json.dumps({'percentage': 0, 'status': 'timeout'})) self.redis_client.set(state_key, json.dumps({'status': 'interrupted'})) raise RuntimeError("HLS processing timed out") except Exception as e: self.redis_client.set(f"processing_progress:{stream_id}", json.dumps({'percentage': 0, 'status': 'error'})) self.redis_client.set(state_key, json.dumps({'status': 'interrupted'})) raise RuntimeError(f"Failed to process video: {str(e)}")