Spaces:
Sleeping
Sleeping
| """ | |
| app.py | |
| ------ | |
| DeepFake Detection System - Flask web application. | |
| Routes: | |
| GET / Home page (upload image / video) | |
| POST /detect Run detection, save to SQLite, render result page | |
| GET /history Dashboard + detection history | |
| POST /history/clear Wipe history | |
| GET /report/<id> Download PDF report for a detection | |
| GET /health Health check for Render | |
| Security measures: | |
| - Extension whitelist + size limits (per type and via MAX_CONTENT_LENGTH) | |
| - Content sniffing: images must open with Pillow, videos with OpenCV | |
| - werkzeug secure_filename + random UUID storage names | |
| - Uploaded files are never executed and are deleted right after analysis | |
| """ | |
| import os | |
| import time | |
| import uuid | |
| from flask import ( | |
| Flask, | |
| abort, | |
| flash, | |
| redirect, | |
| render_template, | |
| request, | |
| send_file, | |
| url_for, | |
| ) | |
| from werkzeug.utils import secure_filename | |
| import database | |
| # NOTE: detector/pdf_generator are imported lazily inside the routes that need | |
| # them. They pull in torch + OpenCV + transformers (~400 MB RSS), which would | |
| # otherwise crash small hosts (Render free tier = 512 MB) at boot time. | |
| BASE_DIR = os.path.dirname(os.path.abspath(__file__)) | |
| UPLOAD_FOLDER = os.path.join(BASE_DIR, "uploads") | |
| IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".bmp"} | |
| VIDEO_EXTENSIONS = {".mp4", ".mov", ".avi", ".mkv", ".webm"} | |
| IMAGE_MAX_BYTES = 10 * 1024 * 1024 # 10 MB | |
| VIDEO_MAX_BYTES = 100 * 1024 * 1024 # 100 MB | |
| app = Flask(__name__) | |
| app.config["SECRET_KEY"] = os.environ.get( | |
| "FLASK_SECRET_KEY", "dev-secret-change-in-production" | |
| ) | |
| app.config["MAX_CONTENT_LENGTH"] = 110 * 1024 * 1024 # hard cap ~110 MB | |
| app.config["UPLOAD_FOLDER"] = UPLOAD_FOLDER | |
| os.makedirs(UPLOAD_FOLDER, exist_ok=True) | |
| database.init_db() | |
| # ---------------------------------------------------------------------- # | |
| # Helpers # | |
| # ---------------------------------------------------------------------- # | |
| def _split_ext(filename): | |
| return os.path.splitext(filename)[1].lower() | |
| def _validate_upload(file_storage): | |
| """Return ('image'|'video', safe_original_name) or raise ValueError.""" | |
| if file_storage is None or not file_storage.filename: | |
| raise ValueError("Please choose a file to analyze.") | |
| original = file_storage.filename | |
| ext = _split_ext(original) | |
| if ext in IMAGE_EXTENSIONS: | |
| kind, limit = "image", IMAGE_MAX_BYTES | |
| elif ext in VIDEO_EXTENSIONS: | |
| kind, limit = "video", VIDEO_MAX_BYTES | |
| else: | |
| raise ValueError( | |
| f"Unsupported file type '{ext}'. " | |
| f"Allowed images: {sorted(IMAGE_EXTENSIONS)} | " | |
| f"videos: {sorted(VIDEO_EXTENSIONS)}." | |
| ) | |
| file_storage.stream.seek(0, os.SEEK_END) | |
| size = file_storage.stream.tell() | |
| file_storage.stream.seek(0) | |
| if size == 0: | |
| raise ValueError("The uploaded file is empty.") | |
| if size > limit: | |
| raise ValueError( | |
| f"File is too large ({size / 1e6:.1f} MB). " | |
| f"Limit for {kind}s is {limit / 1e6:.0f} MB." | |
| ) | |
| return kind, original | |
| def _sniff_content(path, kind): | |
| """Verify the bytes actually decode as the claimed media type.""" | |
| if kind == "image": | |
| from PIL import Image | |
| try: | |
| with Image.open(path) as img: | |
| img.verify() | |
| except Exception: | |
| raise ValueError("This file is not a valid image.") | |
| else: | |
| import cv2 | |
| cap = cv2.VideoCapture(path) | |
| opened = cap.isOpened() | |
| if opened: | |
| ok, _ = cap.read() | |
| opened = ok | |
| cap.release() | |
| if not opened: | |
| raise ValueError("This file is not a readable video.") | |
| # ---------------------------------------------------------------------- # | |
| # Routes # | |
| # ---------------------------------------------------------------------- # | |
| def index(): | |
| stats = database.get_stats() | |
| return render_template("index.html", stats=stats) | |
| def detect(): | |
| try: | |
| kind, original_name = _validate_upload(request.files.get("file")) | |
| except ValueError as exc: | |
| flash(str(exc), "error") | |
| return redirect(url_for("index")) | |
| safe_name = secure_filename(original_name) or "upload" | |
| stored_name = f"{uuid.uuid4().hex}_{safe_name}" | |
| stored_path = os.path.join(UPLOAD_FOLDER, stored_name) | |
| request.files["file"].save(stored_path) | |
| try: | |
| try: | |
| _sniff_content(stored_path, kind) | |
| except ValueError: | |
| raise | |
| except Exception: | |
| raise ValueError("The uploaded file could not be decoded.") | |
| from detector import analyze_image, analyze_video # lazy: heavy imports | |
| result = analyze_image(stored_path) if kind == "image" \ | |
| else analyze_video(stored_path) | |
| except ValueError as exc: | |
| flash(str(exc), "error") | |
| return redirect(url_for("index")) | |
| except Exception: | |
| app.logger.exception("Detection failed") | |
| flash("Something went wrong while analyzing the file. " | |
| "Please try another one.", "error") | |
| return redirect(url_for("index")) | |
| finally: | |
| try: | |
| if os.path.exists(stored_path): | |
| os.remove(stored_path) | |
| except OSError: | |
| pass | |
| det_id = database.insert_detection( | |
| filename=safe_name, | |
| file_type=kind, | |
| prediction=result["prediction"], | |
| confidence=result["confidence"], | |
| frames_analyzed=result.get("frames_analyzed", 0), | |
| processing_time=result.get("processing_time", 0.0), | |
| ) | |
| record = { | |
| "id": det_id, | |
| "filename": safe_name, | |
| "file_type": kind, | |
| **result, | |
| } | |
| return render_template("result.html", r=record) | |
| def history(): | |
| stats = database.get_stats() | |
| rows = database.get_history(limit=50) | |
| return render_template("history.html", stats=stats, rows=rows) | |
| def clear_history(): | |
| database.clear_history() | |
| flash("Detection history cleared.", "success") | |
| return redirect(url_for("history")) | |
| def report(det_id): | |
| record = database.get_detection(det_id) | |
| if record is None: | |
| abort(404) | |
| from pdf_generator import build_report # lazy: heavy imports | |
| pdf_stream = build_report(record) | |
| download_name = f"deepfake_report_{det_id:06d}.pdf" | |
| return send_file( | |
| pdf_stream, | |
| mimetype="application/pdf", | |
| as_attachment=True, | |
| download_name=download_name, | |
| ) | |
| def health(): | |
| return {"status": "ok"}, 200 | |
| def not_found(_e): | |
| return render_template("index.html", stats=database.get_stats()), 404 | |
| def too_large(_e): | |
| flash("Upload rejected: file exceeds the maximum allowed size.", "error") | |
| return redirect(url_for("index")) | |
| if __name__ == "__main__": | |
| port = int(os.environ.get("PORT", 5000)) | |
| debug = os.environ.get("FLASK_DEBUG", "1") == "1" | |
| # Threaded so the UI stays responsive while the model runs on CPU. | |
| app.run(host="127.0.0.1", port=port, debug=debug, threaded=True) | |