Spaces:
Sleeping
Sleeping
File size: 4,717 Bytes
60d70d0 119d4e9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | import base64
import os
import contextlib
import logging
from typing import Optional
import uvicorn
from fastapi import FastAPI, Header, HTTPException, Request, BackgroundTasks
from fastapi.responses import JSONResponse
from fastapi.concurrency import run_in_threadpool
from pydantic import BaseModel, Field
# Setup Logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - [AudioShield] - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Import our detection engine
from detect import detector
# load_dotenv() - REMOVED per user request for Render deployment
# LIFESPAN MANAGER (Resolves Cold Start)
@contextlib.asynccontextmanager
async def lifespan(app: FastAPI):
# Startup Logic
logger.info("--- Warming up the AI engine... ---")
try:
# Trigger model loading in a threadpool so it doesn't block startup completely if async
# But here we want to block until ready.
# Run a dummy inference to ensure weights are on device
dummy_audio = b'\x00' * 16000 # 1 sec silent
await run_in_threadpool(detector.analyze_audio, dummy_audio, "English")
logger.info("--- AI Engine Ready & Warmed Up! ---")
except Exception as e:
logger.error(f"Warmup failed: {e}")
yield
# Shutdown Logic
logger.info("--- Shutting down AudioShield ---")
app = FastAPI(
title="AudioShield AI: Voice Fraud Detector",
version="2.0",
docs_url="/docs",
lifespan=lifespan
)
# CONFIGURATION
# Default key from problem statement example: sk_test_123456789
VALID_API_KEY = os.getenv("API_KEY", "sk_test_123456789")
# MODELS (Strict Adherence to Spec)
class VoiceDetectionRequest(BaseModel):
language: str = Field(..., description="Language: Tamil, English, Hindi, Malayalam, Telugu")
audioFormat: str = Field(..., pattern="^(?i)mp3$", description="Must be 'mp3'")
audioBase64: str = Field(..., description="Base64 encoded MP3 audio")
class VoiceDetectionResponse(BaseModel):
status: str
language: str
classification: str # AI_GENERATED or HUMAN
confidenceScore: float
explanation: str
# ROUTES
@app.post("/api/voice-detection", response_model=VoiceDetectionResponse)
async def detect_voice(
request: VoiceDetectionRequest
):
# 1. Security Check - REMOVED for Public Access per user request
# logger.info(f"Public Access: Processing request for {request.language}")
try:
# 2. Basic Validation (Logic)
if request.audioFormat.lower() != "mp3":
# Just to be perfectly safe, though Pydantic regex handles it
raise ValueError("Only MP3 format is supported.")
# 3. Decode Base64
try:
audio_data = base64.b64decode(request.audioBase64)
except Exception:
raise ValueError("Invalid Base64 encoding.")
if not audio_data:
raise ValueError("Empty audio data.")
# 4. Perform Detection (Non-Blocking)
# We run the synchronous detector.analyze_audio in a threadpool
# so the API remains responsive to other requests.
logger.info(f"Processing request for language: {request.language}")
result = await run_in_threadpool(detector.analyze_audio, audio_data, request.language)
if "error" in result:
# If internal analysis failed, we still want to return a strict error format if possible,
# or map it to the error response.
raise ValueError(result["error"])
# 5. Return formatted response (Strict JSON)
return VoiceDetectionResponse(
status="success",
language=request.language,
classification=result["classification"],
confidenceScore=result["confidenceScore"],
explanation=result["explanation"]
)
except ValueError as ve:
logger.error(f"Validation Error: {ve}")
return JSONResponse(
status_code=400,
content={"status": "error", "message": str(ve)}
)
except Exception as e:
logger.error(f"Internal Error: {e}")
return JSONResponse(
status_code=500,
content={"status": "error", "message": "Internal server error processing audio."}
)
@app.get("/")
def health_check():
return {
"status": "online",
"service": "AudioShield AI (Hackathon Edition)",
"models_loaded": len(detector.pipelines) if hasattr(detector, 'pipelines') else 0
}
# Standard execution for HF Spaces (uvicorn launched via Docker CMD)
|