[AI딥페이크] 5. 딥페이크 탐지 프로젝트(1) - AI 모델 학습

# [중요] Colab의 최신 TensorFlow가 구버전 방식(Keras 2)으로 작동하도록 강제 설정 # 이 셀을 실행한 후에는 반드시 [런타임] -> [세션 다시 시작]을 해야 적용됩니다. # 1. 호환성 패키지 설치 !pip install tf_keras # 2. 환경변수 설정 (TensorFlow 임포트 전에 해야 함) import os os.environ["TF_USE_LEGACY_KERAS"] = "1" import tensorflow as tf print("현재 설정된 버전:", tf.__version__)

# DeepVoice 모델 훈련을 위한 필수 라이브러리 설치 !pip install tensorflow keras librosa numpy scikit-learn pandas import os import librosa import numpy as np import pandas as pd import tensorflow as tf from sklearn.model_selection import train_test_split from sklearn.metrics import f1_score, classification_report, accuracy_score from tensorflow.keras.models import Sequential from tensorflow.keras.layers import Conv1D, MaxPooling1D, Flatten, Dense, Dropout, BatchNormalization # GPU 사용 확인 및 TensorFlow 로깅 레벨 설정 print("TensorFlow Version:", tf.__version__) print("Num GPUs Available: ", len(tf.config.experimental.list_physical_devices('GPU'))) # 딥러닝 모델 저장 경로 MODEL_PATH = 'model_audio.h5'

# MFCC 특징 추출 함수 정의 (변동 없음) def extract_mfcc(file_path, n_mfcc=40, max_len=100): # 음성 파일에서 MFCC 특징을 추출하고 패딩을 적용함. # :param file_path: 음성 파일 경로 # :param n_mfcc: 추출할 MFCC 개수 # :param max_len: 시퀀스 최대 길이 (패딩/절단 기준) # :return: 패딩된 MFCC 배열 try: # 1. Librosa를 사용해 음성 파일 로드 # NOTE: 음성 파일의 샘플링 레이트(sr)는 None으로 설정하여 원본 sr을 사용합니다. y, sr = librosa.load(file_path, sr=None) # 2. MFCC 특징 추출 mfccs = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=n_mfcc) mfccs = mfccs.T # 3. 패딩/절단 (최대 길이에 맞춤) if mfccs.shape[0] > max_len: mfccs = mfccs[:max_len, :] # 절단 elif mfccs.shape[0] < max_len: # 패딩: 부족한 부분을 0으로 채움 pad_width = max_len - mfccs.shape[0] # MFCCs는 (time, n_mfcc) 형태이므로, time 축에만 패딩을 적용 mfccs = np.pad(mfccs, ((0, pad_width), (0, 0)), mode='constant') return mfccs except Exception as e: # 에러 발생 시 None 반환 print(f"Error processing {file_path}: {e}") return None # 데이터셋 로드 및 레이블 지정 (CSV 기반) # Colab 환경에서 '딥보이스 학습데이터'가 마운트되었다고 가정하고 경로 설정 # train.csv와 train 폴더가 같은 경로에 있다고 가정합니다. base_data_dir = './drive/MyDrive/딥보이스 학습데이터' csv_path = os.path.join(base_data_dir, 'train.csv') # 1. Pandas를 사용하여 CSV 파일 로드 print(f"CSV 파일 로드 경로: {csv_path}") try: df = pd.read_csv(csv_path) except FileNotFoundError: print("오류: train.csv 파일을 찾을 수 없음. 경로를 확인하십시오.") df = pd.DataFrame() # --- [수정된 핵심 부분: 데이터 샘플링] --- # 전체 5만 개 데이터를 처리하는 것은 시간이 오래 걸리므로, 100개만 실습함. SAMPLE_SIZE = 100 if len(df) > SAMPLE_SIZE: # stratify=y를 적용하기 위해, 샘플링 전 레이블을 먼저 이진화 df['is_fake'] = df['label'].apply(lambda x: 1 if x == 'fake' else 0) # 레이블(real/fake) 분포를 유지하며 2000개 샘플 추출 (stratified sampling) df_sampled = df.groupby('is_fake').apply(lambda x: x.sample(n=SAMPLE_SIZE // 2, random_state=42)).reset_index(drop=True) else: df_sampled = df print(f"데이터셋 전체 크기: {len(df)}") print(f"실습을 위해 추출된 샘플 크기: {len(df_sampled)}") # --------------------------------------------- # 'path' 컬럼이 상대 경로('./train/...')를 가지고 있다고 가정 audio_folder_path = os.path.join(base_data_dir, 'train') features = [] labels = [] max_len = 100 # 모든 시퀀스의 길이를 100으로 통일함 n_mfcc = 40 # 2. 샘플링된 데이터프레임을 순회하며 MFCC 특징 추출 print("MFCC 특징 추출 및 데이터 로딩을 시작함...") for index, row in df_sampled.iterrows(): relative_path = row['path'] # 예: ./train/AAACWK.ogg label_text = row['label'] # 예: real 또는 fake # 실제 파일 경로 조합 (파일 이름만 추출하여 실제 폴더 경로와 조합) filename = os.path.basename(relative_path) file_path = os.path.join(audio_folder_path, filename) # 레이블을 숫자로 변환: fake=1, real=0 label = 1 if label_text == 'fake' else 0 mfcc = extract_mfcc(file_path, n_mfcc=n_mfcc, max_len=max_len) if mfcc is not None: features.append(mfcc) labels.append(label) # NumPy 배열로 변환 및 데이터셋 분리 X = np.array(features) y = np.array(labels) # Train, Validation, Test 셋 분리 X_train, X_temp, y_train, y_temp = train_test_split(X, y, test_size=0.3, random_state=42, stratify=y) X_val, X_test, y_val, y_test = train_test_split(X_temp, y_temp, test_size=0.5, random_state=42, stratify=y_temp) print(f"총 데이터셋 크기: {len(X)}") print(f"훈련 셋 크기: {len(X_train)}") print(f"검증 셋 크기: {len(X_val)}") print(f"테스트 셋 크기: {len(X_test)}") print(f"MFCC 특징 모양 (Shape): {X_train.shape[1:]}")

# 1D CNN 모델 구축 def build_1d_cnn_model(input_shape): model = Sequential([ # 첫 번째 Conv1D 레이어: 32 필터, 커널 사이즈 5 Conv1D(filters=32, kernel_size=5, activation='relu', input_shape=input_shape), BatchNormalization(), MaxPooling1D(pool_size=2), Dropout(0.3), # 두 번째 Conv1D 레이어: 64 필터, 커널 사이즈 3 Conv1D(filters=64, kernel_size=3, activation='relu'), BatchNormalization(), MaxPooling1D(pool_size=2), Dropout(0.3), # 특징 평탄화 Flatten(), # Dense 레이어 Dense(128, activation='relu'), Dropout(0.5), # 출력 레이어: 이진 분류이므로 1개 뉴런, Sigmoid 사용 Dense(1, activation='sigmoid') ]) # 모델 컴파일 model.compile( optimizer='adam', # 이진 분류이므로 binary_crossentropy 손실 함수 사용 loss='binary_crossentropy', metrics=['accuracy'] ) return model # 모델 생성 # NOTE: X_train.shape[1:]는 2단계 실행 후 정의되므로, 2단계 실행이 필수임. mfcc_input_shape = X_train.shape[1:] audio_model = build_1d_cnn_model(mfcc_input_shape) audio_model.summary()

# 성능 평가 함수 def evaluate_model(model, X, y): """모델을 평가하고 정확도와 F1-Score를 출력함.""" y_pred_proba = model.predict(X) # Sigmoid 출력(0~1)을 이진 값(0 또는 1)으로 변환 (임계값 0.5) y_pred = (y_pred_proba > 0.5).astype(int) # F1-Score 계산 f1 = f1_score(y, y_pred) acc = accuracy_score(y, y_pred) # Classification Report 출력 (정밀도, 재현율, F1-Score 포함) report = classification_report(y, y_pred, target_names=['Real (0)', 'Fake (1)'], output_dict=True) print(f"\n--- 성능 평가 결과 ---") print(f"테스트 셋 정확도: {acc:.4f}") print(f"테스트 셋 F1-Score: {f1:.4f}") print(classification_report(y, y_pred, target_names=['Real (0)', 'Fake (1)'])) return f1 # 1. 미비한 학습 (Epoch 5) 진행 print("--- 1. 미비한 학습 시작 (Epoch 5) ---") # 주의: build_1d_cnn_model 함수는 3단계에서 정의됨 audio_model_5 = build_1d_cnn_model(X_train.shape[1:]) # 새 모델 생성 history_5 = audio_model_5.fit( X_train, y_train, epochs=5, batch_size=32, validation_data=(X_val, y_val), verbose=1 ) f1_5 = evaluate_model(audio_model_5, X_test, y_test) print(f"최종 (Epoch 5) F1-Score: {f1_5:.4f}") # 2. 충분한 학습 (Epoch 50) 진행 print("\n--- 2. 충분한 학습 시작 (Epoch 50) ---") # 주의: build_1d_cnn_model 함수는 3단계에서 정의됨 audio_model_50 = build_1d_cnn_model(X_train.shape[1:]) # 새 모델 생성 history_50 = audio_model_50.fit( X_train, y_train, epochs=50, batch_size=32, validation_data=(X_val, y_val), verbose=0 # 출력 간소화 ) f1_50 = evaluate_model(audio_model_50, X_test, y_test) print(f"최종 (Epoch 50) F1-Score: {f1_50:.4f}")

# 50 Epoch 모델을 최종 파일로 저장 # 파일명: model_audio.h5 (6일차 FastAPI 서버용) # NOTE: audio_model_50은 4단계 실행 후 정의됨 audio_model_50.save('model_audio.h5') print("최종 딥보이스 모델 'model_audio.h5' 저장을 완료함.")

# [1블럭 수정] Keras 2 (Legacy) 강제 설정 # 주의: 이 셀 실행 후 혹시 뭔가 꼬이면 [런타임] -> [세션 다시 시작]을 하세요. import os # 1. 환경 변수 설정 (무조건 tensorflow import보다 먼저!) os.environ["TF_USE_LEGACY_KERAS"] = "1" # 2. 패키지 설치 !pip install tf_keras import tensorflow as tf import tf_keras # 3. 설정 확인 (에러 없이 경로 확인) print(f"TensorFlow Version: {tf.__version__}") print(f"Current tf.keras implementation: {tf.keras}") # tf.keras 출력 결과에 'tf_keras' 라는 단어가 포함되어 있으면 성공입니다. if "tf_keras" in str(tf.keras): print("\n" + "="*50) print("✅ 설정 성공! 구버전(Legacy) Keras가 적용되었습니다.") print("이제 'batch_shape' 오류 없이 서버에서 모델을 읽을 수 있습니다.") print("="*50 + "\n") else: print("\n" + "!"*50) print("⚠️ 경고: 설정이 완벽하지 않을 수 있습니다.") print("하지만 일단 진행해 보세요. 에러 메시지가 'tf_keras'를 가리키고 있다면 작동할 겁니다.") print("!"*50 + "\n")

# 필요한 라이브러리 임포트 import cv2 # 영상/이미지 처리 (OpenCV) import os import numpy as np import tensorflow as tf import matplotlib.pyplot as plt # 학습 추이 시각화용 from tensorflow.keras.preprocessing.image import ImageDataGenerator from tensorflow.keras.models import Sequential from tensorflow.keras.layers import Flatten, Dense, Dropout from tensorflow.keras.applications import Xception # 전이 학습 모델 (XceptionNet) 로드 from tensorflow.keras.callbacks import ModelCheckpoint # 딥러닝 모델 저장 경로 MODEL_IMAGE_PATH = 'model_image.h5' # 이미지 해상도 설정 (XceptionNet 표준 권장 해상도에 맞춰 244x244 or 299x299 사용) IMG_SIZE = (299, 299) BATCH_SIZE = 32

# ---------------------------------------------------------------------- # 3단계: 이미지 데이터 로드 및 증강 (ImageDataGenerator) # ---------------------------------------------------------------------- import tensorflow as tf from tensorflow.keras.preprocessing.image import ImageDataGenerator import os # 1. 경로 및 하이퍼파라미터 설정 # 2단계에서 추출된 이미지가 저장된 상위 폴더 경로 EXTRACTED_IMAGE_PATH = './drive/MyDrive/딥페이크 학습데이터/deepfake_frames' # XceptionNet 권장 입력 크기 (224) IMG_SIZE = (224, 224) BATCH_SIZE = 32 # 데이터가 제대로 준비되었는지 경로 확인 if not os.path.exists(EXTRACTED_IMAGE_PATH): print(f"오류: 데이터 경로 '{EXTRACTED_IMAGE_PATH}'가 존재하지 않습니다. 2단계 실습을 먼저 완료해 주세요.") else: print(f"데이터 경로 확인됨: {EXTRACTED_IMAGE_PATH}") print(f"내부 폴더: {os.listdir(EXTRACTED_IMAGE_PATH)}") print("\n3단계: 데이터 증강(Data Augmentation) 설정 및 Generator 생성을 시작합니다...") # 2. ImageDataGenerator 정의 (데이터 증강 및 전처리) # - rescale: 픽셀 값을 0~255에서 0~1 사이로 정규화 # - rotation, shift, flip: 이미지를 무작위로 변형하여 데이터 다양성 확보 (과적합 방지) # - validation_split: 전체 데이터의 20%를 검증용으로 자동 분리 train_datagen = ImageDataGenerator( rescale=1./255, rotation_range=20, # 랜덤 회전 (20도 이내) width_shift_range=0.2, # 랜덤 가로 이동 height_shift_range=0.2, # 랜덤 세로 이동 horizontal_flip=True, # 랜덤 좌우 반전 fill_mode='nearest', # 빈 공간 채우기 방식 validation_split=0.2 # 20%는 검증 데이터로 사용 ) # 3. 학습용 데이터 생성기 (Training Generator) # subset='training'으로 설정하여 학습 데이터만 로드 print("\n[학습 데이터셋 로드 중...]") train_generator = train_datagen.flow_from_directory( EXTRACTED_IMAGE_PATH, target_size=IMG_SIZE, batch_size=BATCH_SIZE, class_mode='binary', # Real(0) vs Fake(1) 이진 분류 subset='training' # 학습용 데이터 ) # 4. 검증용 데이터 생성기 (Validation Generator) # subset='validation'으로 설정하여 검증 데이터만 로드 # 주의: 검증 데이터에는 증강을 적용하지 않는 것이 일반적이나, # 여기서는 코드 간소화를 위해 동일한 datagen 객체의 split 기능을 사용합니다. print("\n[검증 데이터셋 로드 중...]") validation_generator = train_datagen.flow_from_directory( EXTRACTED_IMAGE_PATH, target_size=IMG_SIZE, batch_size=BATCH_SIZE, class_mode='binary', # Real(0) vs Fake(1) 이진 분류 subset='validation' # 검증용 데이터 ) # 5. 클래스 인덱스 확인 # 폴더 이름(Real, Fake)이 어떻게 0과 1로 매핑되었는지 확인 print("\n클래스 매핑 정보:", train_generator.class_indices) print("3단계: 데이터셋 준비 완료. 이제 모델 구축 단계로 넘어갈 수 있습니다.")

# ---------------------------------------------------------------------- # 4단계: 딥페이크 탐지 모델 구축 (XceptionNet 기반 전이 학습) # ---------------------------------------------------------------------- import tensorflow as tf from tensorflow.keras.applications import Xception from tensorflow.keras.models import Sequential from tensorflow.keras.layers import Flatten, Dense, Dropout # 1단계에서 설정한 IMG_SIZE 사용 # 만약 1단계 실행을 안 했다면 아래 주석 해제 # IMG_SIZE = (299, 299) def build_xception_model(input_shape): # 1. 사전 학습된 모델 로드 (ImageNet 가중치 사용, Top 층 제외) base_model = Xception(weights='imagenet', include_top=False, input_shape=input_shape) # 2. 핵심: 베이스 모델의 가중치 동결 (Freeze) # 학습된 특징 추출 능력을 유지하기 위해 업데이트를 막음 base_model.trainable = False # 3. 새로운 분류기(Classifier) 추가 model = Sequential([ base_model, Flatten(), Dense(512, activation='relu'), Dropout(0.5), # 과적합 방지 Dense(1, activation='sigmoid') # 이진 분류 (Real vs Fake) ]) # 4. 모델 컴파일 model.compile( optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'] ) return model print("3단계: 모델 구축 시작...") # 모델 생성 image_input_shape = IMG_SIZE + (3,) # (299, 299, 3) image_model = build_xception_model(image_input_shape) # 모델 구조 및 파라미터 확인 image_model.summary() print("\n[확인 포인트] Summary 하단의 'Non-trainable params' 숫자를 확인하십시오.") print("이 숫자가 0이 아니라면, XceptionNet의 가중치가 성공적으로 동결된 것입니다.")

# ---------------------------------------------------------------------- # 4단계: 성능 비교 실습 (미비 vs 충분) # ---------------------------------------------------------------------- import matplotlib.pyplot as plt import numpy as np from sklearn.metrics import classification_report, f1_score, accuracy_score from tensorflow.keras.preprocessing.image import ImageDataGenerator # 1. 학습 추이 시각화 함수 def plot_history(history, title): """Loss 및 Accuracy 변화 그래프 출력.""" plt.figure(figsize=(12, 4)) # Loss 그래프 plt.subplot(1, 2, 1) plt.plot(history.history['loss'], label='Training Loss') plt.plot(history.history['val_loss'], label='Validation Loss') plt.title(f'{title} - Loss Trend') plt.xlabel('Epoch') plt.ylabel('Loss (Binary Crossentropy)') plt.legend() # Accuracy 그래프 plt.subplot(1, 2, 2) plt.plot(history.history['accuracy'], label='Training Accuracy') plt.plot(history.history['val_accuracy'], label='Validation Accuracy') plt.title(f'{title} - Accuracy Trend') plt.xlabel('Epoch') plt.ylabel('Accuracy') plt.legend() plt.show() # 2. 성능 평가 함수 def evaluate_generator_metrics(model, generator): """모델을 평가하고 정확도 및 F1-Score를 출력함.""" # 중요: 제너레이터 초기화 (처음부터 순서대로 읽기 위함) generator.reset() # 1. Prediction (순서대로 예측) steps = generator.samples // generator.batch_size + (generator.samples % generator.batch_size > 0) y_pred_proba = model.predict(generator, steps=steps, verbose=0) # 2. 정답지(Label) 가져오기 (셔플되지 않은 순서) y_true = generator.classes[generator.index_array] # 3. 확률을 0과 1로 변환 (0.5 기준) y_pred = (y_pred_proba > 0.5).astype(int).flatten() # 4. 점수 계산 f1 = f1_score(y_true, y_pred, average='weighted') acc = accuracy_score(y_true, y_pred) class_names = list(generator.class_indices.keys()) print(f"\n--- 성능 평가 결과 ---") print(f"테스트 셋 정확도: {acc:.4f}") print(f"테스트 셋 F1-Score (가중평균): {f1:.4f}") print(classification_report(y_true, y_pred, target_names=class_names)) return f1 print("평가용 데이터셋(Test Generator)을 생성합니다 (Shuffle=False)...") # train_datagen은 3단계에서 정의된 객체를 사용합니다. test_generator = train_datagen.flow_from_directory( EXTRACTED_IMAGE_PATH, target_size=IMG_SIZE, batch_size=BATCH_SIZE, class_mode='binary', subset='validation', # 검증 데이터 사용 shuffle=False # [중요] 절대 섞지 않음! 순서대로 채점하기 위함 ) # --- 1. Epoch 5 훈련 (미비 학습 비교) --- print("\n4단계: 1. 미비한 학습 시작 (Epoch 5) ---") # 새 모델 생성 model_5 = build_xception_model(image_input_shape) # 학습 시에는 validation_generator(섞여도 됨) 사용 history_5 = model_5.fit( train_generator, epochs=5, validation_data=validation_generator ) plot_history(history_5, "Epoch 5 (Underfitting)") # [수정] 평가 시에는 test_generator(섞이지 않음) 사용 f1_5 = evaluate_generator_metrics(model_5, test_generator) print(f"최종 (Epoch 5) F1-Score: {f1_5:.4f}") # --- 2. Epoch 50 훈련 (충분 학습) --- print("\n4단계: 2. 충분한 학습 시작 (Epoch 50) ---") # 새 모델 생성 model_50 = build_xception_model(image_input_shape) history_50 = model_50.fit( train_generator, epochs=50, validation_data=validation_generator, verbose=0 # 로그 간소화 ) plot_history(history_50, "Epoch 50 (Overfitting)") # [수정] 평가 시에는 test_generator(섞이지 않음) 사용 f1_50 = evaluate_generator_metrics(model_50, test_generator) print(f"최종 (Epoch 50) F1-Score: {f1_50:.4f}")

# 50 Epoch 모델을 최종 파일로 저장 # 파일명: model_image.h5 (6일차 FastAPI 서버용) model_50.save('model_image.h5') print("최종 딥페이크 이미지 모델 'model_image.h5' 저장을 완료함.")