import streamlit as st from PIL import Image, ImageChops, ImageEnhance, ImageDraw, ImageFilter import numpy as np import matplotlib.pyplot as plt import matplotlib.pyplot as plt import matplotlib.patches as patches from scipy import ndimage from skimage import feature, measure import io import cv2 import os import cv2 as cv from mtcnn import MTCNN from tensorflow.keras.models import load_model from tensorflow.keras.preprocessing import image as keras_image import keras # Load models @st.cache_resource def load_image_forgery_model(): return load_model("imageforgerydetection.h5") @st.cache_resource def load_deepfake_image_model(): return load_model("deepfake_image_detection.h5") @st.cache_resource def load_video_forgery_model(): return load_model("videoforgerydetection.keras") # Constants IMG_SIZE = 224 MAX_SEQ_LENGTH = 20 NUM_FEATURES = 2048 @st.cache_resource def load_deepfake_model(): return load_model('video_classifier_full_model.h5') # Load pre-trained models and processor deepfake_model = load_deepfake_model() vocabulary2 = np.load('label_processor_vocabulary.npy', allow_pickle=True) label_processor2 = keras.layers.StringLookup(num_oov_indices=0, vocabulary=vocabulary2.tolist()) # Helper functions # Image Forgery Detection Functions def convert_to_ela_image(image, quality=90): temp_filename = 'temp_file_name.jpg' ela_filename = 'temp_ela.png' if image.mode != 'RGB': image = image.convert('RGB') image.save(temp_filename, 'JPEG', quality=quality) temp_image = Image.open(temp_filename) ela_image = ImageChops.difference(image, temp_image) extrema = ela_image.getextrema() max_diff = max([ex[1] for ex in extrema]) max_diff = max_diff if max_diff != 0 else 1 scale = 255.0 / max_diff ela_image = ImageEnhance.Brightness(ela_image).enhance(scale) return ela_image def prepare_image_for_forgery(image): ela_image = convert_to_ela_image(image, 90).resize((128, 128)) return np.array(ela_image).flatten() / 255.0 # Individual Analysis Functions def create_ela_analysis(image): """Create ELA analysis visualization""" fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 6)) fig.suptitle('Error Level Analysis (ELA)', fontsize=14, fontweight='bold') # Original image ax1.imshow(image) ax1.set_title('Original Image') ax1.axis('off') # ELA image ela_image = convert_to_ela_image(image, 90) ax2.imshow(ela_image) ax2.set_title('ELA Result (Bright areas indicate potential editing)') ax2.axis('off') plt.tight_layout() buffer = io.BytesIO() plt.savefig(buffer, format='png', dpi=150, bbox_inches='tight') buffer.seek(0) plt.close() return buffer # Deepfake Image Detection def predict_deepfake_image(image_path, model): img = keras_image.load_img(image_path, target_size=(256, 256)) img_array = keras_image.img_to_array(img) / 255.0 img_array = np.expand_dims(img_array, axis=0) prediction = model.predict(img_array) return 'Real' if prediction[0] > 0.5 else 'Fake' # Video Forgery Detection # Configuration target_height, target_width = 240, 320 threshold = 30 # Threshold for freeze/duplicate detection def predict_video_forgery_cnn(video_path, model): """CNN-based video forgery detection""" vid = [] sumframes = 0 cap = cv2.VideoCapture(video_path) while cap.isOpened(): ret, frame = cap.read() if not ret: break # Resize frame to target dimensions frame = cv2.resize(frame, (target_width, target_height)) sumframes += 1 vid.append(frame) cap.release() if sumframes == 0: return False, 0, 0 Xtest = np.array(vid) output = model.predict(Xtest) output = output.reshape((-1)) # Check if any frame is predicted as forged forged_frames = sum(1 for i in output if i > 0.5) is_forged = any(i > 0.5 for i in output) return is_forged, forged_frames, sumframes def analyze_video_tampering(video_path): """Frame difference analysis for tampering detection""" cap = cv2.VideoCapture(video_path) if not cap.isOpened(): return False, [], [] prev_frame = None frame_differences = [] suspected_frames = [] while cap.isOpened(): ret, frame = cap.read() if not ret: break gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) if prev_frame is not None: diff = cv2.absdiff(gray, prev_frame) non_zero = np.count_nonzero(diff) frame_differences.append(non_zero) if non_zero < threshold: current_frame = int(cap.get(cv2.CAP_PROP_POS_FRAMES)) suspected_frames.append(current_frame) prev_frame = gray cap.release() # Simple rule: if any frame is suspected, flag as tampered is_tampered = len(suspected_frames) > 0 return is_tampered, frame_differences, suspected_frames def plot_frame_analysis(frame_differences): """Create a simple plot of frame differences""" plt.figure(figsize=(10, 4)) plt.plot(frame_differences, color='blue', linewidth=1) plt.axhline(y=threshold, color='red', linestyle='--', label=f"Threshold ({threshold})") plt.xlabel("Frame Number") plt.ylabel("Pixel Differences") plt.title("Frame Difference Analysis") plt.legend() plt.grid(True, alpha=0.3) # Add statistics if frame_differences: mean_val = np.mean(frame_differences) std_val = np.std(frame_differences) plt.text(0.02, 0.98, f"Mean: {mean_val:.1f}\nStd: {std_val:.1f}", transform=plt.gca().transAxes, verticalalignment='top', bbox=dict(boxstyle='round', facecolor='white', alpha=0.8)) return plt def combined_video_forgery_detection(video_path, model): """Combined detection using both CNN and frame analysis""" # Method 1: CNN-based detection cnn_forged, cnn_forged_frames, total_frames = predict_video_forgery_cnn(video_path, model) # Method 2: Frame analysis tampering detection frame_tampered, frame_differences, suspected_frames = analyze_video_tampering(video_path) # Results results = { 'cnn_forged': cnn_forged, 'cnn_forged_frames': cnn_forged_frames, 'frame_tampered': frame_tampered, 'suspected_frames': len(suspected_frames), 'total_frames': total_frames, 'frame_differences': frame_differences } # Simple decision logic if cnn_forged and frame_tampered: verdict = "FORGED - Detected by both CNN and Frame Analysis" confidence = "High" elif cnn_forged: verdict = "FORGED - Detected by CNN" confidence = "Medium" elif frame_tampered: verdict = "FORGED - Detected by Frame Analysis" confidence = "Medium" else: verdict = "NOT TAMPERED - No Forgery detected" confidence = "High" return verdict, confidence, results # Deepfake Video Detection def build_feature_extractor(): feature_extractor = keras.applications.InceptionV3( weights="imagenet", include_top=False, pooling="avg", input_shape=(IMG_SIZE, IMG_SIZE, 3), ) preprocess_input = keras.applications.inception_v3.preprocess_input inputs = keras.Input((IMG_SIZE, IMG_SIZE, 3)) preprocessed = preprocess_input(inputs) outputs = feature_extractor(preprocessed) return keras.Model(inputs, outputs, name="feature_extractor") feature_extractor = build_feature_extractor() detector = MTCNN() def load_video(path, max_frames=0, resize=(IMG_SIZE, IMG_SIZE), skip_frames=2): cap = cv.VideoCapture(path) frames = [] frame_count = 0 previous_box = None while True: ret, frame = cap.read() if not ret: break if frame_count % skip_frames == 0: frame, previous_box = get_face_region_first_frame(frame, previous_box) if frame is not None: frame = cv.resize(frame, resize) frame = frame[:, :, [2, 1, 0]] frames.append(frame) if len(frames) == max_frames: break frame_count += 1 while len(frames) < max_frames and frames: frames.append(frames[-1]) cap.release() return np.array(frames) def get_face_region_first_frame(frame, previous_box=None): if previous_box is None: detections = detector.detect_faces(frame) if detections: x, y, width, height = detections[0]['box'] previous_box = (x, y, width, height) else: return None, None else: x, y, width, height = previous_box face_region = frame[y:y+height, x:x+width] return face_region, previous_box def prepare_single_video(frames): frames = frames[None, ...] frame_mask = np.zeros(shape=(1, MAX_SEQ_LENGTH,), dtype="bool") frame_features = np.zeros(shape=(1, MAX_SEQ_LENGTH, NUM_FEATURES), dtype="float32") for i, batch in enumerate(frames): video_length = batch.shape[0] length = min(MAX_SEQ_LENGTH, video_length) for j in range(length): frame_features[i, j, :] = feature_extractor.predict(batch[None, j, :]) frame_mask[i, :length] = 1 return frame_features, frame_mask def sequence_prediction(video_path): class_vocab = label_processor2.get_vocabulary() frames = load_video(video_path) if len(frames) == 0: st.error("Could not process video. Please try another file.") return None frame_features, frame_mask = prepare_single_video(frames) probabilities = deepfake_model.predict([frame_features, frame_mask])[0] predictions = {class_vocab[i]: probabilities[i] * 100 for i in np.argsort(probabilities)[::-1]} return predictions # Streamlit App st.title("Fraudulent Image and Video Detection System") # Sidebar for model selection task = st.sidebar.selectbox("Choose a detection task:", [ "Image Forgery Detection", "Deepfake Image Detection", "Video Forgery Detection", "Deepfake Video Detection" ]) # Main Streamlit App if task == "Image Forgery Detection": uploaded_file = st.file_uploader("Upload an image", type=['jpg', 'jpeg', 'png']) if uploaded_file: image = Image.open(uploaded_file) # Fixed size display - adjust width as needed (300-600 pixels work well) st.image(image, caption="Uploaded Image", width=400) # Original prediction prepared_image = prepare_image_for_forgery(image).reshape(-1, 128, 128, 3) model = load_image_forgery_model() prediction = model.predict(prepared_image) confidence_real = prediction[0][1] * 100 confidence_fake = prediction[0][0] * 100 if confidence_real > confidence_fake: st.success(f"Result: Real Image with {confidence_real:.2f}% confidence") else: st.error(f"Result: Forged Image with {confidence_fake:.2f}% confidence") # Add ELA analysis option st.markdown("---") st.subheader("🔍 Additional Analysis") # Show ELA option checkbox show_ela = st.checkbox("View Error Level Analysis (ELA)", value=False) if show_ela: st.markdown("### Error Level Analysis") st.info("**ELA**: Reveals compression artifacts. Bright areas indicate potential editing or manipulation.") col1, col2 = st.columns([1, 3]) with col1: analyze_button = st.button("Run ELA Analysis", type="primary", use_container_width=True) if analyze_button: with st.spinner("Running Error Level Analysis..."): try: analysis_buffer = create_ela_analysis(image) # Fixed size for analysis results st.image(analysis_buffer, caption="ELA Analysis Results", width=500) # Download button st.download_button( label="Download ELA Results", data=analysis_buffer.getvalue(), file_name="ela_analysis.png", mime="image/png", use_container_width=True ) except Exception as e: st.error(f"Error during ELA analysis: {str(e)}") st.info("ELA analysis may not work with all image types. Try with a different image if needed.") elif task == "Deepfake Image Detection": uploaded_file = st.file_uploader("Upload an image", type=['jpg', 'jpeg', 'png']) if uploaded_file: with open("temp_image.jpg", "wb") as f: f.write(uploaded_file.getbuffer()) # Fixed size display for deepfake detection st.image(uploaded_file, caption="Uploaded Image", width=400) model = load_deepfake_image_model() result = predict_deepfake_image("temp_image.jpg", model) if result == 'Real': st.success("Prediction: Real") else: st.error("Prediction: Fake") os.remove("temp_image.jpg") if task == "Video Forgery Detection": uploaded_file = st.file_uploader("Upload a video", type=['mp4', 'avi', 'mov', 'mkv']) if uploaded_file: # Save uploaded file with open("temp_video.mp4", "wb") as f: f.write(uploaded_file.getbuffer()) st.video("temp_video.mp4") st.write("Analyzing the video for forgery...") # Load model and run combined detection model = load_video_forgery_model() verdict, confidence, results = combined_video_forgery_detection("temp_video.mp4", model) # Display results if "FORGED" in verdict: st.error(f"🚨 {verdict}") else: st.success(f"✅ {verdict}") st.write(f"**Confidence Level:** {confidence}") # Show detailed results col1, col2 = st.columns(2) with col1: st.write("**CNN Analysis:**") if results['cnn_forged']: st.write(f"- Status: Forged ❌") st.write(f"- Forged Frames: {results['cnn_forged_frames']}/{results['total_frames']}") else: st.write(f"- Status: Not Forged ✅") with col2: st.write("**Frame Analysis:**") if results['frame_tampered']: st.write(f"- Status: Tampered ❌") st.write(f"- Suspected Frames: {results['suspected_frames']}") else: st.write(f"- Status: Not Tampered ✅") # Plot frame differences if available if results['frame_differences']: st.write("**Frame Difference Analysis:**") fig = plot_frame_analysis(results['frame_differences']) st.pyplot(fig) plt.close() # Cleanup os.remove("temp_video.mp4") elif task == "Deepfake Video Detection": uploaded_file = st.file_uploader("Upload a video", type=["mp4", "avi", "mov"]) if uploaded_file is not None: with open("temp_video.mp4", "wb") as f: f.write(uploaded_file.read()) st.video("temp_video.mp4") st.write("Analyzing the video...") frames = load_video("temp_video.mp4") if len(frames) == 0: st.error("Could not process video. Please try another file.") else: frame_features, frame_mask = prepare_single_video(frames) probabilities = deepfake_model.predict([frame_features, frame_mask])[0] predictions = {label_processor2.get_vocabulary()[i]: probabilities[i] * 100 for i in np.argsort(probabilities)[::-1]} if predictions: highest_label = max(predictions, key=predictions.get) highest_prob = predictions[highest_label] if highest_label.lower() == "real": st.success(f"The video is real with a confidence of {highest_prob:.2f}%.") elif highest_label.lower() == "fake": st.error(f"This video is a deepfake with a confidence of {highest_prob:.2f}%.") else: st.warning(f"Uncertain prediction: {highest_label} with {highest_prob:.2f}% confidence.")