Update app.py

app.py

@@ -1,281 +1,209 @@
-import os
-import json
-import requests
-import cloudinary
-import cloudinary.uploader
-from requests.adapters import HTTPAdapter
-from urllib3.util.retry import Retry
-from fastapi import FastAPI, Body, HTTPException, BackgroundTasks
-from dotenv import load_dotenv
-from datetime import datetime
-from pipeline import run_intervision_pipeline
-
-# --- Setup Retry Strategy ---
-retry_strategy = Retry(
-    total=3,
-    backoff_factor=1, # Wait 1s, 2s, 4s between retries
-    status_forcelist=[429, 500, 502, 503, 504],
-)
-adapter = HTTPAdapter(max_retries=retry_strategy)
-http = requests.Session()
-http.mount("https://", adapter)
-http.mount("http://", adapter)
-
-# Load environment variables from .env file
-load_dotenv()
-
-app = FastAPI(title="Intervision AI Engine")
-
-# Cloudinary Configuration
-cloudinary.config( 
-    cloud_name = os.getenv("CLOUDINARY_CLOUD_NAME"), 
-    api_key = os.getenv("CLOUDINARY_API_KEY"), 
-    api_secret = os.getenv("CLOUDINARY_API_SECRET") 
-)
-
-# Directory Setup
-RESULT_DIR = "temp_data/results"
-UPLOAD_DIR = "temp_data/uploads"
-os.makedirs(RESULT_DIR, exist_ok=True)
-os.makedirs(UPLOAD_DIR, exist_ok=True)
-
-def time_to_seconds(t_str: str) -> int:
-    """Converts HH:MM:SS timestamp format to total seconds."""
-    if not t_str: return 0
-    h, m, s = map(int, t_str.split(':'))
-    return h * 3600 + m * 60 + s
-
-def background_processing(session_data: dict):
-    """
-    Handles heavy AI processing: video download, pipeline execution, 
-    result upload, and backend notification (callback).
-    """
-    session_id = session_data.get('sessionId')
-    video_url = session_data.get('originalVideoUrl')
-    callback_url = session_data.get('callbackBaseUrl')
-    
-    print(f"[LOG] Processing started for session: {session_id}")
-    
-    # 1. Download the original video from the provided URL
-    local_input_path = os.path.join(UPLOAD_DIR, f"{session_id}_input.mp4")
-# 1. Download with increased timeout and Retry logic
-    try:
-        print(f"[LOG] Downloading video: {video_url}")
-        # Increased timeout to 300s (5 minutes) for large files
-        response = http.get(video_url, stream=True, timeout=300)
-        response.raise_for_status()
-        with open(local_input_path, 'wb') as f:
-            for chunk in response.iter_content(chunk_size=1024*1024):
-                f.write(chunk)
-    except Exception as e:
-        print(f"[DOWNLOAD ERROR]: {e}")
-        # Notify backend that it failed due to download
-        return
-
-    # 2. Prepare question list for the AI Pipeline
-    final_questions = []
-    skipped_failed_reports = []
-    
-    for q in session_data.get('answers', []):
-        if q.get('isAnswered'):
-            final_questions.append({
-                "question_id": q['aiQuestionId'],
-                "question_text": q['questionText'],
-                "ideal_answer": q['expectedAnswer'],
-                "start_time": time_to_seconds(q['startedAt']),
-                "end_time": time_to_seconds(q['submittedAt'])
-            })
-        else:
-            # Handle questions that weren't answered during the session
-            skipped_failed_reports.append({
-                "questionId": q['aiQuestionId'],
-                "userAnswerText": "N/A",
-                "score": 0.0,                       
-                "relevance": 0.0,
-                "confidence": 0.0,
-                "stress": 0.0,
-                "clarity": 0.0,
-                "pauses": 0.0,
-                "toneOfVoice": "N/A",
-                "status": "skipped" if q.get('isSkipped') else "failed"
-            })
-
-    # 3. Execute AI Pipeline (Analysis & Visualization)
-    ai_results = []
-    if final_questions:
-        # run_intervision_pipeline generates Intervision_Final_Result.mp4
-        run_intervision_pipeline(local_input_path, final_questions, RESULT_DIR)
-        report_path = os.path.join(RESULT_DIR, "report.json")
-        if os.path.exists(report_path):
-            with open(report_path, "r") as f:
-                ai_results = json.load(f).get("listOfAnswerReport", [])
-
-    # 4. Upload the processed video to Cloudinary
-    final_video_path = os.path.join(RESULT_DIR, "Intervision_Final_Result.mp4")
-    final_video_url = None
-    if os.path.exists(final_video_path):
-        try:
-            upload_res = cloudinary.uploader.upload(
-                final_video_path, 
-                public_id=f"res_{session_id}", 
-                folder="intervision_results",
-                resource_type="video",
-                chunk_size=6000000
-            )
-            final_video_url = upload_res.get("secure_url")
-        except Exception as e:
-            print(f"[UPLOAD ERROR]: {e}")
-
-    # 5. Construct final payload and notify Backend via Callback
-    final_payload = {
-        "sessionId": session_id,
-        "finalVideoUrl": final_video_url,
-        "report": ai_results + skipped_failed_reports
-    }
-    
-    try:
-        # Notify backend that processing is complete
-        cb_response = requests.post(f"{callback_url}/api/ai-callback", json=final_payload, timeout=30)
-        print(f"[LOG] Callback sent to {callback_url}. Status: {cb_response.status_code}")
-        
-        # 6. Local Cleanup: Remove files to save disk space
-        if os.path.exists(local_input_path): os.remove(local_input_path)
-        if os.path.exists(final_video_path): os.remove(final_video_path)
-            
-    except Exception as e:
-        print(f"[CALLBACK ERROR]: {e}")
-
-@app.post("/process-interview/")
-async def process_interview(background_tasks: BackgroundTasks, data: dict = Body(...)):
-    """Entry point to start the AI analysis asynchronously."""
-    background_tasks.add_task(background_processing, data)
-    return {"message": "Processing started", "sessionId": data.get('sessionId')}
-
-@app.post("/delete-video-by-url/")
-async def delete_video_by_url(data: dict = Body(...)):
-    """
-    Deletes a video from Cloudinary based on its URL.
-    Input JSON: {"videoUrl": "https://..."}
-    """
-    video_url = data.get("videoUrl")
-    if not video_url:
-        raise HTTPException(status_code=400, detail="videoUrl is required")
-
-    try:
-        # Logic to extract the public_id from a Cloudinary URL
-        # Example: .../folder/public_id.mp4 -> folder/public_id
-        url_parts = video_url.split('/')
-        filename_with_ext = url_parts[-1]
-        filename = filename_with_ext.split('.')[0]
-        
-        # Check if the video is inside the results folder
-        folder = url_parts[-2] if "intervision_results" in url_parts[-2] else ""
-        public_id = f"{folder}/{filename}" if folder else filename
-
-        # Trigger deletion from Cloudinary
-        result = cloudinary.uploader.destroy(public_id, resource_type="video")
-
-        if result.get("result") == "ok":
-            return {"status": "success", "message": f"Deleted {public_id}"}
-        return {"status": "failed", "details": result}
-
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
-
-if __name__ == "__main__":
-    import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=8000)
-    
-# @app.post("/process-interview-test/")
-# async def process_test(data: dict = Body(...)):
-#     try:
-#         print(f"--- [TEST LOG] Processing Session: {data['sessionId']} ---")
-        
-#         # 1. Path Check
-#         local_path = r"D:\FayzaAhmed\Graduation_project\models\MultiModal\deployment\interview_test.mp4" 
-#         if not os.path.exists(local_path):
-#             return {"error": f"Video file not found at {local_path}"}
-
-#         # 2. Prepare Data
-#         final_questions = []
-#         for q in data['answers']:
-#             if q.get('isAnswered'):
-#                 final_questions.append({
-#                     "question_id": q['aiQuestionId'],
-#                     "question_text": q['questionText'],
-#                     "ideal_answer": q['expectedAnswer'],
-#                     "start_time": time_to_seconds(q['startedAt']),
-#                     "end_time": time_to_seconds(q['submittedAt'])
-#                 })
-
-#         # 3. Run Pipeline
-#         print("[LOG] Running AI Pipeline...")
-#         run_intervision_pipeline(local_path, final_questions, RESULT_DIR)
-        
-#         # 4. Upload
-#         print("[LOG] Uploading to Cloudinary...")
-#         final_video_path = os.path.join(RESULT_DIR, "Intervision_Final_Result.mp4")
-#         upload_res = cloudinary.uploader.upload(
-#                     final_video_path, 
-#                     public_id=f"{data['sessionId']}_test",
-#                     folder="intervision_tests",
-#                     resource_type="video"  # This is the important part
-# )
-
-#         # 5. Load Report
-#         report_path = os.path.join(RESULT_DIR, "report.json")
-#         if not os.path.exists(report_path):
-#             return {"error": "Pipeline finished but report.json was not created."}
-
-#         with open(report_path, "r") as f:
-#             ai_results = json.load(f)["listOfAnswerReport"]
-
-#         return {
-#             "status": "Success",
-#             "videoUrl": upload_res.get("secure_url"),
-#             "report": ai_results
-#         }
-
-#     except Exception as e:
-#         print(f"[CRITICAL ERROR]: {str(e)}")
-#         return {"error": str(e), "traceback": "Check Terminal for details"}
-
-"""QUESTIONS_CONFIG = 
-[
-    {
-        "question_id": 1,
-        "question_text": "how do you describe yourself",
-        "ideal_answer": "Being different means you have to work at belonging...",
-        "start_time": 0,
-        "end_time": 15,
-    },
-    {
-        "question_id": 2,
-        "question_text": "Tell us about your biggest achievement",
-        "ideal_answer": "I am proud of accomplishing...",
-        "start_time": 15,
-        "end_time": 24,
-    }
-]
-"""
-
-"""
-{
-    "sessionId": "test-session-123",
-    "originalVideoUrl": "local_test_no_url",
-    "callbackBaseUrl": "http://localhost:8000",
-    "answers": [
-        {
-        "questionId": "q-1",
-        "aiQuestionId": 1,
-        "questionText": "How does the speaker encourage people to deal with their differences and uniqueness?",
-        "expectedAnswer": "When you're different, you have to work at the longing. Everybody wants to feel valued and accepted, and we think it should happen spontaneously, but it doesn't. Sometimes society tells us, and we tell ourselves, we don't fit the mold. Take a piece of paper and write down what makes you different. And I want you to celebrate it today and every day. Shout it from the rooftops. What makes me different is what has made me stand out and be successful. I also encourage you to be curious and ask, what is on other people's pieces of paper? What makes them different? Let's celebrate those imperfections that make us special. I hope that it teaches you that nobody has a claim on the word normal. We are all different. We are all quirky and unique, and that is what makes us wonderful.",
-        "isAnswered": true,
-        "isSkipped": false,
-        "isFailed": false,
-        "startedAt": "00:00:00",
-        "submittedAt": "00:00:55"
-        }
-    ]
-}
-"""
+import os
+import json
+import requests
+import cloudinary
+import cloudinary.uploader
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+from fastapi import FastAPI, Body, HTTPException, BackgroundTasks
+from pydantic import BaseModel, HttpUrl
+from typing import List, Optional
+from dotenv import load_dotenv
+from pipeline import run_intervision_pipeline
+
+# --- 1. Setup Retry Strategy for Robust Downloads ---
+# This ensures that if the video download fails momentarily, it retries 3 times.
+retry_strategy = Retry(
+    total=3,
+    backoff_factor=1, 
+    status_forcelist=[429, 500, 502, 503, 504],
+)
+adapter = HTTPAdapter(max_retries=retry_strategy)
+http = requests.Session()
+http.mount("https://", adapter)
+http.mount("http://", adapter)
+
+# Load environment variables from .env file
+load_dotenv()
+
+app = FastAPI(
+    title="Intervision AI Engine",
+    description="Asynchronous AI Pipeline for Interview Analysis",
+    version="1.1.0"
+)
+
+# --- 2. Cloudinary Configuration ---
+cloudinary.config(
+    cloud_name=os.getenv("CLOUDINARY_CLOUD_NAME"),
+    api_key=os.getenv("CLOUDINARY_API_KEY"),
+    api_secret=os.getenv("CLOUDINARY_API_SECRET")
+)
+
+# --- 3. Directory Setup ---
+RESULT_DIR = "temp_data/results"
+UPLOAD_DIR = "temp_data/uploads"
+os.makedirs(RESULT_DIR, exist_ok=True)
+os.makedirs(UPLOAD_DIR, exist_ok=True)
+
+# --- 4. Pydantic Models (Fixes the "additionalProp1" issue) ---
+class AnswerDetail(BaseModel):
+    aiQuestionId: int
+    questionText: str
+    expectedAnswer: str
+    isAnswered: bool
+    isSkipped: bool
+    startedAt: str  # Expected format "HH:MM:SS"
+    submittedAt: str # Expected format "HH:MM:SS"
+
+class InterviewRequest(BaseModel):
+    sessionId: str
+    originalVideoUrl: HttpUrl
+    callbackBaseUrl: HttpUrl
+    answers: List[AnswerDetail]
+
+# --- 5. Helper Functions ---
+def time_to_seconds(t_str: str) -> int:
+    """Converts HH:MM:SS or MM:SS timestamp format to total seconds."""
+    if not t_str or ":" not in t_str: return 0
+    parts = list(map(int, t_str.split(':')))
+    if len(parts) == 3:
+        h, m, s = parts
+        return h * 3600 + m * 60 + s
+    elif len(parts) == 2:
+        m, s = parts
+        return m * 60 + s
+    return 0
+
+# --- 6. Background Processing Logic ---
+def background_processing(session_data: InterviewRequest):
+    """
+    Handles heavy AI processing: video download, pipeline execution, 
+    result upload, and backend notification (callback).
+    """
+    session_id = session_data.sessionId
+    video_url = str(session_data.originalVideoUrl)
+    callback_url = str(session_data.callbackBaseUrl)
+    
+    print(f"[LOG] Processing started for session: {session_id}")
+    
+    local_input_path = os.path.join(UPLOAD_DIR, f"{session_id}_input.mp4")
+
+    # Step 1: Download the original video
+    try:
+        print(f"[LOG] Downloading video: {video_url}")
+        response = http.get(video_url, stream=True, timeout=300)
+        response.raise_for_status()
+        with open(local_input_path, 'wb') as f:
+            for chunk in response.iter_content(chunk_size=1024*1024):
+                f.write(chunk)
+    except Exception as e:
+        print(f"[DOWNLOAD ERROR]: {e}")
+        return
+
+    # Step 2: Prepare questions for the Pipeline
+    final_questions = []
+    skipped_failed_reports = []
+    
+    for q in session_data.answers:
+        if q.isAnswered:
+            final_questions.append({
+                "question_id": q.aiQuestionId,
+                "question_text": q.questionText,
+                "ideal_answer": q.expectedAnswer,
+                "start_time": time_to_seconds(q.startedAt),
+                "end_time": time_to_seconds(q.submittedAt)
+            })
+        else:
+            skipped_failed_reports.append({
+                "questionId": q.aiQuestionId,
+                "userAnswerText": "N/A",
+                "score": 0.0,
+                "relevance": 0.0,
+                "confidence": 0.0,
+                "stress": 0.0,
+                "clarity": 0.0,
+                "pauses": 0.0,
+                "toneOfVoice": 3, # Default to Natural/N/A
+                "status": "skipped" if q.isSkipped else "failed"
+            })
+
+    # Step 3: Execute AI Pipeline
+    ai_results = []
+    if final_questions:
+        run_intervision_pipeline(local_input_path, final_questions, RESULT_DIR)
+        report_path = os.path.join(RESULT_DIR, "report.json")
+        if os.path.exists(report_path):
+            with open(report_path, "r") as f:
+                ai_results = json.load(f).get("listOfAnswerReport", [])
+
+    # Step 4: Upload processed video to Cloudinary
+    final_video_path = os.path.join(RESULT_DIR, "Intervision_Final_Result.mp4")
+    final_video_url = None
+    if os.path.exists(final_video_path):
+        try:
+            upload_res = cloudinary.uploader.upload(
+                final_video_path,
+                public_id=f"res_{session_id}",
+                folder="intervision_results",
+                resource_type="video"
+            )
+            final_video_url = upload_res.get("secure_url")
+        except Exception as e:
+            print(f"[UPLOAD ERROR]: {e}")
+
+    # Step 5: Construct final payload and notify Backend
+    final_payload = {
+        "sessionId": session_id,
+        "finalVideoUrl": final_video_url,
+        "report": ai_results + skipped_failed_reports
+    }
+    
+    try:
+        cb_response = requests.post(f"{callback_url}/api/ai-callback", json=final_payload, timeout=30)
+        print(f"[LOG] Callback sent. Status: {cb_response.status_code}")
+        
+        # Cleanup local files
+        if os.path.exists(local_input_path): os.remove(local_input_path)
+        if os.path.exists(final_video_path): os.remove(final_video_path)
+    except Exception as e:
+        print(f"[CALLBACK ERROR]: {e}")
+
+# --- 7. API Routes ---
+
+@app.get("/")
+async def root():
+    """Health check endpoint to verify the service is running."""
+    return {
+        "status": "online", 
+        "message": "Intervision AI Engine is running",
+        "documentation": "/docs"
+    }
+
+@app.post("/process-interview/")
+async def process_interview(background_tasks: BackgroundTasks, data: InterviewRequest):
+    """
+    Entry point to start the AI analysis asynchronously.
+    Receives validated data via InterviewRequest model.
+    """
+    background_tasks.add_task(background_processing, data)
+    return {"message": "Processing started", "sessionId": data.sessionId}
+
+@app.post("/delete-video-by-url/")
+async def delete_video_by_url(payload: dict = Body(...)):
+    """Deletes a video from Cloudinary based on its URL."""
+    video_url = payload.get("videoUrl")
+    if not video_url:
+        raise HTTPException(status_code=400, detail="videoUrl is required")
+    try:
+        url_parts = video_url.split('/')
+        filename = url_parts[-1].split('.')[0]
+        folder = "intervision_results" if "intervision_results" in video_url else ""
+        public_id = f"{folder}/{filename}" if folder else filename
+        result = cloudinary.uploader.destroy(public_id, resource_type="video")
+        return {"status": result.get("result"), "public_id": public_id}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)