import sys print("DEBUG: main.py is being executed!", flush=True) from fastapi import FastAPI, UploadFile, File, BackgroundTasks from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, FileResponse from fastapi.staticfiles import StaticFiles import shutil import os import traceback import concurrent.futures from uuid import uuid4 from fastapi import HTTPException, Security, Depends, Request from fastapi.security.api_key import APIKeyHeader from fastapi.responses import StreamingResponse import json import asyncio from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded app = FastAPI(title="Deepfake Forensics API", version="2.0.0") # Rate Limiting limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # Security Configuration API_KEY = os.environ.get("API_KEY") or "deepforensics-dev-key" API_KEY_NAME = "x-api-key" api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=True) async def get_api_key(api_key_header: str = Security(api_key_header)): if api_key_header == API_KEY: return api_key_header print(f"DEBUG: Received API Key '{api_key_header}', expected '{API_KEY}'") raise HTTPException(status_code=401, detail="Invalid API Key") # Allow CORS for specific origins ALLOWED_ORIGINS = os.environ.get("ALLOWED_ORIGINS", "*").split(",") app.add_middleware( CORSMiddleware, allow_origins=ALLOWED_ORIGINS, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.get("/") async def root(): return { "status": "online", "message": "DeepForensics API is running. Please use the Vercel frontend to interact with this service." } UPLOAD_DIR = "uploads" REPORT_DIR = "reports" os.makedirs(UPLOAD_DIR, exist_ok=True) os.makedirs(REPORT_DIR, exist_ok=True) # Mount the uploads directory to serve images to the frontend app.mount("/uploads", StaticFiles(directory="uploads"), name="uploads") # In-memory storage for analysis status analysis_jobs = {} # Lazy Loading Models _models = {} def get_detector(): if "detector" not in _models: print("Lazy loading DeepfakeDetector...") from pipeline.models import DeepfakeDetector _models["detector"] = DeepfakeDetector() return _models["detector"] def get_explainer(): if "explainer" not in _models: print("Lazy loading XAIExplainer...") from pipeline.xai_explainer import XAIExplainer _models["explainer"] = XAIExplainer(get_detector().model) return _models["explainer"] def get_meta_classifier(): if "meta_classifier" not in _models: print("Lazy loading DeepfakeMetaClassifier...") from pipeline.ensemble_classifier import DeepfakeMetaClassifier _models["meta_classifier"] = DeepfakeMetaClassifier() _models["meta_classifier"].load_model() return _models["meta_classifier"] @app.post("/api/analyze") @limiter.limit("5/minute") async def analyze_video(request: Request, background_tasks: BackgroundTasks, file: UploadFile = File(...), api_key: str = Depends(get_api_key)): job_id = str(uuid4()) file_extension = file.filename.split(".")[-1].lower() # 1. File Type & True MIME-Type Validation ALLOWED_EXTENSIONS = {"mp4", "avi", "mov", "mkv", "webm", "png", "jpg", "jpeg"} if file_extension not in ALLOWED_EXTENSIONS: raise HTTPException(status_code=400, detail=f"Invalid file type. Allowed: {', '.join(ALLOWED_EXTENSIONS)}") # 2. File Size Validation & Saving (100 MB Limit) file_path = os.path.join(UPLOAD_DIR, f"{job_id}.{file_extension}") MAX_SIZE_BYTES = 100 * 1024 * 1024 file_size = 0 with open(file_path, "wb") as buffer: while chunk := await file.read(1024 * 1024): # Read in 1MB chunks asynchronously file_size += len(chunk) if file_size > MAX_SIZE_BYTES: buffer.close() os.remove(file_path) raise HTTPException(status_code=413, detail="File too large. Maximum size is 100 MB.") buffer.write(chunk) # 3. True MIME-Type Validation (Executed AFTER stream consumption to prevent TCP RST on Uvicorn) try: import magic mime_type = magic.from_file(file_path, mime=True) if not mime_type.startswith(('video/', 'image/')): os.remove(file_path) raise HTTPException(status_code=400, detail=f"Malicious payload detected. File is disguised as {file_extension} but is actually {mime_type}.") except ImportError: print("Warning: python-magic not installed, skipping strict MIME validation.") analysis_jobs[job_id] = {"status": "processing", "progress": 0, "result": None, "file_path": file_path} # Run the heavy processing in the background background_tasks.add_task(run_analysis_pipeline, job_id, file_path) return {"job_id": job_id, "status": "processing"} @app.get("/api/status/{job_id}") async def get_status(job_id: str, api_key: str = Depends(get_api_key)): if job_id not in analysis_jobs: return JSONResponse(status_code=404, content={"message": "Job not found"}) job_data = analysis_jobs[job_id] # Dynamically compute REAL telemetry try: import torch if torch.cuda.is_available(): mem_alloc = torch.cuda.memory_allocated() / 1e9 mem_total = torch.cuda.get_device_properties(0).total_memory / 1e9 vram_alloc = f"{mem_alloc:.1f} GB / {mem_total:.1f} GB" backend_name = f"CUDA 12.1 ({torch.cuda.get_device_name(0)})" else: vram_alloc = "CPU Mode" backend_name = "CPU (PyTorch)" except Exception: vram_alloc = "Unknown" backend_name = "Unknown" is_video = str(job_data.get("file_path", "")).lower().endswith(("mp4", "avi", "mov", "mkv")) job_data["telemetry"] = { "active_model": "Ensemble (EfficientNet + SyncNet)", "vram_allocation": vram_alloc, "hardware_backend": backend_name, "batch_processing": "32 Frames/sec" if is_video else "1 Image/batch" } progress = job_data.get("progress", 0) logs = [] if progress > 0: logs.append({"type": "OK", "msg": "Upload verified. File hash matches."}) if progress >= 5: logs.append({"type": "INFO", "msg": "Extracting raw data stream..."}) if progress >= 15: logs.append({"type": "WAIT", "msg": f"Loading weights to {backend_name}..."}) if progress >= 35: logs.append({"type": "OK", "msg": "Neural net inference complete."}) if progress >= 45: logs.append({"type": "INFO", "msg": "Computing SHAP & GradCAM gradients..."}) if progress >= 55: logs.append({"type": "INFO", "msg": "Running DCT on 8x8 blocks..."}) if progress >= 65: logs.append({"type": "OK", "msg": "Frequency domain mapped."}) if progress >= 75: logs.append({"type": "WAIT", "msg": "Meta-Classifier aggregating 15 sensors..."}) if progress >= 85: logs.append({"type": "INFO", "msg": "Synthesizing explainability PDF..."}) job_data["logs"] = logs return job_data @app.get("/api/status/{job_id}/stream") async def stream_status(job_id: str, request: Request, api_key: str = Depends(get_api_key)): if job_id not in analysis_jobs: raise HTTPException(status_code=404, detail="Job not found") async def event_generator(): # Force flush NGINX proxy buffers by sending 2KB of dummy padding yield ": " + " " * 2048 + "\n\n" while True: if await request.is_disconnected(): break job_data = analysis_jobs[job_id] current_progress = job_data.get("progress", 0) status = job_data.get("status", "processing") import copy data_to_send = copy.deepcopy(job_data) try: import torch if torch.cuda.is_available(): mem_alloc = torch.cuda.memory_allocated() / 1e9 mem_total = torch.cuda.get_device_properties(0).total_memory / 1e9 vram_alloc = f"{mem_alloc:.1f} GB / {mem_total:.1f} GB" backend_name = f"CUDA 12.1 ({torch.cuda.get_device_name(0)})" else: vram_alloc = "CPU Mode" backend_name = "CPU (PyTorch)" except Exception: vram_alloc = "Unknown" backend_name = "Unknown" is_video = str(data_to_send.get("file_path", "")).lower().endswith(("mp4", "avi", "mov", "mkv")) data_to_send["telemetry"] = { "active_model": "Ensemble (EfficientNet + SyncNet)", "vram_allocation": vram_alloc, "hardware_backend": backend_name, "batch_processing": "32 Frames/batch" if is_video else "1 Image/batch" } logs = [] if current_progress > 0: logs.append({"type": "OK", "msg": "Upload verified. File hash matches."}) if current_progress >= 5: logs.append({"type": "INFO", "msg": "Extracting raw data stream..."}) if current_progress >= 15: logs.append({"type": "WAIT", "msg": f"Loading weights to {backend_name}..."}) if current_progress >= 35: logs.append({"type": "OK", "msg": "Neural net inference complete."}) if current_progress >= 45: logs.append({"type": "INFO", "msg": "Computing SHAP & GradCAM gradients..."}) if current_progress >= 55: logs.append({"type": "INFO", "msg": "Running DCT on 8x8 blocks..."}) if current_progress >= 65: logs.append({"type": "OK", "msg": "Frequency domain mapped."}) if current_progress >= 75: logs.append({"type": "WAIT", "msg": "Meta-Classifier aggregating 15 sensors..."}) if current_progress >= 85: logs.append({"type": "INFO", "msg": "Synthesizing explainability PDF..."}) data_to_send["logs"] = logs yield f"data: {json.dumps(data_to_send)}\n\n" if status in ["completed", "failed"]: break await asyncio.sleep(0.5) return StreamingResponse( event_generator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no" } ) @app.get("/api/reports/{job_id}/pdf") async def download_report(job_id: str): pdf_path = os.path.join(REPORT_DIR, f"{job_id}.pdf") if not os.path.exists(pdf_path): return JSONResponse(status_code=404, content={"message": "Report not found"}) return FileResponse(pdf_path, media_type="application/pdf", filename=f"Forensic_Report_{job_id}.pdf") def run_analysis_pipeline(job_id: str, file_path: str): # ============================================= # LAZY IMPORTS: Only load heavy libraries when an analysis job actually starts. # This allows FastAPI to boot in <1 second on HF Spaces free tier. # ============================================= import cv2 import torch import numpy as np import shutil frames_dir = None audio_path = None from pipeline.video_processor import process_video from pipeline.pdf_reporter import generate_pdf_report from pipeline.frequency_analysis import analyze_frequency_domain from pipeline.ela_analysis import analyze_ela from pipeline.face_geometry import analyze_face_geometry from pipeline.noise_analysis import analyze_sensor_noise from pipeline.color_analysis import analyze_chrominance from pipeline.audio_sync import analyze_audio_visual_sync from pipeline.metadata_analysis import analyze_metadata from pipeline.rppg_analysis import extract_rppg_signal from pipeline.lighting_analysis import analyze_lighting from pipeline.eye_analysis import analyze_eye_movements from pipeline.voice_spoofing import analyze_voice_spoofing from pipeline.optical_flow import analyze_optical_flow from pipeline.cfa_analysis import analyze_cfa_artifacts from pipeline.corneal_analysis import analyze_corneal_reflections try: # ============================================= # STAGE 1: Extract frames and audio (0-10%) # ============================================= analysis_jobs[job_id]["progress"] = 5 frames_dir, audio_path = process_video(file_path, job_id) analysis_jobs[job_id]["progress"] = 10 frame_files = sorted([os.path.join(frames_dir, f) for f in os.listdir(frames_dir) if f.endswith(".jpg")]) if not frame_files: raise ValueError("No frames could be extracted for analysis.") # Load the first frame for single-frame analyses first_frame = cv2.imread(frame_files[0]) first_frame_rgb = cv2.cvtColor(first_frame, cv2.COLOR_BGR2RGB) # CRITICAL FIX: The neural network was trained on CROPPED FACES. # Squishing a 1080p full frame into 380x380 destroys all facial high-frequency # artifacts and causes the model to blindly predict "Authentic". def get_face_bbox(img_rgb): from pipeline.face_geometry import detect_face try: landmarks = detect_face(img_rgb) if landmarks and "face_bbox" in landmarks: return landmarks["face_bbox"] except Exception as e: print(f"Face bbox detection failed: {e}") return None def crop_from_bbox(img_rgb, bbox): if bbox is None: # Fallback to center crop fh, fw = img_rgb.shape[:2] min_dim = min(fh, fw) y1, x1 = (fh - min_dim) // 2, (fw - min_dim) // 2 return img_rgb[y1:y1+min_dim, x1:x1+min_dim] x, y, w, h = bbox exp = int(0.2 * w) # 20% expansion margin x1, y1 = max(0, int(x - exp)), max(0, int(y - exp)) x2, y2 = min(img_rgb.shape[1], int(x + w + exp)), min(img_rgb.shape[0], int(y + h + exp)) return img_rgb[y1:y2, x1:x2] first_bbox = get_face_bbox(first_frame_rgb) first_frame_cropped = crop_from_bbox(first_frame_rgb, first_bbox) first_frame_resized = cv2.resize(first_frame_cropped, (380, 380)) # Extract File Metadata file_size_bytes = os.path.getsize(file_path) if os.path.exists(file_path) else 0 original_resolution = f"{first_frame.shape[1]} × {first_frame.shape[0]}" has_audio = audio_path is not None and os.path.exists(audio_path) # Image Quality Assessment (IQA) first_frame_gray = cv2.cvtColor(first_frame, cv2.COLOR_BGR2GRAY) # Crop to the center 50% to evaluate sharpness. Highly textured backgrounds (like curtains) # can trick the Laplacian Variance into thinking a blurry webcam image is a sharp DSLR image! h, w = first_frame_gray.shape center_crop = first_frame_gray[int(h*0.25):int(h*0.75), int(w*0.25):int(w*0.75)] laplacian_var = cv2.Laplacian(center_crop, cv2.CV_64F).var() # Base sharpness threshold (e.g. 100 is blurry, 400 is sharp) # Normalize to a quality_multiplier between 0.3 and 1.3 quality_multiplier = float(np.clip(laplacian_var / 250.0, 0.3, 1.3)) image_quality_str = "High Quality (Sharp)" if quality_multiplier > 0.8 else "Low Quality (Blurry/Webcam)" # ============================================= # STAGE 2: Neural Network Prediction (15-30%) # ============================================= analysis_jobs[job_id]["progress"] = 15 try: # Multi-frame scoring: run all extracted frames through the model # OPTIMIZATION: Multi-frame scoring using a single batched PyTorch inference # OPTIMIZATION: Face Tracking to avoid running Mediapipe on every frame frame_tensors = [] import torchvision.transforms.functional as TF tracker = None if hasattr(cv2, 'TrackerKCF_create'): tracker = cv2.TrackerKCF_create() elif hasattr(cv2, 'TrackerCSRT_create'): tracker = cv2.TrackerCSRT_create() if tracker is not None and first_bbox is not None: tracker.init(first_frame, first_bbox) current_bbox = first_bbox for idx, frame_path in enumerate(frame_files): frame = cv2.imread(frame_path) frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) if idx == 0: frame_cropped = first_frame_cropped else: if tracker is not None and current_bbox is not None: success, new_bbox = tracker.update(frame) if success: current_bbox = new_bbox else: current_bbox = get_face_bbox(frame_rgb) if current_bbox is not None: tracker = cv2.TrackerKCF_create() if hasattr(cv2, 'TrackerKCF_create') else cv2.TrackerCSRT_create() tracker.init(frame, current_bbox) else: current_bbox = get_face_bbox(frame_rgb) frame_cropped = crop_from_bbox(frame_rgb, current_bbox) frame_resized = cv2.resize(frame_cropped, (380, 380)) ft = torch.from_numpy(frame_resized).permute(2, 0, 1).float() / 255.0 ft = TF.normalize(ft, mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]) frame_tensors.append(ft) # OOM PREVENTION: Process frames in chunks (sliding window) BATCH_SIZE = 32 all_frame_scores = [] for i in range(0, len(frame_tensors), BATCH_SIZE): batch_chunk = frame_tensors[i:i + BATCH_SIZE] batch_tensor = torch.stack(batch_chunk) # Run inference on the current batch probs = get_detector().predict(batch_tensor) all_frame_scores.extend([float(p) for p in probs]) del batch_chunk del batch_tensor del probs nn_score = sum(all_frame_scores) / len(all_frame_scores) if all_frame_scores else 0.5 except Exception as e: print(f"Error in Neural Network prediction: {e}") traceback.print_exc() nn_score = 0.5 all_frame_scores = [0.5] analysis_jobs[job_id]["progress"] = 30 import gc gc.collect() # ============================================= # STAGE 3: GradCAM Visual Explanations (30-45%) # ============================================= analysis_jobs[job_id]["progress"] = 35 heatmaps = [] try: # Preprocess: normalize WITH ImageNet mean/std because the Kaggle model expects it! import torchvision.transforms.functional as TF input_tensor = torch.from_numpy(first_frame_resized).permute(2, 0, 1).unsqueeze(0).float() / 255.0 input_tensor = TF.normalize(input_tensor, mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]) heatmap_path = os.path.join(frames_dir, "heatmap_0.jpg") hp, guided_hp = get_explainer().generate_heatmap(input_tensor, first_frame_resized, heatmap_path) if hp: heatmaps.append(hp.replace("\\", "/")) if guided_hp: heatmaps.append(guided_hp.replace("\\", "/")) except Exception as e: print(f"Error in GradCAM generation: {e}") traceback.print_exc() import gc gc.collect() analysis_jobs[job_id]["progress"] = 45 # ============================================= # PARALLEL STAGES 4-7: Freq, ELA, Face, Sync # ============================================= module_errors = [] def run_with_fallback(func, fallback_val, *args, **kwargs): try: return func(*args, **kwargs) except Exception as e: import traceback error_msg = f"{func.__name__} failed: {str(e)}" print(error_msg) module_errors.append(error_msg) with open("error_log.txt", "a") as f: f.write(error_msg + "\n") traceback.print_exc(file=f) return fallback_val is_video = len(frame_files) > 1 # OPTIMIZATION: Removed artificial max_workers=3 bottleneck, but capped to 4 for Hugging Face Spaces stability. # Python will safely scale across the 2 available CPU cores without thrashing memory. with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor: future_freq = executor.submit( run_with_fallback, analyze_frequency_domain, {"score": 0.5, "visualizations": []}, first_frame_rgb, frames_dir, prefix="freq", quality_multiplier=quality_multiplier ) future_ela = executor.submit( run_with_fallback, analyze_ela, {"score": 0.5, "visualizations": []}, first_frame_rgb, frames_dir, prefix="ela", quality_multiplier=quality_multiplier ) future_face = executor.submit( run_with_fallback, analyze_face_geometry, {"score": 0.5, "visualizations": []}, first_frame_rgb, frames_dir, prefix="face", frame_files=frame_files ) future_sync = executor.submit( run_with_fallback, analyze_audio_visual_sync, {"sync_score": 0.5, "error": "fallback"}, file_path, audio_path, frames_dir, prefix="sync" ) future_noise = executor.submit( run_with_fallback, analyze_sensor_noise, {"noise_score": 0.5}, first_frame_rgb, frames_dir, prefix="noise", quality_multiplier=quality_multiplier ) future_color = executor.submit( run_with_fallback, analyze_chrominance, {"color_anomaly_score": 0.5}, first_frame_rgb, frames_dir, prefix="color", quality_multiplier=quality_multiplier ) future_metadata = executor.submit( run_with_fallback, analyze_metadata, {"metadata_anomaly_score": 0.5, "warnings": []}, file_path ) future_rppg = executor.submit( run_with_fallback, extract_rppg_signal, {"rppg_anomaly_score": 0.5, "has_pulse": False}, file_path, frames_dir, prefix="rppg" ) future_lighting = executor.submit( run_with_fallback, analyze_lighting, {"lighting_anomaly_score": 0.5}, first_frame_rgb, frames_dir, prefix="lighting", quality_multiplier=quality_multiplier ) future_eye = executor.submit( run_with_fallback, analyze_eye_movements, {"eye_anomaly_score": 0.5, "warnings": []}, file_path, frames_dir, prefix="eye" ) if is_video else None future_voice = executor.submit( run_with_fallback, analyze_voice_spoofing, {"voice_anomaly_score": 0.5, "warnings": []}, audio_path, frames_dir, prefix="voice" ) if has_audio else None future_flow = executor.submit( run_with_fallback, analyze_optical_flow, {"flow_anomaly_score": 0.5, "warnings": []}, file_path, frames_dir, prefix="flow" ) if is_video else None future_cfa = executor.submit( run_with_fallback, analyze_cfa_artifacts, {"cfa_score": 0.5, "warnings": []}, frame_files[0], save_dir=frames_dir, face_results=None, quality_multiplier=quality_multiplier ) future_corneal = executor.submit( run_with_fallback, analyze_corneal_reflections, {"corneal_score": 0.5, "warnings": []}, frame_files[0], save_dir=frames_dir, face_results=None, quality_multiplier=quality_multiplier ) freq_results = future_freq.result() analysis_jobs[job_id]["progress"] = 55 ela_results = future_ela.result() analysis_jobs[job_id]["progress"] = 65 face_results = future_face.result() analysis_jobs[job_id]["progress"] = 75 sync_results = future_sync.result() sync_score = sync_results.get("sync_score", 0.5) if isinstance(sync_results, dict) else 0.5 analysis_jobs[job_id]["progress"] = 78 noise_results = future_noise.result() analysis_jobs[job_id]["progress"] = 80 color_results = future_color.result() metadata_results = future_metadata.result() rppg_results = future_rppg.result() lighting_results = future_lighting.result() eye_results = future_eye.result() if future_eye else {"eye_anomaly_score": 0.5} voice_results = future_voice.result() if future_voice else {"voice_anomaly_score": 0.5} flow_results = future_flow.result() if future_flow else {"flow_anomaly_score": 0.5} cfa_results = future_cfa.result() corneal_results = future_corneal.result() analysis_jobs[job_id]["progress"] = 82 # ============================================= # STAGE 8: Compute Ensemble Score (82-90%) # ============================================= analysis_jobs[job_id]["progress"] = 85 # Individual scores nn_score = float(np.mean(all_frame_scores)) if all_frame_scores else 0.5 spectral_score = freq_results.get("spectral_anomaly_score", 0.5) ela_score = ela_results.get("ela_score", 0.5) geometry_anomaly = face_results.get("geometry_anomaly_score", 0.5) if face_results.get("face_detected") else 0.5 noise_score = noise_results.get("noise_score", 0.5) color_score = color_results.get("color_anomaly_score", 0.5) metadata_score = metadata_results.get("metadata_anomaly_score", 0.1) rppg_score = rppg_results.get("rppg_anomaly_score", 0.5) lighting_score = lighting_results.get("lighting_anomaly_score", 0.5) eye_score = eye_results.get("eye_anomaly_score", 0.5) voice_score = voice_results.get("voice_anomaly_score", 0.5) flow_score = flow_results.get("flow_anomaly_score", 0.5) cfa_score = cfa_results.get("cfa_score", 0.5) corneal_score = corneal_results.get("corneal_score", 0.5) # Blur Detection using Laplacian Variance first_frame_gray = cv2.cvtColor(first_frame, cv2.COLOR_BGR2GRAY) blur_score = cv2.Laplacian(first_frame_gray, cv2.CV_64F).var() if blur_score < 100: print(f"Blur detected (score: {blur_score:.2f}). Adjusting spectral penalty.") spectral_score = spectral_score * 0.4 freq_results["spectral_anomaly_score"] = spectral_score # ============================================= # EXPLAINABLE AI HEURISTIC: Grounding the Black Box # ============================================= # Collect all physical and biological anomaly scores all_physical_scores = [spectral_score, ela_score, noise_score, color_score, lighting_score, geometry_anomaly] if is_video: all_physical_scores.append(rppg_score) all_physical_scores.append(eye_score) all_physical_scores.append(flow_score) if has_audio: all_physical_scores.append(sync_score) all_physical_scores.append(voice_score) else: all_physical_scores.append(cfa_score) all_physical_scores.append(corneal_score) max_physical_anomaly = max(all_physical_scores) avg_physical_score = sum(all_physical_scores) / len(all_physical_scores) # Build feature dictionary for the Meta-Classifier # We need to map sync_score correctly, since high sync = real. classifier_features = { "nn_score": nn_score, "spectral_score": spectral_score, "ela_score": ela_score, "geometry_anomaly": geometry_anomaly, "noise_score": noise_score, "color_score": color_score, "metadata_score": metadata_score, "rppg_score": rppg_score if is_video else 0.5, "lighting_score": lighting_score, "eye_score": eye_score if is_video else 0.5, "voice_score": voice_score if has_audio else 0.5, "flow_score": flow_score if is_video else 0.5, "cfa_score": cfa_score, "corneal_score": corneal_score } # Determine sync_score if has_audio: classifier_features["sync_score"] = sync_score else: classifier_features["sync_score"] = 0.5 # Use PyTorch Meta-Classifier to compute final confidence fake_prob = get_meta_classifier().predict(classifier_features) # ============================================= # EXPLAINABLE AI HEURISTIC: Catching Flawless Fakes # ============================================= # Deepfakes only need to fail ONE critical biological/physical test to be proven fake. # If the Meta-Classifier averages the score down, we override it to catch the fake. critical_scores = [] if is_video: critical_scores.append(geometry_anomaly) # Head pose snapping / 3D solvePnP violations critical_scores.append(eye_score) # Lack of blinking / unnatural gaze if has_audio: critical_scores.append(sync_score) # Audio-visual desync critical_scores.append(voice_score) # Vocoder audio spoofing if critical_scores and max(critical_scores) > 0.80: print(f"XAI Intervention: Boosting Fake Probability due to critical sensor failure (max {max(critical_scores):.2f})") fake_prob = max(fake_prob, max(critical_scores)) # Determine verdict based on meta-classifier output if fake_prob > 0.70: verdict = "High Confidence Deepfake" elif fake_prob > 0.55: verdict = "Suspected Manipulation" elif fake_prob > 0.40: verdict = "Inconclusive - Manual Review Recommended" else: verdict = "Likely Authentic" ensemble_score = float(np.clip(fake_prob, 0.0, 1.0)) print(f"Meta-Classifier Final Fake Probability: {ensemble_score:.4f}") # Frame-level statistics frame_scores_std = float(np.std(all_frame_scores)) if len(all_frame_scores) > 1 else 0.0 temporal_consistency = "Consistent" if frame_scores_std < 0.15 else "Inconsistent" # True SHAP (SHapley Additive exPlanations) shap_features = generate_shap_features(classifier_features, has_audio) analysis_jobs[job_id]["progress"] = 90 # ============================================= # STAGE 9: Generate Explainable AI Report # ============================================== result_data = { # Core verdict "overall_score": ensemble_score, "verdict": verdict, "frames_analyzed": len(frame_files), # Individual detector scores "nn_score": round(nn_score, 4), "spectral_anomaly_score": round(spectral_score, 4), "ela_score": round(ela_score, 4), "geometry_anomaly_score": round(geometry_anomaly, 4), "noise_score": round(noise_score, 4), "color_score": round(color_score, 4), "sync_score": sync_score, "metadata_score": round(metadata_score, 4), "rppg_score": round(rppg_score, 4), "lighting_score": round(lighting_score, 4), "eye_score": round(eye_score, 4), "voice_score": round(voice_score, 4), "flow_score": round(flow_score, 4), "cfa_score": round(cfa_score, 4), "corneal_score": round(corneal_score, 4), # Multi-frame analysis "frame_scores": [round(s, 4) for s in all_frame_scores], "frame_scores_std": round(frame_scores_std, 4), "temporal_consistency": temporal_consistency, # Sub-module results "frequency_analysis": freq_results, "ela_analysis": ela_results, "face_geometry": face_results, "noise_analysis": noise_results, "color_analysis": color_results, "sync_analysis": sync_results if isinstance(sync_results, dict) else {}, "metadata_analysis": metadata_results, "rppg_analysis": rppg_results, "lighting_analysis": lighting_results, "eye_analysis": eye_results, "voice_analysis": voice_results, "flow_analysis": flow_results, "cfa_analysis": cfa_results, "corneal_analysis": corneal_results, # XAI "shap_top_features": shap_features, "heatmaps": heatmaps, # Add dynamic weights for frontend to display "weights": None, # Error tracking "module_errors": module_errors, # Metadata "file_metadata": { "file_size_bytes": file_size_bytes, "original_resolution": original_resolution, "has_audio": has_audio, "image_quality": image_quality_str, "laplacian_variance": round(laplacian_var, 2) } } pdf_path = os.path.join(REPORT_DIR, f"{job_id}.pdf") generate_pdf_report(result_data, pdf_path) analysis_jobs[job_id]["status"] = "completed" analysis_jobs[job_id]["progress"] = 100 analysis_jobs[job_id]["result"] = result_data except Exception as e: import traceback traceback.print_exc() analysis_jobs[job_id]["status"] = "failed" analysis_jobs[job_id]["error"] = str(e) # NOTE: We cannot immediately delete frames_dir or file_path here # because the frontend React dashboard needs to fetch these images # (heatmaps, waveforms) via HTTP to render the report. # A background cron job should be used to delete files older than 1 hour instead. def generate_shap_features(classifier_features, has_audio): """ Generate true SHAP (SHapley Additive exPlanations) values for the Meta-Classifier prediction. """ try: import shap import numpy as np import torch meta_model = get_meta_classifier() if not meta_model.is_trained: return ["Meta-Classifier is untrained (Fallback mode)"] # Feature names strictly matching the order in ensemble_classifier.py feature_order = [ "nn_score", "spectral_score", "ela_score", "geometry_anomaly", "noise_score", "color_score", "sync_score", "metadata_score", "rppg_score", "lighting_score", "eye_score", "voice_score", "flow_score", "cfa_score", "corneal_score" ] feature_descriptions = { "nn_score": "Neural network pixel-level artifact detection", "spectral_score": "Frequency domain spectral anomalies (DCT/FFT)", "ela_score": "JPEG compression inconsistency (Error Level Analysis)", "geometry_anomaly": "Facial boundary texture mismatch", "noise_score": "Sensor noise (PRNU) inconsistency", "color_score": "Chrominance (YCbCr) color space bleeding", "metadata_score": "Suspicious file EXIF/metadata footprint", "rppg_score": "Lack of biological heart pulse (rPPG)", "lighting_score": "Illumination divergence across composited elements", "eye_score": "Unnatural blink rate or gaze asymmetry", "voice_score": "High-frequency vocoder artifact (Audio Spoofing)", "flow_score": "Blocky temporal motion jitter (Optical Flow)", "cfa_score": "Missing or disrupted Bayer filter (CFA) pattern", "corneal_score": "Physically impossible mismatched corneal light reflections", "sync_score": "Audio-video temporal desynchronization" } x_vector = [classifier_features.get(key, 0.5) for key in feature_order] def shap_predict(X_numpy): # KernelExplainer passes a 2D numpy array [batch_size, num_features] X_tensor = torch.FloatTensor(X_numpy).to(meta_model.device) with torch.no_grad(): preds = meta_model.network(X_tensor) return preds.cpu().numpy().flatten() # Background dataset representing total uncertainty (0.5 for all 15 features) background = np.full((1, 15), 0.5) explainer = shap.KernelExplainer(shap_predict, background) # Explain the current instance shap_values = explainer.shap_values(np.array([x_vector]), silent=True) # In newer SHAP versions, explainer.shap_values might return an Explanation object # or a numpy array. For a single output single instance, it's usually 1D or 2D array. if hasattr(shap_values, "values"): shap_values = shap_values.values # If the output is wrapped in another dimension, extract it if isinstance(shap_values, list): shap_values = shap_values[0] if len(np.shape(shap_values)) > 1: shap_values = shap_values[0] shap_contributions = [] for idx, feature_name in enumerate(feature_order): contrib = float(shap_values[idx]) if abs(contrib) > 0.001: shap_contributions.append((contrib, feature_descriptions[feature_name])) # Sort by highest absolute contribution shap_contributions.sort(key=lambda x: abs(x[0]), reverse=True) features_list = [] total_shap_abs = sum(abs(c[0]) for c in shap_contributions) for contrib, desc in shap_contributions[:5]: percentage = (abs(contrib) / total_shap_abs) * 100 if total_shap_abs > 0 else 0 direction = "→ FAKE" if contrib > 0 else "→ AUTHENTIC" features_list.append(f"{desc} ({percentage:.0f}% {direction})") if not features_list: features_list.append("Baseline Confidence (100%)") return features_list except Exception as e: import traceback traceback.print_exc() return [f"SHAP Explainer Error: {str(e)}"] # ============================================= # FRONTEND MOUNTING (MUST BE AT THE BOTTOM) # ============================================= # Serve static files (React build) if os.path.exists("static"): app.mount("/", StaticFiles(directory="static", html=True), name="static") # SPA Fallback for React Router (if a user refreshes on a subpath) @app.exception_handler(404) async def fallback_to_index(request: Request, exc: HTTPException): # Only fallback if the request isn't specifically targeting the API if not request.url.path.startswith("/api/"): return FileResponse("static/index.html") return JSONResponse(status_code=404, content={"message": "API Route Not Found"}) else: # Fallback health check if the frontend isn't built @app.get("/") async def root(): return {"status": "running", "message": "Deepfake Forensics API is online (GUI missing)."} @app.get("/health") async def health(): return {"status": "healthy"}