Deepfake Detection: Identifying AI-Generated Media

python3 -m venv .venv-deepfake
source .venv-deepfake/bin/activate
pip install --upgrade pip
pip install librosa soundfile numpy pandas scikit-learn
pip install tensorflow keras
pip install matplotlib seaborn
pip install flask flask-cors

import librosa
import numpy as np
import pandas as pd
from pathlib import Path

class AudioFeatureExtractor:
    """Extract features from audio for deepfake detection"""
    
    def __init__(self, sample_rate=22050):
        self.sample_rate = sample_rate
    
    def extract_mfcc(self, audio_path: str, n_mfcc=13):
        """Extract MFCC (Mel-frequency cepstral coefficients) features"""
        y, sr = librosa.load(audio_path, sr=self.sample_rate)
        mfccs = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=n_mfcc)
        return {
            "mfcc_mean": np.mean(mfccs, axis=1).tolist(),
            "mfcc_std": np.std(mfccs, axis=1).tolist(),
            "mfcc_max": np.max(mfccs, axis=1).tolist(),
            "mfcc_min": np.min(mfccs, axis=1).tolist()
        }
    
    def extract_spectral_features(self, audio_path: str):
        """Extract spectral features"""
        y, sr = librosa.load(audio_path, sr=self.sample_rate)
        
        # Spectral centroid
        spectral_centroids = librosa.feature.spectral_centroid(y=y, sr=sr)[0]
        
        # Spectral rolloff
        spectral_rolloff = librosa.feature.spectral_rolloff(y=y, sr=sr)[0]
        
        # Zero crossing rate
        zcr = librosa.feature.zero_crossing_rate(y)[0]
        
        # Chroma features
        chroma = librosa.feature.chroma_stft(y=y, sr=sr)
        
        return {
            "spectral_centroid_mean": np.mean(spectral_centroids),
            "spectral_centroid_std": np.std(spectral_centroids),
            "spectral_rolloff_mean": np.mean(spectral_rolloff),
            "spectral_rolloff_std": np.std(spectral_rolloff),
            "zcr_mean": np.mean(zcr),
            "zcr_std": np.std(zcr),
            "chroma_mean": np.mean(chroma, axis=1).tolist()
        }
    
    def extract_prosody_features(self, audio_path: str):
        """Extract prosody (rhythm, stress, intonation) features"""
        y, sr = librosa.load(audio_path, sr=self.sample_rate)
        
        # Tempo
        tempo, _ = librosa.beat.beat_track(y=y, sr=sr)
        
        # Onset detection
        onset_frames = librosa.onset.onset_detect(y=y, sr=sr)
        onset_times = librosa.frames_to_time(onset_frames, sr=sr)
        
        # Pitch (fundamental frequency)
        pitches, magnitudes = librosa.piptrack(y=y, sr=sr)
        
        return {
            "tempo": float(tempo),
            "onset_count": len(onset_times),
            "onset_rate": len(onset_times) / (len(y) / sr),
            "pitch_mean": np.mean(pitches[pitches > 0]) if np.any(pitches > 0) else 0,
            "pitch_std": np.std(pitches[pitches > 0]) if np.any(pitches > 0) else 0
        }
    
    def extract_all_features(self, audio_path: str):
        """Extract all features for deepfake detection"""
        features = {}
        
        # MFCC features
        mfcc_features = self.extract_mfcc(audio_path)
        features.update(mfcc_features)
        
        # Spectral features
        spectral_features = self.extract_spectral_features(audio_path)
        features.update(spectral_features)
        
        # Prosody features
        prosody_features = self.extract_prosody_features(audio_path)
        features.update(prosody_features)
        
        return features

# Example usage
extractor = AudioFeatureExtractor()

# For demonstration, create a synthetic audio file
# In real usage, you would load actual audio files
print("Audio feature extractor ready")
print("Features: MFCC, Spectral, Prosody")

import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, classification_report, confusion_matrix
from sklearn.preprocessing import StandardScaler
import pickle
import json

class DeepfakeDetector:
    """Machine learning model for deepfake audio detection"""
    
    def __init__(self):
        self.model = None
        self.scaler = StandardScaler()
        self.feature_names = []
    
    def prepare_features(self, features_list: list):
        """Prepare features for model training"""
        # Flatten nested features
        flattened = []
        for feat_dict in features_list:
            flat = {}
            for key, value in feat_dict.items():
                if isinstance(value, list):
                    for i, v in enumerate(value):
                        flat[f"{key}_{i}"] = v
                else:
                    flat[key] = value
            flattened.append(flat)
        
        df = pd.DataFrame(flattened)
        self.feature_names = df.columns.tolist()
        return df
    
    def train(self, real_features: list, fake_features: list):
        """Train deepfake detection model"""
        # Prepare data
        real_df = self.prepare_features(real_features)
        fake_df = self.prepare_features(fake_features)
        
        # Combine and label
        real_df["label"] = 0  # Real
        fake_df["label"] = 1  # Fake
        
        df = pd.concat([real_df, fake_df], ignore_index=True)
        X = df.drop("label", axis=1)
        y = df["label"]
        
        # Scale features
        X_scaled = self.scaler.fit_transform(X)
        
        # Split data
        X_train, X_test, y_train, y_test = train_test_split(
            X_scaled, y, test_size=0.2, random_state=42, stratify=y
        )
        
        # Train model
        self.model = GradientBoostingClassifier(
            n_estimators=100,
            learning_rate=0.1,
            max_depth=5,
            random_state=42
        )
        self.model.fit(X_train, y_train)
        
        # Evaluate
        y_pred = self.model.predict(X_test)
        accuracy = accuracy_score(y_test, y_pred)
        
        print(f"Model accuracy: {accuracy:.3f}")
        print("\nClassification Report:")
        print(classification_report(y_test, y_pred))
        print("\nConfusion Matrix:")
        print(confusion_matrix(y_test, y_pred))
        
        return accuracy
    
    def predict(self, features: dict):
        """Predict if audio is deepfake"""
        if self.model is None:
            raise ValueError("Model not trained. Call train() first.")
        
        # Prepare features
        flat = {}
        for key, value in features.items():
            if isinstance(value, list):
                for i, v in enumerate(value):
                    flat[f"{key}_{i}"] = v
            else:
                flat[key] = value
        
        # Create DataFrame with same columns
        df = pd.DataFrame([flat])
        
        # Ensure all feature columns exist
        for col in self.feature_names:
            if col not in df.columns:
                df[col] = 0
        
        # Reorder columns
        df = df[self.feature_names]
        
        # Scale and predict
        X_scaled = self.scaler.transform(df)
        prediction = self.model.predict(X_scaled)[0]
        probability = self.model.predict_proba(X_scaled)[0]
        
        return {
            "is_deepfake": bool(prediction),
            "confidence": float(max(probability)),
            "real_probability": float(probability[0]),
            "fake_probability": float(probability[1])
        }
    
    def save(self, model_path: str):
        """Save model and scaler"""
        with open(model_path, "wb") as f:
            pickle.dump({
                "model": self.model,
                "scaler": self.scaler,
                "feature_names": self.feature_names
            }, f)
        print(f"Model saved to {model_path}")
    
    def load(self, model_path: str):
        """Load model and scaler"""
        with open(model_path, "rb") as f:
            data = pickle.load(f)
            self.model = data["model"]
            self.scaler = data["scaler"]
            self.feature_names = data["feature_names"]
        print(f"Model loaded from {model_path}")

# Example: Generate synthetic training data
# In production, use real audio files
from audio_features import AudioFeatureExtractor

extractor = AudioFeatureExtractor()

# Simulate feature extraction (replace with real audio files)
print("Deepfake detector ready")
print("Train with real and fake audio features")

import numpy as np
import cv2
from typing import List, Dict

class VideoDeepfakeDetector:
    """Detect deepfakes in video using frame analysis"""
    
    def extract_frame_features(self, frame: np.ndarray) -> Dict:
        """Extract features from a video frame"""
        # Convert to grayscale
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        
        # Face detection (simplified - in production use face detection library)
        # For now, analyze entire frame
        
        # Texture analysis
        from skimage.feature import local_binary_pattern
        lbp = local_binary_pattern(gray, 8, 1, method='uniform')
        lbp_hist, _ = np.histogram(lbp.ravel(), bins=10, range=(0, 10))
        
        # Frequency domain analysis
        fft = np.fft.fft2(gray)
        fft_magnitude = np.abs(fft)
        
        # Edge detection
        edges = cv2.Canny(gray, 50, 150)
        edge_density = np.sum(edges > 0) / edges.size
        
        return {
            "lbp_histogram": lbp_hist.tolist(),
            "fft_mean": float(np.mean(fft_magnitude)),
            "fft_std": float(np.std(fft_magnitude)),
            "edge_density": float(edge_density),
            "brightness_mean": float(np.mean(gray)),
            "brightness_std": float(np.std(gray))
        }
    
    def detect_inconsistencies(self, frames: List[np.ndarray]) -> Dict:
        """Detect inconsistencies across frames (common in deepfakes)"""
        features_list = [self.extract_frame_features(frame) for frame in frames]
        
        # Calculate frame-to-frame variations
        variations = []
        for i in range(1, len(features_list)):
            prev = features_list[i-1]
            curr = features_list[i]
            
            # Calculate difference
            diff = abs(curr["brightness_mean"] - prev["brightness_mean"])
            variations.append(diff)
        
        # Deepfakes often have inconsistent frame transitions
        variation_mean = np.mean(variations)
        variation_std = np.std(variations)
        
        # High variation suggests deepfake
        is_suspicious = variation_std > np.mean(variations) * 0.5
        
        return {
            "frame_count": len(frames),
            "variation_mean": float(variation_mean),
            "variation_std": float(variation_std),
            "is_suspicious": bool(is_suspicious),
            "suspicion_score": float(min(variation_std / (variation_mean + 1e-6), 1.0))
        }

print("Video deepfake detector ready")
print("Analyze video frames for deepfake artifacts")

#!/usr/bin/env python3
"""
Deepfake Voice Message Detection System
Complete implementation with audio processing, ML detection, and API
"""

import os
import librosa
import numpy as np
import pandas as pd
from pathlib import Path
from flask import Flask, request, jsonify, send_file
from flask_cors import CORS
from werkzeug.utils import secure_filename
import pickle
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, classification_report
import json
from datetime import datetime

app = Flask(__name__)
CORS(app)
app.config['UPLOAD_FOLDER'] = 'uploads'
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024  # 16MB max

os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)

class AudioFeatureExtractor:
    """Extract features from audio for deepfake detection"""
    
    def __init__(self, sample_rate=22050):
        self.sample_rate = sample_rate
    
    def extract_all_features(self, audio_path: str) -> dict:
        """Extract comprehensive audio features"""
        try:
            y, sr = librosa.load(audio_path, sr=self.sample_rate, duration=10)
            
            features = {}
            
            # MFCC features
            mfccs = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=13)
            features.update({
                f"mfcc_{i}_mean": float(np.mean(mfccs[i])) for i in range(13)
            })
            features.update({
                f"mfcc_{i}_std": float(np.std(mfccs[i])) for i in range(13)
            })
            
            # Spectral features
            spectral_centroids = librosa.feature.spectral_centroid(y=y, sr=sr)[0]
            spectral_rolloff = librosa.feature.spectral_rolloff(y=y, sr=sr)[0]
            zcr = librosa.feature.zero_crossing_rate(y)[0]
            
            features.update({
                "spectral_centroid_mean": float(np.mean(spectral_centroids)),
                "spectral_centroid_std": float(np.std(spectral_centroids)),
                "spectral_rolloff_mean": float(np.mean(spectral_rolloff)),
                "spectral_rolloff_std": float(np.std(spectral_rolloff)),
                "zcr_mean": float(np.mean(zcr)),
                "zcr_std": float(np.std(zcr))
            })
            
            # Prosody features
            tempo, _ = librosa.beat.beat_track(y=y, sr=sr)
            pitches, magnitudes = librosa.piptrack(y=y, sr=sr)
            
            features.update({
                "tempo": float(tempo),
                "pitch_mean": float(np.mean(pitches[pitches > 0])) if np.any(pitches > 0) else 0.0,
                "pitch_std": float(np.std(pitches[pitches > 0])) if np.any(pitches > 0) else 0.0
            })
            
            return features
        
        except Exception as e:
            raise ValueError(f"Feature extraction failed: {str(e)}")

class DeepfakeDetector:
    """ML model for deepfake audio detection"""
    
    def __init__(self):
        self.model = None
        self.scaler = StandardScaler()
        self.feature_names = []
        self.is_trained = False
    
    def train(self, real_audio_dir: str, fake_audio_dir: str):
        """Train model on real and fake audio samples"""
        extractor = AudioFeatureExtractor()
        
        # Extract features from real audio
        real_features = []
        for audio_file in Path(real_audio_dir).glob("*.wav"):
            try:
                features = extractor.extract_all_features(str(audio_file))
                real_features.append(features)
            except Exception as e:
                print(f"Error processing {audio_file}: {e}")
        
        # Extract features from fake audio
        fake_features = []
        for audio_file in Path(fake_audio_dir).glob("*.wav"):
            try:
                features = extractor.extract_all_features(str(audio_file))
                fake_features.append(features)
            except Exception as e:
                print(f"Error processing {audio_file}: {e}")
        
        if len(real_features) == 0 or len(fake_features) == 0:
            raise ValueError("Need both real and fake audio samples for training")
        
        # Prepare data
        real_df = pd.DataFrame(real_features)
        fake_df = pd.DataFrame(fake_features)
        
        real_df["label"] = 0
        fake_df["label"] = 1
        
        df = pd.concat([real_df, fake_df], ignore_index=True)
        X = df.drop("label", axis=1)
        y = df["label"]
        
        # Ensure consistent feature columns
        self.feature_names = sorted(X.columns.tolist())
        X = X[self.feature_names]
        
        # Scale features
        X_scaled = self.scaler.fit_transform(X)
        
        # Split and train
        X_train, X_test, y_train, y_test = train_test_split(
            X_scaled, y, test_size=0.2, random_state=42, stratify=y
        )
        
        self.model = GradientBoostingClassifier(
            n_estimators=100,
            learning_rate=0.1,
            max_depth=5,
            random_state=42
        )
        self.model.fit(X_train, y_train)
        
        # Evaluate
        y_pred = self.model.predict(X_test)
        accuracy = accuracy_score(y_test, y_pred)
        
        print(f"Model trained with {len(real_features)} real and {len(fake_features)} fake samples")
        print(f"Accuracy: {accuracy:.3f}")
        print(classification_report(y_test, y_pred))
        
        self.is_trained = True
        return accuracy
    
    def predict(self, audio_path: str) -> dict:
        """Predict if audio is deepfake"""
        if not self.is_trained:
            raise ValueError("Model not trained")
        
        extractor = AudioFeatureExtractor()
        features = extractor.extract_all_features(audio_path)
        
        # Prepare features
        df = pd.DataFrame([features])
        
        # Ensure all feature columns exist
        for col in self.feature_names:
            if col not in df.columns:
                df[col] = 0.0
        
        df = df[self.feature_names]
        
        # Scale and predict
        X_scaled = self.scaler.transform(df)
        prediction = self.model.predict(X_scaled)[0]
        probability = self.model.predict_proba(X_scaled)[0]
        
        return {
            "is_deepfake": bool(prediction),
            "confidence": float(max(probability)),
            "real_probability": float(probability[0]),
            "fake_probability": float(probability[1]),
            "timestamp": datetime.now().isoformat()
        }
    
    def save(self, model_path: str):
        """Save trained model"""
        with open(model_path, "wb") as f:
            pickle.dump({
                "model": self.model,
                "scaler": self.scaler,
                "feature_names": self.feature_names
            }, f)
    
    def load(self, model_path: str):
        """Load trained model"""
        with open(model_path, "rb") as f:
            data = pickle.load(f)
            self.model = data["model"]
            self.scaler = data["scaler"]
            self.feature_names = data["feature_names"]
            self.is_trained = True

# Initialize detector
detector = DeepfakeDetector()

# API Routes
@app.route('/health', methods=['GET'])
def health():
    return jsonify({"status": "healthy", "model_trained": detector.is_trained})

@app.route('/predict', methods=['POST'])
def predict():
    """Predict if uploaded audio is deepfake"""
    if 'file' not in request.files:
        return jsonify({"error": "No file provided"}), 400
    
    file = request.files['file']
    if file.filename == '':
        return jsonify({"error": "No file selected"}), 400
    
    if not detector.is_trained:
        return jsonify({"error": "Model not trained"}), 503
    
    try:
        filename = secure_filename(file.filename)
        filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename)
        file.save(filepath)
        
        result = detector.predict(filepath)
        
        # Cleanup
        os.remove(filepath)
        
        return jsonify(result)
    
    except Exception as e:
        return jsonify({"error": str(e)}), 500

@app.route('/batch', methods=['POST'])
def batch_predict():
    """Batch process multiple audio files"""
    if 'files' not in request.files:
        return jsonify({"error": "No files provided"}), 400
    
    files = request.files.getlist('files')
    if not detector.is_trained:
        return jsonify({"error": "Model not trained"}), 503
    
    results = []
    for file in files:
        try:
            filename = secure_filename(file.filename)
            filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename)
            file.save(filepath)
            
            result = detector.predict(filepath)
            result["filename"] = filename
            results.append(result)
            
            os.remove(filepath)
        
        except Exception as e:
            results.append({
                "filename": file.filename,
                "error": str(e)
            })
    
    return jsonify({"results": results})

if __name__ == '__main__':
    # For demonstration, create a simple model
    # In production, train on real datasets
    print("Deepfake Voice Detection API")
    print("Train model with: detector.train('real_audio/', 'fake_audio/')")
    print("Or load existing: detector.load('model.pkl')")
    print("Starting API server on http://localhost:5000")
    app.run(debug=True, host='0.0.0.0', port=5000)

deactivate || true
rm -rf .venv-deepfake *.py *.pkl uploads/ __pycache__/