test_video_accuracy.py

"""
test_video_accuracy.py — Video authenticity detector test suite.

Tests the video detector by downloading real and deepfake video clips,
running frame-level analysis, and reporting per-model scores.
"""

import os
import sys
import requests
import tempfile
import uuid
import logging

# Silence noisy loggers
logging.basicConfig(level=logging.WARNING)

from app import get_video_detector
from image_authenticity.utils.video import extract_frames

# ──────────────────────────────────────────────────────────────────────────────
# Test cases
# Each entry: name, url, expected ("REAL" or "FAKE")
# ──────────────────────────────────────────────────────────────────────────────
TEST_CASES = [
    {
        "name": "Real Video - Big Buck Bunny clip (animated, open source)",
        "url": "https://www.w3schools.com/html/mov_bbb.mp4",
        "expected": "REAL",
        "note": "Open-source animated movie — should score REAL (no deepfake artifacts)"
    },
    {
        "name": "Real Video - Sample nature footage",
        "url": "https://www.learningcontainer.com/wp-content/uploads/2020/05/sample-mp4-file.mp4",
        "expected": "REAL",
        "note": "Real-world sample video"
    },
    {
        "name": "Real Video - HTML5 test video",
        "url": "https://www.w3schools.com/html/movie.mp4",
        "expected": "REAL",
        "note": "Standard real video test clip"
    },
]

HEADERS = {"User-Agent": "Mozilla/5.0"}


def download_video(url: str, dest_path: str) -> bool:
    """Download a video to dest_path. Returns True on success."""
    try:
        r = requests.get(url, headers=HEADERS, stream=True, timeout=30)
        r.raise_for_status()
        with open(dest_path, "wb") as f:
            for chunk in r.iter_content(chunk_size=8192):
                f.write(chunk)
        size_mb = os.path.getsize(dest_path) / (1024 * 1024)
        print(f"  Downloaded: {size_mb:.2f} MB")
        return True
    except Exception as e:
        print(f"  [Download Error] {e}")
        return False


def run_tests():
    detector = get_video_detector()
    threshold = detector.ensemble.fake_threshold
    print(f"\n{'='*65}")
    print(f" VIDEO AUTHENTICITY TEST  |  threshold={threshold}")
    print(f"{'='*65}")

    correct = 0
    total = 0

    with tempfile.TemporaryDirectory() as tmp_dir:
        for case in TEST_CASES:
            print(f"\n>> {case['name']}")
            print(f"  Expected : {case['expected']}")
            print(f"  Note     : {case.get('note', '')}")

            # Download
            vid_path = os.path.join(tmp_dir, f"{uuid.uuid4().hex}.mp4")
            if not download_video(case["url"], vid_path):
                print("  [SKIP] Could not download video.")
                continue

            # Extract frames
            try:
                frames = extract_frames(vid_path, num_frames=8)
                print(f"  Frames   : {len(frames)} extracted")
            except Exception as e:
                print(f"  [SKIP] Frame extraction failed: {e}")
                continue

            # Score each frame
            frame_results = []
            for frame in frames:
                res = detector.predict(frame)
                frame_results.append(res)

            # Aggregate: top 30% most suspicious frames
            frame_results.sort(key=lambda r: r.get("fake_prob", 0.0), reverse=True)
            top_k = max(1, len(frame_results) // 3)
            top_results = frame_results[:top_k]

            avg_fake = sum(r["fake_prob"] for r in top_results) / top_k
            avg_real = 1.0 - avg_fake
            label = "FAKE" if avg_fake >= threshold else "REAL"

            # Aggregate model scores
            model_keys = list(frame_results[0]["scores"].keys()) if frame_results else []
            avg_scores = {}
            for k in model_keys:
                avg_scores[k] = sum(r["scores"].get(k, 0) for r in top_results) / top_k

            # Report
            status = "[OK]" if label == case["expected"] else "[FAIL]"
            print(f"\n  Result   : {label}  ({status})")
            print(f"  Fake Prob: {avg_fake:.4f}  (top-{top_k} frame avg)")
            print(f"  Real Prob: {avg_real:.4f}")
            print(f"  — Per-model scores (top-{top_k} frame avg):")
            for k, v in avg_scores.items():
                print(f"      {k:<14}: {v:.4f}")

            # Also show all frames fake_probs for debug
            all_probs = [f"{r['fake_prob']:.3f}" for r in frame_results]
            print(f"  All frame fake_probs (sorted desc): {all_probs}")

            if label == case["expected"]:
                correct += 1
            total += 1

    print(f"\n{'='*65}")
    print(f" ACCURACY: {correct}/{total} ({correct/total*100:.1f}%)" if total > 0 else " No tests ran.")
    print(f"{'='*65}\n")


if __name__ == "__main__":
    run_tests()