Update dependencies in requirements.txt; enhance FFmpegEncoder with availability check and improve logging in worker

requirements.txt

@@ -1,7 +1,6 @@
-fastapi[all]==0.115.12
-uvicorn==0.34.0
-valkey==6.1.0
-rq>=1.10.0
-ffmpeg-python==0.2.0
-python-multipart==0.0.20
-python-dotenv==1.1.0
+fastapi>=0.68.0
+uvicorn>=0.15.0
+python-multipart>=0.0.5
+redis>=4.3.4
+rq>=1.10.1
+python-dotenv>=0.19.0

video_encoder/worker/encoder.py

@@ -1,5 +1,6 @@
 import subprocess
 import os
+import shutil
 from pathlib import Path
 from typing import List, Dict
 from ..config import EncodingConfig
@@ -9,6 +10,9 @@ logger = logging.getLogger(__name__)
 
 class FFmpegEncoder:
     def __init__(self, input_path: str, output_dir: str):
+        # Verify FFmpeg is available
+        if not shutil.which("ffmpeg"):
+            raise RuntimeError("FFmpeg not found in PATH")
         self.input_path = input_path
         self.output_dir = output_dir
         self.base_name = Path(input_path).stem
@@ -42,22 +46,26 @@ class FFmpegEncoder:
             commands.append(cmd)
         return commands
 
-    def encode(self) -> str:
-        """Execute encoding commands and return master playlist path"""
-        master_playlist = os.path.join(self.output_dir, f"{self.base_name}_master.m3u8")
+def encode(self) -> str:
+    """Execute encoding commands and return master playlist path"""
+    master_playlist = os.path.join(self.output_dir, f"{self.base_name}_master.m3u8")
 
-        with open(master_playlist, "w") as f:
-            f.write("#EXTM3U\n")
-            for res in reversed(EncodingConfig.RESOLUTIONS):
-                f.write(f"#EXT-X-STREAM-INF:BANDWIDTH={res['video_bitrate'].replace('k', '000')},"
-                        f"RESOLUTION={res['width']}x{res['height']}\n")
-                f.write(f"{self.base_name}_{res['name']}.m3u8\n")
+    with open(master_playlist, "w") as f:
+        f.write("#EXTM3U\n")
+        for res in reversed(EncodingConfig.RESOLUTIONS):
+            f.write(f"#EXT-X-STREAM-INF:BANDWIDTH={res['video_bitrate'].replace('k', '000')},"
+                    f"RESOLUTION={res['width']}x{res['height']}\n")
+            f.write(f"{self.base_name}_{res['name']}.m3u8\n")
 
-        for cmd in self.generate_commands():
-            result = subprocess.run(cmd, capture_output=True)
-            logger.debug(f"FFmpeg command: {' '.join(cmd)}")
-            logger.debug(f"FFmpeg output: {result.stdout.decode()}")
-            logger.debug(f"FFmpeg error: {result.stderr.decode()}")
-            result.check_returncode()
+    for cmd in self.generate_commands():
+        logger.info(f"Executing FFmpeg command: {' '.join(cmd)}")
+        result = subprocess.run(cmd, capture_output=True)
+        logger.debug(f"FFmpeg command: {' '.join(cmd)}")
+        logger.debug(f"FFmpeg output: {result.stdout.decode()}")
+        logger.debug(f"FFmpeg error: {result.stderr.decode()}")
+        if result.returncode != 0:
+            logger.error(f"FFmpeg command failed with return code {result.returncode}")
+            raise subprocess.CalledProcessError(result.returncode, cmd)
+        result.check_returncode()
 
-        return master_playlist
+    return master_playlist

video_encoder/worker/tasks.py

@@ -23,8 +23,11 @@ def init_job(job_id: str):
     })
 
 def update_job_progress(job_id: str, progress: float):
-    logger.info(f"Updating progress for job {job_id} to {progress}")
-    RedisConfig.get_connection().hset(f"job:{job_id}", "progress", progress)
+    try:
+        logger.info(f"Updating progress for job {job_id} to {progress}")
+        RedisConfig.get_connection().hset(f"job:{job_id}", "progress", progress)
+    except Exception as e:
+        logger.error(f"Failed to update progress for job {job_id}: {str(e)}")
 
 def encode_video_task(job_id: str, input_path: str) -> Optional[str]:
     max_retries = 3

video_encoder/worker/worker.py

@@ -1,6 +1,14 @@
 from rq import Worker
 from video_encoder.config import RedisConfig
 
+import logging
+logger = logging.getLogger(__name__)
+
 if __name__ == "__main__":
-    worker = Worker([RedisConfig.QUEUE_NAME], connection=RedisConfig.get_connection(), default_result_ttl=3600)
-    worker.work()
+    try:
+        logger.info("Starting RQ worker")
+        worker = Worker([RedisConfig.QUEUE_NAME], connection=RedisConfig.get_connection(), default_result_ttl=3600)
+        logger.info("RQ worker started successfully")
+        worker.work()
+    except Exception as e:
+        logger.error(f"Failed to start RQ worker: {str(e)}")