Forensische deepfake-analyse

"""
Forensische deepfake-analyse in het frequentiedomein.
 
Analyseert de spectrale eigenschappen van beelden om
artefacten te detecteren die karakteristiek zijn voor AI-generatie.
"""
import numpy as np
from typing import Any
 
def compute_frequency_spectrum(image_gray: np.ndarray) -> np.ndarray:
    """
    Bereken het 2D-frequentiespectrum van een grijswaardenbeeld.
 
    Het magnitudespectrum onthult artefacten in het frequentiedomein
    van generatieve modelarchitecturen.
    """
    # Pas 2D-FFT toe
    f_transform = np.fft.fft2(image_gray.astype(np.float64))
    f_shifted = np.fft.fftshift(f_transform)
 
    # Bereken het log-magnitudespectrum
    magnitude = np.log1p(np.abs(f_shifted))
    return magnitude
 
def analyze_spectral_artifacts(
    magnitude_spectrum: np.ndarray,
) -> dict[str, Any]:
    """
    Analyseer het frequentiespectrum op GAN-/diffusiemodel-artefacten.
 
    Bekende artefacten zijn onder meer:
    - GAN-schaakbordpatroon: periodieke pieken uit transpose-convolutie
    - Diffusiemodel: ongebruikelijke energieverdeling bij hoge frequenties
    - Upsampling-artefacten: spectrale kopieën bij Nyquist-gerelateerde frequenties
    """
    h, w = magnitude_spectrum.shape
    center_y, center_x = h // 2, w // 2
 
    # Bereken het radiale gemiddelde (azimutaal gemiddeld vermogensspectrum)
    y_coords, x_coords = np.ogrid[:h, :w]
    distances = np.sqrt((y_coords - center_y) ** 2 + (x_coords - center_x) ** 2)
    max_radius = min(center_y, center_x)
 
    radial_bins = np.linspace(0, max_radius, 100)
    radial_profile = np.zeros(len(radial_bins) - 1)
 
    for i in range(len(radial_bins) - 1):
        mask = (distances >= radial_bins[i]) & (distances < radial_bins[i + 1])
        if mask.any():
            radial_profile[i] = magnitude_spectrum[mask].mean()
 
    # Controleer op schaakbord-artefact (GAN-signatuur)
    # Een schaakbordpatroon creëert pieken op regelmatige frequentie-intervallen
    if len(radial_profile) > 10:
        profile_diff = np.diff(radial_profile)
        # Hoge variantie in de afgeleide wijst op periodieke pieken
        checkerboard_score = float(np.std(profile_diff) / (np.mean(np.abs(profile_diff)) + 1e-10))
    else:
        checkerboard_score = 0.0
 
    # Analyseer de energieratio bij hoge frequenties
    mid_point = len(radial_profile) // 2
    low_freq_energy = float(np.sum(radial_profile[:mid_point]))
    high_freq_energy = float(np.sum(radial_profile[mid_point:]))
    total_energy = low_freq_energy + high_freq_energy
 
    hf_ratio = high_freq_energy / total_energy if total_energy > 0 else 0.0
 
    return {
        "checkerboard_score": round(checkerboard_score, 4),
        "high_freq_ratio": round(hf_ratio, 4),
        "radial_profile": radial_profile.tolist(),
        "likely_gan_artifact": checkerboard_score > 2.0,
        "likely_synthetic": hf_ratio < 0.15 or checkerboard_score > 2.0,
    }

def detect_blending_artifacts(
    image_rgb: np.ndarray,
    face_mask: np.ndarray,
    edge_width: int = 10,
) -> dict:
    """
    Detecteer mengartefacten op gezichtsranden in een vermoedelijke deepfake.
 
    Face-swap-deepfakes moeten het gegenereerde gezicht in het oorspronkelijke
    frame mengen. Dit mengproces laat detecteerbare artefacten achter, waaronder:
    - Discontinuïteiten in de kleurverdeling op de mengrand
    - Veranderingen in de textuurfrequentie op de rand
    - Inconsistente ruispatronen tussen gezicht en achtergrond
    """
    # Extraheer het randgebied
    from scipy.ndimage import binary_dilation, binary_erosion
 
    dilated = binary_dilation(face_mask, iterations=edge_width)
    eroded = binary_erosion(face_mask, iterations=edge_width)
    boundary_mask = dilated & ~eroded
 
    if not boundary_mask.any():
        return {"error": "No valid boundary region found"}
 
    # Analyseer kleurstatistieken binnen versus buiten het gezicht
    face_pixels = image_rgb[eroded].astype(np.float64)
    bg_pixels = image_rgb[dilated & ~face_mask].astype(np.float64)
    boundary_pixels = image_rgb[boundary_mask].astype(np.float64)
 
    if len(face_pixels) == 0 or len(bg_pixels) == 0:
        return {"error": "Insufficient pixels for analysis"}
 
    # Vergelijking van de kleurverdeling per kanaal
    channel_discontinuities = []
    for c in range(3):
        face_mean = float(np.mean(face_pixels[:, c]))
        face_std = float(np.std(face_pixels[:, c]))
        boundary_mean = float(np.mean(boundary_pixels[:, c]))
        bg_mean = float(np.mean(bg_pixels[:, c]))
 
        # Controleer of de randkleur soepel overgaat
        expected_boundary = (face_mean + bg_mean) / 2
        discontinuity = abs(boundary_mean - expected_boundary) / max(face_std, 1.0)
        channel_discontinuities.append(float(discontinuity))
 
    # Ruisanalyse: bereken de lokale variantie in gezicht versus achtergrond
    def local_variance(pixels: np.ndarray) -> float:
        return float(np.mean(np.var(pixels, axis=0)))
 
    face_noise = local_variance(face_pixels)
    bg_noise = local_variance(bg_pixels)
    noise_ratio = face_noise / max(bg_noise, 1e-10)
 
    return {
        "channel_discontinuities": channel_discontinuities,
        "mean_discontinuity": float(np.mean(channel_discontinuities)),
        "face_noise_level": face_noise,
        "background_noise_level": bg_noise,
        "noise_ratio": round(noise_ratio, 4),
        "blending_artifacts_detected": (
            np.mean(channel_discontinuities) > 0.5
            or abs(noise_ratio - 1.0) > 0.5
        ),
    }

def check_semantic_consistency(
    detection_results: dict,
) -> dict:
    """
    Evalueer indicatoren voor semantische consistentie uit een pijplijn
    voor objectdetectie en landmark-analyse.
 
    Verwacht vooraf berekende resultaten van een gezichts-landmark-detector
    en een objectdetector.
    """
    findings = []
 
    # Controleer de consistentie van de oogreflectie
    left_eye_reflection = detection_results.get("left_eye_specular")
    right_eye_reflection = detection_results.get("right_eye_specular")
    if left_eye_reflection is not None and right_eye_reflection is not None:
        # In natuurlijke foto's zouden oogreflecties consistent moeten zijn
        reflection_similarity = float(np.corrcoef(
            np.array(left_eye_reflection).flatten(),
            np.array(right_eye_reflection).flatten(),
        )[0, 1])
        if reflection_similarity < 0.7:
            findings.append({
                "check": "eye_reflection_consistency",
                "result": "FAIL",
                "detail": f"Eye reflection correlation: {reflection_similarity:.2f}",
            })
 
    # Controleer de symmetrie van de gezichts-landmarks
    landmarks = detection_results.get("face_landmarks")
    if landmarks:
        # Meet de bilaterale symmetrie
        left_landmarks = landmarks.get("left_side", [])
        right_landmarks = landmarks.get("right_side", [])
        if left_landmarks and right_landmarks:
            symmetry_score = _compute_symmetry_score(left_landmarks, right_landmarks)
            if symmetry_score < 0.6:
                findings.append({
                    "check": "facial_symmetry",
                    "result": "SUSPICIOUS",
                    "detail": f"Abnormal asymmetry score: {symmetry_score:.2f}",
                })
 
    # Controleer op handanomalieën (vaak in diffusiemodel-output)
    hand_detections = detection_results.get("hands", [])
    for i, hand in enumerate(hand_detections):
        finger_count = hand.get("finger_count", 5)
        if finger_count != 5:
            findings.append({
                "check": f"hand_{i}_finger_count",
                "result": "FAIL",
                "detail": f"Detected {finger_count} fingers (expected 5)",
            })
 
    return {
        "checks_performed": len(findings) + 3,  # Inclusief geslaagde controles
        "anomalies_found": len(findings),
        "findings": findings,
        "overall_assessment": (
            "LIKELY_SYNTHETIC" if len(findings) >= 2
            else "SUSPICIOUS" if len(findings) == 1
            else "NO_ANOMALIES_DETECTED"
        ),
    }
 
def _compute_symmetry_score(
    left: list[tuple[float, float]],
    right: list[tuple[float, float]],
) -> float:
    """Bereken de bilaterale symmetriescore voor gezichts-landmarks."""
    if len(left) != len(right) or not left:
        return 0.0
    left_arr = np.array(left)
    right_arr = np.array(right)
    # Spiegel de rechterkant
    right_mirrored = right_arr.copy()
    right_mirrored[:, 0] = -right_mirrored[:, 0]
    distances = np.linalg.norm(left_arr - right_mirrored, axis=1)
    max_dist = np.max(np.abs(left_arr))
    return float(1.0 - np.mean(distances) / max(max_dist, 1e-10))

def analyze_audio_spectrum(
    audio_samples: np.ndarray,
    sample_rate: int,
) -> dict:
    """
    Analyseer audio op voice-cloning-artefacten.
 
    Onderzoekt de spectrale envelop, formant-consistentie en
    micro-temporele kenmerken die verschillen tussen natuurlijke en
    synthetische spraak.
    """
    # Bereken het spectrogram
    window_size = int(0.025 * sample_rate)  # vensters van 25 ms
    hop_size = int(0.010 * sample_rate)  # hop van 10 ms
 
    # Handmatige STFT-berekening
    num_frames = (len(audio_samples) - window_size) // hop_size + 1
    window = np.hanning(window_size)
    spectrogram = np.zeros((window_size // 2 + 1, num_frames))
 
    for i in range(num_frames):
        start = i * hop_size
        frame = audio_samples[start:start + window_size] * window
        spectrum = np.abs(np.fft.rfft(frame))
        spectrogram[:, i] = spectrum
 
    # Analyseer de spectrale vlakheid (Wiener-entropie)
    # Synthetische spraak heeft vaak een uniformere spectrale energie
    log_spec = np.log(spectrogram + 1e-10)
    spectral_flatness = np.exp(np.mean(log_spec, axis=0)) / (np.mean(spectrogram, axis=0) + 1e-10)
    mean_flatness = float(np.mean(spectral_flatness))
 
    # Analyseer de temporele gladheid
    # Voice cloning kan te gladde of te schokkerige overgangen produceren
    temporal_diff = np.diff(spectrogram, axis=1)
    temporal_smoothness = float(np.mean(np.abs(temporal_diff)))
 
    # Controleer op vocoder-artefacten in hoge frequenties
    # Neurale vocoders produceren vaak karakteristieke patronen boven 8 kHz
    high_freq_start = int(8000 * window_size / sample_rate)
    if high_freq_start < spectrogram.shape[0]:
        high_freq_energy = float(np.mean(spectrogram[high_freq_start:]))
        total_energy = float(np.mean(spectrogram))
        hf_ratio = high_freq_energy / max(total_energy, 1e-10)
    else:
        hf_ratio = 0.0
 
    return {
        "spectral_flatness": round(mean_flatness, 4),
        "temporal_smoothness": round(temporal_smoothness, 4),
        "high_freq_ratio": round(hf_ratio, 4),
        "likely_synthetic": mean_flatness > 0.3 or hf_ratio < 0.05,
        "analysis_notes": _interpret_audio_analysis(mean_flatness, hf_ratio),
    }
 
def _interpret_audio_analysis(flatness: float, hf_ratio: float) -> str:
    if flatness > 0.3 and hf_ratio < 0.05:
        return "High spectral flatness and low high-frequency content suggest neural vocoder synthesis"
    if flatness > 0.3:
        return "Elevated spectral flatness may indicate synthetic generation"
    if hf_ratio < 0.05:
        return "Reduced high-frequency content may indicate voice cloning with bandwidth limitations"
    return "Spectral properties consistent with natural speech"