thread path

app/services/encoder_service.py

@@ -7,9 +7,8 @@ import json
 from datetime import datetime
 import re
 import threading
-import queue
 import signal
-import time  # Added for retry delays
+import time
 
 # Configure logging
 logging.basicConfig(level=logging.INFO)
@@ -87,13 +86,14 @@ class EncoderService:
                 'settings': qualities
             }
 
-            # Start encoding in a separate thread
+            # Start encoding in a separate thread.
+            # Note: Removing the daemon flag to ensure the thread completes.
             thread = threading.Thread(
                 target=self._encode_video,
                 args=(filename, job_id)
             )
-            thread.daemon = True
             thread.start()
+            logger.info(f"Started encoding thread for job {job_id}")
 
             return {'status': 'pending', 'job_id': job_id}
         except Exception as e:
@@ -111,6 +111,7 @@ class EncoderService:
 
             # Create output directory
             output_dir.mkdir(parents=True, exist_ok=True)
+            logger.info(f"Created output directory {output_dir} for job {job_id}")
 
             qualities = self.jobs[job_id]['settings']
             total_steps = len(qualities)
@@ -121,6 +122,7 @@ class EncoderService:
             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")
 
             for quality, settings in qualities.items():
                 max_retries = 3
@@ -128,7 +130,7 @@ class EncoderService:
                 success = False
 
                 while attempts < max_retries and not success:
-                    logger.info(f"Starting encoding for quality {quality}, attempt {attempts+1}")
+                    logger.info(f"Starting encoding for quality {quality}, attempt {attempts+1} for job {job_id}")
                     self.jobs[job_id].update({
                         'status': 'processing',
                         'current_quality': quality,
@@ -170,19 +172,28 @@ class EncoderService:
                         )
                         self.active_processes[job_id] = process
 
-                        # Monitor FFmpeg progress
+                        # Monitor FFmpeg progress with a timeout if no output in 30 seconds
+                        last_output_time = time.time()
                         while True:
-                            line = process.stdout.readline()
-                            if line == '' and process.poll() is not None:
-                                break
+                            if process.stdout.readable():
+                                line = process.stdout.readline()
+                            else:
+                                line = ''
                             if line:
+                                last_output_time = time.time()
                                 progress = self._parse_ffmpeg_progress(line, duration)
                                 if progress is not None:
                                     quality_progress = ((completed_steps + progress / 100) / total_steps) * 100
                                     self.jobs[job_id]['progress'] = quality_progress
+                            # If no new output for 30 seconds, break the loop and retry
+                            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}")
+                                break
+                            if process.poll() is not None:
+                                break
 
                         if process.returncode == 0:
-                            logger.info(f"Encoding successful for quality {quality} on attempt {attempts+1}")
+                            logger.info(f"Encoding successful for quality {quality} on attempt {attempts+1} for job {job_id}")
                             outputs.append({
                                 'quality': quality,
                                 'path': str(output_file),
@@ -192,14 +203,14 @@ class EncoderService:
                             completed_steps += 1
                         else:
                             error_output = process.stderr.read()
-                            logger.error(f"FFmpeg error on quality {quality}, attempt {attempts+1}: {error_output}")
+                            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}: {str(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})")
+                            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:
                             self.jobs[job_id].update({
@@ -217,9 +228,10 @@ class EncoderService:
                 'outputs': outputs,
                 'completion_time': datetime.now().isoformat()
             })
+            logger.info(f"Job {job_id} completed successfully")
 
         except Exception as e:
-            logger.error(f"Encoding failed: {str(e)}")
+            logger.error(f"Encoding failed for job {job_id}: {str(e)}")
             self.jobs[job_id].update({
                 'status': 'failed',
                 'error': str(e)
@@ -271,22 +283,18 @@ class EncoderService:
     def clean_job(self, job_id):
         """Clean up all files related to a job"""
         try:
-            # Get paths
             upload_path = Path(os.getenv('UPLOAD_FOLDER', 'uploads'))
             encoded_path = Path(os.getenv('ENCODED_FOLDER', 'encoded'))
             
-            # Clean up source file if it exists
             if job_id in self.jobs:
                 source_file = upload_path / self.jobs[job_id]['filename']
                 if source_file.exists():
                     source_file.unlink()
 
-            # Clean up encoded files
             job_output_dir = encoded_path / job_id
             if job_output_dir.exists():
                 shutil.rmtree(job_output_dir)
 
-            # Remove job from jobs dict
             if job_id in self.jobs:
                 del self.jobs[job_id]
 
@@ -305,7 +313,6 @@ class EncoderService:
             encoded_path = Path(os.getenv('ENCODED_FOLDER', 'encoded'))
             job_path = encoded_path / job_id
 
-            # Get file sizes
             files_info = []
             if job_path.exists():
                 for output in job.get('outputs', []):
@@ -322,7 +329,7 @@ class EncoderService:
                 'files': files_info
             }
         except Exception as e:
-            logger.error(f"Error getting job info: {str(e)}")
+            logger.error(f"Error getting job info for {job_id}: {str(e)}")
             return None
 
 encoder_service = EncoderService()