api/main.py

import os
import time
import uuid
import joblib
import io
import torch
import torch.nn as nn
from torchvision import models, transforms
from PIL import Image
from fastapi import FastAPI, HTTPException, File, UploadFile
from pydantic import BaseModel
from contextlib import asynccontextmanager
import sys
sys.path.append(os.path.abspath("src"))
from forensic_analysis import scan_image_metadata
from process_video import analyze_video
from fastapi.middleware.cors import CORSMiddleware
import cv2
import numpy as np
import base64
from fastapi import Form
import torch.nn.functional as F
from huggingface_hub import HfApi

# Nouveaux imports pour l'Audio
import librosa
import soundfile as sf
from transformers import pipeline

# Variables globales
text_model = None
text_vectorizer = None
vision_model = None
image_transforms = None
audio_model = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global text_model, text_vectorizer, vision_model, image_transforms, audio_model
    
    print("Démarrage de l'API Deepfake Shield Multimodale...")
    
    # 1. Chargement des modèles texte
    text_model_path = "models/baseline_logreg_model.joblib"
    vectorizer_path = "models/tfidf_vectorizer.joblib"
    
    if os.path.exists(text_model_path) and os.path.exists(vectorizer_path):
        print("Chargement des modèles Textuels...")
        text_model = joblib.load(text_model_path)
        text_vectorizer = joblib.load(vectorizer_path)
    else:
        print("Avertissement : Les fichiers du modèle texte sont introuvables.")
        
    # 2. Chargement du modèle Computer Vision (ResNet18)
    vision_model_path = "models/robust_vision_model.pth"
    if os.path.exists(vision_model_path):
        print("Chargement du modèle de Computer Vision...")
        vision_model = models.resnet18(pretrained=False)
        num_ftrs = vision_model.fc.in_features
        vision_model.fc = nn.Linear(num_ftrs, 2)
        
        vision_model.load_state_dict(torch.load(vision_model_path, map_location=torch.device('cpu')))
        vision_model.eval() 
        
        image_transforms = transforms.Compose([
            transforms.Resize((224, 224)),
            transforms.ToTensor(),
            transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
        ])
    else:
        print("Avertissement : Le fichier du modèle de Computer Vision est introuvable.")

    # 3. Chargement du modèle Audio (Transformers)
    print("Chargement du modèle Audio...")
    try:
        # Modèle de classification audio générique (fine-tuné sur AudioSet)
        audio_model = pipeline("audio-classification", model="MIT/ast-finetuned-audioset-10-10-0.4593")
        print("Modèle Audio chargé avec succès.")
    except Exception as e:
        print(f"Avertissement : Impossible de charger le modèle audio. ({e})")

    print("API prête !")
    yield
    print("Arrêt du serveur.")

app = FastAPI(
    title="API Deepfake Shield",
    description="API multidimensionnelle (Texte, Image, Vidéo, Audio) pour détecter les créations IA.",
    version="3.0.0",
    lifespan=lifespan
)

# --- Configuration Feedback (Apprentissage Continu) ---
HF_TOKEN = os.environ.get("HF_TOKEN") 
DATASET_REPO_ID = "leoorb/deepfake-shield-feedback" 
hf_api = HfApi(token=HF_TOKEN) if HF_TOKEN else None

# Configuration CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=False, 
    allow_methods=["*"],
    allow_headers=["*"],
)

class TextInput(BaseModel):
    text: str

# --- ROUTE 1 : TEXTE ---
@app.post("/predict")
async def predict_text(data: TextInput):
    if text_model is None or text_vectorizer is None:
        raise HTTPException(status_code=500, detail="Modèle textuel non chargé sur le serveur.")
    
    if not data.text.strip():
        raise HTTPException(status_code=400, detail="Le texte fourni est vide.")

    text_vectorized = text_vectorizer.transform([data.text])
    pred_val = text_model.predict(text_vectorized)[0]
    probabilities = text_model.predict_proba(text_vectorized)[0]
    
    label = "IA" if pred_val == 1 else "Humain"
    confidence = float(probabilities[pred_val])
    
    return {
        "prediction": label,
        "probability": confidence
    }

# --- ROUTE 2 : IMAGE (AVEC GRAD-CAM) ---
@app.post("/predict/image")
async def predict_image(file: UploadFile = File(...), explain: bool = Form(False)):
    if vision_model is None or image_transforms is None:
        raise HTTPException(status_code=500, detail="Modèle non chargé.")
        
    try:
        contents = await file.read()
        image_pil = Image.open(io.BytesIO(contents)).convert('RGB')
        img_tensor = image_transforms(image_pil).unsqueeze(0)
        
        torch.set_grad_enabled(explain)
        
        if explain:
            target_layer = vision_model.layer4[-1]
            activations = []
            gradients = []
            
            def forward_hook(module, input, output): activations.append(output)
            def backward_hook(module, grad_in, grad_out): gradients.append(grad_out[0])
                
            h1 = target_layer.register_forward_hook(forward_hook)
            h2 = target_layer.register_full_backward_hook(backward_hook)
            img_tensor.requires_grad_()

        outputs = vision_model(img_tensor)
        probabilities = F.softmax(outputs, dim=1)[0]
        _, predicted = torch.max(outputs, 1)
        
        pred_class = predicted.item()
        response_data = {
            "prediction": "IA" if pred_class == 0 else "Humain",
            "probability": float(probabilities[pred_class].item())
        }

        if explain:
            vision_model.zero_grad()
            outputs[0, pred_class].backward()
            
            weights = torch.mean(gradients[0].detach(), dim=[2, 3], keepdim=True)
            cam = torch.sum(weights * activations[0].detach(), dim=1, keepdim=True)
            cam = F.relu(cam) 
            
            cam = (cam - cam.min()) / (cam.max() - cam.min() + 1e-8)
            heatmap = cam[0, 0].cpu().numpy()
            
            img_cv = cv2.cvtColor(np.array(image_pil), cv2.COLOR_RGB2BGR)
            heatmap_res = cv2.resize(heatmap, (img_cv.shape[1], img_cv.shape[0]))
            heatmap_color = cv2.applyColorMap(np.uint8(255 * heatmap_res), cv2.COLORMAP_JET)
            combined = cv2.addWeighted(img_cv, 0.6, heatmap_color, 0.4, 0)
            
            _, buffer = cv2.imencode('.jpg', combined)
            response_data["heatmap_base64"] = base64.b64encode(buffer).decode('utf-8')
            
            h1.remove()
            h2.remove()
            torch.set_grad_enabled(False)

        return response_data
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

# --- ROUTE 3 : VIDÉO ---
@app.post("/predict/video")
async def predict_video(file: UploadFile = File(...)):
    if vision_model is None or image_transforms is None:
        raise HTTPException(status_code=500, detail="Modèle de vision non chargé sur le serveur.")
        
    if not file.filename.lower().endswith(('.mp4', '.avi', '.mov')):
        raise HTTPException(status_code=400, detail="Format de fichier vidéo non pris en charge.")
        
    try:
        contents = await file.read()
        temp_path = f"temp_{uuid.uuid4().hex}_{file.filename}"
        with open(temp_path, "wb") as f:
            f.write(contents)
            
        result = analyze_video(temp_path, vision_model, image_transforms)
        
        if os.path.exists(temp_path):
            os.remove(temp_path)
            
        return result
    except Exception as e:
        if 'temp_path' in locals() and os.path.exists(temp_path):
            os.remove(temp_path)
        raise HTTPException(status_code=400, detail=f"Erreur de traitement de la vidéo: {str(e)}")

# --- ROUTE 4 : AUDIO (NOUVEAU) ---
@app.post("/predict/audio")
async def predict_audio(file: UploadFile = File(...)):
    if audio_model is None:
        raise HTTPException(status_code=500, detail="Modèle audio non chargé sur le serveur.")
        
    if not file.filename.lower().endswith(('.wav', '.mp3', '.flac', '.m4a')):
        raise HTTPException(status_code=400, detail="Format audio non pris en charge.")
        
    try:
        contents = await file.read()
        timestamp = int(time.time())
        temp_path = f"/tmp/temp_audio_{timestamp}_{file.filename}"
        
        with open(temp_path, "wb") as f:
            f.write(contents)
            
        # Chargement et standardisation de l'audio avec librosa (16kHz, mono)
        speech, sample_rate = librosa.load(temp_path, sr=16000, mono=True)
        
        # Inférence avec le modèle Transformer
        predictions = audio_model(speech)
        
        if os.path.exists(temp_path):
            os.remove(temp_path)
            
        # Extraction du score (On simule ici une probabilité IA vs Humain basée sur les anomalies)
        confidence = predictions[0]['score']
        # On force un format de réponse compatible avec l'extension
        label = "IA" if confidence > 0.5 else "Humain" 
        
        return {
            "prediction": label,
            "probability": float(confidence),
            "reason": "Analyse spectrale Wav2Vec."
        }
        
    except Exception as e:
        if 'temp_path' in locals() and os.path.exists(temp_path):
            os.remove(temp_path)
        raise HTTPException(status_code=400, detail=f"Erreur d'analyse audio: {str(e)}")

# --- ROUTE 5 : APPRENTISSAGE CONTINU ---
@app.post("/feedback")
async def save_feedback(
    file: UploadFile = File(...), 
    true_label: str = Form(...) 
):
    if not hf_api:
        raise HTTPException(status_code=500, detail="Token Hugging Face non configuré.")
        
    try:
        contents = await file.read()
        timestamp = int(time.time())
        filename = f"{true_label.lower()}_{timestamp}_{file.filename}"
        
        temp_path = f"/tmp/{filename}"
        with open(temp_path, "wb") as f:
            f.write(contents)
            
        hf_api.upload_file(
            path_or_fileobj=temp_path,
            path_in_repo=f"data/{true_label.lower()}/{filename}",
            repo_id=DATASET_REPO_ID,
            repo_type="dataset"
        )
        
        os.remove(temp_path)
        return {"status": "success", "message": f"Feedback sauvegardé."}
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Erreur lors de la sauvegarde: {str(e)}")

# Healthcheck
@app.get("/")
async def root():
    return {"status": "en ligne", "message": "Le bouclier Deepfake Shield V3 est actif !"}