Deploy clean backend to Hugging Face

.gitattributes

@@ -0,0 +1 @@
+backend/models/*.pth filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1,5 @@
+backend/venv/
+backend/uploads/
+backend/results/
+__pycache__/
+*.pyc

Dockerfile

@@ -0,0 +1,23 @@
+FROM python:3.10-slim
+
+# Install system dependencies (including ffmpeg for audio extraction and git-lfs for model weights)
+RUN apt-get update && apt-get install -y \
+    ffmpeg \
+    git \
+    git-lfs \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+# Copy requirements and install
+COPY backend/requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy the backend code
+COPY backend/ .
+
+# Expose port (Hugging Face Spaces listens on port 7860 by default)
+EXPOSE 7860
+
+# Run Flask using gunicorn
+CMD ["gunicorn", "-b", "0.0.0.0:7860", "--timeout", "120", "app:app"]

backend/.env

@@ -0,0 +1,7 @@
+MONGO_URI=mongodb://localhost:27017/deepshield
+IMAGE_MODEL_PATH=models/image_model_best.pth
+AUDIO_MODEL_PATH=models/audio_model_best.pth
+VIDEO_MODEL_PATH=models/deepfake_model.pth
+THRESHOLD=0.5
+FLASK_ENV=development
+PORT=5000

backend/.env.example

@@ -0,0 +1,7 @@
+MONGO_URI=mongodb://localhost:27017/deepshield
+IMAGE_MODEL_PATH=models/image_model_best.pth
+AUDIO_MODEL_PATH=models/audio_model_best.pth
+VIDEO_MODEL_PATH=models/deepfake_model.pth
+THRESHOLD=0.5
+FLASK_ENV=development
+PORT=5000

backend/Procfile

@@ -0,0 +1 @@
+web: gunicorn app:app

backend/app.py

@@ -0,0 +1,220 @@
+"""
+DeepShield — Multi-Modal Deepfake Detection API (Flask)
+Flask server handling Video, Image, and Audio deepfake detection with MongoDB integration.
+"""
+import os
+import shutil
+import uuid
+from pathlib import Path
+from dotenv import load_dotenv
+
+from flask import Flask, request, jsonify, send_from_directory
+from flask_cors import CORS
+from werkzeug.utils import secure_filename
+
+from multimodal_detector import MultiModalDetector
+from db import save_detection, get_recent_detections
+
+load_dotenv()
+
+# ── Setup directories ─────────────────────────────────────────────
+for d in ["uploads", "results", "models"]:
+    Path(d).mkdir(exist_ok=True)
+
+# ── App ───────────────────────────────────────────────────────────
+app = Flask(__name__)
+CORS(app)  # Enable CORS for all routes
+
+app.config['UPLOAD_FOLDER'] = 'uploads'
+app.config['MAX_CONTENT_LENGTH'] = 500 * 1024 * 1024  # 500 MB max limit
+
+# ── Load detector once at startup ─────────────────────────────────
+VIDEO_MODEL = os.getenv("VIDEO_MODEL_PATH", "models/deepfake_model.pth")
+IMAGE_MODEL = os.getenv("IMAGE_MODEL_PATH", "models/image_model_best.pth")
+AUDIO_MODEL = os.getenv("AUDIO_MODEL_PATH", "models/audio_model_best.pth")
+THRESHOLD   = float(os.getenv("THRESHOLD", "0.5"))
+
+detector = MultiModalDetector(
+    video_model_path = VIDEO_MODEL if Path(VIDEO_MODEL).exists() else None,
+    image_model_path = IMAGE_MODEL if Path(IMAGE_MODEL).exists() else None,
+    audio_model_path = AUDIO_MODEL if Path(AUDIO_MODEL).exists() else None,
+    threshold        = THRESHOLD,
+)
+
+# ── Allowed file types ────────────────────────────────────────────
+VIDEO_EXTS = {".mp4", ".avi", ".mov", ".mkv", ".webm"}
+IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".bmp"}
+AUDIO_EXTS = {".wav", ".mp3", ".flac", ".ogg", ".m4a"}
+MAX_MB     = {"video": 500, "image": 20, "audio": 50}
+
+def save_upload(file, session_id, ext):
+    upload_dir = Path("uploads") / session_id
+    upload_dir.mkdir(parents=True, exist_ok=True)
+    filename = secure_filename(file.filename)
+    dest = upload_dir / f"input{ext}"
+    file.save(str(dest))
+    return dest, os.path.getsize(str(dest)), filename
+
+# ═══════════════════════════════════════════════════════════════════
+# ENDPOINTS
+# ═══════════════════════════════════════════════════════════════════
+
+@app.route("/health", methods=["GET"])
+def health():
+    return jsonify({
+        "status":          "ok",
+        "system":          "DeepShield v2.0 (Flask)",
+        "device":          detector.device,
+        "video_model":     "Loaded" if Path(VIDEO_MODEL).exists() else "Demo mode",
+        "image_model":     "Loaded" if Path(IMAGE_MODEL).exists() else "Demo mode",
+        "audio_available": detector.audio_available,
+        "modalities":      ["video", "image", "audio"],
+    })
+
+@app.route("/history", methods=["GET"])
+def get_history():
+    limit = request.args.get('limit', default=50, type=int)
+    history = get_recent_detections(limit)
+    return jsonify({
+        "total": len(history),
+        "history": history
+    })
+
+# ── VIDEO Detection ───────────────────────────────────────────────
+@app.route("/detect/video", methods=["POST"])
+def detect_video():
+    if 'file' not in request.files:
+        return jsonify({"detail": "No file part"}), 400
+    
+    file = request.files['file']
+    if file.filename == '':
+        return jsonify({"detail": "No selected file"}), 400
+
+    suffix = Path(file.filename).suffix.lower()
+    if suffix not in VIDEO_EXTS:
+        return jsonify({"detail": f"Unsupported video type: {suffix}. Allowed: {VIDEO_EXTS}"}), 400
+
+    session_id = str(uuid.uuid4())
+    dest, size, orig_filename = save_upload(file, session_id, suffix)
+    size_mb = size / (1024 * 1024)
+    if size_mb > MAX_MB["video"]:
+        shutil.rmtree(dest.parent, ignore_errors=True)
+        return jsonify({"detail": f"File too large ({size_mb:.1f} MB). Max {MAX_MB['video']} MB."}), 413
+
+    try:
+        result = detector.analyze_video(str(dest), session_id=session_id)
+        # Save to MongoDB
+        save_detection(
+            session_id=session_id,
+            filename=orig_filename,
+            modality="video",
+            verdict=result.get("verdict"),
+            confidence=result.get("confidence"),
+            details={"frames_analyzed": result.get("frames_analyzed")}
+        )
+    except Exception as e:
+        shutil.rmtree(dest.parent, ignore_errors=True)
+        return jsonify({"detail": f"Detection error: {e}"}), 500
+
+    return jsonify(result)
+
+# ── IMAGE Detection ───────────────────────────────────────────────
+@app.route("/detect/image", methods=["POST"])
+def detect_image():
+    if 'file' not in request.files:
+        return jsonify({"detail": "No file part"}), 400
+        
+    file = request.files['file']
+    if file.filename == '':
+        return jsonify({"detail": "No selected file"}), 400
+
+    suffix = Path(file.filename).suffix.lower()
+    if suffix not in IMAGE_EXTS:
+        return jsonify({"detail": f"Unsupported image type: {suffix}. Allowed: {IMAGE_EXTS}"}), 400
+
+    session_id = str(uuid.uuid4())
+    dest, size, orig_filename = save_upload(file, session_id, suffix)
+    size_mb = size / (1024 * 1024)
+    if size_mb > MAX_MB["image"]:
+        shutil.rmtree(dest.parent, ignore_errors=True)
+        return jsonify({"detail": f"File too large ({size_mb:.1f} MB). Max {MAX_MB['image']} MB."}), 413
+
+    try:
+        result = detector.analyze_image(str(dest))
+        result["session_id"] = session_id
+        # Save to MongoDB
+        save_detection(
+            session_id=session_id,
+            filename=orig_filename,
+            modality="image",
+            verdict=result.get("verdict"),
+            confidence=result.get("confidence"),
+            details={"faces_detected": result.get("faces_detected", 1)}
+        )
+    except Exception as e:
+        shutil.rmtree(dest.parent, ignore_errors=True)
+        return jsonify({"detail": f"Detection error: {e}"}), 500
+
+    return jsonify(result)
+
+# ── AUDIO Detection ───────────────────────────────────────────────
+@app.route("/detect/audio", methods=["POST"])
+def detect_audio():
+    if 'file' not in request.files:
+        return jsonify({"detail": "No file part"}), 400
+        
+    file = request.files['file']
+    if file.filename == '':
+        return jsonify({"detail": "No selected file"}), 400
+
+    suffix = Path(file.filename).suffix.lower()
+    if suffix not in AUDIO_EXTS:
+        return jsonify({"detail": f"Unsupported audio type: {suffix}. Allowed: {AUDIO_EXTS}"}), 400
+
+    session_id = str(uuid.uuid4())
+    dest, size, orig_filename = save_upload(file, session_id, suffix)
+    size_mb = size / (1024 * 1024)
+    if size_mb > MAX_MB["audio"]:
+        shutil.rmtree(dest.parent, ignore_errors=True)
+        return jsonify({"detail": f"File too large ({size_mb:.1f} MB). Max {MAX_MB['audio']} MB."}), 413
+
+    try:
+        result = detector.analyze_audio(str(dest))
+        result["session_id"] = session_id
+        # Save to MongoDB
+        save_detection(
+            session_id=session_id,
+            filename=orig_filename,
+            modality="audio",
+            verdict=result.get("verdict"),
+            confidence=result.get("confidence")
+        )
+    except Exception as e:
+        shutil.rmtree(dest.parent, ignore_errors=True)
+        return jsonify({"detail": f"Detection error: {e}"}), 500
+
+    return jsonify(result)
+
+# ── Results & cleanup ─────────────────────────────────────────────
+@app.route("/results/<session_id>/<filename>", methods=["GET"])
+def get_result_file(session_id, filename):
+    result_dir = Path("results") / session_id
+    return send_from_directory(str(result_dir), filename)
+
+@app.route("/results/<session_id>", methods=["GET"])
+def get_results(session_id):
+    result_dir = Path("results") / session_id
+    if not result_dir.exists():
+        return jsonify({"detail": "Session not found."}), 404
+    files = [f.name for f in result_dir.glob("*.jpg")]
+    return jsonify({"session_id": session_id, "heatmaps": files})
+
+@app.route("/session/<session_id>", methods=["DELETE"])
+def delete_session(session_id):
+    for d in ["uploads", "results"]:
+        shutil.rmtree(Path(d) / session_id, ignore_errors=True)
+    return jsonify({"deleted": session_id})
+
+if __name__ == "__main__":
+    port = int(os.environ.get("PORT", 5000))
+    app.run(host="0.0.0.0", port=port, debug=True)

backend/audio_detector.py

@@ -0,0 +1,163 @@
+"""
+AudioDeepfakeDetector — Wav2Vec2-based audio deepfake detection
+Uses Facebook's Wav2Vec2-base for feature extraction + lightweight classification head.
+Detects AI-generated or voice-cloned audio (TTS, VC spoofing).
+"""
+import torch
+import torch.nn as nn
+import numpy as np
+from pathlib import Path
+from typing import Optional
+
+try:
+    import librosa
+    LIBROSA_AVAILABLE = True
+except ImportError:
+    LIBROSA_AVAILABLE = False
+
+try:
+    from transformers import Wav2Vec2Model, Wav2Vec2FeatureExtractor
+    WAV2VEC_AVAILABLE = True
+except ImportError:
+    WAV2VEC_AVAILABLE = False
+
+
+# ─────────────────────────────────────────────────────────────────
+# Classifier Head (on top of Wav2Vec2 hidden states)
+# ─────────────────────────────────────────────────────────────────
+class AudioClassifierHead(nn.Module):
+    """
+    Pools temporal hidden states from Wav2Vec2 and classifies Real/Fake.
+    Input:  (B, T, D)  — Wav2Vec2 last_hidden_state
+    Output: (B, 1)     — raw logits (apply sigmoid for probability)
+    """
+    def __init__(self, input_dim: int = 768):
+        super().__init__()
+        self.attention_pool = nn.Sequential(
+            nn.Linear(input_dim, 128),
+            nn.Tanh(),
+            nn.Linear(128, 1),
+        )
+        self.classifier = nn.Sequential(
+            nn.Linear(input_dim, 256),
+            nn.LayerNorm(256),
+            nn.ReLU(inplace=True),
+            nn.Dropout(0.4),
+            nn.Linear(256, 64),
+            nn.ReLU(inplace=True),
+            nn.Dropout(0.2),
+            nn.Linear(64, 1),
+        )
+
+    def forward(self, hidden_states: torch.Tensor) -> torch.Tensor:
+        # Attention-weighted pooling
+        weights = torch.softmax(self.attention_pool(hidden_states), dim=1)  # (B, T, 1)
+        pooled  = (hidden_states * weights).sum(dim=1)                       # (B, D)
+        return self.classifier(pooled)                                       # (B, 1)
+
+
+# ─────────────────────────────────────────────────────────────────
+# Main Audio Deepfake Detector
+# ─────────────────────────────────────────────────────────────────
+class AudioDeepfakeDetector(nn.Module):
+    """
+    Wav2Vec2-base + Attention-Pooling Classifier for audio deepfake detection.
+
+    Pipeline:
+      Raw waveform (16 kHz mono)
+        → Wav2Vec2 feature extractor (normalisation)
+        → Wav2Vec2 transformer encoder  → hidden states (T × 768)
+        → Attention-weighted pooling    → 768-dim vector
+        → FC classifier head            → fake probability
+    """
+    SAMPLE_RATE      = 16_000
+    MAX_DURATION_SEC = 10.0       # Clip to 10 s for speed
+
+    def __init__(self, pretrained: bool = True, freeze_base: bool = True):
+        super().__init__()
+
+        if not WAV2VEC_AVAILABLE:
+            raise RuntimeError(
+                "transformers library not installed.\n"
+                "Run: pip install transformers>=4.30.0"
+            )
+
+        model_name = "facebook/wav2vec2-base"
+        self.feature_extractor = Wav2Vec2FeatureExtractor.from_pretrained(model_name)
+        self.wav2vec2           = Wav2Vec2Model.from_pretrained(model_name)
+
+        # Optionally freeze Wav2Vec2 backbone for efficient fine-tuning
+        if freeze_base:
+            for param in self.wav2vec2.parameters():
+                param.requires_grad = False
+
+        hidden_dim = self.wav2vec2.config.hidden_size   # 768
+        self.head  = AudioClassifierHead(hidden_dim)
+
+    # ── Forward ─────────────────────────────────────────────────
+    def forward(
+        self,
+        input_values:   torch.Tensor,
+        attention_mask: Optional[torch.Tensor] = None,
+    ) -> torch.Tensor:
+        outputs = self.wav2vec2(input_values, attention_mask=attention_mask)
+        return self.head(outputs.last_hidden_state)   # (B, 1) logits
+
+    # ── Inference helpers ────────────────────────────────────────
+    def predict_proba(self, waveform: np.ndarray, device: str = "cpu") -> float:
+        """
+        Args:
+            waveform: 1-D numpy float32 array at 16 kHz
+            device:   'cpu' or 'cuda'
+        Returns:
+            Fake probability in [0, 1]
+        """
+        max_samples = int(self.MAX_DURATION_SEC * self.SAMPLE_RATE)
+        if len(waveform) > max_samples:
+            waveform = waveform[:max_samples]
+
+        inputs = self.feature_extractor(
+            waveform,
+            sampling_rate=self.SAMPLE_RATE,
+            return_tensors="pt",
+            padding=True,
+        )
+        input_values = inputs.input_values.to(device)
+
+        self.eval()
+        with torch.no_grad():
+            logits = self.forward(input_values)
+            prob   = torch.sigmoid(logits).squeeze().item()
+        return float(prob)
+
+    # ── Static helpers ───────────────────────────────────────────
+    @staticmethod
+    def load_audio(path: str, target_sr: int = 16_000) -> np.ndarray:
+        """Load any audio/video file and resample to 16 kHz mono."""
+        if not LIBROSA_AVAILABLE:
+            raise RuntimeError("librosa not installed. Run: pip install librosa>=0.10.0")
+        waveform, _ = librosa.load(path, sr=target_sr, mono=True)
+        return waveform.astype(np.float32)
+
+    @staticmethod
+    def load(path: str, device: str = "cpu") -> "AudioDeepfakeDetector":
+        """Load a trained detector from a .pth file."""
+        model = AudioDeepfakeDetector(pretrained=False, freeze_base=False)
+        state = torch.load(path, map_location=device, weights_only=True)
+        model.load_state_dict(state)
+        model.eval()
+        return model
+
+
+# ── Quick sanity check ────────────────────────────────────────────
+if __name__ == "__main__":
+    if WAV2VEC_AVAILABLE:
+        model  = AudioDeepfakeDetector(pretrained=False, freeze_base=False)
+        dummy  = torch.randn(2, 16_000)  # 1-second batch
+        logits = model(dummy)
+        total  = sum(p.numel() for p in model.parameters())
+        print(f"Output shape : {logits.shape}")
+        print(f"Total params : {total:,}")
+        print("AudioDeepfakeDetector OK ✓")
+    else:
+        print("transformers not installed — skipping sanity check.")

backend/db.py

@@ -0,0 +1,61 @@
+import os
+import datetime
+from pymongo import MongoClient
+from pymongo.server_api import ServerApi
+from dotenv import load_dotenv
+
+load_dotenv()
+
+MONGO_URI = os.getenv("MONGO_URI", "mongodb://localhost:27017/")
+DB_NAME = "deepshield"
+COLLECTION_NAME = "detections"
+
+_db_client = None
+
+def get_db():
+    global _db_client
+    if _db_client is None:
+        try:
+            # Connect to MongoDB Atlas or local MongoDB
+            _db_client = MongoClient(MONGO_URI, server_api=ServerApi('1') if "mongodb.net" in MONGO_URI else None)
+            # Send a ping to confirm a successful connection
+            _db_client.admin.command('ping')
+            print("Successfully connected to MongoDB!")
+        except Exception as e:
+            print(f"MongoDB connection error: {e}")
+            return None
+    
+    return _db_client[DB_NAME]
+
+def save_detection(session_id, filename, modality, verdict, confidence, details=None):
+    db = get_db()
+    if db is None:
+        return False
+        
+    try:
+        record = {
+            "session_id": session_id,
+            "filename": filename,
+            "modality": modality,
+            "verdict": verdict,
+            "confidence": confidence,
+            "details": details or {},
+            "timestamp": datetime.datetime.utcnow()
+        }
+        db[COLLECTION_NAME].insert_one(record)
+        return True
+    except Exception as e:
+        print(f"Error saving to MongoDB: {e}")
+        return False
+
+def get_recent_detections(limit=50):
+    db = get_db()
+    if db is None:
+        return []
+        
+    try:
+        cursor = db[COLLECTION_NAME].find({}, {"_id": 0}).sort("timestamp", -1).limit(limit)
+        return list(cursor)
+    except Exception as e:
+        print(f"Error reading from MongoDB: {e}")
+        return []

backend/detector.py

@@ -0,0 +1,124 @@
+"""
+DeepfakeDetector — orchestrates the full video analysis pipeline
+"""
+import os, uuid, time
+import numpy as np
+import torch
+import torchvision.transforms as T
+from pathlib import Path
+from typing import Optional
+
+from model import HybridDeepfakeDetector
+from face_extractor import FaceExtractor
+from gradcam import GradCAM
+
+# ── Image pre-processing ─────────────────────────────────────────────
+MEAN = [0.485, 0.456, 0.406]
+STD  = [0.229, 0.224, 0.225]
+
+transform = T.Compose([
+    T.ToTensor(),
+    T.Normalize(mean=MEAN, std=STD),
+])
+
+
+class DeepfakeDetector:
+    """
+    End-to-end video deepfake detector.
+    Steps:
+      1. Sample frames from video
+      2. Crop face regions
+      3. Run hybrid model inference
+      4. Generate Grad-CAM heatmaps
+      5. Aggregate per-frame scores into video-level verdict
+    """
+
+    def __init__(
+        self,
+        model_path: Optional[str] = None,
+        device: Optional[str] = None,
+        max_frames: int = 32,
+        threshold: float = 0.5,
+    ):
+        self.device    = device or ("cuda" if torch.cuda.is_available() else "cpu")
+        self.threshold = threshold
+        self.max_frames = max_frames
+
+        # Load model
+        self.model = HybridDeepfakeDetector(pretrained=(model_path is None))
+        if model_path and Path(model_path).exists():
+            state = torch.load(model_path, map_location=self.device)
+            self.model.load_state_dict(state)
+            print(f"[Detector] Loaded weights from {model_path}")
+        else:
+            print("[Detector] Using ImageNet pretrained weights (demo mode). "
+                  "Train on FF++ for research-quality results.")
+
+        self.model.to(self.device).eval()
+
+        self.extractor = FaceExtractor()
+        self.gradcam   = GradCAM(self.model)
+
+    # ── Single frame inference ───────────────────────────────────────
+    def _infer_frame(self, face_rgb: np.ndarray) -> float:
+        """Returns fake probability for one face crop."""
+        tensor = transform(face_rgb).unsqueeze(0).to(self.device)
+        with torch.no_grad():
+            prob = self.model.predict_proba(tensor).item()
+        return float(prob)
+
+    # ── Main analysis ────────────────────────────────────────────────
+    def analyze(self, video_path: str, session_id: Optional[str] = None) -> dict:
+        t0 = time.time()
+        session_id = session_id or str(uuid.uuid4())
+
+        results_dir = Path("results") / session_id
+        results_dir.mkdir(parents=True, exist_ok=True)
+
+        # 1. Extract faces
+        faces = self.extractor.extract(video_path, max_frames=self.max_frames)
+        if not faces:
+            return {
+                "session_id": session_id,
+                "verdict": "UNKNOWN",
+                "confidence": 0.0,
+                "error": "No faces detected in video.",
+                "frames": [],
+            }
+
+        # 2. Per-frame inference + Grad-CAM
+        frame_results = []
+        scores = []
+
+        for i, item in enumerate(faces):
+            face_rgb = item["face"]
+            fidx     = item["frame_idx"]
+            prob     = self._infer_frame(face_rgb)
+            scores.append(prob)
+
+            # Generate and save Grad-CAM heatmap
+            cam_path = str(results_dir / f"frame_{i:04d}_cam.jpg")
+            self.gradcam.generate(face_rgb, cam_path)
+
+            frame_results.append({
+                "frame_idx": fidx,
+                "fake_prob": round(prob, 4),
+                "verdict":   "FAKE" if prob >= self.threshold else "REAL",
+                "cam_path":  f"/results/{session_id}/frame_{i:04d}_cam.jpg",
+            })
+
+        # 3. Video-level score
+        video_score = float(np.mean(scores))
+        verdict = "FAKE" if video_score >= self.threshold else "REAL"
+
+        elapsed = round(time.time() - t0, 2)
+
+        return {
+            "session_id":   session_id,
+            "verdict":      verdict,
+            "confidence":   round(video_score * 100, 2),
+            "fake_prob":    round(video_score, 4),
+            "frames_analyzed": len(faces),
+            "elapsed_sec":  elapsed,
+            "frame_scores": frame_results,
+        }

backend/download_audio_dataset.py

@@ -0,0 +1,104 @@
+"""
+Download Audio Deepfake Dataset - Smart approach using HuggingFace parquet cache
+Uses Audio(decode=False) to get raw bytes without torchcodec, avoids rate limiting
+"""
+
+import os
+import sys
+import tempfile
+import soundfile as sf
+from datasets import load_dataset, Audio
+
+OUTPUT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "datasets", "audio"))
+REAL_TRAIN_DIR   = os.path.join(OUTPUT_DIR, "train", "real")
+REAL_VAL_DIR     = os.path.join(OUTPUT_DIR, "val", "real")
+FAKE_TRAIN_DIR   = os.path.join(OUTPUT_DIR, "train", "fake")
+FAKE_VAL_DIR     = os.path.join(OUTPUT_DIR, "val", "fake")
+
+os.makedirs(REAL_TRAIN_DIR, exist_ok=True)
+os.makedirs(REAL_VAL_DIR, exist_ok=True)
+os.makedirs(FAKE_TRAIN_DIR, exist_ok=True)
+os.makedirs(FAKE_VAL_DIR, exist_ok=True)
+
+print("=" * 60)
+print("  Audio Deepfake Dataset Downloader (Smart Mode)")
+print("  Dataset: garystafford/deepfake-audio-detection")
+print("=" * 60)
+
+# --- Step 1: Load dataset WITHOUT auto-decoding -------------------
+# decode=False gives us raw bytes directly from the parquet
+# No torchcodec needed, no individual file downloads, no rate limit!
+print("\n[1/3] Loading dataset from HuggingFace (parquet only)...")
+print("      Using cached parquet if available...")
+
+ds = load_dataset(
+    "garystafford/deepfake-audio-detection",
+    split="train",
+    trust_remote_code=False
+)
+
+# Disable audio decoding - get raw bytes instead
+ds = ds.cast_column("audio", Audio(decode=False))
+
+print(f"      [OK] Dataset loaded! Total samples: {len(ds)}")
+
+# --- Step 2: Save each sample as WAV -----------------------------
+print(f"\n[2/3] Converting and saving WAV files...")
+print(f"      Output: {OUTPUT_DIR}")
+
+real_count = 0
+fake_count = 0
+skip_count = 0
+
+for i, sample in enumerate(ds):
+    label      = sample["label"]       # 0 = real, 1 = fake
+    audio_info = sample["audio"]       # dict: {"bytes": b"...", "path": "..."}
+
+    audio_bytes = audio_info.get("bytes") if isinstance(audio_info, dict) else None
+    audio_path  = audio_info.get("path", f"sample_{i}") if isinstance(audio_info, dict) else f"sample_{i}"
+
+    if not audio_bytes:
+        skip_count += 1
+        continue
+
+    # Write raw bytes to a temp FLAC file, read it, save as WAV
+    try:
+        with tempfile.NamedTemporaryFile(suffix=".flac", delete=False) as tmp:
+            tmp.write(audio_bytes)
+            tmp_path = tmp.name
+
+        data, sr = sf.read(tmp_path)
+        os.unlink(tmp_path)
+
+        if label == 0:
+            is_val = (real_count % 5 == 0)
+            target_dir = REAL_VAL_DIR if is_val else REAL_TRAIN_DIR
+            out_path = os.path.join(target_dir, f"real_{real_count:04d}.wav")
+            real_count += 1
+        else:
+            is_val = (fake_count % 5 == 0)
+            target_dir = FAKE_VAL_DIR if is_val else FAKE_TRAIN_DIR
+            out_path = os.path.join(target_dir, f"fake_{fake_count:04d}.wav")
+            fake_count += 1
+
+        sf.write(out_path, data, sr)
+
+    except Exception as e:
+        skip_count += 1
+        if os.path.exists(tmp_path):
+            os.unlink(tmp_path)
+        continue
+
+    # Show progress every 50 files
+    total = real_count + fake_count
+    if total % 50 == 0 or total == 1:
+        sys.stdout.write(f"\r      Progress: {total}/1866  (real={real_count}, fake={fake_count}, skipped={skip_count})")
+        sys.stdout.flush()
+
+total = real_count + fake_count
+print(f"\n\n[3/3] COMPLETE!")
+print(f"      Real WAV files : {real_count}  ->  {REAL_TRAIN_DIR} and {REAL_VAL_DIR}")
+print(f"      Fake WAV files : {fake_count}  ->  {FAKE_TRAIN_DIR} and {FAKE_VAL_DIR}")
+print(f"      Skipped        : {skip_count}")
+print(f"      Total saved    : {total}")
+print("=" * 60)

backend/face_extractor.py

@@ -0,0 +1,111 @@
+"""
+Face Extractor — OpenCV-based face detection and frame sampling
+"""
+import cv2
+import numpy as np
+from pathlib import Path
+from typing import List, Tuple, Optional
+
+
+class FaceExtractor:
+    """
+    Extracts and crops face regions from video frames.
+    Uses OpenCV's Haar cascade (no extra dependencies).
+    """
+
+    def __init__(self):
+        cascade_path = cv2.data.haarcascades + "haarcascade_frontalface_default.xml"
+        self.detector = cv2.CascadeClassifier(cascade_path)
+        if self.detector.empty():
+            raise RuntimeError("Failed to load Haar cascade classifier.")
+
+    # ── Frame sampling ──────────────────────────────────────────────
+    def sample_frames(
+        self,
+        video_path: str,
+        max_frames: int = 32,
+        sample_fps: float = 2.0,
+    ) -> List[Tuple[int, np.ndarray]]:
+        """
+        Returns a list of (frame_index, BGR_frame) tuples.
+        """
+        cap = cv2.VideoCapture(str(video_path))
+        if not cap.isOpened():
+            raise ValueError(f"Cannot open video: {video_path}")
+
+        video_fps   = cap.get(cv2.CAP_PROP_FPS) or 25.0
+        total       = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+        interval    = max(1, int(video_fps / sample_fps))
+
+        frames: List[Tuple[int, np.ndarray]] = []
+        idx = 0
+
+        while cap.isOpened() and len(frames) < max_frames:
+            ret, frame = cap.read()
+            if not ret:
+                break
+            if idx % interval == 0:
+                frames.append((idx, frame))
+            idx += 1
+
+        cap.release()
+        return frames
+
+    # ── Face crop ───────────────────────────────────────────────────
+    def crop_face(
+        self,
+        frame: np.ndarray,
+        target_size: Tuple[int, int] = (224, 224),
+        padding_ratio: float = 0.25,
+    ) -> Optional[np.ndarray]:
+        """
+        Detects the largest face and returns a padded RGB crop.
+        Returns None if no face detected (caller should decide what to do).
+        """
+        gray  = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+        faces = self.detector.detectMultiScale(
+            gray, scaleFactor=1.1, minNeighbors=4, minSize=(48, 48)
+        )
+
+        h, w = frame.shape[:2]
+
+        if len(faces) == 0:
+            # Fallback: centre-crop as square
+            size = min(h, w)
+            y0 = (h - size) // 2
+            x0 = (w - size) // 2
+            crop = frame[y0 : y0 + size, x0 : x0 + size]
+        else:
+            # Largest face
+            fx, fy, fw, fh = max(faces, key=lambda r: r[2] * r[3])
+            pad = int(max(fw, fh) * padding_ratio)
+            x1 = max(0, fx - pad)
+            y1 = max(0, fy - pad)
+            x2 = min(w, fx + fw + pad)
+            y2 = min(h, fy + fh + pad)
+            crop = frame[y1:y2, x1:x2]
+
+        if crop.size == 0:
+            return None
+
+        crop = cv2.resize(crop, target_size, interpolation=cv2.INTER_AREA)
+        crop = cv2.cvtColor(crop, cv2.COLOR_BGR2RGB)
+        return crop
+
+    # ── Full pipeline ────────────────────────────────────────────────
+    def extract(
+        self,
+        video_path: str,
+        max_frames: int = 32,
+        sample_fps: float = 2.0,
+    ) -> List[dict]:
+        """
+        Returns list of dicts: {frame_idx, face_rgb (H,W,3 uint8)}
+        """
+        raw_frames = self.sample_frames(video_path, max_frames, sample_fps)
+        results = []
+        for fidx, frame in raw_frames:
+            face = self.crop_face(frame)
+            if face is not None:
+                results.append({"frame_idx": fidx, "face": face})
+        return results

backend/gradcam.py

@@ -0,0 +1,114 @@
+"""
+Grad-CAM — Gradient-weighted Class Activation Mapping
+Generates heatmap overlays showing which facial regions influenced the prediction.
+Essential for IEEE paper explainability section.
+"""
+import cv2
+import numpy as np
+import torch
+import torch.nn.functional as F
+import torchvision.transforms as T
+from PIL import Image
+
+
+MEAN = [0.485, 0.456, 0.406]
+STD  = [0.229, 0.224, 0.225]
+
+transform = T.Compose([T.ToTensor(), T.Normalize(mean=MEAN, std=STD)])
+
+
+class GradCAM:
+    """
+    Hooks into the last convolutional layer of the spatial branch
+    to produce a class activation map.
+    """
+
+    def __init__(self, model):
+        self.model    = model
+        self.device   = next(model.parameters()).device
+        self._fmaps   = None
+        self._grads   = None
+        self._hook_fwd = None
+        self._hook_bwd = None
+        self._register_hooks()
+
+    def _register_hooks(self):
+        # Target: last conv block of EfficientNet spatial branch
+        try:
+            if hasattr(self.model.spatial, "backbone"):
+                target = self.model.spatial.backbone
+                # timm efficientnet: last block
+                if hasattr(target, "blocks"):
+                    target_layer = target.blocks[-1]
+                elif hasattr(target, "features"):
+                    target_layer = target.features[-1]
+                else:
+                    target_layer = list(target.children())[-2]
+            else:
+                target_layer = list(self.model.spatial.children())[-2]
+
+            self._hook_fwd = target_layer.register_forward_hook(self._save_fmaps)
+            self._hook_bwd = target_layer.register_full_backward_hook(self._save_grads)
+        except Exception:
+            pass  # graceful fallback — still saves plain overlay
+
+    def _save_fmaps(self, module, inp, out):
+        self._fmaps = out.detach()
+
+    def _save_grads(self, module, grad_in, grad_out):
+        self._grads = grad_out[0].detach()
+
+    def _compute_cam(self, face_rgb: np.ndarray) -> np.ndarray:
+        """Returns cam as uint8 (224,224,3) or plain red overlay on failure."""
+        if self._hook_fwd is None:
+            return self._plain_overlay(face_rgb)
+
+        tensor = transform(face_rgb).unsqueeze(0).to(self.device)
+        self.model.zero_grad()
+
+        logits = self.model(tensor)
+        score  = torch.sigmoid(logits)
+        score.backward()
+
+        if self._fmaps is None or self._grads is None:
+            return self._plain_overlay(face_rgb)
+
+        weights = self._grads.mean(dim=[2, 3], keepdim=True)   # (1, C, 1, 1)
+        cam     = (weights * self._fmaps).sum(dim=1, keepdim=True)  # (1, 1, H, W)
+        cam     = F.relu(cam)
+        cam     = cam.squeeze().cpu().numpy()
+
+        # Normalise
+        if cam.max() > 0:
+            cam = cam / cam.max()
+
+        cam_resized = cv2.resize(cam, (224, 224))
+        heatmap     = cv2.applyColorMap(np.uint8(255 * cam_resized), cv2.COLORMAP_JET)
+        heatmap     = cv2.cvtColor(heatmap, cv2.COLOR_BGR2RGB)
+
+        # Overlay on original face
+        overlay = cv2.addWeighted(face_rgb, 0.6, heatmap, 0.4, 0)
+        return overlay
+
+    @staticmethod
+    def _plain_overlay(face_rgb: np.ndarray) -> np.ndarray:
+        """Fallback: red-tinted overlay when hooks unavailable."""
+        tint = np.zeros_like(face_rgb)
+        tint[:, :, 0] = 100
+        return cv2.addWeighted(face_rgb, 0.8, tint, 0.2, 0)
+
+    def generate(self, face_rgb: np.ndarray, save_path: str):
+        """Compute Grad-CAM and save to disk."""
+        try:
+            overlay = self._compute_cam(face_rgb)
+        except Exception:
+            overlay = face_rgb
+
+        img = Image.fromarray(overlay.astype(np.uint8))
+        img.save(save_path, quality=85)
+
+    def remove_hooks(self):
+        if self._hook_fwd:
+            self._hook_fwd.remove()
+        if self._hook_bwd:
+            self._hook_bwd.remove()

backend/image_detector.py

@@ -0,0 +1,246 @@
+"""
+ImageDeepfakeDetector — EfficientNetV2-S + MTCNN for static image deepfake detection
+Detects facial inconsistencies, texture/pixel artifacts in images (JPG, PNG, WEBP).
+"""
+import torch
+import torch.nn as nn
+import numpy as np
+import cv2
+from PIL import Image
+from pathlib import Path
+from typing import Optional, Tuple
+import torchvision.transforms as T
+
+try:
+    import timm
+    TIMM_AVAILABLE = True
+except ImportError:
+    TIMM_AVAILABLE = False
+
+try:
+    # facenet-pytorch works functionally even with version warning vs torch>=2.4
+    import warnings
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore")
+        from facenet_pytorch import MTCNN as FacenetMTCNN
+    MTCNN_AVAILABLE = True
+except (ImportError, Exception):
+    MTCNN_AVAILABLE = False
+
+
+# ─────────────────────────────────────────────────────────────────
+# Image pre-processing
+# ─────────────────────────────────────────────────────────────────
+MEAN = [0.485, 0.456, 0.406]
+STD  = [0.229, 0.224, 0.225]
+
+inference_transform = T.Compose([
+    T.Resize((224, 224)),
+    T.ToTensor(),
+    T.Normalize(mean=MEAN, std=STD),
+])
+
+train_transform = T.Compose([
+    T.Resize((256, 256)),
+    T.RandomCrop(224),
+    T.RandomHorizontalFlip(),
+    T.ColorJitter(brightness=0.2, contrast=0.2, saturation=0.1),
+    T.ToTensor(),
+    T.Normalize(mean=MEAN, std=STD),
+])
+
+
+# ─────────────────────────────────────────────────────────────────
+# EfficientNetV2-S Backbone + Classifier Head
+# ─────────────────────────────────────────────────────────────────
+class EfficientNetV2Detector(nn.Module):
+    """
+    EfficientNetV2-S backbone for fake image detection.
+    ImageNet-pretrained → fine-tuned on deepfake face datasets.
+    Detects:
+      - Facial boundary blending artifacts
+      - Texture inconsistencies (blurriness, over-smoothing)
+      - GAN frequency fingerprints in pixel space
+    """
+    def __init__(self, pretrained: bool = True):
+        super().__init__()
+
+        if TIMM_AVAILABLE:
+            # Use tf_efficientnetv2_s which has pretrained weights in all timm versions
+            try:
+                self.backbone = timm.create_model(
+                    "tf_efficientnetv2_s",
+                    pretrained=pretrained,
+                    num_classes=0,
+                    global_pool="avg",
+                )
+            except Exception:
+                self.backbone = timm.create_model(
+                    "efficientnet_b4",
+                    pretrained=pretrained,
+                    num_classes=0,
+                    global_pool="avg",
+                )
+            self.feat_dim = self.backbone.num_features   # 1280
+        else:
+            # Fallback: EfficientNet-B0 via torchvision
+            import torchvision.models as tv
+            net = tv.efficientnet_b0(pretrained=pretrained)
+            self.backbone = nn.Sequential(
+                net.features,
+                nn.AdaptiveAvgPool2d(1),
+                nn.Flatten(),
+            )
+            self.feat_dim = 1280
+
+        self.classifier = nn.Sequential(
+            nn.Linear(self.feat_dim, 512),
+            nn.BatchNorm1d(512),
+            nn.ReLU(inplace=True),
+            nn.Dropout(0.5),
+            nn.Linear(512, 128),
+            nn.ReLU(inplace=True),
+            nn.Dropout(0.3),
+            nn.Linear(128, 1),
+        )
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        features = self.backbone(x)
+        return self.classifier(features)   # (B, 1) logits
+
+    def predict_proba(self, x: torch.Tensor) -> torch.Tensor:
+        with torch.no_grad():
+            return torch.sigmoid(self.forward(x)).squeeze(1)   # (B,)
+
+    @staticmethod
+    def load(path: str, device: str = "cpu") -> "EfficientNetV2Detector":
+        model = EfficientNetV2Detector(pretrained=False)
+        state = torch.load(path, map_location=device, weights_only=True)
+        model.load_state_dict(state)
+        model.eval()
+        return model
+
+
+# ─────────────────────────────────────────────────────────────────
+# MTCNN Face Extractor Wrapper
+# ─────────────────────────────────────────────────────────────────
+class MTCNNExtractor:
+    """
+    MTCNN-based face extractor using facenet-pytorch.
+    Localises and aligns faces before passing to the classifier.
+    """
+    def __init__(self, device: str = "cpu"):
+        if MTCNN_AVAILABLE:
+            self.mtcnn = FacenetMTCNN(
+                image_size=224,
+                margin=30,
+                keep_all=False,
+                post_process=False,
+                device=device,
+            )
+        else:
+            self.mtcnn = None
+            print("[MTCNN] facenet-pytorch not installed — using full image fallback.")
+
+    def extract(self, image_rgb: np.ndarray) -> Tuple[np.ndarray, bool]:
+        """
+        Returns:
+            face_rgb   : (H, W, 3) uint8 numpy array — face crop or full image
+            detected   : bool — True if MTCNN found a face
+        """
+        if self.mtcnn is None:
+            return image_rgb, False
+
+        pil_img = Image.fromarray(image_rgb)
+        try:
+            face_tensor = self.mtcnn(pil_img)   # (C, H, W) float in [0, 255]
+            if face_tensor is not None:
+                face_np = face_tensor.permute(1, 2, 0).byte().numpy()
+                return face_np, True
+        except Exception:
+            pass
+        return image_rgb, False
+
+
+# ─────────────────────────────────────────────────────────────────
+# Full Image Detection Pipeline
+# ─────────────────────────────────────────────────────────────────
+class ImageDeepfakeDetector:
+    """
+    End-to-end pipeline:
+      1. Load image from disk
+      2. Detect & crop face with MTCNN
+      3. Classify with EfficientNetV2-S
+      4. Return verdict + confidence
+    """
+
+    def __init__(
+        self,
+        model_path: Optional[str] = None,
+        device: Optional[str]     = None,
+        threshold: float          = 0.5,
+    ):
+        self.device    = device or ("cuda" if torch.cuda.is_available() else "cpu")
+        self.threshold = threshold
+
+        # EfficientNetV2 model
+        self.model = EfficientNetV2Detector(pretrained=(model_path is None))
+        if model_path and Path(model_path).exists():
+            state = torch.load(model_path, map_location=self.device, weights_only=True)
+            self.model.load_state_dict(state)
+            print(f"[ImageDetector] Loaded weights from {model_path}")
+        else:
+            print("[ImageDetector] Using ImageNet pretrained weights (demo mode).")
+        self.model.to(self.device).eval()
+
+        # MTCNN face extractor
+        self.extractor = MTCNNExtractor(device=self.device)
+
+    def analyze(self, image_path: str) -> dict:
+        """Run deepfake detection on a single image file."""
+        # ── Load ──────────────────────────────────────────────
+        img_bgr = cv2.imread(str(image_path))
+        if img_bgr is None:
+            return {
+                "verdict":       "ERROR",
+                "error":         "Cannot read image file.",
+                "confidence":    0.0,
+                "fake_prob":     0.0,
+                "face_detected": False,
+                "modality":      "image",
+            }
+
+        img_rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
+        h, w    = img_rgb.shape[:2]
+
+        # ── Face extraction ───────────────────────────────────
+        face_rgb, face_detected = self.extractor.extract(img_rgb)
+
+        # ── Pre-process & infer ───────────────────────────────
+        pil_img = Image.fromarray(face_rgb)
+        tensor  = inference_transform(pil_img).unsqueeze(0).to(self.device)
+
+        with torch.no_grad():
+            prob = self.model.predict_proba(tensor).item()
+
+        verdict = "FAKE" if prob >= self.threshold else "REAL"
+
+        return {
+            "verdict":        verdict,
+            "confidence":     round(prob, 4),
+            "fake_prob":      round(prob, 4),
+            "faces_detected": 1 if face_detected else 0,
+            "image_size":     f"{w}×{h}",
+            "modality":       "image",
+        }
+
+
+# ── Quick sanity check ────────────────────────────────────────────
+if __name__ == "__main__":
+    model = EfficientNetV2Detector(pretrained=False)
+    dummy = torch.randn(4, 3, 224, 224)
+    out   = model.predict_proba(dummy)
+    total = sum(p.numel() for p in model.parameters())
+    print(f"Output shape : {out.shape}")
+    print(f"Total params : {total:,}")
+    print("EfficientNetV2Detector OK ✓")

backend/metrics.py

@@ -0,0 +1,49 @@
+"""
+Metrics — AUC, Accuracy, EER computation for IEEE evaluation
+"""
+import numpy as np
+from sklearn.metrics import roc_auc_score, roc_curve, accuracy_score
+from typing import List, Tuple
+
+
+def compute_auc(y_true: List[int], y_scores: List[float]) -> float:
+    """AUC-ROC score. y_true: 0=real, 1=fake."""
+    return float(roc_auc_score(y_true, y_scores))
+
+
+def compute_accuracy(y_true: List[int], y_scores: List[float], threshold: float = 0.5) -> float:
+    preds = [1 if s >= threshold else 0 for s in y_scores]
+    return float(accuracy_score(y_true, preds))
+
+
+def compute_eer(y_true: List[int], y_scores: List[float]) -> float:
+    """
+    Equal Error Rate — threshold where FPR == FNR.
+    Lower is better.
+    """
+    fpr, tpr, thresholds = roc_curve(y_true, y_scores, pos_label=1)
+    fnr = 1.0 - tpr
+    # Find the threshold where |FPR - FNR| is minimised
+    idx = np.nanargmin(np.abs(fpr - fnr))
+    eer = float((fpr[idx] + fnr[idx]) / 2.0)
+    return eer
+
+
+def compute_all(y_true: List[int], y_scores: List[float]) -> dict:
+    """Returns dict with all metrics — suitable for IEEE results table."""
+    auc = compute_auc(y_true, y_scores)
+    acc = compute_accuracy(y_true, y_scores)
+    eer = compute_eer(y_true, y_scores)
+    return {
+        "auc":      round(auc * 100, 2),
+        "accuracy": round(acc * 100, 2),
+        "eer":      round(eer * 100, 2),
+    }
+
+
+if __name__ == "__main__":
+    # Quick smoke test
+    rng = np.random.default_rng(42)
+    labels = [0] * 50 + [1] * 50
+    scores = list(rng.uniform(0, 0.4, 50)) + list(rng.uniform(0.6, 1.0, 50))
+    print(compute_all(labels, scores))

backend/model.py

@@ -0,0 +1,164 @@
+"""
+HybridDeepfakeDetector — Dual-branch Spatial + Frequency CNN
+Architecture for IEEE Research Paper on Deepfake Video Detection
+"""
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+
+try:
+    import timm
+    TIMM_AVAILABLE = True
+except ImportError:
+    TIMM_AVAILABLE = False
+
+import torchvision.models as tv_models
+
+
+# ─────────────────────────────────────────────
+# Frequency Analysis Branch
+# ─────────────────────────────────────────────
+class FrequencyBranch(nn.Module):
+    """
+    Extracts GAN fingerprint artifacts from the frequency domain.
+    GAN generators leave periodic patterns in the DCT/FFT spectrum
+    that are invisible to the human eye but detectable by CNNs.
+    """
+    def __init__(self, out_dim: int = 128):
+        super().__init__()
+        self.conv_layers = nn.Sequential(
+            nn.Conv2d(1, 32, kernel_size=3, padding=1, bias=False),
+            nn.BatchNorm2d(32),
+            nn.ReLU(inplace=True),
+            nn.MaxPool2d(2),                          # 112x112
+
+            nn.Conv2d(32, 64, kernel_size=3, padding=1, bias=False),
+            nn.BatchNorm2d(64),
+            nn.ReLU(inplace=True),
+            nn.MaxPool2d(2),                          # 56x56
+
+            nn.Conv2d(64, 128, kernel_size=3, padding=1, bias=False),
+            nn.BatchNorm2d(128),
+            nn.ReLU(inplace=True),
+            nn.MaxPool2d(2),                          # 28x28
+
+            nn.Conv2d(128, out_dim, kernel_size=3, padding=1, bias=False),
+            nn.BatchNorm2d(out_dim),
+            nn.ReLU(inplace=True),
+            nn.AdaptiveAvgPool2d((1, 1)),             # 1x1
+        )
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        # Convert to grayscale: (B, 1, H, W)
+        gray = 0.299 * x[:, 0:1] + 0.587 * x[:, 1:2] + 0.114 * x[:, 2:3]
+
+        # 2D FFT → log-magnitude spectrum
+        fft = torch.fft.fft2(gray)
+        magnitude = torch.abs(fft)
+        magnitude = torch.log(magnitude + 1e-8)
+
+        # Normalize per sample
+        b = magnitude.shape[0]
+        m = magnitude.view(b, -1)
+        mn = m.mean(dim=1, keepdim=True).view(b, 1, 1, 1)
+        std = m.std(dim=1, keepdim=True).view(b, 1, 1, 1) + 1e-8
+        magnitude = (magnitude - mn) / std
+
+        return self.conv_layers(magnitude).flatten(1)   # (B, out_dim)
+
+
+# ─────────────────────────────────────────────
+# Spatial Branch (EfficientNet-B0 backbone)
+# ─────────────────────────────────────────────
+class SpatialBranch(nn.Module):
+    def __init__(self, pretrained: bool = True):
+        super().__init__()
+        if TIMM_AVAILABLE:
+            self.backbone = timm.create_model(
+                "efficientnet_b0",
+                pretrained=pretrained,
+                num_classes=0,          # Remove classifier head
+                global_pool="avg",
+            )
+            self.out_dim = self.backbone.num_features   # 1280
+        else:
+            # Fallback: MobileNetV3-Small from torchvision
+            backbone = tv_models.mobilenet_v3_small(pretrained=pretrained)
+            self.backbone = nn.Sequential(*list(backbone.children())[:-2],
+                                          nn.AdaptiveAvgPool2d(1), nn.Flatten())
+            self.out_dim = 576
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        return self.backbone(x)     # (B, out_dim)
+
+
+# ─────────────────────────────────────────────
+# Fusion Classifier
+# ─────────────────────────────────────────────
+class FusionClassifier(nn.Module):
+    def __init__(self, spatial_dim: int, freq_dim: int):
+        super().__init__()
+        combined = spatial_dim + freq_dim
+        self.fc = nn.Sequential(
+            nn.Linear(combined, 512),
+            nn.BatchNorm1d(512),
+            nn.ReLU(inplace=True),
+            nn.Dropout(0.5),
+            nn.Linear(512, 128),
+            nn.ReLU(inplace=True),
+            nn.Dropout(0.3),
+            nn.Linear(128, 1),
+        )
+
+    def forward(self, spatial_feat, freq_feat):
+        x = torch.cat([spatial_feat, freq_feat], dim=1)
+        return self.fc(x)   # (B, 1) — raw logits
+
+
+# ─────────────────────────────────────────────
+# HybridDeepfakeDetector (Main Model)
+# ─────────────────────────────────────────────
+class HybridDeepfakeDetector(nn.Module):
+    """
+    Novel dual-branch architecture combining:
+    - Spatial branch (EfficientNet-B0): captures texture/semantic artifacts
+    - Frequency branch (FFT-CNN): captures GAN frequency fingerprints
+    Fused via FC layers for binary Real/Fake classification.
+    """
+    def __init__(self, pretrained: bool = True, freq_dim: int = 128):
+        super().__init__()
+        self.spatial = SpatialBranch(pretrained=pretrained)
+        self.freq    = FrequencyBranch(out_dim=freq_dim)
+        self.fusion  = FusionClassifier(self.spatial.out_dim, freq_dim)
+
+    def forward(self, x: torch.Tensor):
+        s = self.spatial(x)
+        f = self.freq(x)
+        return self.fusion(s, f)        # (B, 1) logits
+
+    def predict_proba(self, x: torch.Tensor) -> torch.Tensor:
+        """Returns fake probability in [0, 1]."""
+        with torch.no_grad():
+            logits = self.forward(x)
+            return torch.sigmoid(logits).squeeze(1)  # (B,)
+
+    @staticmethod
+    def load(path: str, device: str = "cpu") -> "HybridDeepfakeDetector":
+        model = HybridDeepfakeDetector(pretrained=False)
+        state = torch.load(path, map_location=device)
+        model.load_state_dict(state)
+        model.eval()
+        return model
+
+
+# ─────────────────────────────────────────────
+# Quick sanity check
+# ─────────────────────────────────────────────
+if __name__ == "__main__":
+    model = HybridDeepfakeDetector(pretrained=False)
+    dummy = torch.randn(4, 3, 224, 224)
+    out = model.predict_proba(dummy)
+    total = sum(p.numel() for p in model.parameters())
+    print(f"Output shape : {out.shape}")
+    print(f"Total params : {total:,}")
+    print("Model OK ✓")

backend/models/deepfake_model.pth

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:9071a0de0a2dfc7f7e5b1954ad4b457602b01c72132ab85c60933db1f6b1fb6b
+size 84509270

backend/models/image_model_best.pth

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:c93144e2c6245674ea2fcd4f444dfdc0bd44635797db13d3c38cf45f2d8928f3
+size 84520278

backend/multimodal_detector.py

@@ -0,0 +1,206 @@
+"""
+MultiModalDetector — Unified orchestrator for DeepShield
+Handles Video (visual + audio fusion), Image, and Audio detection.
+"""
+import os, uuid, time, subprocess
+import numpy as np
+from pathlib import Path
+from typing import Optional
+import torch
+
+from detector       import DeepfakeDetector        # visual video branch
+from audio_detector import AudioDeepfakeDetector
+from image_detector import ImageDeepfakeDetector
+
+
+# ─────────────────────────────────────────────────────────────────
+# Utility: Extract audio from video using ffmpeg
+# ─────────────────────────────────────────────────────────────────
+def extract_audio_from_video(video_path: str, output_wav: str) -> bool:
+    """
+    Extracts the audio track from a video file and saves as 16 kHz mono WAV.
+    Returns True on success, False if the video has no audio or ffmpeg fails.
+    """
+    try:
+        cmd = [
+            "ffmpeg", "-y",
+            "-i",  video_path,
+            "-vn",                     # No video
+            "-ar", "16000",            # Resample to 16 kHz
+            "-ac", "1",                # Mono
+            "-f",  "wav",
+            output_wav,
+        ]
+        result = subprocess.run(cmd, capture_output=True, timeout=120)
+        return result.returncode == 0 and Path(output_wav).exists() and Path(output_wav).stat().st_size > 0
+    except Exception as e:
+        print(f"[AudioExtract] ffmpeg failed: {e}")
+        return False
+
+
+# ─────────────────────────────────────────────────────────────────
+# MultiModalDetector
+# ─────────────────────────────────────────────────────────────────
+class MultiModalDetector:
+    """
+    Unified deepfake detector for three modalities:
+      - Video  : EfficientNet (visual) + Wav2Vec2 (audio) → fused score
+      - Image  : EfficientNetV2-S + MTCNN face detection
+      - Audio  : Wav2Vec2-base + attention-pooling classifier
+
+    Fusion strategy for video:
+      fused_score = 0.60 × visual_score + 0.40 × audio_score
+      (if no audio track, fused_score = visual_score)
+    """
+
+    VISUAL_WEIGHT = 0.60
+    AUDIO_WEIGHT  = 0.40
+
+    def __init__(
+        self,
+        video_model_path: Optional[str] = None,
+        image_model_path: Optional[str] = None,
+        audio_model_path: Optional[str] = None,
+        device:           Optional[str] = None,
+        threshold:        float          = 0.5,
+    ):
+        self.device    = device or ("cuda" if torch.cuda.is_available() else "cpu")
+        self.threshold = threshold
+
+        print(f"[DeepShield] Initializing MultiModalDetector on {self.device}")
+
+        # ── Video (visual) detector ───────────────────────────
+        self.video_detector = DeepfakeDetector(
+            model_path=video_model_path,
+            device=self.device,
+            max_frames=32,
+            threshold=threshold,
+        )
+
+        # ── Image detector ────────────────────────────────────
+        self.image_detector = ImageDeepfakeDetector(
+            model_path=image_model_path,
+            device=self.device,
+            threshold=threshold,
+        )
+
+        # ── Audio detector ────────────────────────────────────
+        self.audio_available = False
+        self.audio_model     = None
+        try:
+            from audio_detector import AudioDeepfakeDetector, WAV2VEC_AVAILABLE
+            if WAV2VEC_AVAILABLE:
+                self.audio_model = AudioDeepfakeDetector(
+                    pretrained=True,
+                    freeze_base=True,
+                )
+                if audio_model_path and Path(audio_model_path).exists():
+                    state = torch.load(
+                        audio_model_path, map_location=self.device, weights_only=True
+                    )
+                    self.audio_model.load_state_dict(state)
+                    print(f"[AudioDetector] Loaded weights from {audio_model_path}")
+                else:
+                    print("[AudioDetector] Using pretrained Wav2Vec2 features (demo mode).")
+                self.audio_model.to(self.device).eval()
+                self.audio_available = True
+            else:
+                print("[AudioDetector] transformers not installed — audio branch disabled.")
+        except Exception as e:
+            print(f"[AudioDetector] Init failed: {e}")
+
+    # ─��� VIDEO ─────────────────────────────────────────────────────
+    def analyze_video(self, video_path: str, session_id: Optional[str] = None) -> dict:
+        """
+        Full multimodal video analysis:
+          1. Visual branch: face extraction → EfficientNet inference → Grad-CAM
+          2. Audio branch : ffmpeg extract → Wav2Vec2 inference
+          3. Score fusion : 60/40 weighted average
+        """
+        t0         = time.time()
+        session_id = session_id or str(uuid.uuid4())
+
+        # 1. Visual analysis (existing pipeline)
+        visual_result = self.video_detector.analyze(video_path, session_id=session_id)
+        visual_score  = visual_result.get("fake_prob", 0.5)
+
+        # 2. Audio analysis
+        audio_info = {"audio_available": False, "audio_fake_prob": None}
+        if self.audio_available and self.audio_model is not None:
+            audio_wav = str(Path("uploads") / session_id / "audio.wav")
+            has_audio = extract_audio_from_video(video_path, audio_wav)
+            if has_audio:
+                try:
+                    waveform   = AudioDeepfakeDetector.load_audio(audio_wav)
+                    audio_prob = self.audio_model.predict_proba(waveform, self.device)
+                    audio_info = {
+                        "audio_available":   True,
+                        "audio_fake_prob":   round(audio_prob, 4),
+                        "audio_verdict":     "FAKE" if audio_prob >= self.threshold else "REAL",
+                        "audio_confidence":  round(audio_prob * 100, 2),
+                    }
+                except Exception as e:
+                    audio_info = {"audio_available": False, "audio_error": str(e)}
+            else:
+                audio_info = {"audio_available": False, "audio_error": "No audio track found."}
+
+        # 3. Fusion
+        if audio_info.get("audio_available") and audio_info.get("audio_fake_prob") is not None:
+            fused_score = (
+                self.VISUAL_WEIGHT * visual_score
+                + self.AUDIO_WEIGHT * audio_info["audio_fake_prob"]
+            )
+        else:
+            fused_score = visual_score
+
+        fused_verdict = "FAKE" if fused_score >= self.threshold else "REAL"
+
+        return {
+            **visual_result,
+            **audio_info,
+            "visual_fake_prob":   round(visual_score, 4),
+            "visual_confidence":  round(visual_score * 100, 2),
+            "visual_verdict":     "FAKE" if visual_score >= self.threshold else "REAL",
+            "fused_fake_prob":    round(fused_score, 4),
+            "fused_confidence":   round(fused_score, 4),
+            "fused_verdict":      fused_verdict,
+            "verdict":            fused_verdict,
+            "confidence":         round(fused_score, 4),
+            "fake_prob":          round(fused_score, 4),
+            "elapsed_sec":        round(time.time() - t0, 2),
+            "modality":           "video",
+        }
+
+    # ── IMAGE ─────────────────────────────────────────────────────
+    def analyze_image(self, image_path: str) -> dict:
+        """EfficientNetV2-S + MTCNN image pipeline."""
+        return self.image_detector.analyze(image_path)
+
+    # ── AUDIO ─────────────────────────────────────────────────────
+    def analyze_audio(self, audio_path: str) -> dict:
+        """Wav2Vec2 audio-only pipeline."""
+        if not self.audio_available or self.audio_model is None:
+            return {
+                "verdict":    "ERROR",
+                "error":      "Audio detection unavailable. Install: transformers, librosa.",
+                "confidence": 0.0,
+                "fake_prob":  0.0,
+                "modality":   "audio",
+            }
+        try:
+            waveform = AudioDeepfakeDetector.load_audio(audio_path)
+            prob     = self.audio_model.predict_proba(waveform, self.device)
+            return {
+                "verdict":    "FAKE" if prob >= self.threshold else "REAL",
+                "confidence": round(prob, 4),
+                "fake_prob":  round(prob, 4),
+                "modality":   "audio",
+            }
+        except Exception as e:
+            return {
+                "verdict":    "ERROR",
+                "error":      str(e),
+                "confidence": 0.0,
+                "fake_prob":  0.0,
+                "modality":   "audio",
+            }

backend/preprocess_celebdf.py

@@ -0,0 +1,156 @@
+"""
+preprocess_celebdf.py — Extract face crops from Celeb-DF v2 for training
+Reads from:
+  Celeb-DF/Celeb-real/       → label 0 (REAL)
+  Celeb-DF/YouTube-real/     → label 0 (REAL)
+  Celeb-DF/Celeb-synthesis/  → label 1 (FAKE)
+
+Outputs to:
+  data/train/real/  and  data/train/fake/
+  data/val/real/    and  data/val/fake/
+
+Usage:
+  python preprocess_celebdf.py --dataset_dir ../Celeb-DF --out_dir ../data
+                               --frames_per_video 15 --val_split 0.15
+"""
+import argparse
+import os
+import random
+import sys
+from pathlib import Path
+
+import cv2
+
+# ── Face detector (OpenCV, no extra deps) ────────────────────────────
+CASCADE = cv2.data.haarcascades + "haarcascade_frontalface_default.xml"
+face_det = cv2.CascadeClassifier(CASCADE)
+
+
+def extract_faces(video_path: str, n_frames: int = 15, size: int = 224):
+    """Sample n_frames evenly, detect face, return list of BGR crops."""
+    cap = cv2.VideoCapture(video_path)
+    total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+    if total <= 0:
+        cap.release()
+        return []
+
+    indices = sorted(random.sample(range(total), min(n_frames, total)))
+    crops = []
+
+    for idx in indices:
+        cap.set(cv2.CAP_PROP_POS_FRAMES, idx)
+        ret, frame = cap.read()
+        if not ret:
+            continue
+
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+        faces = face_det.detectMultiScale(gray, 1.1, 4, minSize=(48, 48))
+        h, w = frame.shape[:2]
+
+        if len(faces) == 0:
+            # centre crop fallback
+            s = min(h, w)
+            y0, x0 = (h - s) // 2, (w - s) // 2
+            crop = frame[y0:y0+s, x0:x0+s]
+        else:
+            fx, fy, fw, fh = max(faces, key=lambda r: r[2]*r[3])
+            pad = int(max(fw, fh) * 0.2)
+            x1, y1 = max(0, fx-pad), max(0, fy-pad)
+            x2, y2 = min(w, fx+fw+pad), min(h, fy+fh+pad)
+            crop = frame[y1:y2, x1:x2]
+
+        if crop.size == 0:
+            continue
+        crops.append(cv2.resize(crop, (size, size)))
+
+    cap.release()
+    return crops
+
+
+def save_crops(crops, out_dir: Path, stem: str):
+    saved = 0
+    for i, crop in enumerate(crops):
+        p = out_dir / f"{stem}_{i:03d}.jpg"
+        cv2.imwrite(str(p), crop, [cv2.IMWRITE_JPEG_QUALITY, 90])
+        saved += 1
+    return saved
+
+
+def collect_videos(folders):
+    videos = []
+    for folder in folders:
+        p = Path(folder)
+        if p.exists():
+            videos += list(p.glob("*.mp4")) + list(p.glob("*.avi"))
+    return videos
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--dataset_dir", default="../datasets/video",
+                        help="Path to video dataset root (contains real/celeb_real, real/youtube_real, fake/celeb_synthesis)")
+    parser.add_argument("--out_dir",     default="../datasets/video_crops",
+                        help="Output directory for face crops")
+    parser.add_argument("--frames_per_video", type=int, default=15,
+                        help="Face crops to extract per video")
+    parser.add_argument("--val_split",   type=float, default=0.15,
+                        help="Fraction of videos held out for validation")
+    parser.add_argument("--seed",        type=int, default=42)
+    args = parser.parse_args()
+
+    random.seed(args.seed)
+    ds = Path(args.dataset_dir)
+
+    # ── Source folders ──────────────────────────────────────────────
+    real_folders = [ds / "real" / "celeb_real", ds / "real" / "youtube_real"]
+    fake_folders = [ds / "fake" / "celeb_synthesis"]
+
+    real_videos = collect_videos(real_folders)
+    fake_videos = collect_videos(fake_folders)
+
+    print(f"Found {len(real_videos)} REAL videos | {len(fake_videos)} FAKE videos")
+
+    def split(vids):
+        random.shuffle(vids)
+        n_val = max(1, int(len(vids) * args.val_split))
+        return vids[n_val:], vids[:n_val]   # train, val
+
+    real_train, real_val = split(real_videos)
+    fake_train, fake_val = split(fake_videos)
+
+    out = Path(args.out_dir)
+    splits = {
+        ("train", "real"): real_train,
+        ("train", "fake"): fake_train,
+        ("val",   "real"): real_val,
+        ("val",   "fake"): fake_val,
+    }
+
+    # ── Extract ─────────────────────────────────────────────────────
+    total_saved = 0
+    for (split_name, label), vids in splits.items():
+        out_dir = out / split_name / label
+        out_dir.mkdir(parents=True, exist_ok=True)
+        print(f"\n[{split_name}/{label}]  Processing {len(vids)} videos → {out_dir}")
+
+        for i, vpath in enumerate(vids):
+            crops = extract_faces(str(vpath), n_frames=args.frames_per_video)
+            stem  = vpath.stem
+            saved = save_crops(crops, out_dir, stem)
+            total_saved += saved
+            print(f"  [{i+1:04d}/{len(vids)}] {vpath.name}  →  {saved} crops", end="\r")
+
+        print()  # newline after carriage-returns
+
+    print(f"\n✅ Done! Total face crops saved: {total_saved}")
+    print(f"   Output: {out.resolve()}")
+
+    # Summary
+    for (s, l), _ in splits.items():
+        d = out / s / l
+        n = len(list(d.glob("*.jpg")))
+        print(f"   {s}/{l}: {n} images")
+
+
+if __name__ == "__main__":
+    main()

backend/render.yaml

@@ -0,0 +1,13 @@
+services:
+  - type: web
+    name: deepshield-api
+    env: python
+    buildCommand: "pip install -r requirements.txt"
+    startCommand: "gunicorn app:app"
+    envVars:
+      - key: PYTHON_VERSION
+        value: 3.10.0
+      - key: MONGO_URI
+        sync: false
+      - key: FLASK_ENV
+        value: production

backend/requirements.txt

@@ -0,0 +1,18 @@
+flask>=3.0.0
+flask-cors>=4.0.0
+pymongo>=4.6.0
+dnspython>=2.6.0
+--extra-index-url https://download.pytorch.org/whl/cpu
+torch>=2.0.0
+torchvision>=0.15.0
+torchaudio>=2.0.0
+timm>=0.9.16
+facenet-pytorch>=2.5.2
+transformers>=4.30.0
+librosa>=0.10.0
+soundfile>=0.12.1
+opencv-python-headless>=4.9.0
+numpy>=1.24.0
+Pillow>=10.3.0
+python-dotenv>=1.0.1
+gunicorn>=21.2.0

backend/runtime.txt

@@ -0,0 +1 @@
+python-3.11.0

backend/train.py

@@ -0,0 +1,194 @@
+"""
+Training Script — IEEE Research Experiments
+Trains HybridDeepfakeDetector on FaceForensics++ or Celeb-DF v2
+
+Usage:
+  python train.py --data_dir /path/to/dataset --epochs 30 --batch_size 32
+"""
+import argparse, time, os
+try:
+    from tqdm import tqdm
+except ImportError:
+    tqdm = None
+from pathlib import Path
+
+import numpy as np
+import torch
+import torch.nn as nn
+import torch.optim as optim
+from torch.utils.data import Dataset, DataLoader, WeightedRandomSampler
+import torchvision.transforms as T
+from PIL import Image
+from sklearn.metrics import roc_auc_score
+
+from model import HybridDeepfakeDetector
+
+# ── Dataset ─────────────────────────────────────────────────────────
+class DeepfakeDataset(Dataset):
+    """
+    Expects directory layout:
+      data_dir/
+        real/   ← real face crops (PNG/JPG)
+        fake/   ← fake face crops
+    """
+    TRAIN_TF = T.Compose([
+        T.Resize((224, 224)),
+        T.RandomHorizontalFlip(),
+        T.ColorJitter(brightness=0.2, contrast=0.2, saturation=0.1),
+        T.RandomRotation(10),
+        T.GaussianBlur(kernel_size=3, sigma=(0.1, 1.5)),
+        T.ToTensor(),
+        T.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]),
+    ])
+    VAL_TF = T.Compose([
+        T.Resize((224, 224)),
+        T.ToTensor(),
+        T.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]),
+    ])
+
+    def __init__(self, data_dir: str, split: str = "train"):
+        self.tf = self.TRAIN_TF if split == "train" else self.VAL_TF
+        self.samples = []
+        for label, folder in [(0, "real"), (1, "fake")]:
+            p = Path(data_dir) / folder
+            if p.exists():
+                for img in p.rglob("*.jpg"):
+                    self.samples.append((str(img), label))
+                for img in p.rglob("*.png"):
+                    self.samples.append((str(img), label))
+
+        if not self.samples:
+            raise ValueError(f"No images found in {data_dir}. "
+                             f"Ensure real/ and fake/ sub-directories exist under {data_dir}.")
+
+    def __len__(self):
+        return len(self.samples)
+
+    def __getitem__(self, idx):
+        path, label = self.samples[idx]
+        img = Image.open(path).convert("RGB")
+        return self.tf(img), torch.tensor(label, dtype=torch.float32)
+
+
+def get_sampler(dataset: DeepfakeDataset) -> WeightedRandomSampler:
+    labels  = [s[1] for s in dataset.samples]
+    counts  = [labels.count(0), labels.count(1)]
+    weights = [1.0 / counts[l] for l in labels]
+    return WeightedRandomSampler(weights, len(weights))
+
+
+# ── Training loop ───────────────────────────────────────────────────
+def train_epoch(model, loader, optimizer, criterion, device, scaler):
+    model.train()
+    total_loss, n = 0.0, 0
+    total_batches = len(loader)
+
+    iterator = tqdm(loader, desc="  Training", unit="batch",
+                    bar_format="{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}, {rate_fmt}] loss={postfix}") \
+               if tqdm else loader
+
+    for batch_idx, (imgs, labels) in enumerate(iterator):
+        imgs, labels = imgs.to(device), labels.to(device)
+        optimizer.zero_grad()
+        with torch.amp.autocast(device_type=device, enabled=scaler is not None):
+            logits = model(imgs).squeeze(1)
+            loss   = criterion(logits, labels)
+        if scaler:
+            scaler.scale(loss).backward()
+            scaler.step(optimizer)
+            scaler.update()
+        else:
+            loss.backward()
+            optimizer.step()
+        total_loss += loss.item() * imgs.size(0)
+        n += imgs.size(0)
+        # Update tqdm with current avg loss
+        if tqdm and hasattr(iterator, 'set_postfix_str'):
+            iterator.set_postfix_str(f"{total_loss/n:.4f}")
+        elif not tqdm and (batch_idx % 50 == 0 or batch_idx == total_batches - 1):
+            pct = (batch_idx + 1) / total_batches * 100
+            print(f"    Batch {batch_idx+1}/{total_batches} ({pct:.0f}%)  avg_loss={total_loss/n:.4f}",
+                  flush=True)
+    return total_loss / n
+
+
+@torch.no_grad()
+def evaluate(model, loader, device):
+    model.eval()
+    all_probs, all_labels = [], []
+    for imgs, labels in loader:
+        imgs = imgs.to(device)
+        probs = torch.sigmoid(model(imgs).squeeze(1)).cpu().numpy()
+        all_probs.extend(probs)
+        all_labels.extend(labels.numpy())
+    auc = roc_auc_score(all_labels, all_probs)
+    preds = [1 if p >= 0.5 else 0 for p in all_probs]
+    acc = np.mean(np.array(preds) == np.array(all_labels))
+    return {"auc": auc, "acc": acc}
+
+
+# ── Main ─────────────────────────────────────────────────────────────
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--data_dir",   default="../datasets/video_crops")
+    parser.add_argument("--epochs",     type=int,   default=30)
+    parser.add_argument("--batch_size", type=int,   default=32)
+    parser.add_argument("--lr",         type=float, default=1e-4)
+    parser.add_argument("--save_dir",   default="models")
+    parser.add_argument("--device",     default=None)
+    args = parser.parse_args()
+
+    device = args.device or ("cuda" if torch.cuda.is_available() else "cpu")
+    print(f"[Train] Device: {device}")
+
+    # Datasets
+    # data_dir should be the base data/ folder; train/val subdirs are appended internally
+    train_split_dir = str(Path(args.data_dir) / "train")
+    val_split_dir   = str(Path(args.data_dir) / "val")
+    train_ds = DeepfakeDataset(train_split_dir, "train")
+    val_ds   = DeepfakeDataset(val_split_dir,   "val")
+    sampler  = get_sampler(train_ds)
+
+    # num_workers=0 avoids Windows multiprocessing issues
+    train_dl = DataLoader(train_ds, batch_size=args.batch_size, sampler=sampler,  num_workers=0, pin_memory=False, drop_last=True)
+    val_dl   = DataLoader(val_ds,   batch_size=args.batch_size, shuffle=False, num_workers=0)
+
+    print(f"[Train] Train samples: {len(train_ds)} | Val samples: {len(val_ds)}")
+
+    # Model
+    model = HybridDeepfakeDetector(pretrained=True).to(device)
+
+    # Loss with label smoothing
+    criterion = nn.BCEWithLogitsLoss(label_smoothing=0.1) if hasattr(
+        nn.BCEWithLogitsLoss, "label_smoothing"
+    ) else nn.BCEWithLogitsLoss()
+
+    optimizer = optim.AdamW(model.parameters(), lr=args.lr, weight_decay=1e-4)
+    scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=args.epochs)
+    scaler    = torch.cuda.amp.GradScaler() if device == "cuda" else None
+
+    os.makedirs(args.save_dir, exist_ok=True)
+    best_auc = 0.0
+
+    for epoch in range(1, args.epochs + 1):
+        t0    = time.time()
+        loss  = train_epoch(model, train_dl, optimizer, criterion, device, scaler)
+        mets  = evaluate(model, val_dl, device)
+        scheduler.step()
+
+        print(f"Epoch {epoch:03d}/{args.epochs}  "
+              f"Loss={loss:.4f}  AUC={mets['auc']*100:.2f}%  "
+              f"ACC={mets['acc']*100:.2f}%  "
+              f"[{time.time()-t0:.1f}s]")
+
+        if mets["auc"] > best_auc:
+            best_auc = mets["auc"]
+            save_path = Path(args.save_dir) / "deepfake_model.pth"
+            torch.save(model.state_dict(), save_path)
+            print(f"  ✓ Best model saved → {save_path}  (AUC={best_auc*100:.2f}%)")
+
+    print(f"\n[Done] Best AUC: {best_auc*100:.2f}%")
+
+
+if __name__ == "__main__":
+    main()

backend/train_audio.py

@@ -0,0 +1,297 @@
+"""
+train_audio.py — Training script for AudioDeepfakeDetector (Wav2Vec2)
+
+Compatible dataset formats:
+  1. ASVspoof 2019 / 2021  (LA / PA protocols)
+  2. Custom folders:
+       data/audio/
+         train/real/  ← bonafide .wav / .flac / .mp3 files
+         train/fake/  ← spoofed  .wav / .flac / .mp3 files
+         val/real/
+         val/fake/
+
+Usage:
+  python train_audio.py \
+    --data_dir  ../data/audio \
+    --epochs    20 \
+    --batch_size 16 \
+    --lr        1e-4 \
+    --save_dir  ../models
+"""
+
+import argparse, os, time, random
+from pathlib import Path
+
+import numpy as np
+import torch
+import torch.nn as nn
+from torch.utils.data import Dataset, DataLoader
+from torch.optim import AdamW
+from torch.optim.lr_scheduler import CosineAnnealingLR
+from sklearn.metrics import roc_auc_score, accuracy_score
+
+try:
+    import librosa
+    LIBROSA_AVAILABLE = True
+except ImportError:
+    print("[ERROR] librosa not installed. Run: pip install librosa>=0.10.0")
+    LIBROSA_AVAILABLE = False
+
+try:
+    from transformers import Wav2Vec2FeatureExtractor
+    TRANSFORMERS_AVAILABLE = True
+except ImportError:
+    print("[ERROR] transformers not installed. Run: pip install transformers>=4.30.0")
+    TRANSFORMERS_AVAILABLE = False
+
+from audio_detector import AudioDeepfakeDetector
+
+# ─────────────────────────────────────────────────────────────────
+SAMPLE_RATE      = 16_000
+MAX_DURATION_SEC = 6.0
+MAX_SAMPLES      = int(MAX_DURATION_SEC * SAMPLE_RATE)
+SEED             = 42
+
+
+def seed_everything(seed: int = SEED):
+    random.seed(seed)
+    np.random.seed(seed)
+    torch.manual_seed(seed)
+    if torch.cuda.is_available():
+        torch.cuda.manual_seed_all(seed)
+
+
+# ─────────────────────────────────────────────────────────────────
+# Dataset
+# ─────────────────────────────────────────────────────────────────
+AUDIO_EXTS = {".wav", ".flac", ".mp3", ".ogg", ".m4a"}
+
+
+class AudioFakeDataset(Dataset):
+    """
+    Folder-based audio dataset.
+    Expects: <root>/<split>/real/  and  <root>/<split>/fake/
+    """
+
+    def __init__(self, root: str, split: str, feature_extractor):
+        self.feature_extractor = feature_extractor
+        self.samples = []
+
+        for label, name in [(0, "real"), (1, "fake")]:
+            folder = Path(root) / split / name
+            if not folder.exists():
+                print(f"[WARN] Folder not found: {folder}")
+                continue
+            for f in folder.rglob("*"):
+                if f.suffix.lower() in AUDIO_EXTS:
+                    self.samples.append((str(f), label))
+
+        random.shuffle(self.samples)
+        real_n = sum(1 for _, l in self.samples if l == 0)
+        fake_n = sum(1 for _, l in self.samples if l == 1)
+        print(f"[Dataset/{split}] real={real_n}  fake={fake_n}  total={len(self.samples)}")
+
+    def __len__(self):
+        return len(self.samples)
+
+    def __getitem__(self, idx):
+        path, label = self.samples[idx]
+        try:
+            waveform, _ = librosa.load(path, sr=SAMPLE_RATE, mono=True)
+        except Exception:
+            waveform = np.zeros(SAMPLE_RATE, dtype=np.float32)
+
+        # Pad / truncate
+        if len(waveform) > MAX_SAMPLES:
+            start    = random.randint(0, len(waveform) - MAX_SAMPLES)
+            waveform = waveform[start: start + MAX_SAMPLES]
+        else:
+            waveform = np.pad(waveform, (0, MAX_SAMPLES - len(waveform)))
+
+        inputs = self.feature_extractor(
+            waveform.astype(np.float32),
+            sampling_rate=SAMPLE_RATE,
+            return_tensors="pt",
+            padding=True,
+        )
+        return inputs.input_values.squeeze(0), torch.tensor(label, dtype=torch.float32)
+
+
+def collate_fn(batch):
+    input_values, labels = zip(*batch)
+    # Pad to the longest sample in the batch
+    max_len = max(x.shape[0] for x in input_values)
+    padded  = torch.stack([
+        torch.nn.functional.pad(x, (0, max_len - x.shape[0])) for x in input_values
+    ])
+    return padded, torch.stack(labels)
+
+
+# ─────────────────────────────────────────────────────────────────
+# Training helpers
+# ─────────────────────────────────────────────────────────────────
+def run_epoch(model, loader, criterion, optimizer, device, training: bool):
+    model.train(training)
+    total_loss, all_probs, all_labels = 0.0, [], []
+
+    for input_values, labels in loader:
+        input_values = input_values.to(device)
+        labels       = labels.to(device)
+
+        with torch.set_grad_enabled(training):
+            logits = model(input_values).squeeze(1)
+            loss   = criterion(logits, labels)
+
+        if training:
+            optimizer.zero_grad()
+            loss.backward()
+            nn.utils.clip_grad_norm_(model.parameters(), 1.0)
+            optimizer.step()
+
+        total_loss += loss.item() * len(labels)
+        probs = torch.sigmoid(logits).detach().cpu().numpy()
+        all_probs.extend(probs.tolist())
+        all_labels.extend(labels.cpu().numpy().tolist())
+
+    avg_loss = total_loss / len(loader.dataset)
+    preds    = [1 if p >= 0.5 else 0 for p in all_probs]
+    acc      = accuracy_score(all_labels, preds)
+    try:
+        auc = roc_auc_score(all_labels, all_probs)
+    except Exception:
+        auc = 0.5
+    return avg_loss, acc, auc
+
+
+# ─────────────────────────────────────────────────────────────────
+# Main
+# ─────────────────────────────────────────────────────────────────
+def main(args):
+    if not LIBROSA_AVAILABLE or not TRANSFORMERS_AVAILABLE:
+        raise SystemExit("Missing dependencies. See error messages above.")
+
+    seed_everything()
+    device = "cuda" if torch.cuda.is_available() else "cpu"
+    print(f"[Train] Device: {device}  |  Data: {args.data_dir}")
+
+    # ── Model ─────────────────────────────────────────────────
+    model = AudioDeepfakeDetector(pretrained=True, freeze_base=True)
+    model.to(device)
+
+    # ── Feature extractor for dataset ─────────────────────────
+    feat_ext = model.feature_extractor
+
+    # ── Datasets & loaders ────────────────────────────────────
+    train_ds = AudioFakeDataset(args.data_dir, "train", feat_ext)
+    val_ds   = AudioFakeDataset(args.data_dir, "val",   feat_ext)
+
+    if len(train_ds) == 0:
+        print("\n[WARN] No training files found!")
+        print("Expected structure:")
+        print("  data/audio/train/real/*.wav")
+        print("  data/audio/train/fake/*.wav")
+        print("  data/audio/val/real/*.wav")
+        print("  data/audio/val/fake/*.wav")
+        print("\nRunning in demo mode (no actual training).")
+        return
+
+    train_loader = DataLoader(
+        train_ds, batch_size=args.batch_size, shuffle=True,
+        num_workers=2, pin_memory=(device == "cuda"),
+        collate_fn=collate_fn, drop_last=True,
+    )
+    val_loader = DataLoader(
+        val_ds, batch_size=args.batch_size, shuffle=False,
+        num_workers=2, collate_fn=collate_fn,
+    )
+
+    # ── Loss, optimiser, scheduler ────────────────────────────
+    # Compute class weights to handle imbalance
+    real_n = sum(1 for _, l in train_ds.samples if l == 0)
+    fake_n = sum(1 for _, l in train_ds.samples if l == 1)
+    pos_weight = torch.tensor([real_n / max(fake_n, 1)], device=device)
+    criterion  = nn.BCEWithLogitsLoss(pos_weight=pos_weight)
+
+    # Only fine-tune the classifier head (Wav2Vec2 frozen)
+    optimizer  = AdamW(
+        filter(lambda p: p.requires_grad, model.parameters()),
+        lr=args.lr, weight_decay=1e-4,
+    )
+    scheduler  = CosineAnnealingLR(optimizer, T_max=args.epochs, eta_min=1e-6)
+
+    # ── Training loop ─────────────────────────────────────────
+    save_dir = Path(args.save_dir)
+    save_dir.mkdir(parents=True, exist_ok=True)
+    best_auc = 0.0
+
+    print("\n" + "=" * 60)
+    print("  DeepShield — Audio Detector Training (Wav2Vec2)")
+    print("=" * 60)
+
+    for epoch in range(1, args.epochs + 1):
+        t0 = time.time()
+
+        tr_loss, tr_acc, tr_auc = run_epoch(model, train_loader, criterion, optimizer, device, training=True)
+        va_loss, va_acc, va_auc = run_epoch(model, val_loader,   criterion, None,      device, training=False)
+        scheduler.step()
+
+        elapsed = time.time() - t0
+        print(
+            f"Epoch {epoch:03d}/{args.epochs}  "
+            f"| train loss={tr_loss:.4f}  acc={tr_acc:.3f}  AUC={tr_auc:.3f}"
+            f"  | val loss={va_loss:.4f}  acc={va_acc:.3f}  AUC={va_auc:.3f}"
+            f"  | {elapsed:.1f}s"
+        )
+
+        if va_auc > best_auc:
+            best_auc = va_auc
+            ckpt = save_dir / "audio_model_best.pth"
+            torch.save(model.state_dict(), ckpt)
+            print(f"  ✔  Best model saved → {ckpt}  (AUC={best_auc:.4f})")
+
+    # Save final checkpoint
+    final = save_dir / "audio_model_final.pth"
+    torch.save(model.state_dict(), final)
+    print(f"\n[Done] Final model saved → {final}")
+    print(f"[Done] Best val AUC: {best_auc:.4f}")
+
+    # ── Phase 2: Unfreeze Wav2Vec2 and fine-tune further ──────
+    if args.unfreeze_epochs > 0:
+        print(f"\n[Phase 2] Unfreezing Wav2Vec2 for {args.unfreeze_epochs} more epochs...")
+        for param in model.wav2vec2.parameters():
+            param.requires_grad = True
+        optimizer2 = AdamW(model.parameters(), lr=args.lr * 0.1, weight_decay=1e-4)
+        scheduler2 = CosineAnnealingLR(optimizer2, T_max=args.unfreeze_epochs, eta_min=1e-7)
+
+        for epoch in range(1, args.unfreeze_epochs + 1):
+            t0 = time.time()
+            tr_loss, tr_acc, tr_auc = run_epoch(model, train_loader, criterion, optimizer2, device, True)
+            va_loss, va_acc, va_auc = run_epoch(model, val_loader,   criterion, None,       device, False)
+            scheduler2.step()
+            elapsed = time.time() - t0
+            print(
+                f"[P2] Epoch {epoch:03d}/{args.unfreeze_epochs}  "
+                f"| val AUC={va_auc:.3f}  acc={va_acc:.3f} | {elapsed:.1f}s"
+            )
+            if va_auc > best_auc:
+                best_auc = va_auc
+                ckpt = save_dir / "audio_model_best.pth"
+                torch.save(model.state_dict(), ckpt)
+                print(f"  ✔  Best model updated → {ckpt}  (AUC={best_auc:.4f})")
+
+        torch.save(model.state_dict(), save_dir / "audio_model_phase2.pth")
+        print(f"\n[Done] Phase-2 model saved. Best AUC = {best_auc:.4f}")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="DeepShield Audio Deepfake Detector Training")
+    parser.add_argument("--data_dir",        type=str, default="../datasets/audio",
+                        help="Root folder with train/val subfolders")
+    parser.add_argument("--epochs",          type=int, default=20)
+    parser.add_argument("--unfreeze_epochs", type=int, default=5,
+                        help="Extra epochs after unfreezing Wav2Vec2 backbone")
+    parser.add_argument("--batch_size",      type=int, default=16)
+    parser.add_argument("--lr",              type=float, default=1e-4)
+    parser.add_argument("--save_dir",        type=str, default="../models")
+    args = parser.parse_args()
+    main(args)

backend/train_image.py

@@ -0,0 +1,278 @@
+"""
+train_image.py — Training script for EfficientNetV2-S image deepfake detector
+Uses MTCNN-preprocessed face crops for training.
+
+Dataset structure:
+  data/images/
+    train/
+      real/  ← real face crops (JPG/PNG)
+      fake/  ← deepfake face crops (JPG/PNG)
+    val/
+      real/
+      fake/
+
+Compatible datasets:
+  - FaceForensics++ (face crops from real/manipulated videos)
+  - Celeb-DF v2 (face crops)
+  - DFDC face crops
+  - Any real/fake image folder pair
+
+Usage:
+  python train_image.py \
+    --data_dir   ../data/images \
+    --epochs     30 \
+    --batch_size 32 \
+    --lr         1e-4 \
+    --save_dir   ../models
+"""
+
+import argparse, os, time, random
+from pathlib import Path
+
+import numpy as np
+import torch
+import torch.nn as nn
+from torch.utils.data import Dataset, DataLoader
+from torch.optim import AdamW
+from torch.optim.lr_scheduler import CosineAnnealingWarmRestarts
+from sklearn.metrics import roc_auc_score, accuracy_score
+from PIL import Image
+
+from image_detector import EfficientNetV2Detector, train_transform, inference_transform
+
+# ─────────────────────────────────────────────────────────────────
+SEED      = 42
+IMG_EXTS  = {".jpg", ".jpeg", ".png", ".webp", ".bmp"}
+
+
+def seed_everything(seed: int = SEED):
+    random.seed(seed)
+    np.random.seed(seed)
+    torch.manual_seed(seed)
+    if torch.cuda.is_available():
+        torch.cuda.manual_seed_all(seed)
+
+
+# ─────────────────────────────────────────────────────────────────
+# Dataset
+# ─────────────────────────────────────────────────────────────────
+class ImageFakeDataset(Dataset):
+    """
+    Folder-based image dataset.
+    Expects: <root>/<split>/real/  and  <root>/<split>/fake/
+    """
+
+    def __init__(self, root: str, split: str, augment: bool = False):
+        self.transform = train_transform if augment else inference_transform
+        self.samples   = []
+
+        for label, name in [(0, "real"), (1, "fake")]:
+            folder = Path(root) / split / name
+            if not folder.exists():
+                print(f"[WARN] Folder not found: {folder}")
+                continue
+            for f in folder.rglob("*"):
+                if f.suffix.lower() in IMG_EXTS:
+                    self.samples.append((str(f), label))
+
+        random.shuffle(self.samples)
+        real_n = sum(1 for _, l in self.samples if l == 0)
+        fake_n = sum(1 for _, l in self.samples if l == 1)
+        print(f"[Dataset/{split}] real={real_n}  fake={fake_n}  total={len(self.samples)}")
+
+    def __len__(self):
+        return len(self.samples)
+
+    def __getitem__(self, idx):
+        path, label = self.samples[idx]
+        try:
+            img = Image.open(path).convert("RGB")
+        except Exception:
+            img = Image.new("RGB", (224, 224), (128, 128, 128))
+        tensor = self.transform(img)
+        return tensor, torch.tensor(label, dtype=torch.float32)
+
+
+# ─────────────────────────────────────────────────────────────────
+# Training helpers
+# ─────────────────────────────────────────────────────────────────
+def mixup(x, y, alpha=0.2):
+    """MixUp augmentation for better generalisation."""
+    if alpha > 0:
+        lam = np.random.beta(alpha, alpha)
+    else:
+        lam = 1.0
+    idx    = torch.randperm(x.size(0))
+    mixed  = lam * x + (1 - lam) * x[idx]
+    y_mix  = lam * y + (1 - lam) * y[idx]
+    return mixed, y_mix
+
+
+from tqdm import tqdm
+
+def run_epoch(model, loader, criterion, optimizer, device, training: bool, use_mixup: bool = False):
+    model.train(training)
+    total_loss, all_probs, all_labels = 0.0, [], []
+    
+    desc = "Train" if training else "Val"
+    pbar = tqdm(loader, desc=desc, leave=False, dynamic_ncols=True)
+
+    for imgs, labels in pbar:
+        imgs   = imgs.to(device)
+        labels = labels.to(device)
+
+        if training and use_mixup:
+            imgs, labels = mixup(imgs, labels, alpha=0.2)
+
+        with torch.set_grad_enabled(training):
+            logits = model(imgs).squeeze(1)
+            loss   = criterion(logits, labels)
+
+        if training:
+            optimizer.zero_grad()
+            loss.backward()
+            nn.utils.clip_grad_norm_(model.parameters(), 1.0)
+            optimizer.step()
+
+        total_loss += loss.item() * len(labels)
+        probs = torch.sigmoid(logits).detach().cpu().numpy()
+        all_probs.extend(probs.tolist())
+        all_labels.extend(labels.cpu().numpy().tolist())
+        
+        # Update progress bar
+        pbar.set_postfix(loss=f"{(total_loss / max(len(all_labels), 1)):.4f}")
+
+    avg_loss = total_loss / max(len(loader.dataset), 1)
+    preds    = [1 if p >= 0.5 else 0 for p in all_probs]
+    int_labels = [round(l) for l in all_labels]
+    acc        = accuracy_score(int_labels, preds)
+    try:
+        auc = roc_auc_score(int_labels, all_probs)
+    except Exception:
+        auc = 0.5
+    return avg_loss, acc, auc
+
+
+# ─────────────────────────────────────────────────────────────────
+# Main
+# ─────────────────────────────────────────────────────────────────
+def main(args):
+    seed_everything()
+    device = "cuda" if torch.cuda.is_available() else "cpu"
+    print(f"[Train] Device: {device}  |  Data: {args.data_dir}")
+
+    # ── Datasets ──────────────────────────────────────────────
+    train_ds = ImageFakeDataset(args.data_dir, "train", augment=True)
+    val_ds   = ImageFakeDataset(args.data_dir, "val",   augment=False)
+
+    if len(train_ds) == 0:
+        print("\n[WARN] No training images found!")
+        print("Expected structure:")
+        print("  data/images/train/real/*.jpg")
+        print("  data/images/train/fake/*.jpg")
+        print("  data/images/val/real/*.jpg")
+        print("  data/images/val/fake/*.jpg")
+        print("\nYou can generate face crops from FaceForensics++ using preprocess_celebdf.py")
+        print("Running in demo mode (no actual training).")
+        return
+
+    train_loader = DataLoader(
+        train_ds, batch_size=args.batch_size, shuffle=True,
+        num_workers=4, pin_memory=(device == "cuda"), drop_last=True,
+    )
+    val_loader = DataLoader(
+        val_ds, batch_size=args.batch_size, shuffle=False, num_workers=4,
+    )
+
+    # ── Model ─────────────────────────────────────────────────
+    model = EfficientNetV2Detector(pretrained=True)
+
+    # Freeze backbone, only train classifier head initially
+    for param in model.backbone.parameters():
+        param.requires_grad = False
+    model.to(device)
+
+    # ── Loss, optimiser, scheduler ────────────────────────────
+    real_n     = sum(1 for _, l in train_ds.samples if l == 0)
+    fake_n     = sum(1 for _, l in train_ds.samples if l == 1)
+    pos_weight = torch.tensor([real_n / max(fake_n, 1)], device=device)
+    criterion  = nn.BCEWithLogitsLoss(pos_weight=pos_weight)
+
+    optimizer = AdamW(
+        filter(lambda p: p.requires_grad, model.parameters()),
+        lr=args.lr, weight_decay=1e-4,
+    )
+    scheduler = CosineAnnealingWarmRestarts(optimizer, T_0=10, T_mult=1)
+
+    # ── Training loop — Phase 1 (frozen backbone) ─────────────
+    save_dir = Path(args.save_dir)
+    save_dir.mkdir(parents=True, exist_ok=True)
+    best_auc = 0.0
+
+    print("\n" + "=" * 65)
+    print("  DeepShield — Image Detector Training (EfficientNetV2-S)")
+    print("=" * 65)
+    print(f"  Phase 1: {args.epochs} epochs with frozen backbone")
+
+    for epoch in range(1, args.epochs + 1):
+        t0 = time.time()
+        tr_loss, tr_acc, tr_auc = run_epoch(model, train_loader, criterion, optimizer, device, True,  use_mixup=True)
+        va_loss, va_acc, va_auc = run_epoch(model, val_loader,   criterion, None,      device, False, use_mixup=False)
+        scheduler.step()
+
+        elapsed = time.time() - t0
+        print(
+            f"  Epoch {epoch:03d}/{args.epochs}  "
+            f"| train loss={tr_loss:.4f}  acc={tr_acc:.3f}  AUC={tr_auc:.3f}"
+            f"  | val loss={va_loss:.4f}  acc={va_acc:.3f}  AUC={va_auc:.3f}"
+            f"  | {elapsed:.1f}s"
+        )
+
+        if va_auc > best_auc:
+            best_auc = va_auc
+            ckpt = save_dir / "image_model_best.pth"
+            torch.save(model.state_dict(), ckpt)
+            print(f"  ✔  Best model saved → {ckpt}  (AUC={best_auc:.4f})")
+
+    # ── Phase 2: Unfreeze + fine-tune whole network ───────────
+    if args.finetune_epochs > 0:
+        print(f"\n  Phase 2: Unfreezing backbone for {args.finetune_epochs} epochs...")
+        for param in model.backbone.parameters():
+            param.requires_grad = True
+
+        optimizer2 = AdamW(model.parameters(), lr=args.lr * 0.1, weight_decay=1e-4)
+        scheduler2 = CosineAnnealingWarmRestarts(optimizer2, T_0=args.finetune_epochs)
+
+        for epoch in range(1, args.finetune_epochs + 1):
+            t0 = time.time()
+            tr_loss, tr_acc, tr_auc = run_epoch(model, train_loader, criterion, optimizer2, device, True)
+            va_loss, va_acc, va_auc = run_epoch(model, val_loader,   criterion, None,       device, False)
+            scheduler2.step()
+            elapsed = time.time() - t0
+            print(
+                f"  [P2] Epoch {epoch:03d}/{args.finetune_epochs}  "
+                f"| val AUC={va_auc:.3f}  acc={va_acc:.3f} | {elapsed:.1f}s"
+            )
+            if va_auc > best_auc:
+                best_auc = va_auc
+                ckpt = save_dir / "image_model_best.pth"
+                torch.save(model.state_dict(), ckpt)
+                print(f"  ✔  Best model updated → {ckpt}  (AUC={best_auc:.4f})")
+
+    torch.save(model.state_dict(), save_dir / "image_model_final.pth")
+    print(f"\n[Done] Final model saved → {save_dir / 'image_model_final.pth'}")
+    print(f"[Done] Best val AUC: {best_auc:.4f}")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="DeepShield Image Deepfake Detector Training")
+    parser.add_argument("--data_dir",        type=str,   default="../datasets/images")
+    parser.add_argument("--epochs",          type=int,   default=30,
+                        help="Phase-1 epochs (frozen backbone)")
+    parser.add_argument("--finetune_epochs", type=int,   default=10,
+                        help="Phase-2 epochs (full fine-tune)")
+    parser.add_argument("--batch_size",      type=int,   default=32)
+    parser.add_argument("--lr",              type=float, default=1e-4)
+    parser.add_argument("--save_dir",        type=str,   default="../models")
+    args = parser.parse_args()
+    main(args)

backend/train_image_improved.py

@@ -0,0 +1,409 @@
+"""
+train_image_improved.py — Improved EfficientNetV2-S image deepfake detector training
+Optimised for small datasets (~2000 images). Key improvements over original:
+
+ 1. Stronger augmentation (RandomErasing, GaussianBlur, Random90°rotations)
+ 2. Label smoothing to prevent overconfident predictions
+ 3. Warmup + CosineAnnealing LR schedule (better convergence)
+ 4. Gradient accumulation (effective larger batch without RAM cost)
+ 5. Early stopping with patience
+ 6. Test-time augmentation (TTA) during validation for better AUC estimates
+ 7. Longer Phase-2 fine-tune with layer-wise LR decay
+
+Dataset structure (same as original):
+  datasets/images/
+    train/
+      real/  ← real face crops (JPG/PNG)
+      fake/  ← deepfake face crops (JPG/PNG)
+    val/
+      real/
+      fake/
+
+Usage:
+  python train_image_improved.py \
+    --data_dir   ../datasets/images \
+    --epochs     40 \
+    --finetune_epochs 20 \
+    --batch_size 16 \
+    --lr         2e-4 \
+    --save_dir   models
+"""
+
+import argparse, os, time, random
+from pathlib import Path
+
+import numpy as np
+import torch
+import torch.nn as nn
+from torch.utils.data import Dataset, DataLoader, WeightedRandomSampler
+from torch.optim import AdamW
+from torch.optim.lr_scheduler import OneCycleLR
+from sklearn.metrics import roc_auc_score, accuracy_score, f1_score
+from PIL import Image
+from tqdm import tqdm
+import torchvision.transforms as T
+
+from image_detector import EfficientNetV2Detector, inference_transform
+
+# ─────────────────────────────────────────────────────────────────
+SEED     = 42
+IMG_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".bmp"}
+MEAN     = [0.485, 0.456, 0.406]
+STD      = [0.229, 0.224, 0.225]
+
+
+def seed_everything(seed: int = SEED):
+    random.seed(seed)
+    np.random.seed(seed)
+    torch.manual_seed(seed)
+    if torch.cuda.is_available():
+        torch.cuda.manual_seed_all(seed)
+
+
+# ─────────────────────────────────────────────────────────────────
+# IMPROVED Augmentation Pipeline
+# ─────────────────────────────────────────────────────────────────
+strong_train_transform = T.Compose([
+    T.Resize((256, 256)),
+    T.RandomCrop(224),
+    T.RandomHorizontalFlip(p=0.5),
+    T.RandomApply([T.RandomRotation(degrees=10)], p=0.4),
+    T.ColorJitter(brightness=0.3, contrast=0.3, saturation=0.2, hue=0.05),
+    T.RandomApply([T.GaussianBlur(kernel_size=5, sigma=(0.1, 2.0))], p=0.3),
+    T.RandomGrayscale(p=0.05),
+    T.ToTensor(),
+    T.Normalize(mean=MEAN, std=STD),
+    T.RandomErasing(p=0.25, scale=(0.02, 0.15)),   # simulate occlusions
+])
+
+# TTA transforms (multiple crops + flips averaged at val time)
+tta_transforms = [
+    T.Compose([T.Resize((224, 224)), T.ToTensor(), T.Normalize(MEAN, STD)]),
+    T.Compose([T.Resize((256, 256)), T.CenterCrop(224), T.ToTensor(), T.Normalize(MEAN, STD)]),
+    T.Compose([T.Resize((224, 224)), T.RandomHorizontalFlip(p=1.0), T.ToTensor(), T.Normalize(MEAN, STD)]),
+]
+
+
+# ─────────────────────────────────────────────────────────────────
+# Dataset
+# ─────────────────────────────────────────────────────────────────
+class ImageFakeDataset(Dataset):
+    def __init__(self, root: str, split: str, augment: bool = False):
+        self.augment   = augment
+        self.transform = strong_train_transform if augment else inference_transform
+        self.samples   = []
+
+        for label, name in [(0, "real"), (1, "fake")]:
+            folder = Path(root) / split / name
+            if not folder.exists():
+                print(f"[WARN] Folder not found: {folder}")
+                continue
+            for f in folder.rglob("*"):
+                if f.suffix.lower() in IMG_EXTS:
+                    self.samples.append((str(f), label))
+
+        random.shuffle(self.samples)
+        real_n = sum(1 for _, l in self.samples if l == 0)
+        fake_n = sum(1 for _, l in self.samples if l == 1)
+        print(f"[Dataset/{split}] real={real_n}  fake={fake_n}  total={len(self.samples)}")
+
+    def __len__(self):
+        return len(self.samples)
+
+    def __getitem__(self, idx):
+        path, label = self.samples[idx]
+        try:
+            img = Image.open(path).convert("RGB")
+        except Exception:
+            img = Image.new("RGB", (224, 224), (128, 128, 128))
+        tensor = self.transform(img)
+        return tensor, torch.tensor(label, dtype=torch.float32)
+
+    def get_weights(self):
+        """Per-sample weights for WeightedRandomSampler (balances classes)."""
+        labels  = [l for _, l in self.samples]
+        n_real  = labels.count(0)
+        n_fake  = labels.count(1)
+        w_real  = 1.0 / n_real if n_real else 1.0
+        w_fake  = 1.0 / n_fake if n_fake else 1.0
+        return [w_real if l == 0 else w_fake for l in labels]
+
+
+# ─────────────────────────────────────────────────────────────────
+# Label-smoothed BCE loss
+# ─────────────────────────────────────────────────────────────────
+class LabelSmoothingBCE(nn.Module):
+    """BCEWithLogitsLoss with label smoothing for better generalisation."""
+    def __init__(self, smoothing: float = 0.1, pos_weight=None):
+        super().__init__()
+        self.smoothing  = smoothing
+        self.pos_weight = pos_weight
+
+    def forward(self, logits, targets):
+        targets = targets * (1 - self.smoothing) + 0.5 * self.smoothing
+        return nn.functional.binary_cross_entropy_with_logits(
+            logits, targets,
+            pos_weight=self.pos_weight,
+        )
+
+
+# ─────────────────────────────────────────────────────────────────
+# TTA inference helper
+# ─────────────────────────────────────────────────────────────────
+def tta_predict(model, pil_img, device):
+    probs = []
+    for tf in tta_transforms:
+        t = tf(pil_img).unsqueeze(0).to(device)
+        with torch.no_grad():
+            p = torch.sigmoid(model(t)).item()
+        probs.append(p)
+    return np.mean(probs)
+
+
+# ─────────────────────────────────────────────────────────────────
+# Epoch runner
+# ─────────────────────────────────────────────────────────────────
+def run_epoch(model, loader, criterion, optimizer, device, training: bool,
+              scheduler=None, accumulation_steps: int = 1):
+    model.train(training)
+    total_loss, all_probs, all_labels = 0.0, [], []
+    desc = "Train" if training else "Val  "
+    pbar = tqdm(loader, desc=desc, leave=False, dynamic_ncols=True)
+
+    optimizer_step_count = 0
+    if training:
+        optimizer.zero_grad()
+
+    for step, (imgs, labels) in enumerate(pbar):
+        imgs   = imgs.to(device)
+        labels = labels.to(device)
+
+        with torch.set_grad_enabled(training):
+            logits = model(imgs).squeeze(1)
+            loss   = criterion(logits, labels)
+
+            if training:
+                (loss / accumulation_steps).backward()
+                if (step + 1) % accumulation_steps == 0:
+                    nn.utils.clip_grad_norm_(model.parameters(), 1.0)
+                    optimizer.step()
+                    if scheduler is not None:
+                        scheduler.step()
+                    optimizer.zero_grad()
+                    optimizer_step_count += 1
+
+        total_loss += loss.item() * len(labels)
+        probs = torch.sigmoid(logits).detach().cpu().numpy()
+        all_probs.extend(probs.tolist())
+        all_labels.extend(labels.cpu().numpy().tolist())
+        pbar.set_postfix(loss=f"{(total_loss / max(len(all_labels), 1)):.4f}")
+
+    # flush remaining gradient
+    if training and (len(loader) % accumulation_steps != 0):
+        nn.utils.clip_grad_norm_(model.parameters(), 1.0)
+        optimizer.step()
+        optimizer.zero_grad()
+
+    avg_loss = total_loss / max(len(loader.dataset), 1)
+    int_labels = [round(l) for l in all_labels]
+    preds      = [1 if p >= 0.5 else 0 for p in all_probs]
+    acc = accuracy_score(int_labels, preds)
+    f1  = f1_score(int_labels, preds, zero_division=0)
+    try:
+        auc = roc_auc_score(int_labels, all_probs)
+    except Exception:
+        auc = 0.5
+    return avg_loss, acc, auc, f1
+
+
+# ─────────────────────────────────────────────────────────────────
+# Main
+# ─────────────────────────────────────────────────────────────────
+def main(args):
+    seed_everything()
+    device = "cuda" if torch.cuda.is_available() else "cpu"
+    print(f"\n[Train] Device: {device}  |  Data: {args.data_dir}")
+    if device == "cpu":
+        print("[WARN] No GPU detected — training will be slow. Consider using Google Colab with GPU.")
+
+    # ── Datasets ──────────────────────────────────────────────
+    train_ds = ImageFakeDataset(args.data_dir, "train", augment=True)
+    val_ds   = ImageFakeDataset(args.data_dir, "val",   augment=False)
+
+    if len(train_ds) == 0:
+        print("\n[WARN] No training images found! Expected:")
+        print("  datasets/images/train/real/*.jpg")
+        print("  datasets/images/train/fake/*.jpg")
+        return
+
+    # Balanced sampler — ensures each batch has ~50/50 real/fake
+    weights  = train_ds.get_weights()
+    sampler  = WeightedRandomSampler(weights, num_samples=len(weights), replacement=True)
+
+    train_loader = DataLoader(
+        train_ds, batch_size=args.batch_size, sampler=sampler,
+        num_workers=0, pin_memory=(device == "cuda"), drop_last=True,
+    )
+    val_loader = DataLoader(
+        val_ds, batch_size=args.batch_size, shuffle=False, num_workers=0,
+    )
+
+    # ── Model ─────────────────────────────────────────────────
+    model = EfficientNetV2Detector(pretrained=True)
+
+    # Phase 1: freeze backbone, only train classifier head
+    for param in model.backbone.parameters():
+        param.requires_grad = False
+    model.to(device)
+
+    # ── Loss ──────────────────────────────────────────────────
+    real_n     = sum(1 for _, l in train_ds.samples if l == 0)
+    fake_n     = sum(1 for _, l in train_ds.samples if l == 1)
+    pos_weight = torch.tensor([real_n / max(fake_n, 1)], device=device)
+    criterion  = LabelSmoothingBCE(smoothing=0.1, pos_weight=pos_weight)
+
+    # ── Phase 1 optimizer + scheduler ─────────────────────────
+    accum_steps = max(1, 32 // args.batch_size)   # effective batch ≈ 32
+    total_steps_p1 = (len(train_loader) // accum_steps) * args.epochs
+
+    optimizer = AdamW(
+        filter(lambda p: p.requires_grad, model.parameters()),
+        lr=args.lr, weight_decay=2e-4,
+    )
+    scheduler = OneCycleLR(
+        optimizer, max_lr=args.lr,
+        total_steps=total_steps_p1,
+        pct_start=0.1,            # 10% warmup
+        anneal_strategy="cos",
+    )
+
+    # ── Training loop – Phase 1 ───────────────────────────────
+    save_dir = Path(args.save_dir)
+    save_dir.mkdir(parents=True, exist_ok=True)
+    best_auc, no_improve = 0.0, 0
+
+    print("\n" + "=" * 70)
+    print("  DeepShield — Improved Image Detector Training (EfficientNetV2-S)")
+    print("=" * 70)
+    print(f"  Phase 1: {args.epochs} epochs | frozen backbone | LR={args.lr}")
+    print(f"  Grad accum steps: {accum_steps} -> effective batch ~{args.batch_size * accum_steps}")
+    print(f"  Early stopping patience: {args.patience}")
+    print("=" * 70)
+
+    for epoch in range(1, args.epochs + 1):
+        t0 = time.time()
+        tr_loss, tr_acc, tr_auc, tr_f1 = run_epoch(
+            model, train_loader, criterion, optimizer, device,
+            training=True, scheduler=scheduler, accumulation_steps=accum_steps
+        )
+        va_loss, va_acc, va_auc, va_f1 = run_epoch(
+            model, val_loader, criterion, None, device,
+            training=False
+        )
+        elapsed = time.time() - t0
+
+        print(
+            f"  [P1] Ep {epoch:03d}/{args.epochs} "
+            f"| train loss={tr_loss:.4f} acc={tr_acc:.3f} AUC={tr_auc:.3f} F1={tr_f1:.3f}"
+            f" | val loss={va_loss:.4f} acc={va_acc:.3f} AUC={va_auc:.3f} F1={va_f1:.3f}"
+            f" | {elapsed:.1f}s"
+        )
+
+        if va_auc > best_auc:
+            best_auc  = va_auc
+            no_improve = 0
+            ckpt = save_dir / "image_model_best.pth"
+            torch.save(model.state_dict(), ckpt)
+            print(f"  [BEST] Model saved -> {ckpt}  (AUC={best_auc:.4f})")
+        else:
+            no_improve += 1
+            if no_improve >= args.patience:
+                print(f"\n  [STOP] Early stopping at epoch {epoch} (no improvement for {args.patience} epochs)")
+                break
+
+    # ── Phase 2: Unfreeze + fine-tune ─────────────────────────
+    if args.finetune_epochs > 0:
+        print(f"\n  Phase 2: Unfreezing all layers for {args.finetune_epochs} epochs...")
+
+        # Reload best checkpoint before fine-tuning
+        best_ckpt = save_dir / "image_model_best.pth"
+        if best_ckpt.exists():
+            model.load_state_dict(torch.load(str(best_ckpt), map_location=device))
+            print(f"  Loaded best checkpoint (AUC={best_auc:.4f}) for fine-tuning")
+
+        for param in model.backbone.parameters():
+            param.requires_grad = True
+
+        # Layer-wise LR: backbone gets 10x lower LR than head
+        backbone_params = list(model.backbone.parameters())
+        head_params     = list(model.classifier.parameters())
+        param_groups    = [
+            {"params": backbone_params, "lr": args.lr * 0.05},
+            {"params": head_params,     "lr": args.lr * 0.5},
+        ]
+
+        total_steps_p2 = (len(train_loader) // accum_steps) * args.finetune_epochs
+        optimizer2 = AdamW(param_groups, weight_decay=2e-4)
+        scheduler2 = OneCycleLR(
+            optimizer2,
+            max_lr=[args.lr * 0.05, args.lr * 0.5],
+            total_steps=total_steps_p2,
+            pct_start=0.15,
+            anneal_strategy="cos",
+        )
+
+        no_improve = 0
+        for epoch in range(1, args.finetune_epochs + 1):
+            t0 = time.time()
+            tr_loss, tr_acc, tr_auc, tr_f1 = run_epoch(
+                model, train_loader, criterion, optimizer2, device,
+                training=True, scheduler=scheduler2, accumulation_steps=accum_steps
+            )
+            va_loss, va_acc, va_auc, va_f1 = run_epoch(
+                model, val_loader, criterion, None, device, training=False
+            )
+            elapsed = time.time() - t0
+
+            print(
+                f"  [P2] Ep {epoch:03d}/{args.finetune_epochs} "
+                f"| val loss={va_loss:.4f} acc={va_acc:.3f} AUC={va_auc:.3f} F1={va_f1:.3f}"
+                f" | {elapsed:.1f}s"
+            )
+
+            if va_auc > best_auc:
+                best_auc   = va_auc
+                no_improve = 0
+                ckpt = save_dir / "image_model_best.pth"
+                torch.save(model.state_dict(), ckpt)
+                print(f"  [BEST] Model updated -> {ckpt}  (AUC={best_auc:.4f})")
+            else:
+                no_improve += 1
+                if no_improve >= args.patience:
+                    print(f"\n  [STOP] Early stopping at fine-tune epoch {epoch}")
+                    break
+
+    torch.save(model.state_dict(), save_dir / "image_model_final.pth")
+    print(f"\n{'='*70}")
+    print(f"  Training Complete!")
+    print(f"  Best Val AUC : {best_auc:.4f}")
+    print(f"  Best model   : {save_dir / 'image_model_best.pth'}")
+    print(f"  Final model  : {save_dir / 'image_model_final.pth'}")
+    print(f"{'='*70}")
+    print("  NOTE: the Flask app auto-loads 'image_model_best.pth'.")
+    print("        Restart Flask after training to use the new weights.\n")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="DeepShield Improved Image Training")
+    parser.add_argument("--data_dir",        type=str,   default="../datasets/images")
+    parser.add_argument("--epochs",          type=int,   default=40,
+                        help="Phase-1 epochs (frozen backbone)")
+    parser.add_argument("--finetune_epochs", type=int,   default=20,
+                        help="Phase-2 epochs (full fine-tune)")
+    parser.add_argument("--batch_size",      type=int,   default=16)
+    parser.add_argument("--lr",              type=float, default=2e-4)
+    parser.add_argument("--save_dir",        type=str,   default="models")
+    parser.add_argument("--patience",        type=int,   default=10,
+                        help="Early stopping patience (epochs without val AUC improvement)")
+    args = parser.parse_args()
+    main(args)