Deepfake 防御 2026：AI 换脸检测技术全景

2017          2020          2023          2025          2026
 │             │             │             │             │
DeepFaceLab → First Order → Diffusion  → Real-time  → 物理一致
              Motion        based        Deepfake     Deepfake
                             Transfer                  (光照/阴影
                                                      /运动一致)

Deepfake 检测
├── 视觉伪影检测
│   ├── 面部边界检测
│   ├── 纹理不一致检测
│   └── 光照不一致检测
├── 时空一致性检测
│   ├── 帧间一致性
│   ├── 运动模式分析
│   └── 时序异常检测
├── 生物特征检测
│   ├── 心跳信号检测
│   ├── 眨眼模式分析
│   └── 微表情分析
├── 频域检测
│   ├── 频谱分析
│   ├── GAN指纹检测
│   └── 生成模型指纹
└── 多模态检测
    ├── 音视频一致性
    ├── 唇形同步检测
    └── 语义一致性

import cv2
import numpy as np
import torch
import torch.nn as nn

class FacialBoundaryDetector:
    """面部边界伪影检测——Deepfake常见缺陷"""
    
    def __init__(self):
        self.face_cascade = cv2.CascadeClassifier(
            cv2.data.haarcascades + 'haarcascade_frontalface_default.xml'
        )
        self.edge_detector = cv2.Canny
    
    def detect(self, image: np.ndarray) -> dict:
        """检测面部边界伪影"""
        # 1. 人脸检测
        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
        faces = self.face_cascade.detectMultiScale(gray, 1.3, 5)
        
        if len(faces) == 0:
            return {'detected': False, 'reason': 'no_face'}
        
        results = []
        for (x, y, w, h) in faces:
            # 2. 提取面部区域和边界区域
            face_region = image[y:y+h, x:x+w]
            border_region = self._extract_border(image, x, y, w, h)
            
            # 3. 分析边界特征
            face_edges = cv2.Canny(face_region, 50, 150)
            border_edges = cv2.Canny(border_region, 50, 150)
            
            # 4. 计算边界一致性
            face_edge_density = np.sum(face_edges > 0) / face_edges.size
            border_edge_density = np.sum(border_edges > 0) / border_edges.size
            
            # Deepfake通常在面部边界处有不自然的边缘
            edge_discontinuity = abs(face_edge_density - border_edge_density)
            
            # 5. 颜色过渡分析
            color_transition = self._analyze_color_transition(
                face_region, border_region
            )
            
            results.append({
                'face_position': (x, y, w, h),
                'edge_discontinuity': edge_discontinuity,
                'color_transition_score': color_transition,
                'is_suspicious': edge_discontinuity > 0.15 or 
                                color_transition > 0.3
            })
        
        return {
            'detected': True,
            'faces': results,
            'deepfake_probability': np.mean([
                r['edge_discontinuity'] + r['color_transition_score'] 
                for r in results
            ])
        }
    
    def _extract_border(self, image, x, y, w, h, border_width=10):
        """提取面部边界区域"""
        h_img, w_img = image.shape[:2]
        x1 = max(0, x - border_width)
        y1 = max(0, y - border_width)
        x2 = min(w_img, x + w + border_width)
        y2 = min(h_img, y + h + border_width)
        return image[y1:y2, x1:x2]
    
    def _analyze_color_transition(self, face, border):
        """分析颜色过渡"""
        face_mean = np.mean(face, axis=(0, 1))
        border_mean = np.mean(border, axis=(0, 1))
        return np.linalg.norm(face_mean - border_mean) / 255

class DeepfakeDetectionModel(nn.Module):
    """基于 EfficientNet 的 Deepfake 检测模型"""
    
    def __init__(self, pretrained=True):
        super().__init__()
        # 使用预训练的 EfficientNet 作为骨干
        from torchvision.models import efficientnet_v2_s
        self.backbone = efficientnet_v2_s(pretrained=pretrained)
        
        # 替换分类头
        in_features = self.backbone.classifier[1].in_features
        self.backbone.classifier = nn.Sequential(
            nn.Dropout(0.3),
            nn.Linear(in_features, 512),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(512, 1),
            nn.Sigmoid()
        )
    
    def forward(self, x):
        return self.backbone(x)


class MultiScaleDeepfakeDetector:
    """多尺度 Deepfake 检测器"""
    
    def __init__(self):
        self.models = {
            'face_level': DeepfakeDetectionModel(),   # 面部级别
            'patch_level': DeepfakeDetectionModel(),   # 局部区域
            'frame_level': DeepfakeDetectionModel(),   # 整帧级别
        }
    
    def detect(self, video_path: str) -> dict:
        """检测视频中的 Deepfake"""
        cap = cv2.VideoCapture(video_path)
        frame_results = []
        
        frame_count = 0
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
            
            if frame_count % 5 == 0:  # 每5帧检测一次
                # 多尺度检测
                face_pred = self._detect_face_level(frame)
                patch_pred = self._detect_patch_level(frame)
                frame_pred = self._detect_frame_level(frame)
                
                # 集成结果
                combined_pred = np.mean([
                    face_pred, patch_pred, frame_pred
                ])
                
                frame_results.append({
                    'frame': frame_count,
                    'face_level': face_pred,
                    'patch_level': patch_pred,
                    'frame_level': frame_pred,
                    'combined': combined_pred
                })
            
            frame_count += 1
        
        cap.release()
        
        # 汇总结果
        avg_score = np.mean([r['combined'] for r in frame_results])
        max_score = np.max([r['combined'] for r in frame_results])
        
        return {
            'is_deepfake': avg_score > 0.5,
            'confidence': avg_score,
            'max_frame_score': max_score,
            'frames_analyzed': len(frame_results),
            'details': frame_results
        }

class TemporalConsistencyDetector:
    """时序一致性检测——分析帧间一致性"""
    
    def detect(self, video_path: str) -> dict:
        cap = cv2.VideoCapture(video_path)
        
        prev_frame = None
        inconsistencies = []
        flow_magnitudes = []
        
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
            
            gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
            
            if prev_frame is not None:
                # 1. 光流分析
                flow = cv2.calcOpticalFlowFarneback(
                    prev_frame, gray, None, 0.5, 3, 15, 3, 5, 1.2, 0
                )
                magnitude = np.sqrt(flow[..., 0]**2 + flow[..., 1]**2)
                flow_magnitudes.append(np.mean(magnitude))
                
                # 2. 检测光流异常（Deepfake 帧间可能有不自然的运动）
                flow_std = np.std(magnitude)
                if flow_std > 10:  # 异常高的光流方差
                    inconsistencies.append({
                        'frame': len(flow_magnitudes),
                        'flow_std': flow_std,
                        'type': 'motion_inconsistency'
                    })
                
                # 3. 人脸区域光流 vs 背景光流
                face_flow = self._get_face_region_flow(magnitude, frame)
                bg_flow = self._get_background_flow(magnitude, frame)
                
                if face_flow > 0 and bg_flow > 0:
                    ratio = face_flow / bg_flow
                    # 真实视频中人脸和背景运动比例应在合理范围
                    if ratio > 5 or ratio < 0.1:
                        inconsistencies.append({
                            'frame': len(flow_magnitudes),
                            'face_bg_ratio': ratio,
                            'type': 'motion_ratio_anomaly'
                        })
            
            prev_frame = gray
        
        cap.release()
        
        return {
            'total_inconsistencies': len(inconsistencies),
            'inconsistency_rate': len(inconsistencies) / max(len(flow_magnitudes), 1),
            'details': inconsistencies[:20],
            'is_suspicious': len(inconsistencies) > len(flow_magnitudes) * 0.1
        }

class rPPGDetector:
    """远程光电容积脉搏波（rPPG）检测
    真实人脸有微小的肤色变化（心跳引起），
    Deepfake 通常无法复制这一信号"""
    
    def detect(self, video_path: str) -> dict:
        cap = cv2.VideoCapture(video_path)
        
        face_signals = []
        frame_count = 0
        
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
            
            # 1. 提取面部区域
            face = self._extract_face(frame)
            if face is None:
                frame_count += 1
                continue
            
            # 2. 计算绿色通道均值（rPPG信号主要在绿色通道）
            green_mean = np.mean(face[:, :, 1])
            face_signals.append(green_mean)
            frame_count += 1
        
        cap.release()
        
        if len(face_signals) < 30:
            return {'detected': False, 'reason': 'insufficient_frames'}
        
        # 3. 信号处理
        signal = np.array(face_signals)
        signal = signal - np.mean(signal)  # 去直流
        signal = signal / (np.std(signal) + 1e-8)  # 归一化
        
        # 4. 带通滤波（0.7-3Hz = 42-180 BPM）
        from scipy.signal import butter, filtfilt
        b, a = butter(3, [0.7, 3.0], btype='band', fs=30)
        filtered = filtfilt(b, a, signal)
        
        # 5. FFT 分析心跳频率
        fft = np.fft.rfft(filtered)
        freqs = np.fft.rfftfreq(len(filtered), 1/30)
        power = np.abs(fft)**2
        
        # 心跳频率应在 0.7-3Hz 范围内有明显峰值
        heart_rate_mask = (freqs >= 0.7) & (freqs <= 3.0)
        heart_rate_power = np.sum(power[heart_rate_mask])
        total_power = np.sum(power)
        
        snr = heart_rate_power / max(total_power - heart_rate_power, 1e-8)
        
        # 6. 判断
        is_real = snr > 2.0  # SNR > 2dB 表示有真实心跳信号
        
        return {
            'is_real': is_real,
            'heartbeat_snr': snr,
            'estimated_bpm': freqs[np.argmax(power[heart_rate_mask])] * 60,
            'confidence': min(1.0, snr / 10),
            'signal_quality': 'good' if snr > 5 else 'medium' if snr > 2 else 'poor'
        }
    
    def _extract_face(self, frame):
        """提取面部区域"""
        face_cascade = cv2.CascadeClassifier(
            cv2.data.haarcascades + 'haarcascade_frontalface_default.xml'
        )
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        faces = face_cascade.detectMultiScale(gray, 1.3, 5)
        
        if len(faces) == 0:
            return None
        
        x, y, w, h = faces[0]
        # 取面部中心区域（避开边界）
        margin = int(w * 0.2)
        return frame[y+margin:y+h-margin, x+margin:x+w-margin]

class BlinkDetector:
    """眨眼模式检测——Deepfake 眨眼频率通常异常"""
    
    def __init__(self):
        # 使用 dlib 的面部关键点
        import dlib
        self.detector = dlib.get_frontal_face_detector()
        self.predictor = dlib.shape_predictor('shape_predictor_68.dat')
    
    def detect(self, video_path: str) -> dict:
        cap = cv2.VideoCapture(video_path)
        
        ear_values = []  # Eye Aspect Ratio
        
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
            
            gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
            faces = self.detector(gray)
            
            for face in faces:
                landmarks = self.predictor(gray, face)
                
                # 计算眼睛纵横比（EAR）
                left_ear = self._calculate_ear(landmarks, 'left')
                right_ear = self._calculate_ear(landmarks, 'right')
                
                avg_ear = (left_ear + right_ear) / 2
                ear_values.append(avg_ear)
        
        cap.release()
        
        if len(ear_values) < 30:
            return {'detected': False, 'reason': 'insufficient_frames'}
        
        # 分析眨眼模式
        ear_array = np.array(ear_values)
        blinks = self._count_blinks(ear_array)
        
        # 正常人眨眼频率：10-20次/分钟
        duration_min = len(ear_values) / (30 * 60)  # 假设30fps
        blink_rate = blinks / max(duration_min, 0.01)
        
        # Deepfake 通常眨眼频率异常
        is_suspicious = blink_rate < 3 or blink_rate > 40
        
        return {
            'blink_count': blinks,
            'blink_rate_per_min': blink_rate,
            'is_suspicious': is_suspicious,
            'reason': 'abnormal_blink_rate' if is_suspicious else 'normal',
            'confidence': 0.8 if is_suspicious else 0.6
        }
    
    def _calculate_ear(self, landmarks, side: str) -> float:
        """计算眼睛纵横比"""
        if side == 'left':
            p1, p2, p3, p4, p5, p6 = 36, 37, 38, 39, 40, 41
        else:
            p1, p2, p3, p4, p5, p6 = 42, 43, 44, 45, 46, 47
        
        pts = [(landmarks.part(i).x, landmarks.part(i).y) for i in [p1,p2,p3,p4,p5,p6]]
        
        # EAR = (|p2-p6| + |p3-p5|) / (2 * |p1-p4|)
        vertical = (np.linalg.norm(np.array(pts[1]) - np.array(pts[5])) +
                   np.linalg.norm(np.array(pts[2]) - np.array(pts[4])))
        horizontal = np.linalg.norm(np.array(pts[0]) - np.array(pts[3]))
        
        return vertical / (2 * horizontal) if horizontal > 0 else 0
    
    def _count_blinks(self, ear_array: np.ndarray) -> int:
        """计算眨眼次数"""
        threshold = 0.25
        blinks = 0
        below = False
        
        for ear in ear_array:
            if ear < threshold and not below:
                below = True
            elif ear >= threshold and below:
                blinks += 1
                below = False
        
        return blinks

class AudioVisualConsistencyDetector:
    """音视频一致性检测"""
    
    def detect(self, video_path: str) -> dict:
        """检测音视频是否一致"""
        # 1. 提取音频
        audio = self._extract_audio(video_path)
        
        # 2. 提取面部运动
        lip_movements = self._extract_lip_movements(video_path)
        
        # 3. 提取语音特征
        speech_envelope = self._extract_speech_envelope(audio)
        
        # 4. 计算唇形-语音相关性
        correlation = self._compute_correlation(
            lip_movements, speech_envelope
        )
        
        # Deepfake 通常唇形与语音不同步
        is_suspicious = correlation < 0.3
        
        return {
            'lip_speech_correlation': correlation,
            'is_suspicious': is_suspicious,
            'confidence': 1 - correlation,
            'lip_sync_quality': 'good' if correlation > 0.6 
                              else 'medium' if correlation > 0.3 
                              else 'poor'
        }

class DeepfakeDefenseSystem:
    """Deepfake 集成防御系统"""
    
    def __init__(self):
        self.detectors = {
            'visual': FacialBoundaryDetector(),
            'temporal': TemporalConsistencyDetector(),
            'rppg': rPPGDetector(),
            'blink': BlinkDetector(),
            'av_sync': AudioVisualConsistencyDetector(),
            'deep_learning': MultiScaleDeepfakeDetector(),
        }
    
    def analyze(self, video_path: str) -> dict:
        """全面分析"""
        results = {}
        scores = []
        weights = {
            'deep_learning': 0.35,
            'rppg': 0.20,
            'av_sync': 0.15,
            'temporal': 0.15,
            'visual': 0.10,
            'blink': 0.05,
        }
        
        for name, detector in self.detectors.items():
            try:
                result = detector.detect(video_path)
                results[name] = result
                
                # 提取风险分数
                if name == 'deep_learning':
                    scores.append((result['confidence'], weights[name]))
                elif name == 'rppg':
                    scores.append((1 - result.get('confidence', 0), weights[name]))
                elif name == 'blink':
                    scores.append((result.get('confidence', 0), weights[name]))
                else:
                    scores.append((result.get('confidence', 0.5), weights[name]))
            except Exception as e:
                results[name] = {'error': str(e)}
        
        # 加权综合
        final_score = sum(score * weight for score, weight in scores)
        
        return {
            'is_deepfake': final_score > 0.5,
            'confidence': final_score,
            'risk_level': 'high' if final_score > 0.7
                         else 'medium' if final_score > 0.5
                         else 'low',
            'detector_results': results,
            'recommendation': self._recommendation(final_score)
        }
    
    def _recommendation(self, score: float) -> str:
        if score > 0.8:
            return "极可能是Deepfake，建议拒绝"
        elif score > 0.6:
            return "疑似Deepfake，建议人工复核"
        elif score > 0.4:
            return "存在风险，建议进一步验证"
        else:
            return "未检测到Deepfake特征"