Spaces:
Sleeping
Sleeping
File size: 4,408 Bytes
96fd859 | 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 133 134 135 136 137 138 139 140 141 142 143 144 145 | """EchoGuard API — FastAPI application entry point."""
import time
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from app.config import settings
from app.routers import analysis
logger = logging.getLogger(__name__)
_startup_time = time.time()
from app.state import model_info
@asynccontextmanager
async def lifespan(app: FastAPI):
model_info.status = "loading"
try:
from transformers import AutoModelForAudioClassification, AutoFeatureExtractor
model_name_gary = "garystafford/wav2vec2-deepfake-voice-detector"
model_info.gary_feature_extractor = AutoFeatureExtractor.from_pretrained(model_name_gary)
model_info.gary_model = AutoModelForAudioClassification.from_pretrained(model_name_gary)
model_name_bisher = "Bisher/wav2vec2_ASV_deepfake_audio_detection"
model_info.bisher_feature_extractor = AutoFeatureExtractor.from_pretrained(model_name_bisher)
model_info.bisher_model = AutoModelForAudioClassification.from_pretrained(model_name_bisher)
model_info.status = "ready"
logger.info("Loaded both DL models for ensemble detection.")
except Exception as e:
model_info.status = "failed"
model_info.error = str(e)
logger.error(f"Failed to load DL models: {e}")
yield
model_info.gary_feature_extractor = None
model_info.gary_model = None
model_info.bisher_feature_extractor = None
model_info.bisher_model = None
app = FastAPI(
title="EchoGuard API",
description=(
"AI-powered deepfake audio detection API. "
"Upload WAV or MP3 files for analysis. "
"Maximum file size: 30 MB. Maximum duration: 5 minutes."
),
version="0.1.0",
docs_url="/api/docs",
redoc_url="/api/redoc",
openapi_url="/api/openapi.json",
lifespan=lifespan,
)
origins = [origin.strip() for origin in settings.cors_origins.split(",")]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
"""Handle Pydantic/FastAPI validation errors with a clean JSON response."""
errors = exc.errors()
detail = "; ".join(
f"{err.get('loc', ['unknown'])[-1]}: {err.get('msg', 'validation error')}"
for err in errors
)
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content={
"error": "validation_error",
"detail": detail,
"status_code": 422,
},
)
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
"""Catch-all handler for unexpected exceptions."""
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={
"error": "internal_server_error",
"detail": "An unexpected error occurred. Please try again later.",
"status_code": 500,
},
)
app.include_router(analysis.router, prefix="/api")
@app.get(
"/",
summary="API Root",
description="Returns basic information about the EchoGuard API.",
tags=["System"],
)
async def root():
"""Root endpoint providing basic API information."""
return {
"name": "EchoGuard Deepfake Detection API",
"version": "0.1.0",
"status": "online",
"documentation": "/api/docs",
"endpoints": {
"health_check": "/api/health",
"analyze_audio": "/api/analyze"
}
}
@app.get(
"/api/health",
summary="Health check",
description="Returns the current health status and uptime of the EchoGuard API.",
tags=["System"],
)
async def health_check():
"""Health check endpoint."""
return {
"status": "healthy" if model_info.status == "ready" else "degraded",
"service": "EchoGuard API",
"version": "0.1.0",
"uptime_seconds": round(time.time() - _startup_time, 2),
"model_status": model_info.status,
"model_error": model_info.error
}
|