Upload 3 files

app.py

@@ -1,4 +1,4 @@
-from fastapi import FastAPI, HTTPException
+from fastapi import FastAPI, HTTPException, BackgroundTasks
 from pydantic import BaseModel
 import os
 import requests
@@ -10,9 +10,14 @@ from pathlib import Path
 from supabase import create_client, Client
 from openai import OpenAI
 import time
+from typing import Dict, Optional
 
 app = FastAPI()
 
+# Global state for background jobs
+# In a production environment, this should be a DB or Redis, but for HF Space singleton, a dict works
+jobs: Dict[str, dict] = {}
+
 class ProcessRequest(BaseModel):
     videoUrl: str
     projectId: str
@@ -20,54 +25,56 @@ class ProcessRequest(BaseModel):
     supabaseKey: str
     openaiKey: str
 
+class JobStatus(BaseModel):
+    job_id: str
+    status: str
+    progress: int
+    message: str
+    result: Optional[dict] = None
+    error: Optional[str] = None
+
 @app.get("/")
 def read_root():
-    return {"status": "Avatar Worker is Online"}
+    return {"status": "Avatar Worker is Online", "active_jobs": len(jobs)}
 
-@app.post("/process")
-async def process_video(req: ProcessRequest):
+@app.get("/status/{job_id}", response_model=JobStatus)
+async def get_status(job_id: str):
+    if job_id not in jobs:
+        raise HTTPException(status_code=404, detail="Job not found")
+    return jobs[job_id]
+
+def background_process(job_id: str, req: ProcessRequest):
     temp_dir = Path(f"/tmp/{uuid.uuid4()}")
     temp_dir.mkdir(parents=True, exist_ok=True)
     
     try:
         # 1. Download Video
+        jobs[job_id].update({"status": "processing", "progress": 10, "message": "Downloading video..."})
         video_path = temp_dir / "input_video.mp4"
-        print(f"Downloading video from {req.videoUrl}...")
         resp = requests.get(req.videoUrl, stream=True)
         if resp.status_code != 200:
-            raise HTTPException(status_code=400, detail="Failed to download video from Supabase")
+            raise Exception("Failed to download video from Supabase")
         
         with open(video_path, 'wb') as f:
             for chunk in resp.iter_content(chunk_size=8192):
                 f.write(chunk)
 
-        # 2. Extract Audio for STT (to save bandwidth/token costs if needed, but Whisper API handles video too)
-        # Actually Whisper API accepts files up to 25MB. If video is larger, we MUST extract audio.
+        # 2. Extract Audio for STT
+        jobs[job_id].update({"progress": 20, "message": "Extracting audio for AI analysis..."})
         audio_path = temp_dir / "audio.mp3"
-        print("Extracting audio for STT...")
         subprocess.run([
             "ffmpeg", "-i", str(video_path), 
             "-vn", "-acodec", "libmp3lame", "-ar", "16000", "-ac", "1", 
             str(audio_path)
         ], check=True, capture_output=True)
 
-        # 3. Initialize Supabase Client
-        print("Initializing Supabase client...")
-        try:
-            supabase: Client = create_client(req.supabaseUrl, req.supabaseKey)
-        except Exception as se:
-            print(f"FAILED to initialize Supabase: {str(se)}")
-            raise HTTPException(status_code=500, detail=f"Supabase Init Error: {str(se)}")
+        # 3. Initialize Clients
+        jobs[job_id].update({"progress": 30, "message": "Preparing AI engines..."})
+        supabase: Client = create_client(req.supabaseUrl, req.supabaseKey)
+        openai_client = OpenAI(api_key=req.openaiKey)
 
         # 4. Get Timestamps from OpenAI Whisper
-        print("Initializing OpenAI client...")
-        try:
-            openai_client = OpenAI(api_key=req.openaiKey)
-        except Exception as oe:
-            print(f"FAILED to initialize OpenAI: {str(oe)}")
-            raise HTTPException(status_code=500, detail=f"OpenAI Init Error: {str(oe)}")
-
-        print("Calling OpenAI Whisper API...")
+        jobs[job_id].update({"progress": 40, "message": "Analyzing speech and timing..."})
         with open(audio_path, "rb") as audio_file:
             transcript = openai_client.audio.transcriptions.create(
                 file=audio_file,
@@ -78,10 +85,11 @@ async def process_video(req: ProcessRequest):
 
         segments = transcript.segments
         if not segments:
-            raise HTTPException(status_code=400, detail="No speech detected in video")
+            raise Exception("No speech detected in video")
 
         # 5. Slice Video and Upload
         processed_slices = []
+        total_segments = len(segments)
         
         for i, segment in enumerate(segments):
             start = segment.start
@@ -89,14 +97,15 @@ async def process_video(req: ProcessRequest):
             text = segment.text.strip()
             duration = end - start
             
-            if duration < 0.5: continue # Skip too short segments
+            if duration < 0.5: continue
             
+            # Update progress within the slicing phase (40% to 90%)
+            step_progress = 40 + int((i / total_segments) * 50)
+            jobs[job_id].update({"progress": step_progress, "message": f"Slicing segment {i+1}/{total_segments}..."})
+
             output_filename = f"slice_{i}.mp4"
             output_path = temp_dir / output_filename
             
-            print(f"Slicing segment {i}: {start}s to {end}s...")
-            
-            # Re-encode to ensure clean timestamps and compatibility (matching our earlier fix)
             subprocess.run([
                 "ffmpeg", "-ss", str(start), "-t", str(duration), "-i", str(video_path),
                 "-c:v", "libx264", "-preset", "ultrafast", "-crf", "28",
@@ -106,8 +115,6 @@ async def process_video(req: ProcessRequest):
 
             # Upload to Supabase
             storage_path = f"{req.projectId}/avatar_{int(time.time())}_{i}.mp4"
-            print(f"Uploading slice {i} to Supabase: {storage_path}")
-            
             with open(output_path, "rb") as f:
                 supabase.storage.from_("projects").upload(
                     path=storage_path,
@@ -115,30 +122,38 @@ async def process_video(req: ProcessRequest):
                     file_options={"content-type": "video/mp4", "x-upsert": "true"}
                 )
             
-            # Get Public URL
             public_url = supabase.storage.from_("projects").get_public_url(storage_path)
-            
-            processed_slices.append({
-                "text": text,
-                "url": public_url,
-                "duration": duration
-            })
-
-        return {
-            "success": True,
-            "slices": processed_slices
-        }
-
-    except subprocess.CalledProcessError as e:
-        print(f"FFmpeg Error: {e.stderr.decode()}")
-        raise HTTPException(status_code=500, detail=f"Video processing failed: {e.stderr.decode()}")
+            processed_slices.append({"text": text, "url": public_url, "duration": duration})
+
+        jobs[job_id].update({
+            "status": "completed", 
+            "progress": 100, 
+            "message": "Processing complete!", 
+            "result": {"slices": processed_slices}
+        })
+
     except Exception as e:
-        print(f"General Error: {str(e)}")
-        raise HTTPException(status_code=500, detail=str(e))
+        print(f"Error in background job {job_id}: {str(e)}")
+        jobs[job_id].update({"status": "failed", "error": str(e)})
     finally:
-        # Cleanup
         shutil.rmtree(temp_dir, ignore_errors=True)
 
+@app.post("/process")
+async def process_video(req: ProcessRequest, background_tasks: BackgroundTasks):
+    job_id = str(uuid.uuid4())
+    jobs[job_id] = {
+        "job_id": job_id,
+        "status": "queued",
+        "progress": 0,
+        "message": "Job received and queued",
+        "result": None,
+        "error": None
+    }
+    
+    background_tasks.add_task(background_process, job_id, req)
+    
+    return {"job_id": job_id}
+
 if __name__ == "__main__":
     import uvicorn
     uvicorn.run(app, host="0.0.0.0", port=7860)