|
import os |
|
import tempfile |
|
import uvicorn |
|
from fastapi import FastAPI, File, UploadFile, HTTPException |
|
from fastapi.middleware.cors import CORSMiddleware |
|
from fastapi.responses import JSONResponse |
|
import shutil |
|
from pydantic import BaseModel |
|
from typing import Optional, Dict, Any, List |
|
|
|
|
|
from app import DeepfakeDetector, convert_audio |
|
|
|
|
|
app = FastAPI( |
|
title="Deepfake Voice Detection API", |
|
description="API for detecting deepfake audio using the MelodyMachine/Deepfake-audio-detection-V2 model", |
|
version="0.1.0", |
|
) |
|
|
|
|
|
app.add_middleware( |
|
CORSMiddleware, |
|
allow_origins=["*"], |
|
allow_credentials=True, |
|
allow_methods=["*"], |
|
allow_headers=["*"], |
|
) |
|
|
|
|
|
detector = None |
|
|
|
@app.on_event("startup") |
|
async def startup_event(): |
|
global detector |
|
detector = DeepfakeDetector() |
|
print("Deepfake Detector model loaded and ready to use") |
|
|
|
class PredictionResponse(BaseModel): |
|
prediction: str |
|
confidence: float |
|
probabilities: Dict[str, float] |
|
|
|
@app.post("/detect/", response_model=PredictionResponse) |
|
async def detect_audio(file: UploadFile = File(...)): |
|
""" |
|
Detect if an audio file contains a deepfake voice |
|
""" |
|
if not file: |
|
raise HTTPException(status_code=400, detail="No file provided") |
|
|
|
|
|
if not file.filename.lower().endswith(('.wav', '.mp3', '.ogg', '.flac')): |
|
raise HTTPException( |
|
status_code=400, |
|
detail="Invalid file format. Only WAV, MP3, OGG, and FLAC files are supported." |
|
) |
|
|
|
try: |
|
|
|
temp_dir = tempfile.gettempdir() |
|
temp_path = os.path.join(temp_dir, file.filename) |
|
|
|
|
|
with open(temp_path, "wb") as buffer: |
|
shutil.copyfileobj(file.file, buffer) |
|
|
|
|
|
processed_audio = convert_audio(temp_path) |
|
|
|
|
|
result = detector.detect(processed_audio) |
|
|
|
|
|
try: |
|
os.remove(temp_path) |
|
os.remove(processed_audio) if processed_audio != temp_path else None |
|
except: |
|
pass |
|
|
|
return result |
|
|
|
except Exception as e: |
|
raise HTTPException(status_code=500, detail=f"Error processing audio: {str(e)}") |
|
|
|
@app.get("/health/") |
|
async def health_check(): |
|
""" |
|
Check if the API is running and the model is loaded |
|
""" |
|
if detector is None: |
|
return JSONResponse( |
|
status_code=503, |
|
content={"status": "error", "message": "Model not loaded"} |
|
) |
|
return {"status": "ok", "model_loaded": True} |
|
|
|
@app.get("/model-info/") |
|
async def model_info(): |
|
""" |
|
Get information about the model being used |
|
""" |
|
return { |
|
"model_id": "MelodyMachine/Deepfake-audio-detection-V2", |
|
"base_model": "facebook/wav2vec2-base", |
|
"performance": { |
|
"loss": 0.0141, |
|
"accuracy": 0.9973 |
|
}, |
|
"description": "Fine-tuned model for binary classification distinguishing between real and deepfake audio" |
|
} |
|
|
|
if __name__ == "__main__": |
|
uvicorn.run("api:app", host="0.0.0.0", port=8000, reload=True) |