097_数字取证高级技术：AI生成内容检测与深度伪造识别实战指南——从特征提取到模型对抗的全面分析方法

import nltk
from collections import Counter
import numpy as np
from scipy import stats

def analyze_text_statistics(text):
    # 分词
    tokens = nltk.word_tokenize(text)
    unique_tokens = set(tokens)
    
    # 计算统计特征
    stats_results = {
        '词汇多样性': len(unique_tokens) / len(tokens),
        '平均词长': np.mean([len(token) for token in tokens]),
        '高频词比例': len([t for t in tokens if t.lower() in nltk.corpus.stopwords.words('english')]) / len(tokens),
        '长句比例': len([s for s in nltk.sent_tokenize(text) if len(s.split()) > 20]) / len(nltk.sent_tokenize(text)),
        '罕见词频率': len([t for t in tokens if len(t) > 10 and t not in nltk.corpus.stopwords.words('english')]) / len(tokens)
    }
    
    # 计算词频分布的熵
    word_counts = Counter(tokens)
    probabilities = [count/len(tokens) for count in word_counts.values()]
    stats_results['词频熵'] = -sum(p * np.log2(p) for p in probabilities)
    
    # 计算句子长度分布的偏度
    sentence_lengths = [len(s.split()) for s in nltk.sent_tokenize(text)]
    stats_results['句子长度偏度'] = stats.skew(sentence_lengths)
    
    return stats_results

import spacy
import networkx as nx

# 加载spaCy模型
nlp = spacy.load('en_core_web_lg')

def analyze_syntax_structure(text):
    doc = nlp(text)
    
    # 计算句法复杂度指标
    sentences = list(doc.sents)
    syntax_features = {
        '平均依存深度': 0,
        '平均句法树高度': 0,
        '并列结构频率': 0,
        '从句密度': 0,
        '被动语态比例': 0
    }
    
    for sent in sentences:
        # 构建依存句法图
        G = nx.DiGraph()
        for token in sent:
            G.add_node(token.i)
            if token.head.i != token.i:
                G.add_edge(token.head.i, token.i)
        
        # 计算依存深度
        try:
            depths = [nx.shortest_path_length(G, source=sent.root.i, target=node) for node in G.nodes()]
            syntax_features['平均依存深度'] += np.mean(depths)
        except:
            pass
        
        # 计算句法树高度
        try:
            height = nx.dag_longest_path_length(G)
            syntax_features['平均句法树高度'] += height
        except:
            pass
        
        # 计算并列结构
        syntax_features['并列结构频率'] += len([token for token in sent if token.dep_ == 'cc'])
        
        # 计算从句密度
        syntax_features['从句密度'] += len([token for token in sent if token.dep_ == 'mark'])
        
        # 计算被动语态
        syntax_features['被动语态比例'] += len([token for token in sent if token.tag_ in ['VBN', 'VBD'] and token.dep_ == 'auxpass'])
    
    # 平均化特征
    if len(sentences) > 0:
        for key in syntax_features:
            syntax_features[key] /= len(sentences)
    
    return syntax_features

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, accuracy_score
import pandas as pd

# 假设我们有标记数据集
def train_text_classifier(real_texts, ai_texts):
    # 准备数据集
    texts = real_texts + ai_texts
    labels = [0] * len(real_texts) + [1] * len(ai_texts)
    
    # 划分训练集和测试集
    X_train, X_test, y_train, y_test = train_test_split(texts, labels, test_size=0.2, random_state=42)
    
    # 特征提取
    vectorizer = TfidfVectorizer(max_features=10000, ngram_range=(1, 3))
    X_train_features = vectorizer.fit_transform(X_train)
    X_test_features = vectorizer.transform(X_test)
    
    # 训练分类器
    clf = RandomForestClassifier(n_estimators=200, random_state=42)
    clf.fit(X_train_features, y_train)
    
    # 评估模型
    y_pred = clf.predict(X_test_features)
    print("准确率:", accuracy_score(y_test, y_pred))
    print("详细报告:")
    print(classification_report(y_test, y_pred))
    
    return clf, vectorizer

# 使用模型预测新文本
def detect_ai_text(text, clf, vectorizer):
    features = vectorizer.transform([text])
    prediction = clf.predict(features)[0]
    confidence = clf.predict_proba(features)[0].max()
    return {
        'is_ai_generated': bool(prediction),
        'confidence': confidence
    }

import torch
from transformers import AutoModelForSequenceClassification, AutoTokenizer

def load_bert_detector(model_name="distilbert-base-uncased-finetuned-sst-2-english"):
    # 加载预训练的BERT模型
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    model = AutoModelForSequenceClassification.from_pretrained(model_name)
    return tokenizer, model

def bert_based_detection(text, tokenizer, model):
    # 标记化输入文本
    inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
    
    # 获取模型预测
    with torch.no_grad():
        outputs = model(**inputs)
        logits = outputs.logits
    
    # 计算概率
    probabilities = torch.nn.functional.softmax(logits, dim=-1)
    
    # 确定是AI生成还是人类生成
    # 这里假设索引1表示AI生成（需要根据具体训练模型调整）
    ai_probability = probabilities[0][1].item()
    
    return {
        'is_ai_generated': ai_probability > 0.5,
        'ai_probability': ai_probability
    }

import torch
from transformers import AutoModel, AutoTokenizer

def extract_transformer_features(texts, model_name="bert-base-uncased", layer=-2):
    # 加载模型和分词器
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    model = AutoModel.from_pretrained(model_name)
    model.eval()
    
    features_list = []
    
    for text in texts:
        # 标记化
        inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
        
        # 获取隐藏状态
        with torch.no_grad():
            outputs = model(**inputs, output_hidden_states=True)
            # 使用指定层的隐藏状态
            hidden_states = outputs.hidden_states[layer]
        
        # 聚合特征
        # 1. 平均池化
        mean_features = hidden_states.mean(dim=1).squeeze().numpy()
        # 2. 最大池化
        max_features = hidden_states.max(dim=1).values.squeeze().numpy()
        # 3. 使用[CLS]标记
        cls_features = hidden_states[:, 0, :].squeeze().numpy()
        
        # 合并特征
        combined_features = np.concatenate([mean_features, max_features, cls_features])
        features_list.append(combined_features)
    
    return np.array(features_list)

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, Dataset

# 对比学习数据集
class ContrastiveTextDataset(Dataset):
    def __init__(self, real_texts, ai_texts, tokenizer, max_length=512):
        self.tokenizer = tokenizer
        self.max_length = max_length
        # 创建正负样本对
        self.pairs = []
        
        # 正样本对（都为人类或都为AI）
        for i in range(len(real_texts) - 1):
            self.pairs.append((real_texts[i], real_texts[i+1], 1))
        for i in range(len(ai_texts) - 1):
            self.pairs.append((ai_texts[i], ai_texts[i+1], 1))
        
        # 负样本对（一个人类，一个AI）
        min_len = min(len(real_texts), len(ai_texts))
        for i in range(min_len):
            self.pairs.append((real_texts[i], ai_texts[i], 0))
    
    def __len__(self):
        return len(self.pairs)
    
    def __getitem__(self, idx):
        text1, text2, label = self.pairs[idx]
        
        # 标记化
        encoded1 = self.tokenizer(text1, truncation=True, max_length=self.max_length, return_tensors='pt')
        encoded2 = self.tokenizer(text2, truncation=True, max_length=self.max_length, return_tensors='pt')
        
        return {
            'input_ids1': encoded1['input_ids'].squeeze(),
            'attention_mask1': encoded1['attention_mask'].squeeze(),
            'input_ids2': encoded2['input_ids'].squeeze(),
            'attention_mask2': encoded2['attention_mask'].squeeze(),
            'label': torch.tensor(label, dtype=torch.float)
        }

# 对比学习模型
class ContrastiveTextDetector(nn.Module):
    def __init__(self, base_model):
        super().__init__()
        self.base_model = base_model
        self.projection = nn.Sequential(
            nn.Linear(768, 256),
            nn.ReLU(),
            nn.Linear(256, 128)
        )
        self.classifier = nn.Linear(128*2, 1)
    
    def forward(self, input_ids1, attention_mask1, input_ids2, attention_mask2):
        # 获取第一个文本的特征
        outputs1 = self.base_model(input_ids1, attention_mask=attention_mask1)
        features1 = outputs1.last_hidden_state.mean(dim=1)
        projected1 = self.projection(features1)
        
        # 获取第二个文本的特征
        outputs2 = self.base_model(input_ids2, attention_mask=attention_mask2)
        features2 = outputs2.last_hidden_state.mean(dim=1)
        projected2 = self.projection(features2)
        
        # 合并特征并分类
        combined = torch.cat([projected1, projected2], dim=1)
        logits = self.classifier(combined).squeeze()
        
        return logits

import cv2
import numpy as np
from mediapipe import solutions
from mediapipe.framework.formats import landmark_pb2

def detect_anatomical_anomalies(image_path):
    # 加载图像
    image = cv2.imread(image_path)
    image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
    
    # 初始化MediaPipe手部检测
    mp_hands = solutions.hands
    hands = mp_hands.Hands(static_image_mode=True, max_num_hands=2)
    
    # 检测手部
    results = hands.process(image_rgb)
    anomalies = {
        'hand_count': 0,
        'finger_anomalies': [],
        'anatomical_issues': []
    }
    
    if results.multi_hand_landmarks:
        anomalies['hand_count'] = len(results.multi_hand_landmarks)
        
        for hand_idx, hand_landmarks in enumerate(results.multi_hand_landmarks):
            # 计算手指数量
            fingers = detect_fingers(hand_landmarks)
            if len(fingers) not in [4, 5]:  # 4或5个手指为正常
                anomalies['finger_anomalies'].append({
                    'hand_index': hand_idx,
                    'finger_count': len(fingers),
                    'confidence': '高'
                })
            
            # 检查手指比例
            finger_lengths = measure_finger_lengths(hand_landmarks)
            length_ratios = calculate_finger_ratios(finger_lengths)
            
            # 检测异常比例
            if check_abnormal_ratios(length_ratios):
                anomalies['anatomical_issues'].append({
                    'hand_index': hand_idx,
                    'issue': '手指比例异常',
                    'ratios': length_ratios
                })
    
    # 检查面部异常
    face_anomalies = detect_face_anomalies(image_rgb)
    if face_anomalies:
        anomalies['face_issues'] = face_anomalies
    
    return anomalies

def detect_fingers(hand_landmarks):
    # 简化的手指检测逻辑
    # 实际实现需要更复杂的算法
    fingers = []
    # 这里仅作为示例框架
    return fingers

def measure_finger_lengths(hand_landmarks):
    # 测量手指长度
    lengths = []
    # 实际实现需要计算骨骼长度
    return lengths

def calculate_finger_ratios(finger_lengths):
    # 计算手指比例
    ratios = []
    # 实际实现需要计算各手指间的比例
    return ratios

def check_abnormal_ratios(ratios, threshold=0.3):
    # 检查比例是否异常
    # 与平均比例相差超过阈值则认为异常
    return False  # 示例返回

def detect_face_anomalies(image_rgb):
    # 面部异常检测
    # 使用面部关键点分析眼睛、鼻子、嘴巴比例
    return None  # 示例返回

import cv2
import numpy as np
from scipy.ndimage import gaussian_filter

def analyze_lighting_consistency(image_path):
    # 加载图像
    image = cv2.imread(image_path)
    gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
    
    # 计算梯度来检测边缘和阴影
    grad_x = cv2.Sobel(gray, cv2.CV_64F, 1, 0, ksize=3)
    grad_y = cv2.Sobel(gray, cv2.CV_64F, 0, 1, ksize=3)
    gradient_magnitude = np.sqrt(grad_x**2 + grad_y**2)
    
    # 计算光照方向
    light_direction = estimate_light_direction(gray)
    
    # 检测阴影不一致
    shadow_regions = detect_shadow_regions(gray)
    shadow_consistency = check_shadow_consistency(shadow_regions, light_direction)
    
    # 检测反射异常
    reflection_analysis = analyze_reflections(image)
    
    # 综合分析
    lighting_analysis = {
        'light_direction': light_direction,
        'shadow_consistency': shadow_consistency,
        'reflection_issues': reflection_analysis,
        'confidence': calculate_confidence(shadow_consistency, reflection_analysis)
    }
    
    return lighting_analysis

def estimate_light_direction(gray_image, blur_size=5):
    # 高斯模糊以减少噪点
    blurred = gaussian_filter(gray_image, blur_size)
    
    # 使用梯度信息估计光源方向
    # 实际实现需要更复杂的算法
    return {'azimuth': 0.0, 'elevation': 0.0}  # 示例返回

def detect_shadow_regions(gray_image, threshold=0.3):
    # 检测图像中的阴影区域
    # 实际实现需要基于物理光照模型
    return []  # 示例返回

def check_shadow_consistency(shadow_regions, light_direction):
    # 检查阴影是否与光源方向一致
    # 实际实现需要验证阴影投射方向
    return True  # 示例返回

def analyze_reflections(image):
    # 分析图像中的反射区域
    # 检测不自然的高光
    return []  # 示例返回

def calculate_confidence(shadow_consistency, reflection_analysis):
    # 计算分析结果的置信度
    return 0.5  # 示例返回

import cv2
import numpy as np
import matplotlib.pyplot as plt

def analyze_frequency_domain(image_path):
    # 加载图像
    image = cv2.imread(image_path, cv2.IMREAD_GRAYSCALE)
    
    # 计算傅里叶变换
    dft = cv2.dft(np.float32(image), flags=cv2.DFT_COMPLEX_OUTPUT)
    dft_shift = np.fft.fftshift(dft)
    magnitude_spectrum = 20 * np.log(cv2.magnitude(dft_shift[:, :, 0], dft_shift[:, :, 1]) + 1)
    
    # 分析频率分布特征
    frequency_features = {
        'low_frequency_energy': calculate_low_freq_energy(magnitude_spectrum),
        'high_frequency_energy': calculate_high_freq_energy(magnitude_spectrum),
        'frequency_decay': analyze_frequency_decay(magnitude_spectrum),
        'spectral_residual': compute_spectral_residual(image)
    }
    
    # 检测频率异常模式
    anomalies = detect_frequency_anomalies(frequency_features)
    
    return {
        'frequency_features': frequency_features,
        'anomalies': anomalies,
        'confidence': calculate_detection_confidence(anomalies)
    }

def calculate_low_freq_energy(magnitude_spectrum, ratio=0.1):
    # 计算低频能量
    h, w = magnitude_spectrum.shape
    center_h, center_w = h // 2, w // 2
    radius = int(min(h, w) * ratio)
    
    # 创建低频区域掩码
    y, x = np.ogrid[-center_h:h-center_h, -center_w:w-center_w]
    mask = x*x + y*y <= radius*radius
    
    # 计算区域内能量
    return np.sum(magnitude_spectrum[mask])

def calculate_high_freq_energy(magnitude_spectrum, ratio=0.1):
    # 计算高频能量
    h, w = magnitude_spectrum.shape
    center_h, center_w = h // 2, w // 2
    radius = int(min(h, w) * ratio)
    
    # 创建高频区域掩码（排除低频区域和中心）
    y, x = np.ogrid[-center_h:h-center_h, -center_w:w-center_w]
    mask = x*x + y*y > radius*radius
    
    # 计算区域内能量
    return np.sum(magnitude_spectrum[mask])

def analyze_frequency_decay(magnitude_spectrum):
    # 分析频率衰减特性
    # 实际实现需要拟合频率衰减曲线
    return 0.5  # 示例返回

def compute_spectral_residual(image):
    # 计算谱残差
    # 用于检测显著性区域
    return 0.0  # 示例返回

def detect_frequency_anomalies(frequency_features):
    # 检测频率异常模式
    anomalies = []
    
    # 检查高频能量是否异常低
    if frequency_features['high_frequency_energy'] < threshold_high_freq:
        anomalies.append({
            'type': 'low_high_frequency',
            'description': '高频能量异常低，可能为AI生成',
            'severity': '高'
        })
    
    return anomalies

def calculate_detection_confidence(anomalies):
    # 计算检测置信度
    if not anomalies:
        return 0.0
    return 0.8  # 示例返回

import pywt
import cv2
import numpy as np

def analyze_wavelet_features(image_path):
    # 加载图像
    image = cv2.imread(image_path, cv2.IMREAD_GRAYSCALE)
    
    # 执行小波变换（多级）
    coeffs = pywt.wavedec2(image, 'db4', level=3)
    cA3, (cH3, cV3, cD3), (cH2, cV2, cD2), (cH1, cV1, cD1) = coeffs
    
    # 分析小波系数特征
    wavelet_features = {
        'approximation_statistics': analyze_coeff_statistics(cA3),
        'detail_statistics': {
            'level3': {
                'horizontal': analyze_coeff_statistics(cH3),
                'vertical': analyze_coeff_statistics(cV3),
                'diagonal': analyze_coeff_statistics(cD3)
            },
            'level2': {
                'horizontal': analyze_coeff_statistics(cH2),
                'vertical': analyze_coeff_statistics(cV2),
                'diagonal': analyze_coeff_statistics(cD2)
            },
            'level1': {
                'horizontal': analyze_coeff_statistics(cH1),
                'vertical': analyze_coeff_statistics(cV1),
                'diagonal': analyze_coeff_statistics(cD1)
            }
        },
        'cross_scale_relations': analyze_cross_scale_relations(coeffs)
    }
    
    # 检测异常模式
    anomalies = detect_wavelet_anomalies(wavelet_features)
    
    return {
        'wavelet_features': wavelet_features,
        'anomalies': anomalies,
        'confidence': calculate_wavelet_confidence(anomalies)
    }

def analyze_coeff_statistics(coeffs):
    # 分析小波系数统计特征
    return {
        'mean': np.mean(coeffs),
        'std': np.std(coeffs),
        'skewness': calculate_skewness(coeffs),
        'kurtosis': calculate_kurtosis(coeffs),
        'energy': np.sum(coeffs**2),
        'sparsity': calculate_sparsity(coeffs)
    }

def calculate_skewness(data):
    # 计算偏度
    return 0.0  # 示例返回

def calculate_kurtosis(data):
    # 计算峰度
    return 0.0  # 示例返回

def calculate_sparsity(data, threshold=1e-5):
    # 计算稀疏度（接近零的系数比例）
    return 0.0  # 示例返回

def analyze_cross_scale_relations(coeffs):
    # 分析跨尺度关系
    # 实际实现需要计算不同尺度间的相关性
    return {}  # 示例返回

def detect_wavelet_anomalies(wavelet_features):
    # 检测小波特征异常
    anomalies = []
    
    # 检查高频细节系数的稀疏度
    level1_details = wavelet_features['detail_statistics']['level1']
    avg_sparsity = np.mean([level1_details[orientation]['sparsity'] 
                          for orientation in ['horizontal', 'vertical', 'diagonal']])
    
    if avg_sparsity < threshold_sparsity:
        anomalies.append({
            'type': 'low_detail_sparsity',
            'description': '高频细节系数稀疏度过低，可能为AI生成',
            'severity': '中'
        })
    
    return anomalies

def calculate_wavelet_confidence(anomalies):
    # 计算小波分析的置信度
    if not anomalies:
        return 0.0
    return 0.7  # 示例返回

import tensorflow as tf
from tensorflow.keras.applications import EfficientNetB0
from tensorflow.keras.layers import Dense, GlobalAveragePooling2D, Dropout
from tensorflow.keras.models import Model
from tensorflow.keras.preprocessing.image import ImageDataGenerator

def build_image_detector(input_shape=(224, 224, 3), num_classes=2):
    # 使用预训练的EfficientNetB0作为基础模型
    base_model = EfficientNetB0(include_top=False, weights='imagenet', input_shape=input_shape)
    
    # 冻结基础模型的层
    base_model.trainable = False
    
    # 添加自定义分类层
    x = base_model.output
    x = GlobalAveragePooling2D()(x)
    x = Dense(1024, activation='relu')(x)
    x = Dropout(0.5)(x)
    x = Dense(512, activation='relu')(x)
    x = Dropout(0.3)(x)
    predictions = Dense(num_classes, activation='softmax')(x)
    
    # 构建完整模型
    model = Model(inputs=base_model.input, outputs=predictions)
    
    # 编译模型
    model.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=0.0001),
                 loss='categorical_crossentropy',
                 metrics=['accuracy'])
    
    return model

def train_image_detector(model, train_dir, val_dir, batch_size=32, epochs=20):
    # 创建图像数据生成器
    train_datagen = ImageDataGenerator(
        rescale=1./255,
        rotation_range=20,
        width_shift_range=0.2,
        height_shift_range=0.2,
        shear_range=0.2,
        zoom_range=0.2,
        horizontal_flip=True,
        fill_mode='nearest'
    )
    
    val_datagen = ImageDataGenerator(rescale=1./255)
    
    # 创建数据加载器
    train_generator = train_datagen.flow_from_directory(
        train_dir,
        target_size=(224, 224),
        batch_size=batch_size,
        class_mode='categorical'
    )
    
    val_generator = val_datagen.flow_from_directory(
        val_dir,
        target_size=(224, 224),
        batch_size=batch_size,
        class_mode='categorical'
    )
    
    # 训练模型
    history = model.fit(
        train_generator,
        steps_per_epoch=train_generator.samples // batch_size,
        epochs=epochs,
        validation_data=val_generator,
        validation_steps=val_generator.samples // batch_size
    )
    
    return history

def detect_ai_image(image_path, model):
    # 预处理图像
    img = tf.keras.preprocessing.image.load_img(image_path, target_size=(224, 224))
    img_array = tf.keras.preprocessing.image.img_to_array(img)
    img_array = tf.expand_dims(img_array, 0)  # 添加批次维度
    img_array = img_array / 255.0  # 归一化
    
    # 预测
    predictions = model.predict(img_array)
    class_idx = tf.argmax(predictions[0]).numpy()
    confidence = predictions[0][class_idx]
    
    return {
        'is_ai_generated': bool(class_idx == 1),  # 假设类别1为AI生成
        'confidence': float(confidence),
        'raw_predictions': predictions[0].tolist()
    }

import tensorflow as tf
from tensorflow.keras import layers, models
import numpy as np

class ContrastiveImageDataset:
    def __init__(self, image_paths, batch_size=32, image_size=(224, 224)):
        self.image_paths = image_paths
        self.batch_size = batch_size
        self.image_size = image_size
        self.augmentation = self._get_augmentation()
    
    def _get_augmentation(self):
        # 定义图像增强操作
        return tf.keras.Sequential([
            layers.RandomFlip("horizontal"),
            layers.RandomRotation(0.1),
            layers.RandomZoom(0.1),
            layers.RandomContrast(0.1),
            layers.RandomBrightness(0.1),
        ])
    
    def _preprocess_image(self, image_path):
        # 加载和预处理图像
        image = tf.io.read_file(image_path)
        image = tf.image.decode_jpeg(image, channels=3)
        image = tf.image.resize(image, self.image_size)
        image = tf.cast(image, tf.float32) / 255.0
        return image
    
    def _create_pairs(self):
        # 创建正样本对（同一图像的不同增强版本）
        dataset = tf.data.Dataset.from_tensor_slices(self.image_paths)
        dataset = dataset.map(self._preprocess_image, num_parallel_calls=tf.data.AUTOTUNE)
        dataset = dataset.map(lambda x: (x, self.augmentation(x)), num_parallel_calls=tf.data.AUTOTUNE)
        dataset = dataset.batch(self.batch_size).prefetch(tf.data.AUTOTUNE)
        return dataset
    
    def get_dataset(self):
        return self._create_pairs()

class SimCLRModel:
    def __init__(self, input_shape=(224, 224, 3), projection_dim=128):
        self.input_shape = input_shape
        self.projection_dim = projection_dim
        self.base_model = self._create_base_model()
        self.encoder = self._create_encoder()
        self.projection_head = self._create_projection_head()
        self.model = self._create_simclr_model()
    
    def _create_base_model(self):
        # 使用预训练模型作为基础编码器
        base_model = tf.keras.applications.ResNet50(include_top=False, weights='imagenet', input_shape=self.input_shape)
        base_model.trainable = True  # 微调整个模型
        return base_model
    
    def _create_encoder(self):
        # 创建编码器模型
        inputs = layers.Input(shape=self.input_shape)
        x = self.base_model(inputs)
        x = layers.GlobalAveragePooling2D()(x)
        return models.Model(inputs, x, name="encoder")
    
    def _create_projection_head(self):
        # 创建投影头
        inputs = layers.Input(shape=(2048,))  # ResNet50特征维度
        x = layers.Dense(2048, activation="relu")(inputs)
        x = layers.BatchNormalization()(x)
        x = layers.Dense(self.projection_dim)(x)
        return models.Model(inputs, x, name="projection_head")
    
    def _create_simclr_model(self):
        # 创建完整的SimCLR模型
        image_a = layers.Input(shape=self.input_shape)
        image_b = layers.Input(shape=self.input_shape)
        
        features_a = self.encoder(image_a)
        features_b = self.encoder(image_b)
        
        projections_a = self.projection_head(features_a)
        projections_b = self.projection_head(features_b)
        
        return models.Model([image_a, image_b], [projections_a, projections_b], name="simclr_model")
    
    def compile(self, temperature=0.1):
        # 编译模型
        self.temperature = temperature
        self.model.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=0.0001),
                          loss=self._contrastive_loss)
    
    def _contrastive_loss(self, projections_1, projections_2):
        # 计算对比损失
        # 归一化投影向量
        projections_1 = tf.math.l2_normalize(projections_1, axis=1)
        projections_2 = tf.math.l2_normalize(projections_2, axis=1)
        
        # 计算相似度矩阵
        similarities = tf.matmul(projections_1, projections_2, transpose_b=True) / self.temperature
        
        # 创建标签（对角线为正对）
        batch_size = tf.shape(similarities)[0]
        labels = tf.range(batch_size)
        
        # 计算交叉熵损失
        loss_1_2 = tf.keras.losses.sparse_categorical_crossentropy(labels, similarities, from_logits=True)
        loss_2_1 = tf.keras.losses.sparse_categorical_crossentropy(labels, tf.transpose(similarities), from_logits=True)
        
        return (loss_1_2 + loss_2_1) / 2

import librosa
import numpy as np

def extract_mfcc_and_logmel(audio_path, sr=16000, n_mfcc=20, n_mels=64):
    y, sr = librosa.load(audio_path, sr=sr)
    # 能量归一化与去静音（简化）
    y = librosa.util.normalize(y)
    # MFCC
    mfcc = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=n_mfcc)
    # Log-Mel 频谱
    mel = librosa.feature.melspectrogram(y=y, sr=sr, n_mels=n_mels)
    logmel = librosa.power_to_db(mel)
    # 统计特征聚合
    feats = {
        'mfcc_mean': np.mean(mfcc, axis=1).tolist(),
        'mfcc_std': np.std(mfcc, axis=1).tolist(),
        'logmel_mean': np.mean(logmel, axis=1).tolist(),
        'logmel_std': np.std(logmel, axis=1).tolist(),
    }
    return feats

from sklearn.metrics.pairwise import cosine_similarity

def speaker_match_score(emb1, emb2):
    # emb1/emb2: 说话人嵌入向量
    sim = cosine_similarity([emb1], [emb2])[0][0]
    return float(sim)

import cv2

def extract_face_crops(video_path, max_frames=128):
    cap = cv2.VideoCapture(video_path)
    crops = []
    count = 0
    while cap.isOpened() and count < max_frames:
        ret, frame = cap.read()
        if not ret:
            break
        # 这里应接入人脸检测/对齐（如MediaPipe/Dlib/MTCNN）
        # 简化处理：直接缩放
        crop = cv2.resize(frame, (224, 224))
        crops.append(crop)
        count += 1
    cap.release()
    return crops

def score_fusion(scores, weights=None):
    # scores: {'text':0.7,'image':0.4,'audio':0.8}
    if weights is None:
        weights = {k:1.0 for k in scores}
    total_w = sum(weights.values())
    fused = sum(scores[k]*weights[k] for k in scores)/total_w
    return fused