app/main.py

"""
AI Voice Detection API
Main FastAPI application entry point
"""


import os
import sys
import subprocess
import traceback
import datetime
from contextlib import asynccontextmanager
from typing import Any, Dict

import numpy as np
from fastapi import FastAPI, Request, UploadFile, File, Form
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from dotenv import load_dotenv

from app.routes.voice_detection import router as voice_router
from app.ml_detector import get_ml_detector
from app.audio.audio_processor import audio_processor
from app.voice_detector import voice_detector

# Load environment variables
load_dotenv()

# ------------------------------------------------------------------
# Torch bootstrap (CPU only, installed at runtime if missing)
# ------------------------------------------------------------------
def ensure_torch():
    try:
        import torch  # noqa
        import torchaudio  # noqa
        print("✅ Torch already installed")
    except ImportError:
        print("⬇️ Installing torch + torchaudio (CPU)")
        subprocess.check_call([
            sys.executable,
            "-m",
            "pip",
            "install",
            "torch",
            "torchaudio",
            "--index-url",
            "https://download.pytorch.org/whl/cpu"
        ])


# ------------------------------------------------------------------
# FastAPI lifespan
# ------------------------------------------------------------------
@asynccontextmanager
async def lifespan(app: FastAPI):
    """
    Lifespan events:
    - Startup: Ensure torch is installed, then load ML models
    - Shutdown: Clean up resources
    """
    print("🚀 Starting up... Pre-loading ML models")
    
    # Ensure torch exists before model loading
    ensure_torch()

    # Preload ML models
    try:
        detector = get_ml_detector()
        detector.load_model()
        print("✅ ML Models loaded successfully")
    except Exception as e:
        print(f"⚠️ Warning: Model loading failed: {e}")

    yield

    print("🛑 Shutting down...")


# ------------------------------------------------------------------
# FastAPI app
# ------------------------------------------------------------------
app = FastAPI(
    title="Voice Detection API",
    lifespan=lifespan,
    description="""
    REST API for detecting AI-generated voices in audio samples.

    ## Features
    - Detects AI-generated vs human voices
    - Supports 5 languages: Tamil, English, Hindi, Malayalam, Telugu
    - Returns confidence scores and explanations
    - Hackathon-compatible root POST endpoint

    ## Usage
    Send a POST request to `/` or `/api/voice-detection` with:
    - multipart form-data with `file` and `language`
    """,
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
)

# CORS (open for hackathon / demo)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# API routes
app.include_router(voice_router)


# ------------------------------------------------------------------
# Health check
# ------------------------------------------------------------------
@app.get("/", tags=["Health"])
async def health_check():
    return {
        "status": "healthy",
        "service": "AI Voice Detection API",
        "version": "1.0.0",
        "languages": ["Tamil", "English", "Hindi", "Malayalam", "Telugu"],
    }


SAFE_CONFIDENCE = 0.75
MIN_AUDIO_BYTES = 100
ALLOWED_AUDIO_EXTENSIONS = {".wav", ".mp3"}


def _success_response(language: str, classification: str, confidence: float, explanation: str) -> Dict[str, Any]:
    return {
        "status": "success",
        "language": language,
        "classification": classification,
        "confidenceScore": round(confidence, 2),
        "explanation": explanation,
    }


def _human_fallback(language: str, explanation: str) -> Dict[str, Any]:
    return _success_response(language, "HUMAN", SAFE_CONFIDENCE, explanation)


def _error_response(message: str) -> Dict[str, str]:
    return {
        "status": "error",
        "message": message,
    }


def _is_supported_audio_file(filename: str | None) -> bool:
    if not filename:
        return False
    extension = os.path.splitext(filename)[1].lower()
    return extension in ALLOWED_AUDIO_EXTENSIONS


# ------------------------------------------------------------------
# Hackathon root POST endpoint
# ------------------------------------------------------------------
@app.post("/", tags=["Detection"])
async def root_detect(
    file: UploadFile = File(...),
    language: str = Form(default="english"),
):
    request_id = datetime.datetime.now().strftime("%H%M%S%f")[:10]
    start_time = datetime.datetime.now()

    print(f"\n{'='*70}")
    print(f"📥 REQUEST #{request_id} | {start_time.isoformat()}")
    print(f"   Language: {language} | File: {file.filename}")

    try:
        if not _is_supported_audio_file(file.filename):
            return _error_response("Unsupported file type. Only .wav and .mp3 are allowed")

        audio_bytes = await file.read()
        audio_bytes_len = len(audio_bytes)
        print(f"   File size: {audio_bytes_len:,} bytes ({audio_bytes_len/1024:.1f} KB)")

        if audio_bytes_len < MIN_AUDIO_BYTES:
            return _human_fallback(language, "Audio sample too short for reliable detection")

        try:
            features, audio_samples, sample_rate = audio_processor.process_audio_file(
                audio_bytes, file.filename or "audio.mp3"
            )
        except Exception as proc_err:
            print(f"   ❌ Audio processing failed: {proc_err}")
            try:
                result = voice_detector.detect({}, audio=None, sr=None, audio_bytes=audio_bytes)
                return _success_response(
                    language,
                    result["classification"],
                    result["confidenceScore"],
                    "Processed with transformers only (audio decode fallback)",
                )
            except Exception:
                return _error_response(f"Audio processing failed: {str(proc_err)}")

        if audio_samples is not None and len(audio_samples) > 0:
            duration = len(audio_samples) / sample_rate
            rms = float(np.sqrt(np.mean(audio_samples ** 2)))
            peak = float(np.max(np.abs(audio_samples)))
            print(f"   Duration: {duration:.2f}s | SR: {sample_rate}Hz | RMS: {rms:.4f} | Peak: {peak:.4f}")

        try:
            result = voice_detector.detect(
                features,
                audio=audio_samples,
                sr=sample_rate,
                audio_bytes=audio_bytes,
            )
        except Exception as detect_err:
            print(f"   ❌ Detection failed: {detect_err}")
            traceback.print_exc()
            return _human_fallback(language, "Detection error, defaulting to HUMAN")

        elapsed = (datetime.datetime.now() - start_time).total_seconds()
        print(f"\n🔍 RESULT #{request_id}:")
        print(f"   Classification: {result['classification']} | Confidence: {result['confidenceScore']:.3f}")
        print(f"   Time: {elapsed:.2f}s")
        print(f"{'='*70}\n")

        return _success_response(
            language,
            result["classification"],
            result["confidenceScore"],
            result["explanation"],
        )

    except Exception as e:
        elapsed = (datetime.datetime.now() - start_time).total_seconds()
        print(f"\n❌ ERROR #{request_id}: {str(e)}")
        traceback.print_exc()
        return _human_fallback(language, "Processing error, defaulting to HUMAN")


# ------------------------------------------------------------------
# Global exception handler
# ------------------------------------------------------------------
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """Global exception handler for unhandled errors."""
    return JSONResponse(
        status_code=500,
        content={
            "status": "error",
            "message": "Internal server error. Please try again later.",
        },
    )


# ------------------------------------------------------------------
# Local dev entrypoint (not used by Docker CMD)
# ------------------------------------------------------------------
if __name__ == "__main__":
    import uvicorn

    port = int(os.getenv("PORT", 8000))
    uvicorn.run("app.main:app", host="0.0.0.0", port=port, reload=True)