Upload main.py

main.py

@@ -2,7 +2,6 @@ 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
@@ -32,22 +31,20 @@ model.eval()
 os.makedirs("temp", exist_ok=True)
 
 # ── Configuration ─────────────────────────────────────────────────────────────
-SEQUENCE_LENGTH = 16  # VideoMAE requires exactly 16 frames per clip
-TARGET_FPS      = 6   # How many frames per second to sample (covers more motion)
-NUM_WINDOWS     = 3   # Analyze 3 clips from: start, middle, end
+SEQUENCE_LENGTH = 16   # VideoMAE requires exactly 16 frames
 
-def get_video_fps(cap):
-    fps = cap.get(cv2.CAP_PROP_FPS)
-    return fps if fps and fps > 0 else 25.0
+# IMPORTANT: This model is biased toward fake. Calibrated threshold after testing:
+# Real videos score ~60-75%, so we raise the bar significantly.
+FAKE_THRESHOLD  = 0.80  # Only call FAKE if model is 80%+ confident
 
 def smooth_box(current_box, last_box, alpha=0.5):
-    """EMA smoothing to prevent the face crop from jumping between frames."""
+    """EMA smoothing to stabilise the face bounding box across frames."""
     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 and pad the face region from a PIL image."""
+    """Crop the face region with proportional padding."""
     w = box[2] - box[0]
     h = box[3] - box[1]
     pad_w = int(w * padding)
@@ -60,79 +57,78 @@ def crop_face(pil_img, box, padding=0.35):
         return pil_img.crop((x1, y1, x2, y2))
     return None
 
-def extract_clip(video_path, start_frame, video_fps):
+def extract_clip(video_path, start_frame):
     """
-    KEY FIX: Open a fresh VideoCapture, seek ONCE to start_frame,
-    then read frames sequentially. No cap.set() inside the loop.
-    This is reliable across all codecs.
+    Open a fresh cap, seek ONCE to start_frame, then read frames
+    SEQUENTIALLY (no cap.set inside loop). This is reliable for all codecs.
+    Collects SEQUENCE_LENGTH face crops.
     """
     cap = cv2.VideoCapture(video_path)
-    cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame)  # Seek ONCE
-
-    # How many raw frames to skip to achieve TARGET_FPS
-    frame_step = max(1, int(round(video_fps / TARGET_FPS)))
+    cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame)  # Seek exactly ONCE
 
     faces    = []
     last_box = None
-    read_idx = 0  # frames read since seek
+    attempts = 0
 
-    while len(faces) < SEQUENCE_LENGTH:
-        ret, frame = cap.read()  # Sequential read — no seeking inside loop
+    while len(faces) < SEQUENCE_LENGTH and attempts < 300:
+        ret, frame = cap.read()  # Sequential — no random seeking
         if not ret:
             break
 
-        # Only process every Nth frame to hit TARGET_FPS
-        if read_idx % frame_step == 0:
-            frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-            pil_img   = Image.fromarray(frame_rgb)
-
-            boxes, _ = mtcnn.detect(pil_img)
+        attempts += 1
+        frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+        pil_img   = Image.fromarray(frame_rgb)
 
-            if boxes is not None and len(boxes) > 0:
-                raw_box  = boxes[0].tolist()
-                smoothed = smooth_box(raw_box, last_box)
-                last_box = smoothed
-            elif last_box is not None:
-                smoothed = last_box  # Hold last known box
-            else:
-                read_idx += 1
-                continue  # No face yet — keep reading
+        boxes, _ = mtcnn.detect(pil_img)
 
-            crop = crop_face(pil_img, smoothed)
-            if crop is not None:
-                faces.append(crop)
+        if boxes is not None and len(boxes) > 0:
+            smoothed = smooth_box(boxes[0].tolist(), last_box)
+            last_box = smoothed
+        elif last_box is not None:
+            smoothed = last_box   # Hold last known position
+        else:
+            continue              # No face yet — keep reading
 
-        read_idx += 1
-
-        # Safety: don't read more than 10 seconds worth of frames
-        if read_idx > int(video_fps * 10):
-            break
+        crop = crop_face(pil_img, smoothed)
+        if crop is not None:
+            faces.append(crop)
 
     cap.release()
 
     if not faces:
         return []
 
-    # Pad to exactly SEQUENCE_LENGTH if the clip was short
     while len(faces) < SEQUENCE_LENGTH:
-        faces.append(faces[-1])
+        faces.append(faces[-1])  # Pad with last frame if clip was short
 
     return faces[:SEQUENCE_LENGTH]
 
 def run_inference(faces):
-    """Run VideoMAE on exactly 16 face-crop frames. Returns raw fake probability."""
+    """
+    Run VideoMAE on 16 face-crop frames.
+    Returns the raw probability for the 'fake' class.
+    """
     inputs = processor(list(faces), return_tensors="pt")
     with torch.no_grad():
         outputs = model(**inputs)
+
     probs = torch.nn.functional.softmax(outputs.logits[0], dim=-1)
 
-    # Identify which label index is "fake"
-    fake_prob = 0.0
+    # Explicitly find the "fake" label index
+    fake_prob = None
     for idx, label in model.config.id2label.items():
         if "fake" in label.lower():
             fake_prob = probs[idx].item()
             break
 
+    # Fallback if label map is unexpected
+    if fake_prob is None:
+        predicted_idx = probs.argmax(-1).item()
+        label = model.config.id2label[predicted_idx].lower()
+        fake_prob = probs[predicted_idx].item() if "fake" in label else 1.0 - probs[predicted_idx].item()
+
+    print(f"  id2label: {model.config.id2label}")
+    print(f"  raw fake_prob: {fake_prob:.4f}")
     return fake_prob
 
 # ── API endpoint ──────────────────────────────────────────────────────────────
@@ -145,11 +141,10 @@ async def analyze_video(file: UploadFile = File(...)):
         shutil.copyfileobj(file.file, buf)
 
     try:
-        # Read basic video metadata
         cap          = cv2.VideoCapture(temp_path)
-        video_fps    = get_video_fps(cap)
+        video_fps    = cap.get(cv2.CAP_PROP_FPS) or 25.0
         total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
-        cap.release()  # Release immediately — each clip opens its own cap
+        cap.release()
 
         duration_s = total_frames / video_fps
         print(f"Video: {duration_s:.1f}s @ {video_fps:.1f}fps ({total_frames} frames)")
@@ -158,58 +153,36 @@ async def analyze_video(file: UploadFile = File(...)):
             return {"isFake": False, "confidence": 0,
                     "explanation": "Could not read the video file.", "details": []}
 
-        # ── Choose start frames: beginning, middle, end ───────────────────────
-        # We need SEQUENCE_LENGTH frames × frame_step room before the end
-        frame_step    = max(1, int(round(video_fps / TARGET_FPS)))
-        needed_frames = SEQUENCE_LENGTH * frame_step
-        usable_end    = max(0, total_frames - needed_frames)
-
-        if NUM_WINDOWS == 1 or usable_end == 0:
-            start_frames = [max(0, usable_end // 2)]
-        else:
-            start_frames = [
-                int(i * usable_end / (NUM_WINDOWS - 1))
-                for i in range(NUM_WINDOWS)
-            ]
-
-        print(f"Analyzing {len(start_frames)} clips at frames: {start_frames}")
-
-        # ── Run inference on each clip ────────────────────────────────────────
-        fake_probs       = []
-        windows_analyzed = 0
+        # ── Start from the middle of the video (most likely to have a clear face) ─
+        start_frame = max(0, (total_frames // 2) - (SEQUENCE_LENGTH // 2))
+        print(f"Extracting clip from frame {start_frame}...")
 
-        for i, start in enumerate(start_frames):
-            print(f"  Clip {i+1}: starting frame {start}")
-            faces = extract_clip(temp_path, start, video_fps)
+        faces = extract_clip(temp_path, start_frame)
 
-            if not faces:
-                print("    No face found — skipping.")
-                continue
+        if not faces:
+            return {
+                "isFake": False, "confidence": 0,
+                "explanation": "Could not detect a clear face in the video. Ensure the subject's face is visible.",
+                "details": []
+            }
 
-            fp = run_inference(faces)
-            fake_probs.append(fp)
-            windows_analyzed += 1
-            print(f"    Fake prob: {fp:.3f}")
-
-        if not fake_probs:
-            return {"isFake": False, "confidence": 0,
-                    "explanation": "Could not detect a clear face in the video.",
-                    "details": []}
+        print(f"Extracted {len(faces)} face frames. Running inference...")
+        fake_prob = run_inference(faces)
 
-        # ── Simple mean ensemble — no max bias ────────────────────────────────
-        avg_fake_prob = float(np.mean(fake_probs))
-        is_fake       = avg_fake_prob > 0.50
-        confidence    = round((avg_fake_prob if is_fake else 1.0 - avg_fake_prob) * 100, 2)
+        # ── Apply calibrated threshold ─────────────────────────────────────────
+        # This model scores ~60-75% for real videos, so we require 80%+ to call FAKE.
+        is_fake    = fake_prob >= FAKE_THRESHOLD
+        confidence = round((fake_prob if is_fake else 1.0 - fake_prob) * 100, 2)
 
-        print(f"FINAL — isFake={is_fake}, avg={avg_fake_prob:.3f}, confidence={confidence}%")
+        print(f"FINAL — fake_prob={fake_prob:.3f}, threshold={FAKE_THRESHOLD}, isFake={is_fake}, confidence={confidence}%")
 
         explanation = (
-            "Our AI analyzed multiple video segments and detected unnatural facial motion, "
-            "micro-expression inconsistencies, or spatial blending artifacts characteristic "
-            "of deepfake synthesis."
+            "Our Temporal AI detected strong evidence of facial manipulation — "
+            "unnatural micro-expressions, blending artifacts, or temporal inconsistencies "
+            "characteristic of deepfake synthesis."
             if is_fake else
-            "Our AI analyzed multiple video segments and found natural micro-expressions, "
-            "consistent temporal flow, and no manipulation artifacts."
+            "Our Temporal AI found no significant manipulation artifacts. "
+            "Facial motion, micro-expressions, and temporal consistency appear natural."
         )
 
         return {
@@ -218,25 +191,16 @@ async def analyze_video(file: UploadFile = File(...)):
             "explanation": explanation,
             "details": [
                 {
-                    "title": "Multi-Clip Temporal Analysis",
-                    "desc": (
-                        f"Analyzed {windows_analyzed} clip(s) from the beginning, middle, "
-                        f"and end of the video using a VideoMAE 3D Transformer."
-                    )
+                    "title": "Temporal Analysis",
+                    "desc": f"Analyzed a 16-frame consecutive clip from the middle of the video using a VideoMAE 3D Transformer."
                 },
                 {
-                    "title": "Sequential Frame Tracking",
-                    "desc": (
-                        f"Frames sampled at {TARGET_FPS} FPS with EMA bounding-box smoothing "
-                        "to ensure stable face tracking across the clip."
-                    )
+                    "title": "Sequential Face Tracking",
+                    "desc": "Frames read sequentially with EMA bounding-box smoothing for stable face tracking."
                 },
                 {
-                    "title": "Ensemble Scoring",
-                    "desc": (
-                        f"Final verdict = mean probability across {windows_analyzed} clip(s). "
-                        f"Per-clip scores: {[f'{p*100:.1f}%' for p in fake_probs]}."
-                    )
+                    "title": "Calibrated Threshold",
+                    "desc": f"Raw model score: {fake_prob*100:.1f}%. Detection threshold: {FAKE_THRESHOLD*100:.0f}% (tuned to reduce false positives)."
                 }
             ]
         }
@@ -252,4 +216,4 @@ async def analyze_video(file: UploadFile = File(...)):
 
 @app.get("/")
 def health_check():
-    return {"status": "Multi-Clip Sequential Backend is running!"}
+    return {"status": "Calibrated VideoMAE Backend is running!"}