debug beta

app/routes.py

@@ -28,7 +28,7 @@ def files_page():
 
 @api_bp.route('/files')
 def list_files():
-    """List all encoded files with their details"""
+    """List all quality files that have been encoded (even if the overall job isn’t finished)"""
     try:
         encoded_dir = Path(current_app.config['ENCODED_FOLDER'])
         files = []
@@ -39,12 +39,14 @@ def list_files():
                     job_id = job_dir.name
                     job_info = encoder_service.get_job_info(job_id)
                     
-                    if job_info and job_info.get('status') == 'completed':
+                    # If there are any quality files available, include this job.
+                    if job_info and job_info.get('files'):
                         files.append({
                             'job_id': job_id,
                             'output_name': job_info.get('output_name', ''),
                             'created_at': job_info.get('start_time'),
-                            'completed_at': job_info.get('completion_time'),
+                            'completed_at': job_info.get('completion_time', None),
+                            'status': job_info.get('status'),
                             'qualities': {
                                 file['quality']: file['size'] 
                                 for file in job_info.get('files', [])
@@ -312,8 +314,7 @@ def serve_video(job_id, quality):
 def allowed_file(filename):
     """Check if the file extension is allowed"""
     ALLOWED_EXTENSIONS = {'mp4', 'mov', 'avi', 'mkv', 'wmv', 'flv', 'webm', '3gp','ts','m4v', 'mpg', 'mpeg'}
-    return '.' in filename and \
-           filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
+    return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
 
 def generate_job_id():
     """Generate a unique job ID"""

app/services/encoder_service.py

@@ -26,7 +26,7 @@ def _read_pipe(pipe, q):
 class EncoderService:
     def __init__(self):
         self.jobs = {}
-        self.threads = {}  # Store encoding thread references
+        self.threads = {}   # Store encoding thread references
         # Optimized settings for web streaming
         self.default_qualities = {
             '480p': {
@@ -36,7 +36,7 @@ class EncoderService:
                 'maxrate': '1500k',
                 'bufsize': '2000k',
                 'audio_bitrate': '128k',
-                'keyframe': '48',  # Keyframe every 2 seconds at 24fps
+                'keyframe': '48',   # Keyframe every 2 seconds at 24fps
                 'preset': 'ultrafast',  # Faster encoding speed
                 'profile': 'main',
                 'level': '3.1',
@@ -110,8 +110,64 @@ class EncoderService:
             logger.error(f"Failed to start encoding job: {str(e)}")
             return {'status': 'failed', 'error': str(e)}
 
+    def _split_video(self, input_file, split_dir, segment_duration=300): # 5 minutes = 300 seconds
+        """Splits the video into segments using ffmpeg."""
+        try:
+            split_cmd = [
+                'ffmpeg',
+                '-i', str(input_file),
+                '-c', 'copy',
+                '-map', '0',
+                '-segment_time', str(segment_duration),
+                '-f', 'segment',
+                '-segment_format', 'mp4',
+                '-reset_timestamps', '1', # Important to reset timestamps for each segment
+                str(split_dir / 'segment_%04d.mp4') # Using 4 digits for segment numbering
+            ]
+            logger.info(f"Splitting video with command: {' '.join(split_cmd)}")
+            subprocess.run(split_cmd, check=True, capture_output=True)
+            segments = sorted([split_dir / f for f in os.listdir(split_dir) if f.startswith('segment_') and f.endswith('.mp4')])
+            return segments
+        except subprocess.CalledProcessError as e:
+            logger.error(f"Error splitting video: {e.stderr.decode()}")
+            raise Exception(f"FFmpeg split failed: {e}")
+        except Exception as e:
+            logger.error(f"Error splitting video: {str(e)}")
+            raise
+
+    def _stitch_segments(self, segment_files, output_file):
+        """Stitches video segments back together using ffmpeg."""
+        try:
+            # Create a list file for concat demuxer
+            list_file_path = output_file.parent / 'segment_list.txt'
+            with open(list_file_path, 'w') as f:
+                for segment in segment_files:
+                    f.write(f"file '{segment}'\n")
+
+            stitch_cmd = [
+                'ffmpeg',
+                '-f', 'concat',
+                '-safe', '0', # Set to 0 if paths are relative, which they are in temp dir
+                '-i', str(list_file_path),
+                '-c', 'copy',
+                str(output_file)
+            ]
+            logger.info(f"Stitching video with command: {' '.join(stitch_cmd)}")
+            subprocess.run(stitch_cmd, check=True, capture_output=True)
+
+            # Clean up list file
+            if list_file_path.exists():
+                list_file_path.unlink()
+
+        except subprocess.CalledProcessError as e:
+            logger.error(f"Error stitching video: {e.stderr.decode()}")
+            raise Exception(f"FFmpeg stitch failed: {e}")
+        except Exception as e:
+            logger.error(f"Error stitching video: {str(e)}")
+            raise
+
     def _encode_video(self, filename, job_id):
-        """Internal method to handle video encoding with retries and enhanced error handling"""
+        """Internal method to handle video encoding with splitting, encoding segments, and stitching."""
         try:
             upload_path = Path(os.getenv('UPLOAD_FOLDER', 'uploads'))
             encoded_path = Path(os.getenv('ENCODED_FOLDER', 'encoded'))
@@ -123,20 +179,39 @@ class EncoderService:
             output_dir.mkdir(parents=True, exist_ok=True)
             logger.info(f"Created output directory {output_dir} for job {job_id}")
 
+            # Create a temporary directory for video segments
+            split_dir = output_dir / 'segments'
+            split_dir.mkdir(exist_ok=True)
+            segment_files = []
+
+            try:
+                segment_files = self._split_video(input_file, split_dir)
+                logger.info(f"Video split into {len(segment_files)} segments for job {job_id}")
+            except Exception as split_err:
+                raise Exception(f"Video splitting failed: {split_err}")
+
             qualities = self.jobs[job_id]['settings']
-            total_steps = len(qualities)
-            completed_steps = 0
+            total_qualities = len(qualities)
             outputs = []
+            global_progress_steps = total_qualities * len(segment_files) + total_qualities # segments encoding + stitching
+            completed_global_steps = 0
+
+            # Get video duration (using the first segment for approximation, or original file if needed)
+            duration = 0
+            if segment_files:
+                duration = self._get_video_duration(segment_files[0]) # Using first segment duration
+            else: # Fallback to original video if no segments (unlikely, but for robustness)
+                duration = self._get_video_duration(input_file)
 
-            # Get video duration
-            duration = self._get_video_duration(input_file)
             if not duration:
                 raise Exception("Could not determine video duration")
-            logger.info(f"Video duration for job {job_id}: {duration} seconds")
+            logger.info(f"Video duration (using segment or original) for job {job_id}: {duration} seconds")
+
 
             for quality, settings in qualities.items():
                 max_retries = 3
                 attempts = 0
+                encoded_segments_for_quality = []
                 success = False
 
                 while attempts < max_retries and not success:
@@ -144,99 +219,135 @@ class EncoderService:
                     self.jobs[job_id].update({
                         'status': 'processing',
                         'current_quality': quality,
-                        'progress': (completed_steps / total_steps) * 100
+                        'progress': (completed_global_steps / global_progress_steps) * 100
                     })
-                    output_file = output_dir / f"{output_name}_{quality}.mp4"
-                    
-                    # Build the FFmpeg command
-                    cmd = [
-                        'ffmpeg', '-y',
-                        '-i', str(input_file),
-                        '-c:v', 'libx264',
-                        '-preset', settings['preset'],
-                        '-profile:v', settings['profile'],
-                        '-level', settings['level'],
-                        '-tune', settings['tune'],
-                        '-b:v', settings['bitrate'],
-                        '-maxrate', settings['maxrate'],
-                        '-bufsize', settings['bufsize'],
-                        '-vf', f"scale={settings['width']}:{settings['height']}",
-                        '-g', settings['keyframe'],
-                        '-keyint_min', settings['keyframe'],
-                        '-sc_threshold', '0',
-                        '-c:a', 'aac',
-                        '-b:a', settings['audio_bitrate'],
-                        '-ar', '48000',
-                        '-ac', '2',
-                        '-movflags', '+faststart',
-                        '-progress', 'pipe:1',
-                        str(output_file)
-                    ]
-
-                    try:
-                        process = subprocess.Popen(
-                            cmd,
-                            stdout=subprocess.PIPE,
-                            stderr=subprocess.PIPE,
-                            universal_newlines=True
-                        )
-                        self.active_processes[job_id] = process
-
-                        # Use a queue and separate thread to read stdout
-                        q = queue.Queue()
-                        reader_thread = threading.Thread(target=_read_pipe, args=(process.stdout, q))
-                        reader_thread.daemon = True
-                        reader_thread.start()
-
-                        last_output_time = time.time()
-                        while True:
-                            try:
-                                # Wait up to 5 seconds for a line
-                                line = q.get(timeout=5)
-                                last_output_time = time.time()
-                                prog = self._parse_ffmpeg_progress(line, duration)
-                                if prog is not None:
-                                    quality_progress = ((completed_steps + prog / 100) / total_steps) * 100
-                                    self.jobs[job_id]['progress'] = quality_progress
-                            except queue.Empty:
-                                # If no output in 5 seconds, check if process is still alive
-                                if time.time() - last_output_time > 30:
-                                    logger.warning(f"No ffmpeg output for 30 seconds on quality {quality}, attempt {attempts+1} for job {job_id}")
+
+                    segment_output_files = []
+                    segment_success = True # Track if all segments for current quality encoded successfully
+                    completed_steps = 0
+
+                    for segment_input_file in segment_files:
+                        segment_output_file = split_dir / f"{segment_input_file.stem}_{quality}.mp4" # Output segment in split_dir
+                        segment_output_files.append(segment_output_file)
+
+                        cmd = [
+                            'ffmpeg', '-y',
+                            '-i', str(segment_input_file),
+                            '-c:v', 'libx264',
+                            '-preset', settings['preset'],
+                            '-profile:v', settings['profile'],
+                            '-level', settings['level'],
+                            '-tune', settings['tune'],
+                            '-b:v', settings['bitrate'],
+                            '-maxrate', settings['maxrate'],
+                            '-bufsize', settings['bufsize'],
+                            '-vf', f"scale={settings['width']}:{settings['height']}",
+                            '-g', settings['keyframe'],
+                            '-keyint_min', settings['keyframe'],
+                            '-sc_threshold', '0',
+                            '-c:a', 'aac',
+                            '-b:a', settings['audio_bitrate'],
+                            '-ar', '48000',
+                            '-ac', '2',
+                            '-movflags', '+faststart',
+                            '-progress', 'pipe:1',
+                            str(segment_output_file)
+                        ]
+
+                        try:
+                            process = subprocess.Popen(
+                                cmd,
+                                stdout=subprocess.PIPE,
+                                stderr=subprocess.PIPE,
+                                universal_newlines=True
+                            )
+                            self.active_processes[job_id] = process
+
+                            q = queue.Queue()
+                            reader_thread = threading.Thread(target=_read_pipe, args=(process.stdout, q))
+                            reader_thread.daemon = True
+                            reader_thread.start()
+
+                            last_output_time = time.time()
+                            segment_progress = 0
+                            while True:
+                                try:
+                                    line = q.get(timeout=5)
+                                    last_output_time = time.time()
+                                    prog = self._parse_ffmpeg_progress(line, duration) # Using segment duration for progress
+                                    if prog is not None:
+                                        segment_progress = prog
+                                        quality_progress = ((completed_global_steps + (completed_steps * 100 + prog) / 100 ) / global_progress_steps) * 100 # Approximation
+                                        self.jobs[job_id]['progress'] = quality_progress # Update overall job progress - this is rough, needs refinement for segments
+                                except queue.Empty:
+                                    if time.time() - last_output_time > 30:
+                                        logger.warning(f"No ffmpeg output for 30 seconds on quality {quality}, segment {segment_input_file.name}, attempt {attempts+1} for job {job_id}")
+                                        break
+                                if process.poll() is not None:
                                     break
-                            if process.poll() is not None:
-                                break
+                            reader_thread.join(timeout=5)
+
+                            if process.returncode == 0:
+                                logger.info(f"Encoding successful for quality {quality}, segment {segment_input_file.name} on attempt {attempts+1} for job {job_id}")
+                                completed_global_steps += 1 # Increment for each successful segment encoding
+                                encoded_segments_for_quality.append(segment_output_file)
+                            else:
+                                error_output = process.stderr.read()
+                                logger.error(f"FFmpeg error on quality {quality}, segment {segment_input_file.name}, attempt {attempts+1} for job {job_id}: {error_output}")
+                                segment_success = False # Indicate segment encoding failure
+                                raise Exception(f"FFmpeg failed for quality {quality}, segment {segment_input_file.name}")
 
-                        reader_thread.join(timeout=5)
-                        # Check if process completed successfully
-                        if process.returncode == 0:
-                            logger.info(f"Encoding successful for quality {quality} on attempt {attempts+1} for job {job_id}")
+                        except Exception as e:
+                            logger.error(f"Error encoding segment {segment_input_file.name} for quality {quality} on attempt {attempts+1} for job {job_id}: {str(e)}")
+                            segment_success = False # Ensure segment_success is False if exception occurs
+                            attempts += 1
+                            if attempts < max_retries:
+                                logger.info(f"Retrying encoding for quality {quality}, segment {segment_input_file.name} (attempt {attempts+2} of {max_retries}) for job {job_id}")
+                                time.sleep(2)
+                            else:
+                                break # Break out of attempts loop for current quality if max retries reached
+                        finally:
+                            if job_id in self.active_processes:
+                                del self.active_processes[job_id]
+                        if not segment_success: # If any segment failed, break segment loop and retry/fail quality
+                            break
+                    if segment_success: # If all segments for current quality were successful
+                        success = True # Mark quality encoding as successful
+                        completed_global_steps += 1 # Increment for successful quality stitching step
+
+                        # Stitch segments for current quality
+                        output_file = output_dir / f"{output_name}_{quality}.mp4"
+                        try:
+                            self._stitch_segments(encoded_segments_for_quality, output_file)
                             outputs.append({
                                 'quality': quality,
                                 'path': str(output_file),
                                 'settings': settings
                             })
-                            success = True
-                            completed_steps += 1
-                        else:
-                            error_output = process.stderr.read()
-                            logger.error(f"FFmpeg error on quality {quality}, attempt {attempts+1} for job {job_id}: {error_output}")
-                            raise Exception(f"FFmpeg failed for quality {quality}")
-
-                    except Exception as e:
-                        logger.error(f"Error encoding {quality} on attempt {attempts+1} for job {job_id}: {str(e)}")
-                        attempts += 1
-                        if attempts < max_retries:
-                            logger.info(f"Retrying encoding for quality {quality} (attempt {attempts+1} of {max_retries}) for job {job_id}")
-                            time.sleep(2)  # Wait before retrying
-                        else:
+                            logger.info(f"Stitching successful for quality {quality} for job {job_id}")
+
+                        except Exception as stitch_err:
+                            logger.error(f"Stitching failed for quality {quality} for job {job_id}: {stitch_err}")
                             self.jobs[job_id].update({
                                 'status': 'failed',
-                                'error': f"Failed encoding {quality} after {max_retries} attempts: {str(e)}"
+                                'error': f"Stitching failed for quality {quality}: {str(stitch_err)}"
                             })
-                            return
-                    finally:
-                        if job_id in self.active_processes:
-                            del self.active_processes[job_id]
+                            return # Early return if stitching fails, consider if retry for stitching needed
+
+                        # Clean up encoded segments for this quality (optional, but good to clean up split_dir)
+                        for seg_file in encoded_segments_for_quality:
+                            if seg_file.exists():
+                                seg_file.unlink()
+
+                        self.jobs[job_id]['progress'] = (completed_global_steps / global_progress_steps) * 100 # Final progress for this quality
+
+
+                if not success: # After max retries for a quality
+                    self.jobs[job_id].update({
+                        'status': 'failed',
+                        'error': f"Failed encoding quality {quality} after {max_retries} attempts"
+                    })
+                    return # Early return if quality encoding fails after retries
 
             self.jobs[job_id].update({
                 'status': 'completed',
@@ -246,6 +357,7 @@ class EncoderService:
             })
             logger.info(f"Job {job_id} completed successfully")
 
+
         except Exception as e:
             logger.error(f"Encoding failed for job {job_id}: {str(e)}")
             self.jobs[job_id].update({
@@ -253,10 +365,14 @@ class EncoderService:
                 'error': str(e)
             })
         finally:
+            # Clean up split directory and segments after processing is done (success or fail)
+            if split_dir.exists():
+                shutil.rmtree(split_dir)
             # Remove thread reference once job completes/fails
             if job_id in self.threads:
                 del self.threads[job_id]
 
+
     def _get_video_duration(self, input_file):
         """Get video duration using FFprobe"""
         try:
@@ -305,7 +421,7 @@ class EncoderService:
         try:
             upload_path = Path(os.getenv('UPLOAD_FOLDER', 'uploads'))
             encoded_path = Path(os.getenv('ENCODED_FOLDER', 'encoded'))
-            
+
             if job_id in self.jobs:
                 source_file = upload_path / self.jobs[job_id]['filename']
                 if source_file.exists():
@@ -328,7 +444,7 @@ class EncoderService:
         try:
             if job_id not in self.jobs:
                 return None
-            
+
             job = self.jobs[job_id]
             encoded_path = Path(os.getenv('ENCODED_FOLDER', 'encoded'))
             job_path = encoded_path / job_id
@@ -352,4 +468,4 @@ class EncoderService:
             logger.error(f"Error getting job info for {job_id}: {str(e)}")
             return None
 
-encoder_service = EncoderService()
+encoder_service = EncoderService()