Deploy deepfake detector backend

Dockerfile

@@ -0,0 +1,13 @@
+FROM python:3.10-slim
+
+WORKDIR /app
+
+COPY . .
+
+RUN pip install --no-cache-dir flask flask-cors onnxruntime numpy Pillow scipy torch timm opencv-python-headless scikit-learn joblib
+
+ENV PORT=7860
+
+EXPOSE 7860
+
+CMD ["python", "app.py"]

app.py

@@ -0,0 +1,229 @@
+"""
+Deepfake detector backend — with tuned threshold.
+
+Pipeline:
+    image (BGR)
+      ├── 4-channel tensor (RGB normalized + FFT power spectrum) → EfficientNet-B0 → 1280
+      ├── FFT azimuthal avg at 256×256                                          →  128
+      └── noise (gray - GaussianBlur) FFT azimuthal avg at 256×256              →  128
+                                                                                 ────
+                                                                                 1536  → StandardScaler → SVM (ONNX)
+
+IMPORTANT: The SVM was trained on StandardScaler-normalized features.
+           scaler.pkl must be present in the backend folder.
+           Without it, predictions collapse to 0%/100%.
+
+Threshold tuned to 0.35 based on independent test set analysis.
+
+Run:
+    pip install flask flask-cors onnxruntime torch timm opencv-python-headless scipy numpy pillow scikit-learn
+    python app.py
+"""
+
+from flask import Flask, request, jsonify
+from flask_cors import CORS
+import onnxruntime as ort
+import numpy as np
+import cv2
+import io
+import torch
+import torch.nn as nn
+import timm
+from torchvision import transforms
+from scipy import ndimage
+from PIL import Image
+import joblib
+import os
+
+app = Flask(__name__)
+CORS(app)
+
+# ---------------------------------------------------------------------------
+# Config
+# ---------------------------------------------------------------------------
+FFT_SIZE = 256
+AZ_BINS = FFT_SIZE // 2           # 128
+
+BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+CNN_WEIGHTS_PATH = os.path.join(BASE_DIR, "best_model.pth")
+SVM_ONNX_PATH    = os.path.join(BASE_DIR, "svm_linear_model.onnx")
+SCALER_PATH      = os.path.join(BASE_DIR, "scaler.pkl")
+
+DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+
+DECISION_THRESHOLD = 0.35
+CALIBRATION_TEMPERATURE = 3.0
+
+# ---------------------------------------------------------------------------
+# CNN feature extractor (4-channel EfficientNet-B0)
+# ---------------------------------------------------------------------------
+class DeepfakeDetector(nn.Module):
+    def __init__(self):
+        super().__init__()
+        self.backbone = timm.create_model("efficientnet_b0", pretrained=False, num_classes=0)
+        old_conv = self.backbone.conv_stem
+        new_conv = nn.Conv2d(
+            4, old_conv.out_channels,
+            kernel_size=old_conv.kernel_size,
+            stride=old_conv.stride,
+            padding=old_conv.padding,
+            bias=old_conv.bias is not None,
+        )
+        self.backbone.conv_stem = new_conv
+        self.head = nn.Sequential(
+            nn.Dropout(0.3), nn.Linear(1280, 256), nn.ReLU(),
+            nn.Dropout(0.2), nn.Linear(256, 1),
+        )
+
+    def forward(self, x):
+        return self.backbone(x)
+
+
+feature_model = DeepfakeDetector().to(DEVICE)
+state = torch.load(CNN_WEIGHTS_PATH, map_location=DEVICE, weights_only=False)
+feature_model.load_state_dict(state, strict=False)
+feature_model.eval()
+print("✅ CNN feature extractor loaded")
+
+svm_session = ort.InferenceSession(SVM_ONNX_PATH, providers=["CPUExecutionProvider"])
+SVM_INPUT_NAME = svm_session.get_inputs()[0].name
+print(f"✅ SVM ONNX loaded — input '{SVM_INPUT_NAME}', expects {svm_session.get_inputs()[0].shape}")
+print(f"✅ Decision threshold: {DECISION_THRESHOLD} (lowered from default 0.5)")
+print("\n--- SVM ONNX Outputs ---")
+for out in svm_session.get_outputs():
+    print(f"  name={out.name!r}  shape={out.shape}  type={out.type}")
+print("------------------------\n")
+
+if os.path.exists(SCALER_PATH):
+    feature_scaler = joblib.load(SCALER_PATH)
+    print(f"✅ StandardScaler loaded from {SCALER_PATH}")
+    print(f"   mean range: [{feature_scaler.mean_.min():.3f}, {feature_scaler.mean_.max():.3f}]")
+    print(f"   scale range: [{feature_scaler.scale_.min():.3f}, {feature_scaler.scale_.max():.3f}]")
+else:
+    feature_scaler = None
+    print(f"⚠️  WARNING: {SCALER_PATH} not found — predictions will be 0% or 100%!")
+
+torch_transform = transforms.Compose([
+    transforms.ToPILImage(),
+    transforms.Resize((224, 224)),
+    transforms.ToTensor(),
+    transforms.Normalize(mean=[0.485, 0.456, 0.406],
+                         std=[0.229, 0.224, 0.225]),
+])
+
+
+def compute_azimuthal_average(spectrum_2d: np.ndarray) -> np.ndarray:
+    h, w = spectrum_2d.shape
+    cy, cx = h // 2, w // 2
+    Y, X = np.ogrid[:h, :w]
+    r = np.sqrt((X - cx) ** 2 + (Y - cy) ** 2).astype(int)
+    max_r = min(cy, cx)
+    return ndimage.mean(spectrum_2d, labels=r, index=np.arange(0, max_r))
+
+
+def extract_combined_features(img_bgr: np.ndarray) -> np.ndarray:
+    img_rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
+    tensor_3ch = torch_transform(img_rgb)
+
+    gray224 = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2GRAY)
+    gray224 = cv2.resize(gray224, (224, 224)).astype(np.float32)
+    f = np.fft.fft2(gray224)
+    f_shift = np.fft.fftshift(f)
+    ps = np.log1p(np.abs(f_shift) ** 2)
+    ps_norm = (ps - ps.min()) / (ps.max() - ps.min() + 1e-8)
+    fft_ch = torch.tensor(ps_norm, dtype=torch.float32).unsqueeze(0)
+    tensor_4ch = torch.cat([tensor_3ch, fft_ch], dim=0).unsqueeze(0).to(DEVICE)
+
+    with torch.no_grad():
+        cnn_feat = feature_model(tensor_4ch).cpu().numpy().flatten()
+
+    gray = cv2.resize(cv2.cvtColor(img_bgr, cv2.COLOR_BGR2GRAY),
+                      (FFT_SIZE, FFT_SIZE)).astype(np.float32)
+
+    f2 = np.fft.fft2(gray)
+    f2_shift = np.fft.fftshift(f2)
+    power = np.log1p(np.abs(f2_shift) ** 2)
+    az_avg = compute_azimuthal_average(power)
+
+    blurred = cv2.GaussianBlur(gray, (5, 5), 1.0)
+    noise = gray - blurred
+    nf = np.fft.fft2(noise)
+    nf_shift = np.fft.fftshift(nf)
+    noise_power = np.log1p(np.abs(nf_shift) ** 2)
+    noise_az = compute_azimuthal_average(noise_power)
+
+    combined = np.concatenate([cnn_feat, az_avg, noise_az]).astype(np.float32)
+    assert combined.shape == (1536,), f"Bad feature dim: {combined.shape}"
+    return combined
+
+
+def get_confidence_level(p_fake: float) -> str:
+    distance = abs(p_fake - DECISION_THRESHOLD)
+    if distance < 0.10:
+        return "uncertain"
+    elif distance < 0.25:
+        return "low"
+    elif distance < 0.40:
+        return "medium"
+    else:
+        return "high"
+
+
+@app.route("/", methods=["GET"])
+def health():
+    return jsonify({"status": "ok", "message": "Deepfake detector backend is running."})
+
+
+@app.route("/analyze", methods=["POST"])
+def analyze():
+    if "image" not in request.files:
+        return jsonify({"error": "No image"}), 400
+
+    try:
+        raw = request.files["image"].read()
+        pil = Image.open(io.BytesIO(raw)).convert("RGB")
+        img_rgb = np.array(pil)
+        img_bgr = cv2.cvtColor(img_rgb, cv2.COLOR_RGB2BGR)
+
+        features = extract_combined_features(img_bgr).reshape(1, -1)
+
+        if feature_scaler is not None:
+            features = feature_scaler.transform(features).astype(np.float32)
+
+        label_arr, prob_list = svm_session.run(None, {SVM_INPUT_NAME: features})
+        for out in svm_session.get_outputs():
+            print(out.name, out.shape, out.type)
+        prob_dict = prob_list[0]
+        p_fake_raw = float(prob_dict[1])
+
+        eps = 1e-6
+        p_clipped = float(np.clip(p_fake_raw, eps, 1.0 - eps))
+        logit_p = np.log(p_clipped / (1.0 - p_clipped))
+        p_fake = float(1.0 / (1.0 + np.exp(-logit_p / CALIBRATION_TEMPERATURE)))
+        p_real = 1.0 - p_fake
+
+        is_fake = bool(int(label_arr[0]) == 1)
+        confidence = get_confidence_level(p_fake)
+
+        display_percent = round(p_fake * 100, 2)
+        print(f"DEBUG: raw={p_fake_raw:.4f}  temp_scaled={p_fake:.4f}  "
+              f"label={'FAKE' if is_fake else 'REAL'}  confidence={confidence}")
+
+        return jsonify({
+            "probability": display_percent,
+            "label": "AI Generated / Fake" if is_fake else "Authentic Media",
+            "is_fake": is_fake,
+            "confidence": confidence,
+            "p_real": round(p_real * 100, 2),
+            "p_fake": round(p_fake * 100, 2),
+            "threshold_used": DECISION_THRESHOLD,
+        })
+
+    except Exception as e:
+        import traceback; traceback.print_exc()
+        return jsonify({"error": f"Failed: {e}"}), 500
+
+
+if __name__ == "__main__":
+    port = int(os.environ.get("PORT", 5000))
+    app.run(host="0.0.0.0", port=port, debug=False)

best_model.pth

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:5f4f2b8f38d45ae4db43f571308f1c005673dd47efb5d3725c6ae1485836f30a
+size 17647947

check_weights.py

@@ -0,0 +1,39 @@
+import torch
+import torch.nn as nn
+import timm
+
+class DeepfakeDetector(nn.Module):
+    def __init__(self):
+        super().__init__()
+        self.backbone = timm.create_model("efficientnet_b0", pretrained=False, num_classes=0)
+        old_conv = self.backbone.conv_stem
+        new_conv = nn.Conv2d(4, old_conv.out_channels,
+            kernel_size=old_conv.kernel_size, stride=old_conv.stride,
+            padding=old_conv.padding, bias=old_conv.bias is not None)
+        self.backbone.conv_stem = new_conv
+        self.head = nn.Sequential(
+            nn.Dropout(0.3), nn.Linear(1280, 256), nn.ReLU(),
+            nn.Dropout(0.2), nn.Linear(256, 1))
+    def forward(self, x):
+        return self.backbone(x)
+
+model = DeepfakeDetector()
+state = torch.load("best_model.pth", map_location="cpu", weights_only=False)
+
+print(f"Keys in checkpoint: {len(state)}")
+print(f"Keys in model:      {len(model.state_dict())}")
+
+result = model.load_state_dict(state, strict=False)
+print(f"\nMissing keys (not loaded, stay random!): {len(result.missing_keys)}")
+for k in result.missing_keys[:20]:
+    print(f"  {k}")
+
+print(f"\nUnexpected keys (in file but unused): {len(result.unexpected_keys)}")
+for k in result.unexpected_keys[:20]:
+    print(f"  {k}")
+
+# Check sample tensor for sanity
+ckpt_first = list(state.keys())[0]
+model_first = list(model.state_dict().keys())[0]
+print(f"\nCheckpoint first key: {ckpt_first}  shape={state[ckpt_first].shape}")
+print(f"Model first key:      {model_first}  shape={model.state_dict()[model_first].shape}")

download_model.py

@@ -0,0 +1,16 @@
+import requests
+import os
+
+# Link to the specific v2 ONNX model
+url = "https://huggingface.co/onnx-community/Deep-Fake-Detector-v2-Model-ONNX/resolve/main/onnx/model.onnx"
+
+def download_model():
+    print("Downloading Deep-Fake-Detector-v2 (ONNX)... This may take a moment.")
+    response = requests.get(url, stream=True)
+    with open("model.onnx", "wb") as f:
+        for chunk in response.iter_content(chunk_size=8192):
+            f.write(chunk)
+    print("Download Complete! 'model.onnx' is ready.")
+
+if __name__ == "__main__":
+    download_model()

requirements.txt

@@ -0,0 +1,9 @@
+flask
+flask-cors
+onnxruntime
+numpy
+Pillow
+scipy
+torch
+timm
+opencv-python-headless

scaler.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:d033f1eadf524df6115c85e801ca028efd6786849750bbb43d5fdc01251b412d
+size 37479

svm_linear_model.onnx

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:3903d7ded59104307e8c419bfb11c00914ee64a5e1609c80d3659bf45a4b32a3
+size 1189973

testbatch.py

@@ -0,0 +1,116 @@
+"""
+Batch-test backend menggunakan satu folder dengan naming convention:
+    r1, r2, r3, ...  → real
+    f1, f2, f3, ...  → fake
+
+Usage:
+    1. Pastiin app.py jalan di terminal lain
+    2. python test_batch.py "C:/Users/yourname/Downloads"
+       atau
+       python test_batch.py ~/Downloads
+"""
+import os, sys, glob, re, requests
+from collections import Counter
+
+if len(sys.argv) < 2:
+    print("Usage: python test_batch.py /path/to/folder")
+    print('Windows example: python test_batch.py "C:\\Users\\YourName\\Downloads"')
+    print("Mac example:     python test_batch.py ~/Downloads")
+    sys.exit(1)
+
+folder = os.path.expanduser(sys.argv[1])
+URL = "http://localhost:5000/analyze"
+
+if not os.path.isdir(folder):
+    print(f"❌ Folder tidak ditemukan: {folder}")
+    sys.exit(1)
+
+# Cari semua gambar
+EXTS = ["jpg", "jpeg", "png", "webp"]
+all_files = []
+for ext in EXTS:
+    all_files.extend(glob.glob(os.path.join(folder, f"*.{ext}")))
+    all_files.extend(glob.glob(os.path.join(folder, f"*.{ext.upper()}")))
+
+# Klasifikasi berdasarkan prefix nama file (r1.jpg → real, f1.jpg → fake)
+pattern = re.compile(r"^([rf])(\d+)\.", re.IGNORECASE)
+labeled_files = []
+for fp in all_files:
+    name = os.path.basename(fp)
+    m = pattern.match(name)
+    if m:
+        prefix = m.group(1).lower()
+        true_label = "real" if prefix == "r" else "fake"
+        idx = int(m.group(2))
+        labeled_files.append((fp, true_label, idx))
+
+if not labeled_files:
+    print(f"❌ Gak ada file dengan format r1.jpg / f1.jpg dst di {folder}")
+    print("   File yang ada di folder:")
+    for fp in sorted(all_files)[:10]:
+        print(f"     {os.path.basename(fp)}")
+    sys.exit(1)
+
+# Sort: real dulu (sesuai index), terus fake
+labeled_files.sort(key=lambda x: (x[1], x[2]))
+print(f"✅ Ditemukan {len(labeled_files)} gambar berlabel di {folder}\n")
+
+stats = Counter()
+mistakes = []
+current_label = None
+
+for fp, true_label, idx in labeled_files:
+    # Print header tiap ganti kategori
+    if true_label != current_label:
+        count = sum(1 for _, t, _ in labeled_files if t == true_label)
+        print(f"\n=== {true_label.upper()} ({count} files) ===")
+        print(f"{'file':<25} {'pred':<6} {'p_fake':>9}  {'ok'}")
+        print("-" * 55)
+        current_label = true_label
+
+    try:
+        with open(fp, "rb") as f:
+            r = requests.post(URL, files={"image": f}, timeout=30)
+    except requests.exceptions.ConnectionError:
+        print(f"\n❌ Backend gak nyala di {URL}")
+        print("   Jalanin dulu: python app.py")
+        sys.exit(1)
+
+    if r.status_code != 200:
+        print(f"  ERROR on {fp}: {r.text}")
+        continue
+
+    d = r.json()
+    pred = "fake" if d["is_fake"] else "real"
+    ok = pred == true_label
+    stats[(true_label, pred)] += 1
+    if not ok:
+        mistakes.append((fp, true_label, pred, d["p_fake"]))
+    mark = "OK" if ok else "WRONG"
+    name = os.path.basename(fp)[:23]
+    print(f"{name:<25} {pred:<6} {d['p_fake']:>8.2f}%  {mark}")
+
+# Summary
+print("\n" + "=" * 55)
+print("CONFUSION MATRIX")
+print("=" * 55)
+print(f"{'':<10} {'pred_real':>12} {'pred_fake':>12}")
+print(f"{'true_real':<10} {stats[('real','real')]:>12} {stats[('real','fake')]:>12}")
+print(f"{'true_fake':<10} {stats[('fake','real')]:>12} {stats[('fake','fake')]:>12}")
+
+total = sum(stats.values())
+correct = stats[("real","real")] + stats[("fake","fake")]
+if total:
+    print(f"\nAccuracy: {correct}/{total} = {100*correct/total:.1f}%")
+
+n_real = stats[("real","real")] + stats[("real","fake")]
+n_fake = stats[("fake","real")] + stats[("fake","fake")]
+if n_real:
+    print(f"Real recall:  {stats[('real','real')]}/{n_real} = {100*stats[('real','real')]/n_real:.1f}%")
+if n_fake:
+    print(f"Fake recall:  {stats[('fake','fake')]}/{n_fake} = {100*stats[('fake','fake')]/n_fake:.1f}%")
+
+if mistakes:
+    print(f"\n{len(mistakes)} MISCLASSIFIED:")
+    for fp, t, p, pf in mistakes:
+        print(f"  {os.path.basename(fp):<25}  true={t:<5} pred={p:<5} p_fake={pf:.2f}%")