""" LectureLens — FastAPI Application 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 BackgroundTasks, Depends, FastAPI, File, Form, HTTPException, UploadFile, status from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from app.alert_engine import ( compute_audio_score, compute_video_score, generate_alerts, load_thresholds, ) from app.analyzers.audio_analyzer import analyze_audio, _get_dnsmos_session from app.config import Settings, get_settings from app.schemas import ( AnalysisResponse, ErrorResponse, HealthResponse, JobStatus, JobSubmitResponse, JobResultResponse, MediaType, ) from app.utils import ( AUDIO_EXTENSIONS, VIDEO_EXTENSIONS, detect_extension, validate_extension, ) # ── Logging ─────────────────────────────────────────────────────────────────── logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)-8s | %(name)s — %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) 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 async def lifespan(app: FastAPI): settings = get_settings() logger.info("🚀 LectureLens API starting up…") # Pre-load heavy models so first request is fast logger.info("Pre-loading DNSMOS model…") _get_dnsmos_session(settings.dnsmos_model_path) # Warm up thresholds cache logger.info("Loading thresholds from %s…", settings.thresholds_config_path) try: load_thresholds(settings.thresholds_config_path) except FileNotFoundError: logger.warning("thresholds.yaml not found — alert engine will use code defaults") logger.info("✅ LectureLens API ready.") yield logger.info("LectureLens API shutting down.") # ── FastAPI App ─────────────────────────────────────────────────────────────── app = FastAPI( title="LectureLens API", description=( "**LectureLens** — Technical audio & video quality analysis for Zoom lecture recordings.\n\n" "Accepts a single audio or video file per request and returns:\n" "- 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" "*(Note: API Key authentication is currently disabled for testing)*" ), version="1.0.0", docs_url="/docs", redoc_url="/redoc", lifespan=lifespan, ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["GET", "POST"], allow_headers=["*"], ) # ── Custom exception handlers ───────────────────────────────────────────────── @app.exception_handler(HTTPException) async def http_exception_handler(request, exc: HTTPException): detail = exc.detail if isinstance(detail, dict): return JSONResponse(status_code=exc.status_code, content=detail) return JSONResponse( status_code=exc.status_code, content={"error_code": "HTTP_ERROR", "detail": str(detail)}, ) @app.exception_handler(Exception) async def generic_exception_handler(request, exc: Exception): logger.exception("Unhandled exception: %s", exc) return JSONResponse( status_code=500, content={"error_code": "PROCESSING_FAILED", "detail": str(exc)}, ) # ── 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...") audio_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(audio_metrics, "audio", settings.thresholds_config_path) logger.info(f"[{job_id}] 🎧 Step 3/3: Computing composite audio score...") audio_score = compute_audio_score(audio_metrics) video_metrics = None video_score = None else: logger.info(f"[{job_id}] 🎬 Step 1/4: Starting video analysis...") from app.analyzers.video_analyzer import analyze_video video_metrics = await analyze_video(tmp_path, settings.frame_sample_rate_seconds) logger.info(f"[{job_id}] 🎬 Step 2/4: Extracting audio and starting audio analysis...") from app.utils import run_ffmpeg_async tmp_audio_path = Path(tempfile.NamedTemporaryFile(delete=False, suffix=".wav").name) try: # Extract audio to 16kHz mono WAV await run_ffmpeg_async([ "-i", str(tmp_path), "-vn", "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1", str(tmp_audio_path) ]) audio_metrics = await analyze_audio(tmp_audio_path, settings.dnsmos_model_path) except Exception as exc: logger.warning(f"[{job_id}] ⚠️ Failed to extract/analyze audio from video: {exc}") audio_metrics = None finally: if tmp_audio_path.exists(): try: os.unlink(tmp_audio_path) except: pass logger.info(f"[{job_id}] 🎬 Step 3/4: Generating alerts against thresholds...") alerts = generate_alerts(video_metrics, "video", settings.thresholds_config_path) if audio_metrics: audio_alerts = generate_alerts(audio_metrics, "audio", settings.thresholds_config_path) alerts.extend(audio_alerts) logger.info(f"[{job_id}] 🎬 Step 4/4: Computing composite video score...") video_score = compute_video_score(video_metrics) audio_score = compute_audio_score(audio_metrics) if audio_metrics else 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, audio_metrics=audio_metrics, video_metrics=video_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", response_model=HealthResponse, summary="Health check", tags=["System"], ) async def health(): """Returns service status and version.""" return HealthResponse() @app.post( "/analyze", 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"}, 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( None, description="Speaker name/role — returned as-is for easy n8n mapping" ), language: str = Form("ar", description="Recording language code (reserved for future use)"), settings: Settings = Depends(get_settings), ): """ **POST /analyze** — Submit a file and receive a `job_id`. The file is processed asynchronously in the background. Use `GET /analyze/{job_id}` to poll for the results to avoid HTTP timeouts. """ # ── 1. File size guard ──────────────────────────────────────────────────── content = await file.read() if len(content) > settings.max_file_size_bytes: raise HTTPException( status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail={ "error_code": "FILE_TOO_LARGE", "detail": ( f"File size {len(content) / 1e6:.1f} MB exceeds " f"the {settings.max_file_size_mb} MB limit." ), }, ) # ── 2. Extension validation ─────────────────────────────────────────────── filename = file.filename or f"upload.{media_type.value}" ext = detect_extension(filename) allowed = AUDIO_EXTENSIONS if media_type == MediaType.audio else VIDEO_EXTENSIONS if ext not in allowed: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail={ "error_code": "INVALID_MEDIA_TYPE", "detail": ( f"Extension '{ext}' is not supported for media_type='{media_type.value}'. " f"Allowed: {sorted(allowed)}" ), }, ) # ── 3. Save to temp file and dispatch background task ───────────────────── try: 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("Failed to queue file '%s': %s", filename, exc) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail={"error_code": "QUEUE_FAILED", "detail": f"Could not queue file: {exc}"} ) @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"] )