test_encoding_real_control.py

"""
CRITICAL CONTROL TEST: Does the encoding approach falsely flag REAL videos?

The encoding-pipeline experiment showed DeCoF's fake_prob rises 0.3-0.6+
when AI-generated videos are re-encoded. Before deploying this as a
pre-processing step, we MUST verify that re-encoding REAL videos does NOT
also push them toward FAKE.

If real videos also flip to FAKE after re-encoding, the approach is unsafe
for deployment (it destroys real/fake discrimination).
"""
import os
import sys
import io
import contextlib
import cv2
import numpy as np
import torch
from pathlib import Path

project_root = Path(__file__).parent
sys.path.insert(0, str(project_root))
sys.path.insert(0, str(project_root / 'sdk'))

import importlib.util
spec = importlib.util.spec_from_file_location(
    'decof_detector', project_root / 'models' / 'video' / 'DeCoF' / 'detector.py'
)
module = importlib.util.module_from_spec(spec)
sys.modules['decof_detector'] = module
spec.loader.exec_module(module)

# Real videos (from media-authenticity/real_videos)
real_videos = [
    'C:/Users/HP/Image-Authenticity/media-authenticity/real_videos/glasses.mp4',
    'C:/Users/HP/Image-Authenticity/media-authenticity/real_videos/video_2026-07-10_12-23-49.mp4',
    'C:/Users/HP/Image-Authenticity/media-authenticity/real_videos/video_2026-07-11_18-57-10.mp4',
]

# Also test AI-generated Veo videos for comparison (same conditions)
fake_videos = [
    'D:/veo/veo/veo_cowboy_sun_1.mp4',
    'D:/veo/veo/veo_example_012_elephant.mp4',
]

TARGET_W, TARGET_H = 1280, 720
SQUARE_SIDE = 720
tmp_dir = project_root / 'encoding_control_tmp'
tmp_dir.mkdir(exist_ok=True)

DECOF_INDICES = np.linspace(0, 31, 8, dtype=int)


def build_8frame_video(frames, fps, dst):
    h, w = frames[0].shape[:2]
    out = cv2.VideoWriter(dst, cv2.VideoWriter_fourcc(*'mp4v'), fps, (w, h))
    for frame in frames:
        out.write(cv2.cvtColor(frame, cv2.COLOR_RGB2BGR))
    out.release()


def resize_video_full(src, dst, target_w, target_h):
    cap = cv2.VideoCapture(src)
    fps = cap.get(cv2.CAP_PROP_FPS)
    out = cv2.VideoWriter(dst, cv2.VideoWriter_fourcc(*'mp4v'), fps, (target_w, target_h))
    while True:
        ret, frame = cap.read()
        if not ret:
            break
        out.write(cv2.resize(frame, (target_w, target_h), interpolation=cv2.INTER_LINEAR))
    cap.release()
    out.release()


def run_on_frames(detector, frames):
    features = detector.extract_clip_features(frames)
    with torch.no_grad():
        logits = detector.small_vit(features)
        probs = torch.softmax(logits, dim=-1)
        return float(probs[0, 1].item())


def run_on_video(detector, video_path):
    with contextlib.redirect_stdout(io.StringIO()):
        result = detector.predict_from_video_path(video_path, threshold=0.5)
    return result.probability


print("=" * 80)
print("CRITICAL CONTROL: Encoding effect on REAL vs FAKE videos")
print("=" * 80)
print("""
Conditions for each video:
  A. Original video -> DeCoF (baseline)
  B. Frames -> 8-frame mp4v re-encoded video -> DeCoF (encoding pipeline)
  C. Full video resized to 1280x720 + mp4v re-encoded -> DeCoF

If REAL videos also flip to FAKE after re-encoding, the approach is UNSAFE.
""")

print("[1] Loading detector...")
detector = module.DeCoFDetector()
detector.load()

all_results = []

# Test real videos
print("\n" + "=" * 80)
print("REAL VIDEOS")
print("=" * 80)
for video_path in real_videos:
    if not os.path.exists(video_path):
        print(f"  SKIP (not found): {video_path}")
        continue
    name = os.path.basename(video_path)
    print(f"\n--- {name} ---")

    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    cap.release()
    print(f"  Resolution: {w}x{h}, fps: {fps:.1f}")

    # A: Baseline
    p_a = run_on_video(detector, video_path)

    # Extract frames, center-crop to square, resize to 720
    frames = detector._extract_decof_frames(video_path)
    square_frames = []
    for frame in frames:
        fh, fw = frame.shape[:2]
        side = min(fw, fh)
        left = (fw - side) // 2
        top = (fh - side) // 2
        square_frames.append(frame[top:top+side, left:left+side])
    resized_frames = [cv2.resize(f, (SQUARE_SIDE, SQUARE_SIDE)) for f in square_frames]

    # B: 8-frame re-encoded video
    b_name = name.replace('.mp4', '_8f.mp4')
    b_path = str(tmp_dir / b_name)
    build_8frame_video(resized_frames, fps, b_path)
    p_b = run_on_video(detector, b_path)

    # C: Full video resize
    c_name = name.replace('.mp4', '_720p.mp4')
    c_path = str(tmp_dir / c_name)
    if not os.path.exists(c_path):
        resize_video_full(video_path, c_path, TARGET_W, TARGET_H)
    p_c = run_on_video(detector, c_path)

    cls_a = "FAKE" if p_a >= 0.5 else "REAL"
    cls_b = "FAKE" if p_b >= 0.5 else "REAL"
    cls_c = "FAKE" if p_c >= 0.5 else "REAL"

    print(f"  A. Original:          {p_a:.4f}  ({cls_a})")
    print(f"  B. 8f mp4v encode:    {p_b:.4f}  ({cls_b})  delta={p_b-p_a:+.4f}")
    print(f"  C. Full 720p encode:  {p_c:.4f}  ({cls_c})  delta={p_c-p_a:+.4f}")

    all_results.append(('REAL', name, p_a, p_b, p_c))

# Test fake videos
print("\n" + "=" * 80)
print("FAKE (AI-GENERATED Veo) VIDEOS")
print("=" * 80)
for video_path in fake_videos:
    name = os.path.basename(video_path)
    print(f"\n--- {name} ---")

    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    cap.release()
    print(f"  Resolution: {w}x{h}, fps: {fps:.1f}")

    p_a = run_on_video(detector, video_path)

    frames = detector._extract_decof_frames(video_path)
    square_frames = []
    for frame in frames:
        fh, fw = frame.shape[:2]
        side = min(fw, fh)
        left = (fw - side) // 2
        top = (fh - side) // 2
        square_frames.append(frame[top:top+side, left:left+side])
    resized_frames = [cv2.resize(f, (SQUARE_SIDE, SQUARE_SIDE)) for f in square_frames]

    b_name = name.replace('.mp4', '_8f.mp4')
    b_path = str(tmp_dir / b_name)
    build_8frame_video(resized_frames, fps, b_path)
    p_b = run_on_video(detector, b_path)

    c_name = name.replace('.mp4', '_720p.mp4')
    c_path = str(tmp_dir / c_name)
    if not os.path.exists(c_path):
        resize_video_full(video_path, c_path, TARGET_W, TARGET_H)
    p_c = run_on_video(detector, c_path)

    cls_a = "FAKE" if p_a >= 0.5 else "REAL"
    cls_b = "FAKE" if p_b >= 0.5 else "REAL"
    cls_c = "FAKE" if p_c >= 0.5 else "REAL"

    print(f"  A. Original:          {p_a:.4f}  ({cls_a})")
    print(f"  B. 8f mp4v encode:    {p_b:.4f}  ({cls_b})  delta={p_b-p_a:+.4f}")
    print(f"  C. Full 720p encode:  {p_c:.4f}  ({cls_c})  delta={p_c-p_a:+.4f}")

    all_results.append(('FAKE', name, p_a, p_b, p_c))

# Summary
print("\n" + "=" * 80)
print("SUMMARY")
print("=" * 80)
print(f"\n{'Type':<6s} {'Video':<45s} {'Orig':>8s} {'Enc8f':>8s} {'720p':>8s}")
print("-" * 80)
for vtype, name, pa, pb, pc in all_results:
    print(f"{vtype:<6s} {name:<45s} {pa:8.4f} {pb:8.4f} {pc:8.4f}")

print("\n" + "=" * 80)
print("VERDICT")
print("=" * 80)
print("""
If REAL videos stay REAL after encoding (B/C below 0.5), the approach
is SAFE to deploy as a pre-processing step.

If REAL videos flip to FAKE after encoding, the approach is UNSAFE -
it exploits DeCoF's compression sensitivity and will produce massive
false-positive rates on genuine videos.
""")