""" 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/ 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 # # ---------------------------------------------------------------------- # @app.route("/") def index(): stats = database.get_stats() return render_template("index.html", stats=stats) @app.route("/detect", methods=["POST"]) 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) @app.route("/history") def history(): stats = database.get_stats() rows = database.get_history(limit=50) return render_template("history.html", stats=stats, rows=rows) @app.route("/history/clear", methods=["POST"]) def clear_history(): database.clear_history() flash("Detection history cleared.", "success") return redirect(url_for("history")) @app.route("/report/") 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, ) @app.route("/health") def health(): return {"status": "ok"}, 200 @app.errorhandler(404) def not_found(_e): return render_template("index.html", stats=database.get_stats()), 404 @app.errorhandler(413) 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)