Upload 2 files

main.py

@@ -2,6 +2,7 @@ import os
 import cv2
 import torch
 import shutil
+import numpy as np
 from fastapi import FastAPI, UploadFile, File
 from fastapi.middleware.cors import CORSMiddleware
 from transformers import VideoMAEImageProcessor, VideoMAEForVideoClassification
@@ -10,151 +11,245 @@ from PIL import Image
 
 app = FastAPI()
 
-# Enable CORS for the React frontend
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=["*"], 
+    allow_origins=["*"],
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
 )
 
-# Initialize Models globally so they load once on startup
+# ── Load models once at startup ───────────────────────────────────────────────
 print("Loading MTCNN Face Detector...")
 mtcnn = MTCNN(keep_all=False, select_largest=True, post_process=False)
 
-print("Loading Hugging Face Temporal Deepfake Detector (VideoMAE)...")
+print("Loading VideoMAE Temporal Deepfake Detector...")
 model_name = "Ammar2k/videomae-base-finetuned-deepfake-subset"
-processor = VideoMAEImageProcessor.from_pretrained(model_name)
-model = VideoMAEForVideoClassification.from_pretrained(model_name)
+processor  = VideoMAEImageProcessor.from_pretrained(model_name)
+model      = VideoMAEForVideoClassification.from_pretrained(model_name)
+model.eval()
 
-# Ensure temp directory exists
 os.makedirs("temp", exist_ok=True)
 
-def extract_faces_sequence(video_path, sequence_length=16):
-    """Extracts a sequence of continuous frames and tracks the face temporally."""
-    cap = cv2.VideoCapture(video_path)
-    frames_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
-    
-    if frames_count == 0:
-        return []
-        
-    # Try to get frames from the middle of the video
-    start_frame = max(0, (frames_count // 2) - (sequence_length // 2))
-    cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame)
-    
-    faces = []
-    last_box = None
-    
-    for _ in range(sequence_length):
+# ── Configuration ─────────────────────────────────────────────────────────────
+SEQUENCE_LENGTH = 16   # VideoMAE requires exactly 16 frames per clip
+TARGET_FPS      = 8    # Sample the video at this framerate (higher = finer detail)
+NUM_WINDOWS     = 4    # Number of independent clips to analyze across the video
+WINDOW_SECONDS  = 2    # Duration (seconds) of each clip
+
+def get_video_fps(cap):
+    fps = cap.get(cv2.CAP_PROP_FPS)
+    return fps if fps and fps > 0 else 25.0
+
+def smooth_box(current_box, last_box, alpha=0.6):
+    """Exponential moving-average smoothing on bounding boxes to reduce jitter."""
+    if last_box is None:
+        return current_box
+    return [
+        alpha * c + (1 - alpha) * p
+        for c, p in zip(current_box, last_box)
+    ]
+
+def crop_face(pil_img, box, padding=0.35):
+    """Crop the face from a PIL image with proportional padding."""
+    w = box[2] - box[0]
+    h = box[3] - box[1]
+    pad_w = int(w * padding)
+    pad_h = int(h * padding)
+    x1 = max(0, int(box[0]) - pad_w)
+    y1 = max(0, int(box[1]) - pad_h)
+    x2 = min(pil_img.width,  int(box[2]) + pad_w)
+    y2 = min(pil_img.height, int(box[3]) + pad_h)
+    if x2 > x1 and y2 > y1:
+        return pil_img.crop((x1, y1, x2, y2))
+    return None
+
+def extract_window(cap, start_frame, video_fps):
+    """
+    Extract one 16-frame clip from [start_frame] sampled at TARGET_FPS.
+    Returns a list of exactly SEQUENCE_LENGTH PIL face crops.
+    """
+    # How many raw video frames to skip between each sample
+    frame_step = max(1, int(video_fps / TARGET_FPS))
+    faces      = []
+    last_box   = None
+
+    frame_idx = start_frame
+    attempts  = 0
+    max_attempts = SEQUENCE_LENGTH * frame_step * 3  # safety cap
+
+    while len(faces) < SEQUENCE_LENGTH and attempts < max_attempts:
+        cap.set(cv2.CAP_PROP_POS_FRAMES, frame_idx)
         ret, frame = cap.read()
         if not ret:
             break
-            
-        # Convert BGR to RGB for MTCNN and PIL
+
         frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-        pil_img = Image.fromarray(frame_rgb)
-        
-        # Detect and crop face
+        pil_img   = Image.fromarray(frame_rgb)
+
         boxes, _ = mtcnn.detect(pil_img)
-        
-        # Bounding box smoothing: fallback to last known box if detection fails on a frame
+
         if boxes is not None and len(boxes) > 0:
-            box = boxes[0]
-            last_box = box
+            raw_box  = boxes[0].tolist()
+            smoothed = smooth_box(raw_box, last_box)
+            last_box = smoothed
         elif last_box is not None:
-            box = last_box
+            smoothed = last_box          # hold last known position
         else:
-            continue # Skip if no face found yet
-            
-        # Add 30% padding around the face for better context
-        w = box[2] - box[0]
-        h = box[3] - box[1]
-        pad_w = int(w * 0.3)
-        pad_h = int(h * 0.3)
-        
-        x1 = max(0, int(box[0]) - pad_w)
-        y1 = max(0, int(box[1]) - pad_h)
-        x2 = min(pil_img.width, int(box[2]) + pad_w)
-        y2 = min(pil_img.height, int(box[3]) + pad_h)
-        
-        if x2 > x1 and y2 > y1:
-            face_crop = pil_img.crop((x1, y1, x2, y2))
-            faces.append(face_crop)
-            
-    cap.release()
-    
-    # VideoMAE requires exactly `sequence_length` frames. Pad by duplicating last frame if short.
-    if len(faces) == 0:
+            frame_idx += frame_step
+            attempts  += 1
+            continue
+
+        crop = crop_face(pil_img, smoothed)
+        if crop is not None:
+            faces.append(crop)
+
+        frame_idx += frame_step
+        attempts  += 1
+
+    if not faces:
         return []
-        
-    while len(faces) < sequence_length:
+
+    # Pad to exactly SEQUENCE_LENGTH by repeating the last frame
+    while len(faces) < SEQUENCE_LENGTH:
         faces.append(faces[-1])
-        
-    return faces
 
+    return faces[:SEQUENCE_LENGTH]
+
+def run_inference_on_clip(faces):
+    """Run VideoMAE on a single 16-frame clip. Returns (is_fake, fake_probability)."""
+    inputs = processor(list(faces), return_tensors="pt")
+    with torch.no_grad():
+        outputs = model(**inputs)
+    probs = torch.nn.functional.softmax(outputs.logits[0], dim=-1)
+    class_idx = probs.argmax(-1).item()
+    label     = model.config.id2label[class_idx].lower()
+    is_fake   = "fake" in label
+    # Normalize: always return the probability assigned to "fake"
+    fake_prob = probs[class_idx].item() if is_fake else 1.0 - probs[class_idx].item()
+    return is_fake, fake_prob
+
+# ── API endpoint ──────────────────────────────────────────────────────────────
 @app.post("/api/analyze")
 async def analyze_video(file: UploadFile = File(...)):
-    print(f"Received file: {file.filename}")
-    
-    temp_video_path = os.path.join("temp", file.filename)
-    with open(temp_video_path, "wb") as buffer:
-        shutil.copyfileobj(file.file, buffer)
-        
+    print(f"Received: {file.filename}")
+    temp_path = os.path.join("temp", file.filename)
+
+    with open(temp_path, "wb") as buf:
+        shutil.copyfileobj(file.file, buf)
+
     try:
-        print("Extracting facial sequence from video...")
-        # Extract 16 consecutive frames
-        faces = extract_faces_sequence(temp_video_path, sequence_length=16)
-        
-        if not faces:
-            return {
-                "isFake": False, 
-                "confidence": 0, 
-                "explanation": "Could not detect a clear face in the video sequence.",
-                "details": []
-            }
-            
-        print(f"Extracted {len(faces)} frame sequence. Running temporal inference...")
-        
-        # Prepare for VideoMAE 3D model
-        inputs = processor(list(faces), return_tensors="pt")
-        
-        # Run inference
-        with torch.no_grad():
-            outputs = model(**inputs)
-            
-        # Video classification models return logits for the whole sequence
-        probabilities = torch.nn.functional.softmax(outputs.logits[0], dim=-1)
-        
-        # Get highest probability label
-        predicted_class_idx = probabilities.argmax(-1).item()
-        label = model.config.id2label[predicted_class_idx].lower()
-        
-        is_fake = 'fake' in label
-        confidence = round(probabilities[predicted_class_idx].item() * 100, 2)
-            
-        explanation = "Our 3D Temporal AI analyzed facial motion and detected unnatural movement, micro-expressions, or spatial inconsistencies typical of deepfakes." if is_fake else "Our 3D Temporal AI analyzed the facial movement and found natural micro-expressions and consistent temporal flow."
-        
-        print(f"Result: isFake={is_fake}, confidence={confidence}%")
-        
+        cap          = cv2.VideoCapture(temp_path)
+        video_fps    = get_video_fps(cap)
+        total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+        duration_s   = total_frames / video_fps
+
+        print(f"Video: {duration_s:.1f}s @ {video_fps:.1f}fps  ({total_frames} frames)")
+
+        if total_frames == 0:
+            return {"isFake": False, "confidence": 0,
+                    "explanation": "Could not read the video file.", "details": []}
+
+        # ── Determine start frames for each analysis window ───────────────────
+        window_frames = int(WINDOW_SECONDS * video_fps)   # raw frames per window
+        usable_end    = max(0, total_frames - window_frames)
+
+        if usable_end == 0:
+            # Very short video: use a single window at the start
+            start_frames = [0]
+        else:
+            # Spread NUM_WINDOWS evenly across the usable range
+            start_frames = [
+                int(i * usable_end / (NUM_WINDOWS - 1)) if NUM_WINDOWS > 1 else usable_end // 2
+                for i in range(NUM_WINDOWS)
+            ]
+            # Remove duplicate starts (can happen with short videos)
+            start_frames = sorted(set(start_frames))
+
+        # ── Run inference on each window ──────────────────────────────────────
+        all_fake_probs  = []
+        windows_analyzed = 0
+
+        for i, start in enumerate(start_frames):
+            print(f"  Window {i+1}/{len(start_frames)} — starting at frame {start}")
+            faces = extract_window(cap, start, video_fps)
+
+            if not faces:
+                print("    No face detected — skipping window.")
+                continue
+
+            _, fake_prob = run_inference_on_clip(faces)
+            all_fake_probs.append(fake_prob)
+            windows_analyzed += 1
+            print(f"    Fake probability: {fake_prob:.3f}")
+
+        cap.release()
+
+        if not all_fake_probs:
+            return {"isFake": False, "confidence": 0,
+                    "explanation": "Could not detect a clear face in the video.",
+                    "details": []}
+
+        # ── Aggregate results (weighted toward the most suspicious window) ────
+        arr            = np.array(all_fake_probs)
+        avg_fake_prob  = float(np.mean(arr))
+        max_fake_prob  = float(np.max(arr))
+        # Blend: 70 % mean + 30 % max  → catches a single damning window
+        blended_prob   = 0.70 * avg_fake_prob + 0.30 * max_fake_prob
+
+        is_fake    = blended_prob > 0.50
+        confidence = round((blended_prob if is_fake else 1.0 - blended_prob) * 100, 2)
+
+        print(f"FINAL — isFake={is_fake}, blended_prob={blended_prob:.3f}, confidence={confidence}%")
+
+        explanation = (
+            "Our 3D Temporal AI analyzed multiple video segments and detected "
+            "unnatural facial motion, micro-expression inconsistencies, or "
+            "spatial blending artifacts characteristic of deepfakes."
+            if is_fake else
+            "Our 3D Temporal AI analyzed multiple video segments and found "
+            "natural micro-expressions, consistent temporal flow, and no "
+            "manipulation artifacts."
+        )
+
         return {
             "isFake": is_fake,
             "confidence": confidence,
             "explanation": explanation,
             "details": [
-                {"title": "Temporal Analysis", "desc": "Analyzed a continuous 16-frame clip using a VideoMAE 3D Transformer."},
-                {"title": "Motion Tracking", "desc": "Tracked facial landmarks across time to detect jitter, blending artifacts, and lip-sync inconsistencies."}
+                {
+                    "title": "Multi-Window Temporal Analysis",
+                    "desc": (
+                        f"Analyzed {windows_analyzed} independent {WINDOW_SECONDS}-second "
+                        f"clip(s) spread across the video using a VideoMAE 3D Transformer."
+                    )
+                },
+                {
+                    "title": "High-FPS Face Tracking",
+                    "desc": (
+                        f"Sampled at {TARGET_FPS} FPS with smoothed bounding-box tracking "
+                        "to capture jitter, blinking anomalies, and lip-sync mismatches."
+                    )
+                },
+                {
+                    "title": "Ensemble Scoring",
+                    "desc": (
+                        f"Final score = 70% average + 30% peak across all windows "
+                        f"(avg {avg_fake_prob*100:.1f}%, peak {max_fake_prob*100:.1f}%)."
+                    )
+                }
             ]
         }
-        
+
     except Exception as e:
-        print(f"Error analyzing video: {e}")
+        print(f"Error: {e}")
+        import traceback; traceback.print_exc()
         return {"error": str(e)}
-        
+
     finally:
-        if os.path.exists(temp_video_path):
-            os.remove(temp_video_path)
+        if os.path.exists(temp_path):
+            os.remove(temp_path)
 
 @app.get("/")
 def health_check():
-    return {"status": "3D Temporal Backend is running!"}
+    return {"status": "Multi-Window Temporal Backend is running!"}

requirements.txt

@@ -7,3 +7,4 @@ torchvision
 transformers
 facenet-pytorch
 Pillow
+numpy