""" Deepfake Detection API — real-time interview frame analysis with Trust Meta-Classifier middleware + GradCAM + Temporal + TTA + Frequency Domain Analysis + Session Reports + Performance Optimizations. Run: uvicorn main:app --host 0.0.0.0 --port 8000 """ import io, base64, time, uuid import torch import torch.nn as nn import numpy as np import tempfile import os import json import urllib.request import traceback try: import cv2 except ImportError: pass from PIL import Image from fastapi import FastAPI, HTTPException, BackgroundTasks, Query from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from typing import Optional from torchvision import transforms from facenet_pytorch import InceptionResnetV1, fixed_image_standardization, MTCNN # ────────────────────────────────────── # Trust Meta-Classifier imports # ────────────────────────────────────── from trust_model import ( load_trust_model, build_trust_input, predict_trust, ) from frame_store import save_frame_record, get_review_stats # ────────────────────────────────────── # Game-Changer modules (additive) # ────────────────────────────────────── from gradcam import generate_gradcam_heatmap from temporal_analyzer import TemporalAnalyzer from tta import predict_with_tta # ────────────────────────────────────── # Priority 4-6: Frequency, Reports, Performance # ────────────────────────────────────── from frequency_analyzer import analyze_frequency from session_report import generate_session_report, list_sessions from performance import FrameDeduplicator WEIGHTS_PATH = "models/best_model.pt" DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu") IMG_SIZE = 299 CONF_THRESH = 0.60 print(f"[SERVER] Device: {DEVICE}") class DeepfakeClassifier(nn.Module): def __init__(self): super().__init__() self.backbone = InceptionResnetV1(classify=False, pretrained='vggface2').to(DEVICE) self.head = nn.Sequential( nn.Linear(512, 256), nn.BatchNorm1d(256), nn.GELU(), nn.Dropout(0.4), nn.Linear(256, 128), nn.BatchNorm1d(128), nn.GELU(), nn.Dropout(0.3), nn.Linear(128, 2) ).to(DEVICE) def forward(self, x): return self.head(self.backbone(x)) model = DeepfakeClassifier() model.load_state_dict(torch.load(WEIGHTS_PATH, map_location=DEVICE)) model.eval() print(f"[SERVER] Primary model loaded") # ────────────────────────────────────── # Initialize Temporal Analyzer (singleton) # ────────────────────────────────────── temporal_analyzer = TemporalAnalyzer() print(f"[SERVER] Temporal Analyzer initialized") # ────────────────────────────────────── # Initialize Frame Deduplicator (performance) # ────────────────────────────────────── frame_deduplicator = FrameDeduplicator(threshold=0.97, ttl_seconds=10.0) print(f"[SERVER] Frame Deduplicator initialized") # ────────────────────────────────────── # Load Trust Meta-Classifier # ────────────────────────────────────── trust_model, TRUST_COLD_START = load_trust_model(device=str(DEVICE)) mtcnn = MTCNN(image_size=IMG_SIZE, keep_all=False, min_face_size=40, device=DEVICE, post_process=False, margin=20) transform = transforms.Compose([ transforms.Resize((IMG_SIZE, IMG_SIZE)), transforms.ToTensor(), fixed_image_standardization, ]) CLASS_NAMES = {0: "FAKE", 1: "REAL"} app = FastAPI(title="Deepfake Detection API", version="2.0") app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) # ────────────────────────────────────── # Request / Response Models # ────────────────────────────────────── class FrameRequest(BaseModel): image_b64: str session_id: Optional[str] = None # Optional: groups frames from one live session class ProcessRequest(BaseModel): documentId: str fileUrl: str mediaType: str callbackUrl: str class PredictionResponse(BaseModel): label: str confidence: float uncertain: bool face_detected: bool latency_ms: float # Trust meta-classifier fields (additive — frontend can ignore) trust_verdict: str = "TRUSTED" trust_score: float = 1.0 # ── GradCAM explainability (additive) ── heatmap_b64: Optional[str] = None # ── Temporal consistency analysis (additive) ── temporal_consistency: float = 1.0 temporal_anomaly: bool = False temporal_drift: float = 0.0 temporal_details: str = "stable" # ── TTA adversarial robustness (additive) ── tta_confidence: Optional[float] = None tta_agreement: float = 1.0 tta_label: Optional[str] = None # ── Frequency domain analysis (additive) ── spectral_score: float = 0.5 spectral_anomaly: bool = False high_freq_energy: float = 0.0 spectral_details: str = "unavailable" # ── Performance dedup (additive) ── dedup_cache_hit: bool = False # ────────────────────────────────────── # Background media processing (unchanged) # ────────────────────────────────────── def process_media_task(req: ProcessRequest): try: req_download = urllib.request.Request(req.fileUrl, headers={'User-Agent': 'Mozilla/5.0'}) response = urllib.request.urlopen(req_download) content = response.read() preds = [] confs = [] if req.mediaType == 'video': fd, temp_path = tempfile.mkstemp(suffix=".mp4") with os.fdopen(fd, 'wb') as f: f.write(content) cap = cv2.VideoCapture(temp_path) frame_count = 0 while True: ret, frame = cap.read() if not ret: break frame_count += 1 if frame_count % 30 == 0: img = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)) face_tensor = mtcnn(img) if face_tensor is not None: face_pil = Image.fromarray(face_tensor.permute(1,2,0).byte().cpu().numpy()) inp = transform(face_pil).unsqueeze(0).to(DEVICE) with torch.no_grad(): probs = torch.softmax(model(inp), dim=1)[0] pred = int(torch.argmax(probs)) conf = float(probs[pred]) preds.append(pred) confs.append(conf) cap.release() os.remove(temp_path) if not preds: raise Exception("No face detected in video") avg_pred = int(round(sum(preds)/len(preds))) avg_conf = sum(confs)/len(confs) final_pred = avg_pred final_conf = avg_conf else: # image img = Image.open(io.BytesIO(content)).convert("RGB") face_tensor = mtcnn(img) if face_tensor is None: raise Exception("No face detected") face_pil = Image.fromarray(face_tensor.permute(1,2,0).byte().cpu().numpy()) inp = transform(face_pil).unsqueeze(0).to(DEVICE) with torch.no_grad(): probs = torch.softmax(model(inp), dim=1)[0] pred = int(torch.argmax(probs)) conf = float(probs[pred]) final_pred = pred final_conf = conf label = CLASS_NAMES[final_pred] # Send callback payload = json.dumps({ "documentId": req.documentId, "result": {"label": label, "confidence": final_conf} }).encode('utf-8') cb_req = urllib.request.Request(req.callbackUrl, data=payload, headers={'Content-Type': 'application/json'}, method='POST') urllib.request.urlopen(cb_req) print(f"[PROCESS] Done process for {req.documentId}") except Exception as e: traceback.print_exc() print(f"[PROCESS] Error processing {req.documentId}: {e}") try: payload = json.dumps({ "documentId": req.documentId, "result": {"error": str(e)} }).encode('utf-8') cb_req = urllib.request.Request(req.callbackUrl, data=payload, headers={'Content-Type': 'application/json'}, method='POST') urllib.request.urlopen(cb_req) except Exception as cb_err: print(f"[PROCESS] Callback error for {req.documentId}: {cb_err}") @app.post("/process") def process_media(req: ProcessRequest, background_tasks: BackgroundTasks): background_tasks.add_task(process_media_task, req) return {"message": "Processing started"} @app.get("/") def read_root(): return {"message": "DeepShield AI Server is running"} @app.get("/health") def health(): return { "status": "ok", "device": str(DEVICE), "trust_cold_start": TRUST_COLD_START, "capabilities": { "gradcam": True, "temporal_analysis": True, "tta_robustness": True, "trust_meta_classifier": not TRUST_COLD_START, "frequency_domain": True, "session_reports": True, "frame_deduplication": True, }, "active_temporal_sessions": len(temporal_analyzer._sessions), "dedup_cache_sessions": frame_deduplicator.active_sessions, } # ────────────────────────────────────── # /predict — with Trust Meta-Classifier # ────────────────────────────────────── @app.post("/predict", response_model=PredictionResponse) def predict(req: FrameRequest, background_tasks: BackgroundTasks): global TRUST_COLD_START t0 = time.perf_counter() session_id = req.session_id or f"session_{uuid.uuid4().hex[:12]}" # Decode image try: raw_bytes = base64.b64decode(req.image_b64) img = Image.open(io.BytesIO(raw_bytes)).convert("RGB") except Exception as e: raise HTTPException(status_code=400, detail=f"Bad image: {e}") # Detect face face_tensor = mtcnn(img) if face_tensor is None: latency = round((time.perf_counter() - t0) * 1000, 1) return PredictionResponse( label="UNKNOWN", confidence=0.0, uncertain=True, face_detected=False, latency_ms=latency, trust_verdict="UNTRUSTED", trust_score=0.0, ) # Prepare input face_pil = Image.fromarray(face_tensor.permute(1, 2, 0).byte().cpu().numpy()) inp = transform(face_pil).unsqueeze(0).to(DEVICE) # ── Step 0: Frame Deduplication (performance optimization) ── dedup_result = None try: dedup_result = frame_deduplicator.check(session_id, face_pil) except Exception as e: print(f"[PREDICT] Dedup check error (non-fatal): {e}") if dedup_result is not None: # Cache hit — skip all inference latency = round((time.perf_counter() - t0) * 1000, 1) return PredictionResponse( label=dedup_result.get("label", "UNKNOWN"), confidence=dedup_result.get("confidence", 0.0), uncertain=dedup_result.get("uncertain", False), face_detected=True, latency_ms=latency, trust_verdict=dedup_result.get("trust_verdict", "TRUSTED"), trust_score=dedup_result.get("trust_score", 1.0), heatmap_b64=dedup_result.get("heatmap_b64"), temporal_consistency=dedup_result.get("temporal_consistency", 1.0), temporal_anomaly=dedup_result.get("temporal_anomaly", False), temporal_drift=dedup_result.get("temporal_drift", 0.0), temporal_details=dedup_result.get("temporal_details", "stable"), tta_confidence=dedup_result.get("tta_confidence"), tta_agreement=dedup_result.get("tta_agreement", 1.0), tta_label=dedup_result.get("tta_label"), spectral_score=dedup_result.get("spectral_score", 0.5), spectral_anomaly=dedup_result.get("spectral_anomaly", False), high_freq_energy=dedup_result.get("high_freq_energy", 0.0), spectral_details=dedup_result.get("spectral_details", "unavailable"), dedup_cache_hit=True, ) # ── Step 1: Primary model inference (split backbone + head) ── with torch.no_grad(): embedding = model.backbone(inp) # 512-dim face embedding logits = model.head(embedding) # classification logits probs = torch.softmax(logits, dim=1)[0] pred = int(torch.argmax(probs)) conf = float(probs[pred]) latency = round((time.perf_counter() - t0) * 1000, 1) # ── Step 2: Trust Meta-Classifier ── trust_input = build_trust_input( embedding=embedding[0], primary_pred_idx=pred, confidence=conf, latency_ms=latency, device=str(DEVICE), ) trust_result = predict_trust(trust_model, trust_input, cold_start=TRUST_COLD_START) # ── Step 3: GradCAM Heatmap (explainability) ── heatmap_b64 = None try: heatmap_b64 = generate_gradcam_heatmap(model, inp, pred, device=str(DEVICE)) except Exception as e: print(f"[PREDICT] GradCAM error (non-fatal): {e}") # ── Step 4: Temporal Consistency Analysis ── temporal_result = { "consistency": 1.0, "anomaly": False, "drift": 0.0, "details": "stable", } try: temporal_result = temporal_analyzer.analyze(session_id, embedding[0]) except Exception as e: print(f"[PREDICT] Temporal error (non-fatal): {e}") # ── Step 5: Test-Time Augmentation (adversarial robustness) ── tta_result = {"tta_confidence": None, "tta_agreement": 1.0, "tta_label": None} try: tta_data = predict_with_tta(face_pil, model, transform, DEVICE) tta_result = { "tta_confidence": tta_data["tta_confidence"], "tta_agreement": tta_data["tta_agreement"], "tta_label": tta_data["tta_label"], } except Exception as e: print(f"[PREDICT] TTA error (non-fatal): {e}") # ── Step 5.5: Frequency Domain Analysis ── freq_result = { "spectral_score": 0.5, "high_freq_energy": 0.0, "spectral_anomaly": False, "spectral_details": "unavailable", } try: freq_result = analyze_frequency(face_pil) except Exception as e: print(f"[PREDICT] Frequency analysis error (non-fatal): {e}") # ── Step 6: Save frame record in background (fire-and-forget) ── # Convert frame to JPEG bytes for storage frame_buffer = io.BytesIO() img.save(frame_buffer, format="JPEG", quality=75) frame_jpeg = frame_buffer.getvalue() background_tasks.add_task( save_frame_record, session_id=session_id, frame_bytes=frame_jpeg, primary_label=CLASS_NAMES[pred], primary_confidence=conf, face_detected=True, latency_ms=latency, embedding_list=embedding[0].cpu().tolist(), trust_verdict=trust_result["trust_verdict"], trust_score=trust_result["trust_score"], ) # ── Step 7: Cache result for dedup ── dedup_cache_data = { "label": CLASS_NAMES[pred], "confidence": round(conf, 3), "uncertain": (conf < CONF_THRESH), "trust_verdict": trust_result["trust_verdict"], "trust_score": trust_result["trust_score"], "heatmap_b64": heatmap_b64, "temporal_consistency": temporal_result.get("consistency", 1.0), "temporal_anomaly": temporal_result.get("anomaly", False), "temporal_drift": temporal_result.get("drift", 0.0), "temporal_details": temporal_result.get("details", "stable"), "tta_confidence": tta_result.get("tta_confidence"), "tta_agreement": tta_result.get("tta_agreement", 1.0), "tta_label": tta_result.get("tta_label"), "spectral_score": freq_result.get("spectral_score", 0.5), "spectral_anomaly": freq_result.get("spectral_anomaly", False), "high_freq_energy": freq_result.get("high_freq_energy", 0.0), "spectral_details": freq_result.get("spectral_details", "unavailable"), } try: frame_deduplicator.store(session_id, face_pil, dedup_cache_data) except Exception as e: print(f"[PREDICT] Dedup store error (non-fatal): {e}") # ── Step 8: Return response (backward-compatible + new fields) ── return PredictionResponse( label=CLASS_NAMES[pred], confidence=round(conf, 3), uncertain=(conf < CONF_THRESH), face_detected=True, latency_ms=latency, trust_verdict=trust_result["trust_verdict"], trust_score=trust_result["trust_score"], # Game-changer fields heatmap_b64=heatmap_b64, temporal_consistency=temporal_result.get("consistency", 1.0), temporal_anomaly=temporal_result.get("anomaly", False), temporal_drift=temporal_result.get("drift", 0.0), temporal_details=temporal_result.get("details", "stable"), tta_confidence=tta_result.get("tta_confidence"), tta_agreement=tta_result.get("tta_agreement", 1.0), tta_label=tta_result.get("tta_label"), # Frequency domain fields spectral_score=freq_result.get("spectral_score", 0.5), spectral_anomaly=freq_result.get("spectral_anomaly", False), high_freq_energy=freq_result.get("high_freq_energy", 0.0), spectral_details=freq_result.get("spectral_details", "unavailable"), dedup_cache_hit=False, ) # ────────────────────────────────────── # Session Report Endpoints # ────────────────────────────────────── @app.get("/recent_threats") def recent_threats(limit: int = Query(default=10, le=50)): """Fetch recent threat detections for the dashboard feed.""" from frame_store import get_recent_threats return get_recent_threats(limit=limit) @app.get("/sessions") def get_sessions(limit: int = Query(default=50, le=200)): """List recent sessions with summary stats.""" return list_sessions(limit=limit) @app.get("/sessions/{session_id}/report") def get_session_report(session_id: str): """Generate a comprehensive forensic report for a specific session.""" report = generate_session_report(session_id) if "error" in report: raise HTTPException(status_code=404, detail=report["error"]) return report # ────────────────────────────────────── # Retraining Endpoints # ────────────────────────────────────── @app.get("/retrain/status") def retrain_status(): """Returns review/training statistics.""" stats = get_review_stats() stats["cold_start"] = TRUST_COLD_START stats["min_frames_for_training"] = 50 stats["ready_to_train"] = stats["available_for_training"] >= 50 return stats @app.post("/retrain") def trigger_retrain(background_tasks: BackgroundTasks): """Triggers retraining of the trust meta-classifier.""" try: from retrain_trust_model import retrain result = retrain(device=str(DEVICE)) # Hot-reload the newly saved model so it applies instantly global trust_model, TRUST_COLD_START trust_model, TRUST_COLD_START = load_trust_model(device=str(DEVICE)) result["hot_reloaded"] = True return result except Exception as e: traceback.print_exc() raise HTTPException(status_code=500, detail=f"Retrain failed: {e}") @app.post("/retrain/approve") def approve_retrain(): """Promotes the candidate model to production.""" global trust_model, TRUST_COLD_START try: from retrain_trust_model import approve result = approve() # Hot-reload the new model trust_model, TRUST_COLD_START = load_trust_model(device=str(DEVICE)) result["hot_reloaded"] = True return result except Exception as e: traceback.print_exc() raise HTTPException(status_code=500, detail=f"Approve failed: {e}")