Add logging and threading support for video encoding worker

run.py

@@ -1,7 +1,20 @@
 import uvicorn
 from video_encoder.api.main import app
+from rq import Worker, Queue, Connection
+from redis import Redis
+from video_encoder.config import RedisConfig
 
 if __name__ == "__main__":
+    import threading
+
+    def start_worker():
+        with Connection(RedisConfig.get_connection()):
+            worker = Worker(map(Queue, [RedisConfig.QUEUE_NAME]))
+            worker.work()
+
+    worker_thread = threading.Thread(target=start_worker)
+    worker_thread.start()
+
     uvicorn.run(
         app,
         host="0.0.0.0",

video_encoder/api/main.py

@@ -10,23 +10,31 @@ from rq import Queue
 from redis import from_url
 from ..config import EncodingConfig, RedisConfig
 from ..worker.tasks import encode_video_task
+import logging
+
+logger = logging.getLogger(__name__)
 
 q = Queue(connection=RedisConfig.get_connection(), default_timeout=3600)
 
 app = FastAPI(title="Video Encoding Service")
 
+@app.on_event("startup")
+async def startup_event():
+    logger.info("Application startup")
+
 @app.post("/upload")
 async def upload_video(file: UploadFile):
     job_id = str(uuid.uuid4())
     temp_dir = os.path.join(EncodingConfig.TEMP_DIR, job_id)
     os.makedirs(temp_dir, exist_ok=True)
-    
+
     input_path = os.path.join(temp_dir, file.filename)
     with open(input_path, "wb") as f:
         f.write(await file.read())
-        
+
+    logger.info(f"Enqueuing job {job_id} for file {file.filename}")
     q.enqueue(encode_video_task, job_id, input_path)
-    
+
     return JSONResponse(
         content={"job_id": job_id},
         status_code=status.HTTP_202_ACCEPTED
@@ -36,10 +44,11 @@ async def upload_video(file: UploadFile):
 async def get_status(job_id: str):
     redis = RedisConfig.get_connection()
     progress = redis.hget(f"job:{job_id}", "progress")
-    
+
     if not progress:
+        logger.warning(f"Job {job_id} not found")
         raise HTTPException(status_code=404, detail="Job not found")
-    
+
     return {"job_id": job_id, "progress": float(progress)}
 
 @app.get("/play/{job_id}")
@@ -49,14 +58,16 @@ async def play_video(job_id: str, token: str):
         job_id.encode(),
         hashlib.sha256
     ).hexdigest()
-    
+
     if not hmac.compare_digest(token, expected_token):
+        logger.warning(f"Invalid token for job {job_id}")
         raise HTTPException(status_code=403, detail="Invalid token")
-    
+
     master_playlist = os.path.join(EncodingConfig.TEMP_DIR, job_id, f"{job_id}_master.m3u8")
     if not os.path.exists(master_playlist):
+        logger.warning(f"Playlist not found for job {job_id}")
         raise HTTPException(status_code=404, detail="Playlist not found")
-        
+
     return FileResponse(
         master_playlist,
         media_type="application/vnd.apple.mpegurl"
@@ -70,5 +81,6 @@ async def generate_token(job_id: str):
         f"{job_id}{timestamp}".encode(),
         hashlib.sha256
     ).hexdigest()
-    
+
+    logger.info(f"Generated token for job {job_id}")
     return {"token": f"{timestamp}:{signature}"}

video_encoder/config.py

@@ -7,8 +7,14 @@ from dotenv import load_dotenv
 
 load_dotenv()
 
+import logging
 class EncodingConfig:
+    DEBUG = os.getenv("DEBUG", "False") == "True"
     TEMP_DIR = "./tmp"
+    LOG_LEVEL = logging.DEBUG if DEBUG else logging.INFO
+
+logging.basicConfig(level=EncodingConfig.LOG_LEVEL)
+logger = logging.getLogger(__name__)
 
 class RedisConfig:
     URL = os.getenv("REDIS_URL")
@@ -18,7 +24,12 @@ class RedisConfig:
     @classmethod
     def get_connection(cls):
         if not cls.CONN_POOL:
-            cls.CONN_POOL = from_url(cls.URL, max_connections=20)
+            try:
+                cls.CONN_POOL = from_url(cls.URL, max_connections=20)
+                logger.info("Successfully connected to Redis")
+            except Exception as e:
+                logger.error(f"Failed to connect to Redis: {e}")
+                raise
         return cls.CONN_POOL
     HMAC_SECRET = os.getenv("VIDEO_HMAC_SECRET", "default-secret-change-me")
     

video_encoder/worker/tasks.py

@@ -9,23 +9,25 @@ from typing import Optional
 from .ffmpeg import FFmpegEncoder
 from ..config import EncodingConfig
 
-logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 q = Queue(connection=RedisConfig.get_connection(), default_timeout=3600)
 
 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)
 
 def encode_video_task(job_id: str, input_path: str) -> Optional[str]:
     try:
+        logger.info(f"Starting encoding for job {job_id}")
         output_dir = os.path.join(EncodingConfig.TEMP_DIR, job_id)
         encoder = FFmpegEncoder(input_path, output_dir)
-        
+
         update_job_progress(job_id, 0.1)
         master_playlist = encoder.encode()
         update_job_progress(job_id, 1.0)
-        
+
+        logger.info(f"Encoding completed for job {job_id}")
         return master_playlist
     except Exception as e:
         logger.error(f"Encoding failed for job {job_id}: {str(e)}")