Detecting AI-Generated Deepfakes: Forensic Computer Vision in the Age of Synthetic Media

import numpy as np
import cv2
from scipy import fftpack, signal
from sklearn.ensemble import RandomForestClassifier

class FrequencyDomainDetector:
    """Deepfake detection via spectral analysis."""
    
    def __init__(self):
        self.classifier = RandomForestClassifier(n_estimators=100, max_depth=20)
        
    def extract_dct_features(self, image, block_size=8):
        """
        Extract DCT coefficient statistics from image blocks.
        JPEG compression artifacts are characteristic of authentic images.
        """
        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) if len(image.shape) == 3 else image
        h, w = gray.shape
        
        # Divide into non-overlapping blocks
        dct_coefficients = []
        for i in range(0, h - block_size + 1, block_size):
            for j in range(0, w - block_size + 1, block_size):
                block = gray[i:i+block_size, j:j+block_size].astype(np.float32)
                
                # Compute 2D DCT
                dct_block = cv2.dct(block)
                dct_coefficients.append(dct_block.flatten())
        
        dct_array = np.array(dct_coefficients)
        
        # Statistical features from DCT coefficients
        features = {
            'dct_mean': np.mean(dct_array, axis=0),
            'dct_std': np.std(dct_array, axis=0),
            'dct_skew': self._skewness(dct_array),
            'dct_kurtosis': self._kurtosis(dct_array)
        }
        
        # Flatten all features into single vector
        feature_vector = np.concatenate([
            features['dct_mean'][:10],  # First 10 DCT coefficients
            features['dct_std'][:10],
            [features['dct_skew'], features['dct_kurtosis']]
        ])
        
        return feature_vector
    
    def extract_fft_features(self, image):
        """
        Extract Fourier spectrum features.
        GANs produce different frequency distributions than real cameras.
        """
        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) if len(image.shape) == 3 else image
        
        # 2D FFT
        fft = fftpack.fft2(gray)
        fft_shifted = fftpack.fftshift(fft)
        magnitude_spectrum = np.abs(fft_shifted)
        
        # Power spectral density
        psd = magnitude_spectrum ** 2
        
        # Radial profile (azimuthal average)
        h, w = psd.shape
        center = (h // 2, w // 2)
        y, x = np.indices(psd.shape)
        r = np.sqrt((x - center[1])**2 + (y - center[0])**2).astype(int)
        
        radial_profile = np.bincount(r.ravel(), weights=psd.ravel()) / np.bincount(r.ravel())
        
        # Features from radial profile
        features = {
            'low_freq_energy': np.sum(radial_profile[:10]),
            'mid_freq_energy': np.sum(radial_profile[10:50]),
            'high_freq_energy': np.sum(radial_profile[50:]),
            'spectral_slope': self._compute_spectral_slope(radial_profile),
            'spectral_flatness': np.exp(np.mean(np.log(radial_profile + 1e-10))) / (np.mean(radial_profile) + 1e-10)
        }
        
        feature_vector = np.array([
            features['low_freq_energy'],
            features['mid_freq_energy'],
            features['high_freq_energy'],
            features['spectral_slope'],
            features['spectral_flatness']
        ])
        
        return feature_vector
    
    def extract_cfa_features(self, image):
        """
        Analyze Color Filter Array patterns.
        Authentic camera images have specific inter-channel correlations.
        """
        if len(image.shape) != 3:
            return np.zeros(6)  # No color channels
        
        b, g, r = cv2.split(image)
        
        # Compute cross-channel correlations
        corr_rg = np.corrcoef(r.flatten(), g.flatten())[0, 1]
        corr_rb = np.corrcoef(r.flatten(), b.flatten())[0, 1]
        corr_gb = np.corrcoef(g.flatten(), b.flatten())[0, 1]
        
        # Green channel should have higher correlation with red/blue
        # (CFA interpolation creates this structure)
        green_dominance = (corr_rg + corr_gb) / 2 - corr_rb
        
        # Compute color difference statistics
        rg_diff = r.astype(np.float32) - g.astype(np.float32)
        gb_diff = g.astype(np.float32) - b.astype(np.float32)
        
        features = np.array([
            corr_rg, corr_rb, corr_gb,
            green_dominance,
            np.std(rg_diff),
            np.std(gb_diff)
        ])
        
        return features
    
    def extract_all_features(self, image):
        """Combine all frequency-domain features."""
        dct_features = self.extract_dct_features(image)
        fft_features = self.extract_fft_features(image)
        cfa_features = self.extract_cfa_features(image)
        
        return np.concatenate([dct_features, fft_features, cfa_features])
    
    def train(self, real_images, fake_images):
        """Train classifier on real and fake images."""
        print("Extracting features from training images...")
        
        real_features = np.array([self.extract_all_features(img) for img in real_images])
        fake_features = np.array([self.extract_all_features(img) for img in fake_images])
        
        X = np.vstack([real_features, fake_features])
        y = np.array([0] * len(real_images) + [1] * len(fake_images))  # 0=real, 1=fake
        
        print(f"Training on {len(X)} samples...")
        self.classifier.fit(X, y)
        
        # Feature importance
        importances = self.classifier.feature_importances_
        print(f"Top 5 discriminative features:")
        top_indices = np.argsort(importances)[-5:][::-1]
        for idx in top_indices:
            print(f"  Feature {idx}: importance = {importances[idx]:.4f}")
    
    def predict(self, image):
        """
        Predict if image is deepfake.
        
        Returns:
            probability: float in [0, 1], where 1 = likely fake
        """
        features = self.extract_all_features(image)
        probability = self.classifier.predict_proba([features])[0][1]
        return probability
    
    def _skewness(self, data):
        """Compute skewness of data."""
        mean = np.mean(data)
        std = np.std(data)
        return np.mean(((data - mean) / (std + 1e-10)) ** 3)
    
    def _kurtosis(self, data):
        """Compute kurtosis of data."""
        mean = np.mean(data)
        std = np.std(data)
        return np.mean(((data - mean) / (std + 1e-10)) ** 4) - 3
    
    def _compute_spectral_slope(self, spectrum):
        """Fit power law to spectrum: S(f) = A * f^(-alpha)."""
        freq = np.arange(1, len(spectrum))
        log_freq = np.log(freq + 1e-10)
        log_spectrum = np.log(spectrum[1:] + 1e-10)
        
        # Linear regression in log-log space
        slope, _ = np.polyfit(log_freq, log_spectrum, 1)
        return slope


# Example usage
detector = FrequencyDomainDetector()

# Load training data (assume we have authentic and deepfake images)
real_images = [cv2.imread(f'real/image_{i}.jpg') for i in range(1000)]
fake_images = [cv2.imread(f'fake/image_{i}.jpg') for i in range(1000)]

# Train detector
detector.train(real_images, fake_images)

# Test on suspicious image
test_image = cv2.imread('suspicious_ceo_video_frame.jpg')
fake_probability = detector.predict(test_image)

print(f"Deepfake probability: {fake_probability:.2%}")
if fake_probability > 0.7:
    print("⚠️ HIGH RISK: Likely AI-generated")
elif fake_probability > 0.4:
    print("⚠️ MODERATE RISK: Requires human review")
else:
    print("✓ LOW RISK: Likely authentic")
                        

import torch
import torch.nn as nn
import torch.nn.functional as F
from torchvision import models
import numpy as np

class DeepfakeDetectorNetwork(nn.Module):
    """
    EfficientNet-based deepfake detector with attention mechanism.
    """
    
    def __init__(self, num_classes=2, dropout=0.5):
        super().__init__()
        
        # Load pretrained EfficientNet-B4
        self.backbone = models.efficientnet_b4(pretrained=True)
        
        # Remove final classification layer
        num_features = self.backbone.classifier[1].in_features
        self.backbone.classifier = nn.Identity()
        
        # Spatial attention module
        self.attention = SpatialAttention()
        
        # Classification head
        self.classifier = nn.Sequential(
            nn.Dropout(dropout),
            nn.Linear(num_features, 512),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(512, num_classes)
        )
        
    def forward(self, x, return_attention=False):
        # Extract features
        features = self.backbone(x)
        
        # Apply attention (optional visualization)
        if return_attention:
            features, attention_map = self.attention(features, return_map=True)
            logits = self.classifier(features)
            return logits, attention_map
        else:
            features = self.attention(features)
            logits = self.classifier(features)
            return logits


class SpatialAttention(nn.Module):
    """Attention module focusing on face boundaries."""
    
    def __init__(self, in_channels=1792):  # EfficientNet-B4 output
        super().__init__()
        self.conv = nn.Conv2d(in_channels, 1, kernel_size=1)
        
    def forward(self, x, return_map=False):
        # x shape: [batch, channels, H, W]
        attention_map = torch.sigmoid(self.conv(x))
        attended_features = x * attention_map
        
        # Global average pooling
        attended_features = F.adaptive_avg_pool2d(attended_features, 1)
        attended_features = attended_features.view(attended_features.size(0), -1)
        
        if return_map:
            return attended_features, attention_map
        return attended_features


class AdversarialTrainer:
    """
    Train deepfake detector with adversarial examples for robustness.
    """
    
    def __init__(self, model, device='cuda'):
        self.model = model.to(device)
        self.device = device
        self.optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4, weight_decay=1e-5)
        self.scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(self.optimizer, T_max=100)
        
    def generate_adversarial_example(self, images, labels, epsilon=0.03):
        """
        Generate adversarial examples using FGSM (Fast Gradient Sign Method).
        
        Args:
            images: input images
            labels: true labels
            epsilon: perturbation magnitude (L_inf norm)
        
        Returns:
            perturbed images designed to fool classifier
        """
        images = images.to(self.device).requires_grad_(True)
        labels = labels.to(self.device)
        
        # Forward pass
        logits = self.model(images)
        loss = F.cross_entropy(logits, labels)
        
        # Backward pass to get gradients
        self.model.zero_grad()
        loss.backward()
        
        # Generate perturbation
        perturbation = epsilon * images.grad.sign()
        
        # Apply perturbation
        adv_images = images + perturbation
        adv_images = torch.clamp(adv_images, 0, 1)  # Keep in valid range
        
        return adv_images.detach()
    
    def train_epoch(self, train_loader, adversarial_ratio=0.5):
        """
        Train for one epoch with mixture of clean and adversarial examples.
        
        Args:
            adversarial_ratio: fraction of batch to generate adversarial examples for
        """
        self.model.train()
        total_loss = 0
        correct = 0
        total = 0
        
        for batch_idx, (images, labels) in enumerate(train_loader):
            images = images.to(self.device)
            labels = labels.to(self.device)
            
            batch_size = images.size(0)
            adv_size = int(batch_size * adversarial_ratio)
            
            # Split batch into clean and adversarial
            clean_images = images[adv_size:]
            clean_labels = labels[adv_size:]
            
            adv_source_images = images[:adv_size]
            adv_source_labels = labels[:adv_size]
            
            # Generate adversarial examples
            if adv_size > 0:
                adv_images = self.generate_adversarial_example(
                    adv_source_images, adv_source_labels, epsilon=0.03
                )
                
                # Combine clean and adversarial
                combined_images = torch.cat([clean_images, adv_images], dim=0)
                combined_labels = torch.cat([clean_labels, adv_source_labels], dim=0)
            else:
                combined_images = clean_images
                combined_labels = clean_labels
            
            # Forward pass
            self.optimizer.zero_grad()
            logits = self.model(combined_images)
            loss = F.cross_entropy(logits, combined_labels)
            
            # Backward pass
            loss.backward()
            
            # Gradient clipping for stability
            torch.nn.utils.clip_grad_norm_(self.model.parameters(), max_norm=1.0)
            
            self.optimizer.step()
            
            # Statistics
            total_loss += loss.item()
            _, predicted = logits.max(1)
            total += combined_labels.size(0)
            correct += predicted.eq(combined_labels).sum().item()
            
            if batch_idx % 10 == 0:
                print(f'Batch {batch_idx}/{len(train_loader)}, '
                      f'Loss: {loss.item():.4f}, '
                      f'Acc: {100.*correct/total:.2f}%')
        
        self.scheduler.step()
        
        epoch_loss = total_loss / len(train_loader)
        epoch_acc = 100. * correct / total
        
        return epoch_loss, epoch_acc
    
    def evaluate(self, test_loader):
        """Evaluate on test set."""
        self.model.eval()
        correct = 0
        total = 0
        
        true_positives = 0  # Correctly identified fakes
        false_positives = 0  # Real flagged as fake
        true_negatives = 0  # Correctly identified real
        false_negatives = 0  # Fake flagged as real
        
        with torch.no_grad():
            for images, labels in test_loader:
                images = images.to(self.device)
                labels = labels.to(self.device)
                
                logits = self.model(images)
                _, predicted = logits.max(1)
                
                total += labels.size(0)
                correct += predicted.eq(labels).sum().item()
                
                # Compute confusion matrix components
                # Assume label 0 = real, 1 = fake
                for pred, label in zip(predicted, labels):
                    if label == 1 and pred == 1:
                        true_positives += 1
                    elif label == 0 and pred == 1:
                        false_positives += 1
                    elif label == 0 and pred == 0:
                        true_negatives += 1
                    elif label == 1 and pred == 0:
                        false_negatives += 1
        
        accuracy = 100. * correct / total
        
        # Compute metrics
        precision = true_positives / (true_positives + false_positives) if (true_positives + false_positives) > 0 else 0
        recall = true_positives / (true_positives + false_negatives) if (true_positives + false_negatives) > 0 else 0
        f1_score = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0
        
        print(f"\n=== Test Set Performance ===")
        print(f"Accuracy: {accuracy:.2f}%")
        print(f"Precision: {precision:.4f} (of flagged deepfakes, what % are truly fake)")
        print(f"Recall: {recall:.4f} (of all deepfakes, what % did we catch)")
        print(f"F1 Score: {f1_score:.4f}")
        print(f"False Positive Rate: {false_positives/(false_positives+true_negatives):.2%}")
        
        return {
            'accuracy': accuracy,
            'precision': precision,
            'recall': recall,
            'f1_score': f1_score,
            'confusion_matrix': {
                'TP': true_positives,
                'FP': false_positives,
                'TN': true_negatives,
                'FN': false_negatives
            }
        }


# Example training loop
model = DeepfakeDetectorNetwork()
trainer = AdversarialTrainer(model, device='cuda')

# Assume we have DataLoaders
# train_loader = ... (FaceForensics++, Celeb-DF, DFDC combined)
# test_loader = ...

print("Training with adversarial examples...")
for epoch in range(50):
    train_loss, train_acc = trainer.train_epoch(train_loader, adversarial_ratio=0.5)
    print(f"\nEpoch {epoch+1}/50")
    print(f"Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.2f}%")
    
    if (epoch + 1) % 5 == 0:
        print(f"\nEvaluating on test set...")
        test_metrics = trainer.evaluate(test_loader)
        
        # Save checkpoint
        torch.save({
            'epoch': epoch,
            'model_state_dict': model.state_dict(),
            'optimizer_state_dict': trainer.optimizer.state_dict(),
            'test_accuracy': test_metrics['accuracy'],
        }, f'deepfake_detector_epoch_{epoch+1}.pth')