feat: implement asynchronous polling system with background tasks

app/main.py

@@ -6,11 +6,15 @@ Main entry point: routes, startup, error handlers.
 from __future__ import annotations
 
 import logging
+import os
+import tempfile
 import time
+import uuid
 from contextlib import asynccontextmanager
+from pathlib import Path
 from typing import Optional
 
-from fastapi import Depends, FastAPI, File, Form, HTTPException, UploadFile, status
+from fastapi import BackgroundTasks, Depends, FastAPI, File, Form, HTTPException, UploadFile, status
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 
@@ -26,13 +30,15 @@ from app.schemas import (
     AnalysisResponse,
     ErrorResponse,
     HealthResponse,
+    JobStatus,
+    JobSubmitResponse,
+    JobResultResponse,
     MediaType,
 )
 from app.utils import (
     AUDIO_EXTENSIONS,
     VIDEO_EXTENSIONS,
     detect_extension,
-    save_upload_to_tempfile,
     validate_extension,
 )
 
@@ -46,6 +52,13 @@ logging.basicConfig(
 logger = logging.getLogger("lecturelens.main")
 
 
+# ── Global Jobs DB (In-Memory) ────────────────────────────────────────────────
+
+# Since HF Spaces free tier sleeps and loses memory, this is ephemeral.
+# Keys are job_ids (UUIDs).
+jobs_db: dict[str, dict] = {}
+
+
 # ── Lifespan (startup/shutdown) ───────────────────────────────────────────────
 
 @asynccontextmanager
@@ -79,7 +92,7 @@ app = FastAPI(
         "- Detailed KPI metrics (loudness, SNR, sharpness, brightness, …)\n"
         "- Structured alerts with severity levels and actionable fixes\n"
         "- A composite quality score (0–1)\n\n"
-        "Authentication: `X-API-Key` header."
+        "*(Note: API Key authentication is currently disabled for testing)*"
     ),
     version="1.0.0",
     docs_url="/docs",
@@ -117,7 +130,76 @@ async def generic_exception_handler(request, exc: Exception):
     )
 
 
-# ── Routes ─────────────────────────────────────────────────────────────────────
+# ── Background Task ───────────────────────────────────────────────────────────
+
+async def process_job_background(
+    job_id: str,
+    tmp_path: Path,
+    media_type: MediaType,
+    participant_label: Optional[str],
+    settings: Settings,
+    filename: str
+):
+    """
+    Executes the analysis in the background, updates jobs_db, and cleans up the temp file.
+    Includes extensive logging for debugging.
+    """
+    logger.info(f"[{job_id}] ⚙️ Started background processing for file: {filename}")
+    t_start = time.perf_counter()
+    
+    try:
+        if media_type == MediaType.audio:
+            logger.info(f"[{job_id}] 🎧 Step 1/3: Starting audio analysis...")
+            metrics = await analyze_audio(tmp_path, settings.dnsmos_model_path)
+            
+            logger.info(f"[{job_id}] 🎧 Step 2/3: Generating alerts against thresholds...")
+            alerts = generate_alerts(metrics, "audio", settings.thresholds_config_path)
+            
+            logger.info(f"[{job_id}] 🎧 Step 3/3: Computing composite audio score...")
+            audio_score = compute_audio_score(metrics)
+            video_score = None
+        else:
+            logger.info(f"[{job_id}] 🎬 Step 1/3: Starting video analysis...")
+            from app.analyzers.video_analyzer import analyze_video
+            metrics = await analyze_video(tmp_path, settings.frame_sample_rate_seconds)
+            
+            logger.info(f"[{job_id}] 🎬 Step 2/3: Generating alerts against thresholds...")
+            alerts = generate_alerts(metrics, "video", settings.thresholds_config_path)
+            
+            logger.info(f"[{job_id}] 🎬 Step 3/3: Computing composite video score...")
+            video_score = compute_video_score(metrics)
+            audio_score = None
+
+        processing_time = round(time.perf_counter() - t_start, 3)
+        logger.info(f"[{job_id}] ✅ Analysis completed successfully in {processing_time}s with {len(alerts)} alert(s).")
+        
+        result = AnalysisResponse(
+            media_type=media_type,
+            participant_label=participant_label,
+            processing_time_seconds=processing_time,
+            metrics=metrics,
+            alerts=alerts,
+            overall_audio_score=audio_score,
+            overall_video_score=video_score,
+        )
+        
+        jobs_db[job_id]["status"] = JobStatus.completed
+        jobs_db[job_id]["result"] = result
+
+    except Exception as exc:
+        logger.exception(f"[{job_id}] ❌ Error processing file {filename}: {exc}")
+        jobs_db[job_id]["status"] = JobStatus.failed
+        jobs_db[job_id]["error_detail"] = str(exc)
+        
+    finally:
+        logger.info(f"[{job_id}] 🧹 Cleaning up temporary file: {tmp_path}")
+        try:
+            os.unlink(tmp_path)
+        except Exception as e:
+            logger.warning(f"[{job_id}] ⚠️ Failed to delete temp file {tmp_path}: {e}")
+
+
+# ── Routes ────────────────────────────────────────────────────────────────────
 
 @app.get(
     "/health",
@@ -126,23 +208,23 @@ async def generic_exception_handler(request, exc: Exception):
     tags=["System"],
 )
 async def health():
-    """Returns service status and version. No authentication required."""
+    """Returns service status and version."""
     return HealthResponse()
 
 
 @app.post(
     "/analyze",
-    response_model=AnalysisResponse,
-    summary="Analyze audio or video file",
+    response_model=JobSubmitResponse,
+    summary="Submit an audio or video file for analysis",
     tags=["Analysis"],
     responses={
         400: {"model": ErrorResponse, "description": "Unsupported file type"},
         413: {"model": ErrorResponse, "description": "File too large"},
-        422: {"model": ErrorResponse, "description": "Corrupted or unreadable file"},
-        500: {"model": ErrorResponse, "description": "Internal processing failure"},
+        500: {"model": ErrorResponse, "description": "Internal queuing failure"},
     },
 )
 async def analyze(
+    background_tasks: BackgroundTasks,
     file: UploadFile = File(..., description="Audio (M4A/WAV/MP3) or video (MP4/MOV) file"),
     media_type: MediaType = Form(..., description="'audio' or 'video'"),
     participant_label: Optional[str] = Form(
@@ -152,15 +234,11 @@ async def analyze(
     settings: Settings = Depends(get_settings),
 ):
     """
-    **POST /analyze** — the single analysis endpoint.
+    **POST /analyze** — Submit a file and receive a `job_id`.
 
-    - Send `media_type=audio` for per-speaker audio files (M4A, WAV, MP3…).
-    - Send `media_type=video` for the full recording video (MP4, MOV…).
-    - Use `participant_label` to tag audio files with the speaker's name;
-      the value is echoed back in the response so n8n can correlate results.
+    The file is processed asynchronously in the background. Use `GET /analyze/{job_id}`
+    to poll for the results to avoid HTTP timeouts.
     """
-    t_start = time.perf_counter()
-
     # ── 1. File size guard ────────────────────────────────────────────────────
     content = await file.read()
     if len(content) > settings.max_file_size_bytes:
@@ -191,47 +269,75 @@ async def analyze(
             },
         )
 
-    # ── 3. Run analysis inside a temp file ────────────────────────────────────
+    # ── 3. Save to temp file and dispatch background task ─────────────────────
     try:
-        async with save_upload_to_tempfile(content, suffix=ext) as tmp_path:
-            if media_type == MediaType.audio:
-                metrics = await analyze_audio(tmp_path, settings.dnsmos_model_path)
-                alerts = generate_alerts(metrics, "audio", settings.thresholds_config_path)
-                audio_score = compute_audio_score(metrics)
-                video_score = None
-            else:
-                from app.analyzers.video_analyzer import analyze_video
-                metrics = await analyze_video(tmp_path, settings.frame_sample_rate_seconds)
-                alerts = generate_alerts(metrics, "video", settings.thresholds_config_path)
-                video_score = compute_video_score(metrics)
-                audio_score = None
+        tmp = tempfile.NamedTemporaryFile(delete=False, suffix=ext)
+        tmp.write(content)
+        tmp.flush()
+        tmp.close()
+        tmp_path = Path(tmp.name)
+        
+        job_id = str(uuid.uuid4())
+        jobs_db[job_id] = {
+            "status": JobStatus.processing,
+            "filename": filename,
+            "result": None,
+            "error_detail": None
+        }
+        
+        logger.info(f"[{job_id}] 📥 Job created for file: {filename} ({len(content) / 1e6:.2f} MB)")
+        
+        background_tasks.add_task(
+            process_job_background,
+            job_id=job_id,
+            tmp_path=tmp_path,
+            media_type=media_type,
+            participant_label=participant_label,
+            settings=settings,
+            filename=filename
+        )
+        
+        return JobSubmitResponse(
+            job_id=job_id,
+            status=JobStatus.processing,
+            message="File successfully queued for background analysis."
+        )
 
     except Exception as exc:
-        logger.exception("Analysis failed for file '%s': %s", filename, exc)
+        logger.exception("Failed to queue file '%s': %s", filename, exc)
         raise HTTPException(
-            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
-            detail={
-                "error_code": "CORRUPTED_FILE",
-                "detail": f"Could not process the uploaded file: {exc}",
-            },
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail={"error_code": "QUEUE_FAILED", "detail": f"Could not queue file: {exc}"}
         )
 
-    processing_time = round(time.perf_counter() - t_start, 3)
-    logger.info(
-        "✅ %s analysis complete — %s | %.1fs | %d alert(s)",
-        media_type.value.upper(),
-        participant_label or filename,
-        processing_time,
-        len(alerts),
-    )
 
-    return AnalysisResponse(
-        status="completed",
-        media_type=media_type,
-        participant_label=participant_label,
-        processing_time_seconds=processing_time,
-        metrics=metrics,
-        alerts=alerts,
-        overall_audio_score=audio_score,
-        overall_video_score=video_score,
+@app.get(
+    "/analyze/{job_id}",
+    response_model=JobResultResponse,
+    summary="Get job status and results",
+    tags=["Analysis"],
+    responses={
+        404: {"model": ErrorResponse, "description": "Job not found"},
+    },
+)
+async def get_job_status(job_id: str):
+    """
+    **GET /analyze/{job_id}** — check the status of a background analysis job.
+    
+    - If `status` is **processing**, wait and poll again.
+    - If `status` is **completed**, the `result` field will contain the full metrics and alerts.
+    - If `status` is **failed**, the `error_detail` field will explain why.
+    """
+    job = jobs_db.get(job_id)
+    if not job:
+        raise HTTPException(
+            status_code=404,
+            detail={"error_code": "NOT_FOUND", "detail": f"Job ID {job_id} not found."}
+        )
+        
+    return JobResultResponse(
+        job_id=job_id,
+        status=job["status"],
+        result=job["result"],
+        error_detail=job["error_detail"]
     )

app/schemas.py

@@ -29,6 +29,12 @@ class AlertCategory(str, Enum):
     video = "video"
 
 
+class JobStatus(str, Enum):
+    processing = "processing"
+    completed = "completed"
+    failed = "failed"
+
+
 # ── Alert ─────────────────────────────────────────────────────────────────────
 
 class Alert(BaseModel):
@@ -109,7 +115,6 @@ class VideoMetrics(BaseModel):
 # ── Analysis Response ─────────────────────────────────────────────────────────
 
 class AnalysisResponse(BaseModel):
-    status: str = "completed"
     media_type: MediaType
     participant_label: Optional[str] = None
     processing_time_seconds: float
@@ -126,6 +131,25 @@ class AnalysisResponse(BaseModel):
     )
 
 
+# ── Job Polling Responses ─────────────────────────────────────────────────────
+
+class JobSubmitResponse(BaseModel):
+    job_id: str
+    status: JobStatus = JobStatus.processing
+    message: str
+
+
+class JobResultResponse(BaseModel):
+    job_id: str
+    status: JobStatus
+    result: Optional[AnalysisResponse] = Field(
+        None, description="Populated only if status is 'completed'"
+    )
+    error_detail: Optional[str] = Field(
+        None, description="Populated only if status is 'failed'"
+    )
+
+
 # ── Health ────────────────────────────────────────────────────────────────────
 
 class HealthResponse(BaseModel):

tests/test_api.py

@@ -7,26 +7,21 @@ Requires all dependencies installed (no Docker).
 from __future__ import annotations
 
 import io
-import struct
+import time
 import wave
-from pathlib import Path
 
 import numpy as np
 import pytest
 from fastapi.testclient import TestClient
 
 from app.main import app
-from app.config import get_settings
 
 
 # ── Fixtures ───────────────────────────────────────────────────────────────────
 
 @pytest.fixture(scope="module")
 def client():
-    """TestClient with valid API key injected."""
-    settings = get_settings()
-    headers = {"X-API-Key": settings.lecturelens_api_key}
-    with TestClient(app, headers=headers) as c:
+    with TestClient(app) as c:
         yield c
 
 
@@ -54,58 +49,43 @@ def test_health(client):
     assert r.json()["status"] == "ok"
 
 
-# ── Authentication ─────────────────────────────────────────────────────────────
-
-def test_missing_api_key_returns_401(wav_bytes):
-    with TestClient(app) as no_auth_client:
-        r = no_auth_client.post(
-            "/analyze",
-            data={"media_type": "audio"},
-            files={"file": ("test.wav", wav_bytes, "audio/wav")},
-        )
-    assert r.status_code == 401
-    assert r.json()["error_code"] == "UNAUTHORIZED"
-
-
-def test_wrong_api_key_returns_401(wav_bytes):
-    with TestClient(app, headers={"X-API-Key": "wrong-key"}) as bad_client:
-        r = bad_client.post(
-            "/analyze",
-            data={"media_type": "audio"},
-            files={"file": ("test.wav", wav_bytes, "audio/wav")},
-        )
-    assert r.status_code == 401
+# ── Polling Analysis ───────────────────────────────────────────────────────────
 
-
-# ── Audio analysis ─────────────────────────────────────────────────────────────
-
-def test_audio_analysis_returns_200(client, wav_bytes):
+def test_audio_analysis_polling(client, wav_bytes):
+    # 1. Submit the job
     r = client.post(
         "/analyze",
         data={"media_type": "audio", "participant_label": "test_speaker"},
         files={"file": ("test.wav", wav_bytes, "audio/wav")},
     )
     assert r.status_code == 200, r.text
-    body = r.json()
-    assert body["status"] == "completed"
-    assert body["media_type"] == "audio"
-    assert body["participant_label"] == "test_speaker"
-    assert "metrics" in body
-    assert "alerts" in body
-    assert body["overall_audio_score"] is not None
-
-
-def test_audio_response_schema(client, wav_bytes):
-    r = client.post(
-        "/analyze",
-        data={"media_type": "audio"},
-        files={"file": ("test.wav", wav_bytes, "audio/wav")},
-    )
-    assert r.status_code == 200
-    m = r.json()["metrics"]
-    # All expected fields present
-    for field in ["clipped_samples_count", "silence_segments"]:
-        assert field in m, f"Missing field: {field}"
+    submit_body = r.json()
+    assert submit_body["status"] == "processing"
+    assert "job_id" in submit_body
+    job_id = submit_body["job_id"]
+
+    # 2. Poll for the result
+    for _ in range(10):
+        r_poll = client.get(f"/analyze/{job_id}")
+        assert r_poll.status_code == 200
+        poll_body = r_poll.json()
+        
+        # TestClient actually runs BackgroundTasks synchronously at the end of the request.
+        # So by the time we poll, it's usually already 'completed'.
+        if poll_body["status"] == "completed":
+            res = poll_body["result"]
+            assert res["media_type"] == "audio"
+            assert res["participant_label"] == "test_speaker"
+            assert "metrics" in res
+            assert "alerts" in res
+            assert res["overall_audio_score"] is not None
+            break
+        elif poll_body["status"] == "failed":
+            pytest.fail(f"Job failed: {poll_body['error_detail']}")
+            
+        time.sleep(0.5)
+    else:
+        pytest.fail("Job did not complete within timeout")
 
 
 # ── Validation ─────────────────────────────────────────────────────────────────
@@ -120,7 +100,12 @@ def test_wrong_extension_returns_400(client, wav_bytes):
     assert r.json()["error_code"] == "INVALID_MEDIA_TYPE"
 
 
-def test_docs_accessible():
-    with TestClient(app) as c:
-        r = c.get("/docs")
+def test_job_not_found(client):
+    r = client.get("/analyze/non-existent-job-id")
+    assert r.status_code == 404
+    assert r.json()["error_code"] == "NOT_FOUND"
+
+
+def test_docs_accessible(client):
+    r = client.get("/docs")
     assert r.status_code == 200