Spaces:
Running
Running
crop to face, run at 380px, drop fabricated method field
Browse files- download_models.py +40 -13
- main.py +16 -45
- models/detector.py +43 -16
- routers/detect.py +124 -84
- routers/health.py +21 -4
- utils/__init__.py +0 -0
- utils/face.py +89 -0
download_models.py
CHANGED
|
@@ -1,31 +1,58 @@
|
|
| 1 |
"""
|
| 2 |
-
|
| 3 |
-
|
|
|
|
|
|
|
|
|
|
| 4 |
"""
|
|
|
|
| 5 |
import os
|
|
|
|
|
|
|
|
|
|
| 6 |
from huggingface_hub import hf_hub_download
|
| 7 |
|
| 8 |
-
HF_REPO
|
| 9 |
MODELS_DIR = "models"
|
|
|
|
| 10 |
os.makedirs(MODELS_DIR, exist_ok=True)
|
| 11 |
|
| 12 |
-
|
|
|
|
| 13 |
dest = os.path.join(MODELS_DIR, save_as or hf_filename)
|
|
|
|
| 14 |
if os.path.exists(dest):
|
| 15 |
-
print(f"
|
| 16 |
-
return
|
| 17 |
-
|
|
|
|
| 18 |
path = hf_hub_download(
|
| 19 |
repo_id=HF_REPO,
|
| 20 |
filename=hf_filename,
|
| 21 |
token=os.getenv("HF_TOKEN"),
|
| 22 |
)
|
| 23 |
-
import shutil
|
| 24 |
shutil.copy(path, dest)
|
| 25 |
-
print(f"
|
|
|
|
|
|
|
| 26 |
|
| 27 |
if __name__ == "__main__":
|
| 28 |
-
|
| 29 |
-
|
| 30 |
-
|
| 31 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
"""
|
| 2 |
+
Fetch model weights from HuggingFace at image build time.
|
| 3 |
+
|
| 4 |
+
If this fails the container must not start. A backend running without the
|
| 5 |
+
fine-tuned checkpoint will happily serve predictions from an untrained head,
|
| 6 |
+
which is worse than being down — the numbers look real and aren't.
|
| 7 |
"""
|
| 8 |
+
|
| 9 |
import os
|
| 10 |
+
import shutil
|
| 11 |
+
import sys
|
| 12 |
+
|
| 13 |
from huggingface_hub import hf_hub_download
|
| 14 |
|
| 15 |
+
HF_REPO = "Sowaiba01/deepguard-ai"
|
| 16 |
MODELS_DIR = "models"
|
| 17 |
+
|
| 18 |
os.makedirs(MODELS_DIR, exist_ok=True)
|
| 19 |
|
| 20 |
+
|
| 21 |
+
def download(hf_filename: str, save_as: str | None = None) -> str:
|
| 22 |
dest = os.path.join(MODELS_DIR, save_as or hf_filename)
|
| 23 |
+
|
| 24 |
if os.path.exists(dest):
|
| 25 |
+
print(f"Already present: {dest}")
|
| 26 |
+
return dest
|
| 27 |
+
|
| 28 |
+
print(f"Downloading {hf_filename} -> {dest}")
|
| 29 |
path = hf_hub_download(
|
| 30 |
repo_id=HF_REPO,
|
| 31 |
filename=hf_filename,
|
| 32 |
token=os.getenv("HF_TOKEN"),
|
| 33 |
)
|
|
|
|
| 34 |
shutil.copy(path, dest)
|
| 35 |
+
print(f"Saved: {dest} ({os.path.getsize(dest) / 1e6:.1f} MB)")
|
| 36 |
+
return dest
|
| 37 |
+
|
| 38 |
|
| 39 |
if __name__ == "__main__":
|
| 40 |
+
try:
|
| 41 |
+
# The detector looks for the v2 filename first, then falls back to v1.
|
| 42 |
+
# Write both so either lookup succeeds.
|
| 43 |
+
v2 = download("efficientnet_b4_deepguard_v2.pth")
|
| 44 |
+
|
| 45 |
+
legacy = os.path.join(MODELS_DIR, "efficientnet_b4_deepguard.pth")
|
| 46 |
+
if not os.path.exists(legacy):
|
| 47 |
+
shutil.copy(v2, legacy)
|
| 48 |
+
print(f"Also wrote legacy name: {legacy}")
|
| 49 |
+
|
| 50 |
+
download("inswapper_128.onnx")
|
| 51 |
+
|
| 52 |
+
except Exception as e:
|
| 53 |
+
print(f"\nFATAL: could not download model weights: {e}")
|
| 54 |
+
print("The backend cannot serve real predictions without them.")
|
| 55 |
+
print("If the model repo is private, set HF_TOKEN as a Space secret.")
|
| 56 |
+
sys.exit(1)
|
| 57 |
+
|
| 58 |
+
print("\nAll models ready.")
|
main.py
CHANGED
|
@@ -2,29 +2,39 @@
|
|
| 2 |
DeepGuard AI - FastAPI Backend
|
| 3 |
Deepfake Detection & Generation API
|
| 4 |
"""
|
|
|
|
| 5 |
from fastapi import FastAPI
|
| 6 |
from fastapi.middleware.cors import CORSMiddleware
|
| 7 |
-
from fastapi.responses import HTMLResponse
|
| 8 |
from contextlib import asynccontextmanager
|
| 9 |
import os
|
| 10 |
from dotenv import load_dotenv
|
|
|
|
| 11 |
from routers import detect, generate, dataset, health
|
|
|
|
| 12 |
load_dotenv()
|
|
|
|
| 13 |
@asynccontextmanager
|
| 14 |
async def lifespan(app: FastAPI):
|
| 15 |
"""Load models on startup."""
|
| 16 |
-
print("DeepGuard
|
| 17 |
from models.detector import load_detection_model
|
| 18 |
-
|
| 19 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 20 |
yield
|
| 21 |
print("Shutting down...")
|
|
|
|
| 22 |
app = FastAPI(
|
| 23 |
title="DeepGuard AI API",
|
| 24 |
-
description="Deepfake
|
| 25 |
version="1.2.0",
|
| 26 |
lifespan=lifespan,
|
| 27 |
)
|
|
|
|
| 28 |
app.add_middleware(
|
| 29 |
CORSMiddleware,
|
| 30 |
allow_origins=os.getenv("ALLOWED_ORIGINS", "http://localhost:3000").split(","),
|
|
@@ -32,51 +42,12 @@ app.add_middleware(
|
|
| 32 |
allow_methods=["*"],
|
| 33 |
allow_headers=["*"],
|
| 34 |
)
|
| 35 |
-
# ------------------------------------------------------------------
|
| 36 |
-
# Landing page (registered BEFORE routers so it takes precedence over
|
| 37 |
-
# any "/" route defined in a router)
|
| 38 |
-
# ------------------------------------------------------------------
|
| 39 |
-
LANDING_HTML = """<!DOCTYPE html>
|
| 40 |
-
<html lang="en">
|
| 41 |
-
<head>
|
| 42 |
-
<meta charset="utf-8">
|
| 43 |
-
<meta name="viewport" content="width=device-width, initial-scale=1">
|
| 44 |
-
<title>DeepGuard AI - Deepfake Detection</title>
|
| 45 |
-
</head>
|
| 46 |
-
<body style="margin:0;font-family:system-ui,-apple-system,sans-serif;background:#0e0b1f;color:#ece9f7;text-align:center;padding:8vh 1.5rem 3rem">
|
| 47 |
-
<h1 style="font-size:2.4rem;margin-bottom:0.4rem">DeepGuard AI</h1>
|
| 48 |
-
<p style="color:#a89fc9;max-width:36rem;margin:0 auto 2.2rem;line-height:1.6">
|
| 49 |
-
Deepfake (face-swap) detection powered by <b>EfficientNet-B4</b> —
|
| 50 |
-
<b>91.54%</b> validation accuracy on the 140K Real & Fake Faces dataset.
|
| 51 |
-
Includes a paired deepfake dataset of <b>10,852 images</b> generated with
|
| 52 |
-
InsightFace inswapper_128 for training and benchmarking detectors.
|
| 53 |
-
</p>
|
| 54 |
-
<a href="https://deep-guard-xai.vercel.app" target="_blank" rel="noopener" style="display:inline-block;background:#8b5cf6;color:#12081f;font-weight:700;padding:0.9rem 2.4rem;border-radius:10px;text-decoration:none;font-size:1.15rem">
|
| 55 |
-
Open Live Demo
|
| 56 |
-
</a>
|
| 57 |
-
<p style="margin-top:2.2rem">
|
| 58 |
-
<a href="/docs" style="color:#b9a6ff;text-decoration:none">API Docs</a>
|
| 59 |
-
·
|
| 60 |
-
<a href="https://github.com/Sowaiba-01/DeepGuard-XAI" target="_blank" rel="noopener" style="color:#b9a6ff;text-decoration:none">GitHub</a>
|
| 61 |
-
·
|
| 62 |
-
<a href="https://huggingface.co/Sowaiba01/deepguard-ai" target="_blank" rel="noopener" style="color:#b9a6ff;text-decoration:none">Model</a>
|
| 63 |
-
·
|
| 64 |
-
<a href="https://huggingface.co/datasets/Sowaiba01/Deepfake" target="_blank" rel="noopener" style="color:#b9a6ff;text-decoration:none">Dataset</a>
|
| 65 |
-
</p>
|
| 66 |
-
<p style="color:#6b6187;font-size:0.8rem;margin-top:3rem">
|
| 67 |
-
For research use only. Generation endpoints are intended for benchmarking detection models.
|
| 68 |
-
</p>
|
| 69 |
-
</body>
|
| 70 |
-
</html>"""
|
| 71 |
|
| 72 |
-
|
| 73 |
-
@app.get("/", response_class=HTMLResponse, include_in_schema=False)
|
| 74 |
-
async def root() -> HTMLResponse:
|
| 75 |
-
return HTMLResponse(content=LANDING_HTML)
|
| 76 |
app.include_router(health.router, tags=["health"])
|
| 77 |
app.include_router(detect.router, prefix="/detect", tags=["detection"])
|
| 78 |
app.include_router(generate.router, prefix="/generate", tags=["generation"])
|
| 79 |
app.include_router(dataset.router, prefix="/dataset", tags=["dataset"])
|
|
|
|
| 80 |
if __name__ == "__main__":
|
| 81 |
import uvicorn
|
| 82 |
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)
|
|
|
|
| 2 |
DeepGuard AI - FastAPI Backend
|
| 3 |
Deepfake Detection & Generation API
|
| 4 |
"""
|
| 5 |
+
|
| 6 |
from fastapi import FastAPI
|
| 7 |
from fastapi.middleware.cors import CORSMiddleware
|
|
|
|
| 8 |
from contextlib import asynccontextmanager
|
| 9 |
import os
|
| 10 |
from dotenv import load_dotenv
|
| 11 |
+
|
| 12 |
from routers import detect, generate, dataset, health
|
| 13 |
+
|
| 14 |
load_dotenv()
|
| 15 |
+
|
| 16 |
@asynccontextmanager
|
| 17 |
async def lifespan(app: FastAPI):
|
| 18 |
"""Load models on startup."""
|
| 19 |
+
print("DeepGuard backend starting...")
|
| 20 |
from models.detector import load_detection_model
|
| 21 |
+
|
| 22 |
+
detector, weights_loaded = load_detection_model()
|
| 23 |
+
app.state.detector = detector
|
| 24 |
+
app.state.detector_weights_loaded = weights_loaded
|
| 25 |
+
|
| 26 |
+
print("Detection model ready" if weights_loaded
|
| 27 |
+
else "Detection model running WITHOUT fine-tuned weights")
|
| 28 |
yield
|
| 29 |
print("Shutting down...")
|
| 30 |
+
|
| 31 |
app = FastAPI(
|
| 32 |
title="DeepGuard AI API",
|
| 33 |
+
description="Deepfake detection and generation API powered by EfficientNet-B4 and SimSwap",
|
| 34 |
version="1.2.0",
|
| 35 |
lifespan=lifespan,
|
| 36 |
)
|
| 37 |
+
|
| 38 |
app.add_middleware(
|
| 39 |
CORSMiddleware,
|
| 40 |
allow_origins=os.getenv("ALLOWED_ORIGINS", "http://localhost:3000").split(","),
|
|
|
|
| 42 |
allow_methods=["*"],
|
| 43 |
allow_headers=["*"],
|
| 44 |
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 45 |
|
|
|
|
|
|
|
|
|
|
|
|
|
| 46 |
app.include_router(health.router, tags=["health"])
|
| 47 |
app.include_router(detect.router, prefix="/detect", tags=["detection"])
|
| 48 |
app.include_router(generate.router, prefix="/generate", tags=["generation"])
|
| 49 |
app.include_router(dataset.router, prefix="/dataset", tags=["dataset"])
|
| 50 |
+
|
| 51 |
if __name__ == "__main__":
|
| 52 |
import uvicorn
|
| 53 |
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)
|
models/detector.py
CHANGED
|
@@ -1,6 +1,9 @@
|
|
| 1 |
"""
|
| 2 |
-
EfficientNet-B4
|
| 3 |
-
|
|
|
|
|
|
|
|
|
|
| 4 |
"""
|
| 5 |
|
| 6 |
import os
|
|
@@ -45,26 +48,50 @@ class DeepfakeDetector(nn.Module):
|
|
| 45 |
def load_detection_model(
|
| 46 |
checkpoint_path: str | None = None,
|
| 47 |
device: str | None = None,
|
| 48 |
-
) -> DeepfakeDetector:
|
| 49 |
"""
|
| 50 |
-
Load the
|
| 51 |
-
|
| 52 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 53 |
"""
|
| 54 |
device = device or os.getenv("DEVICE", "cpu")
|
| 55 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 56 |
|
| 57 |
model = DeepfakeDetector(pretrained=True)
|
|
|
|
| 58 |
|
| 59 |
-
|
| 60 |
-
|
| 61 |
-
|
| 62 |
-
|
| 63 |
-
|
| 64 |
-
|
| 65 |
-
|
| 66 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 67 |
|
| 68 |
model = model.to(device)
|
| 69 |
model.eval()
|
| 70 |
-
return model
|
|
|
|
| 1 |
"""
|
| 2 |
+
EfficientNet-B4 face-swap detector.
|
| 3 |
+
|
| 4 |
+
Binary classifier: 0 = not swapped, 1 = swapped.
|
| 5 |
+
Fine-tuned on 10,852 face crops — 91.54% on the held-out set.
|
| 6 |
+
Trained at 380x380; inference must use the same resolution.
|
| 7 |
"""
|
| 8 |
|
| 9 |
import os
|
|
|
|
| 48 |
def load_detection_model(
|
| 49 |
checkpoint_path: str | None = None,
|
| 50 |
device: str | None = None,
|
| 51 |
+
) -> tuple[DeepfakeDetector, bool]:
|
| 52 |
"""
|
| 53 |
+
Load the detector.
|
| 54 |
+
|
| 55 |
+
Returns (model, weights_loaded). weights_loaded is False when no fine-tuned
|
| 56 |
+
checkpoint was found — the backbone is then ImageNet-pretrained with an
|
| 57 |
+
untrained head, so its output is arbitrary. Callers must surface that
|
| 58 |
+
rather than presenting the scores as real predictions.
|
| 59 |
+
|
| 60 |
+
Tries v2 first, then falls back to v1.
|
| 61 |
"""
|
| 62 |
device = device or os.getenv("DEVICE", "cpu")
|
| 63 |
+
|
| 64 |
+
candidates = []
|
| 65 |
+
if checkpoint_path:
|
| 66 |
+
candidates.append(checkpoint_path)
|
| 67 |
+
elif os.getenv("DETECTION_MODEL_PATH"):
|
| 68 |
+
candidates.append(os.getenv("DETECTION_MODEL_PATH"))
|
| 69 |
+
else:
|
| 70 |
+
candidates = [
|
| 71 |
+
"models/efficientnet_b4_deepguard_v2.pth",
|
| 72 |
+
"models/efficientnet_b4_deepguard.pth",
|
| 73 |
+
]
|
| 74 |
|
| 75 |
model = DeepfakeDetector(pretrained=True)
|
| 76 |
+
weights_loaded = False
|
| 77 |
|
| 78 |
+
for path in candidates:
|
| 79 |
+
if not os.path.exists(path):
|
| 80 |
+
continue
|
| 81 |
+
try:
|
| 82 |
+
state = torch.load(path, map_location=device, weights_only=True)
|
| 83 |
+
model.load_state_dict(state)
|
| 84 |
+
print(f" Loaded fine-tuned weights: {path}")
|
| 85 |
+
weights_loaded = True
|
| 86 |
+
break
|
| 87 |
+
except Exception as e:
|
| 88 |
+
print(f" Could not load {path}: {e}")
|
| 89 |
+
|
| 90 |
+
if not weights_loaded:
|
| 91 |
+
print(" WARNING: no fine-tuned checkpoint found.")
|
| 92 |
+
print(f" Looked in: {', '.join(candidates)}")
|
| 93 |
+
print(" Running on ImageNet weights with an untrained head — predictions are meaningless.")
|
| 94 |
|
| 95 |
model = model.to(device)
|
| 96 |
model.eval()
|
| 97 |
+
return model, weights_loaded
|
routers/detect.py
CHANGED
|
@@ -1,6 +1,9 @@
|
|
| 1 |
"""
|
| 2 |
-
Detection Router — EfficientNet-B4
|
| 3 |
-
|
|
|
|
|
|
|
|
|
|
| 4 |
"""
|
| 5 |
|
| 6 |
import time
|
|
@@ -11,20 +14,26 @@ import tempfile
|
|
| 11 |
import cv2
|
| 12 |
import numpy as np
|
| 13 |
from fastapi import APIRouter, UploadFile, File, HTTPException, Request, Form
|
| 14 |
-
from fastapi.responses import JSONResponse
|
| 15 |
from PIL import Image
|
| 16 |
import torch
|
| 17 |
import torch.nn.functional as F
|
| 18 |
|
|
|
|
|
|
|
| 19 |
router = APIRouter()
|
| 20 |
|
| 21 |
ALLOWED_IMAGE = {"image/jpeg", "image/png", "image/webp"}
|
| 22 |
ALLOWED_VIDEO = {"video/mp4", "video/avi", "video/quicktime", "video/webm"}
|
| 23 |
|
|
|
|
|
|
|
|
|
|
| 24 |
|
| 25 |
-
|
| 26 |
-
|
|
|
|
| 27 |
import torchvision.transforms as T
|
|
|
|
| 28 |
transform = T.Compose([
|
| 29 |
T.Resize((size, size)),
|
| 30 |
T.ToTensor(),
|
|
@@ -35,67 +44,83 @@ def preprocess_image(img: Image.Image, size: int = 224):
|
|
| 35 |
|
| 36 |
def compute_gradcam(model, tensor: torch.Tensor, device: str) -> np.ndarray:
|
| 37 |
"""
|
| 38 |
-
|
| 39 |
-
Returns
|
| 40 |
"""
|
| 41 |
model.eval()
|
| 42 |
|
| 43 |
activations: list[torch.Tensor] = []
|
| 44 |
gradients: list[torch.Tensor] = []
|
| 45 |
|
| 46 |
-
# Hook the last MBConv block — deepest spatial feature map before global pool
|
| 47 |
target_layer = model.backbone.blocks[-1]
|
| 48 |
|
| 49 |
-
|
| 50 |
-
|
| 51 |
-
)
|
| 52 |
-
bwd_handle = target_layer.register_full_backward_hook(
|
| 53 |
-
lambda m, grad_in, grad_out: gradients.append(grad_out[0])
|
| 54 |
-
)
|
| 55 |
|
| 56 |
-
# Forward (no torch.no_grad — need grads)
|
| 57 |
t = tensor.to(device)
|
| 58 |
logit = model(t)
|
| 59 |
|
| 60 |
-
# Backward w.r.t. the predicted class score
|
| 61 |
model.zero_grad()
|
| 62 |
logit.backward()
|
| 63 |
|
| 64 |
-
|
| 65 |
-
|
| 66 |
|
| 67 |
-
grads = gradients[0]
|
| 68 |
-
acts = activations[0]
|
| 69 |
|
| 70 |
-
# Global-average-pool the gradients → channel weights
|
| 71 |
weights = grads.mean(dim=[2, 3], keepdim=True)
|
| 72 |
-
|
| 73 |
-
# Weighted combination + ReLU
|
| 74 |
-
cam = F.relu((weights * acts).sum(dim=1)).squeeze() # [H, W]
|
| 75 |
cam = cam.detach().cpu().float().numpy()
|
| 76 |
|
| 77 |
-
# Normalize to [0, 1]
|
| 78 |
lo, hi = cam.min(), cam.max()
|
| 79 |
-
|
| 80 |
-
return cam
|
| 81 |
|
| 82 |
|
| 83 |
-
def
|
| 84 |
"""
|
| 85 |
-
|
| 86 |
-
|
|
|
|
|
|
|
|
|
|
| 87 |
"""
|
| 88 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 89 |
|
| 90 |
-
|
| 91 |
-
|
| 92 |
-
heatmap_bgr = cv2.applyColorMap(cam_u8, cv2.COLORMAP_JET)
|
| 93 |
-
heatmap_rgb = cv2.cvtColor(heatmap_bgr, cv2.COLOR_BGR2RGB)
|
| 94 |
|
| 95 |
-
|
| 96 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 97 |
|
| 98 |
-
# Encode to base64 PNG
|
| 99 |
buf = io.BytesIO()
|
| 100 |
Image.fromarray(blended).save(buf, format="PNG")
|
| 101 |
return base64.b64encode(buf.getvalue()).decode()
|
|
@@ -107,58 +132,64 @@ async def detect_deepfake(
|
|
| 107 |
file: UploadFile = File(...),
|
| 108 |
gradcam: bool = Form(False),
|
| 109 |
):
|
| 110 |
-
"""
|
| 111 |
-
Detect deepfake in image or video.
|
| 112 |
-
- gradcam=true → also returns GradCAM heatmap as base64 PNG (images only)
|
| 113 |
-
- video files → returns per-frame scores + aggregate verdict
|
| 114 |
-
"""
|
| 115 |
content_type = file.content_type or ""
|
| 116 |
is_image = content_type in ALLOWED_IMAGE
|
| 117 |
is_video = content_type in ALLOWED_VIDEO
|
| 118 |
|
| 119 |
if not (is_image or is_video):
|
| 120 |
-
raise HTTPException(400, "Unsupported file type. Use JPG, PNG, WEBP
|
| 121 |
|
| 122 |
-
start
|
| 123 |
-
content
|
| 124 |
detector = request.app.state.detector
|
| 125 |
device = next(detector.parameters()).device.type
|
| 126 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 127 |
# ── IMAGE ────────────────────────────────────────────────────────────────
|
| 128 |
if is_image:
|
| 129 |
-
img
|
| 130 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 131 |
|
| 132 |
gradcam_b64: str | None = None
|
|
|
|
|
|
|
|
|
|
| 133 |
|
| 134 |
if gradcam:
|
| 135 |
-
# GradCAM needs gradients — cannot use torch.no_grad()
|
| 136 |
-
detector.eval()
|
| 137 |
cam = compute_gradcam(detector, tensor, device)
|
| 138 |
-
gradcam_b64 = overlay_heatmap(
|
| 139 |
-
|
|
|
|
|
|
|
|
|
|
| 140 |
else:
|
| 141 |
-
detector.eval()
|
| 142 |
with torch.no_grad():
|
| 143 |
prob = torch.sigmoid(detector(tensor.to(device))).item()
|
| 144 |
|
| 145 |
label = "FAKE" if prob > 0.5 else "REAL"
|
| 146 |
confidence = prob * 100 if label == "FAKE" else (1 - prob) * 100
|
| 147 |
-
latency_ms = (time.time() - start) * 1000
|
| 148 |
-
|
| 149 |
-
regions = (
|
| 150 |
-
["Eye boundaries", "Jaw edge artifacts", "Skin texture transitions"]
|
| 151 |
-
if label == "FAKE" else []
|
| 152 |
-
)
|
| 153 |
|
| 154 |
return {
|
| 155 |
-
"label":
|
| 156 |
-
"confidence":
|
| 157 |
-
"model":
|
| 158 |
-
"
|
| 159 |
-
"
|
| 160 |
-
"
|
| 161 |
-
"
|
|
|
|
|
|
|
|
|
|
| 162 |
}
|
| 163 |
|
| 164 |
# ── VIDEO ────────────────────────────────────────────────────────────────
|
|
@@ -174,14 +205,14 @@ async def detect_deepfake(
|
|
| 174 |
|
| 175 |
try:
|
| 176 |
cap = cv2.VideoCapture(tmp_path)
|
| 177 |
-
fps
|
| 178 |
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
|
| 179 |
|
| 180 |
-
# Sample 1 frame per second, max 30 frames
|
| 181 |
sample_every = max(1, int(fps))
|
| 182 |
max_samples = 30
|
| 183 |
|
| 184 |
frame_results: list[dict] = []
|
|
|
|
| 185 |
frame_idx = 0
|
| 186 |
|
| 187 |
detector.eval()
|
|
@@ -190,9 +221,14 @@ async def detect_deepfake(
|
|
| 190 |
ret, frame = cap.read()
|
| 191 |
if not ret:
|
| 192 |
break
|
|
|
|
| 193 |
if frame_idx % sample_every == 0:
|
| 194 |
-
img
|
| 195 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
| 196 |
with torch.no_grad():
|
| 197 |
prob = torch.sigmoid(detector(tensor.to(device))).item()
|
| 198 |
|
|
@@ -201,7 +237,9 @@ async def detect_deepfake(
|
|
| 201 |
"timestamp": round(frame_idx / fps, 2),
|
| 202 |
"score": round(prob, 4),
|
| 203 |
"label": "FAKE" if prob > 0.5 else "REAL",
|
|
|
|
| 204 |
})
|
|
|
|
| 205 |
frame_idx += 1
|
| 206 |
|
| 207 |
cap.release()
|
|
@@ -210,7 +248,7 @@ async def detect_deepfake(
|
|
| 210 |
os.unlink(tmp_path)
|
| 211 |
|
| 212 |
if not frame_results:
|
| 213 |
-
raise HTTPException(400, "Could not extract frames from video.")
|
| 214 |
|
| 215 |
scores = [r["score"] for r in frame_results]
|
| 216 |
avg_prob = float(np.mean(scores))
|
|
@@ -218,21 +256,23 @@ async def detect_deepfake(
|
|
| 218 |
|
| 219 |
label = "FAKE" if avg_prob > 0.5 else "REAL"
|
| 220 |
confidence = avg_prob * 100 if label == "FAKE" else (1 - avg_prob) * 100
|
| 221 |
-
latency_ms = (time.time() - start) * 1000
|
| 222 |
|
| 223 |
return {
|
| 224 |
-
"label":
|
| 225 |
-
"confidence":
|
| 226 |
-
"model":
|
| 227 |
-
"
|
| 228 |
-
"
|
| 229 |
-
"
|
|
|
|
|
|
|
|
|
|
| 230 |
"video": {
|
| 231 |
-
"fps":
|
| 232 |
-
"total_frames":
|
| 233 |
-
"frames_analyzed":
|
| 234 |
-
"
|
| 235 |
-
"
|
|
|
|
| 236 |
},
|
| 237 |
-
"gradcam_b64": None,
|
| 238 |
}
|
|
|
|
| 1 |
"""
|
| 2 |
+
Detection Router — EfficientNet-B4 face-swap detection with GradCAM.
|
| 3 |
+
|
| 4 |
+
The model is a binary classifier: swapped vs not swapped. It has no head for
|
| 5 |
+
identifying which tool produced a swap, so this router does not report one.
|
| 6 |
+
Anything returned here is either a model output or derived from one.
|
| 7 |
"""
|
| 8 |
|
| 9 |
import time
|
|
|
|
| 14 |
import cv2
|
| 15 |
import numpy as np
|
| 16 |
from fastapi import APIRouter, UploadFile, File, HTTPException, Request, Form
|
|
|
|
| 17 |
from PIL import Image
|
| 18 |
import torch
|
| 19 |
import torch.nn.functional as F
|
| 20 |
|
| 21 |
+
from utils.face import crop_largest_face
|
| 22 |
+
|
| 23 |
router = APIRouter()
|
| 24 |
|
| 25 |
ALLOWED_IMAGE = {"image/jpeg", "image/png", "image/webp"}
|
| 26 |
ALLOWED_VIDEO = {"video/mp4", "video/avi", "video/quicktime", "video/webm"}
|
| 27 |
|
| 28 |
+
# Must match the training resolution. The model was fine-tuned at 380x380;
|
| 29 |
+
# running inference at a different scale silently degrades accuracy.
|
| 30 |
+
INPUT_SIZE = 380
|
| 31 |
|
| 32 |
+
|
| 33 |
+
def preprocess_image(img: Image.Image, size: int = INPUT_SIZE):
|
| 34 |
+
"""Resize, tensorise and normalise for EfficientNet-B4."""
|
| 35 |
import torchvision.transforms as T
|
| 36 |
+
|
| 37 |
transform = T.Compose([
|
| 38 |
T.Resize((size, size)),
|
| 39 |
T.ToTensor(),
|
|
|
|
| 44 |
|
| 45 |
def compute_gradcam(model, tensor: torch.Tensor, device: str) -> np.ndarray:
|
| 46 |
"""
|
| 47 |
+
GradCAM over the last MBConv block.
|
| 48 |
+
Returns an H×W array normalised to [0, 1].
|
| 49 |
"""
|
| 50 |
model.eval()
|
| 51 |
|
| 52 |
activations: list[torch.Tensor] = []
|
| 53 |
gradients: list[torch.Tensor] = []
|
| 54 |
|
|
|
|
| 55 |
target_layer = model.backbone.blocks[-1]
|
| 56 |
|
| 57 |
+
fwd = target_layer.register_forward_hook(lambda m, i, o: activations.append(o))
|
| 58 |
+
bwd = target_layer.register_full_backward_hook(lambda m, gi, go: gradients.append(go[0]))
|
|
|
|
|
|
|
|
|
|
|
|
|
| 59 |
|
|
|
|
| 60 |
t = tensor.to(device)
|
| 61 |
logit = model(t)
|
| 62 |
|
|
|
|
| 63 |
model.zero_grad()
|
| 64 |
logit.backward()
|
| 65 |
|
| 66 |
+
fwd.remove()
|
| 67 |
+
bwd.remove()
|
| 68 |
|
| 69 |
+
grads = gradients[0]
|
| 70 |
+
acts = activations[0]
|
| 71 |
|
|
|
|
| 72 |
weights = grads.mean(dim=[2, 3], keepdim=True)
|
| 73 |
+
cam = F.relu((weights * acts).sum(dim=1)).squeeze()
|
|
|
|
|
|
|
| 74 |
cam = cam.detach().cpu().float().numpy()
|
| 75 |
|
|
|
|
| 76 |
lo, hi = cam.min(), cam.max()
|
| 77 |
+
return (cam - lo) / (hi - lo + 1e-8)
|
|
|
|
| 78 |
|
| 79 |
|
| 80 |
+
def regions_from_cam(cam: np.ndarray, threshold: float = 0.6) -> list[str]:
|
| 81 |
"""
|
| 82 |
+
Name the face zones the heatmap actually lit up.
|
| 83 |
+
|
| 84 |
+
The image fed to the model is a face crop, so a 3x3 grid over it maps
|
| 85 |
+
reasonably onto facial areas. This is coarse, but unlike a fixed list it
|
| 86 |
+
is derived from the model's own attention.
|
| 87 |
"""
|
| 88 |
+
h, w = cam.shape
|
| 89 |
+
if h < 3 or w < 3:
|
| 90 |
+
return []
|
| 91 |
+
|
| 92 |
+
zones = [
|
| 93 |
+
[("upper left of the face", "forehead and hairline", "upper right of the face")],
|
| 94 |
+
[("left eye and cheek", "nose and mid-face", "right eye and cheek")],
|
| 95 |
+
[("left jaw", "chin and mouth", "right jaw")],
|
| 96 |
+
]
|
| 97 |
|
| 98 |
+
hits: list[tuple[float, str]] = []
|
| 99 |
+
rh, rw = h // 3, w // 3
|
|
|
|
|
|
|
| 100 |
|
| 101 |
+
for r in range(3):
|
| 102 |
+
for c in range(3):
|
| 103 |
+
block = cam[r * rh:(r + 1) * rh, c * rw:(c + 1) * rw]
|
| 104 |
+
if block.size == 0:
|
| 105 |
+
continue
|
| 106 |
+
score = float(block.mean())
|
| 107 |
+
if score >= threshold:
|
| 108 |
+
hits.append((score, zones[r][0][c]))
|
| 109 |
+
|
| 110 |
+
hits.sort(reverse=True)
|
| 111 |
+
return [name for _, name in hits[:3]]
|
| 112 |
+
|
| 113 |
+
|
| 114 |
+
def overlay_heatmap(img: Image.Image, cam: np.ndarray, alpha: float = 0.45) -> str:
|
| 115 |
+
"""Blend the heatmap over the image and return base64 PNG."""
|
| 116 |
+
size = INPUT_SIZE
|
| 117 |
+
orig = np.array(img.convert("RGB").resize((size, size)))
|
| 118 |
+
|
| 119 |
+
cam_u8 = (cv2.resize(cam, (size, size)) * 255).astype(np.uint8)
|
| 120 |
+
heat = cv2.cvtColor(cv2.applyColorMap(cam_u8, cv2.COLORMAP_JET), cv2.COLOR_BGR2RGB)
|
| 121 |
+
|
| 122 |
+
blended = (orig * (1 - alpha) + heat * alpha).clip(0, 255).astype(np.uint8)
|
| 123 |
|
|
|
|
| 124 |
buf = io.BytesIO()
|
| 125 |
Image.fromarray(blended).save(buf, format="PNG")
|
| 126 |
return base64.b64encode(buf.getvalue()).decode()
|
|
|
|
| 132 |
file: UploadFile = File(...),
|
| 133 |
gradcam: bool = Form(False),
|
| 134 |
):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 135 |
content_type = file.content_type or ""
|
| 136 |
is_image = content_type in ALLOWED_IMAGE
|
| 137 |
is_video = content_type in ALLOWED_VIDEO
|
| 138 |
|
| 139 |
if not (is_image or is_video):
|
| 140 |
+
raise HTTPException(400, "Unsupported file type. Use JPG, PNG, WEBP or MP4.")
|
| 141 |
|
| 142 |
+
start = time.time()
|
| 143 |
+
content = await file.read()
|
| 144 |
detector = request.app.state.detector
|
| 145 |
device = next(detector.parameters()).device.type
|
| 146 |
|
| 147 |
+
# Set by load_detection_model — False means no fine-tuned checkpoint was
|
| 148 |
+
# found and the backbone is running on ImageNet weights, so any prediction
|
| 149 |
+
# is meaningless. Surface that rather than hide it.
|
| 150 |
+
weights_loaded = getattr(request.app.state, "detector_weights_loaded", True)
|
| 151 |
+
|
| 152 |
# ── IMAGE ────────────────────────────────────────────────────────────────
|
| 153 |
if is_image:
|
| 154 |
+
img = Image.open(io.BytesIO(content))
|
| 155 |
+
|
| 156 |
+
# The model was trained on face crops. Match that at inference time.
|
| 157 |
+
face_img, bbox = crop_largest_face(img)
|
| 158 |
+
face_detected = face_img is not None
|
| 159 |
+
model_input = face_img if face_detected else img
|
| 160 |
+
|
| 161 |
+
tensor = preprocess_image(model_input)
|
| 162 |
|
| 163 |
gradcam_b64: str | None = None
|
| 164 |
+
regions: list[str] = []
|
| 165 |
+
|
| 166 |
+
detector.eval()
|
| 167 |
|
| 168 |
if gradcam:
|
|
|
|
|
|
|
| 169 |
cam = compute_gradcam(detector, tensor, device)
|
| 170 |
+
gradcam_b64 = overlay_heatmap(model_input, cam)
|
| 171 |
+
with torch.no_grad():
|
| 172 |
+
prob = torch.sigmoid(detector(tensor.to(device))).item()
|
| 173 |
+
if prob > 0.5 and face_detected:
|
| 174 |
+
regions = regions_from_cam(cam)
|
| 175 |
else:
|
|
|
|
| 176 |
with torch.no_grad():
|
| 177 |
prob = torch.sigmoid(detector(tensor.to(device))).item()
|
| 178 |
|
| 179 |
label = "FAKE" if prob > 0.5 else "REAL"
|
| 180 |
confidence = prob * 100 if label == "FAKE" else (1 - prob) * 100
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 181 |
|
| 182 |
return {
|
| 183 |
+
"label": label,
|
| 184 |
+
"confidence": round(confidence, 2),
|
| 185 |
+
"model": "EfficientNet-B4",
|
| 186 |
+
"latency_ms": round((time.time() - start) * 1000, 1),
|
| 187 |
+
"regions": regions,
|
| 188 |
+
"gradcam_b64": gradcam_b64,
|
| 189 |
+
"face_detected": face_detected,
|
| 190 |
+
"face_bbox": bbox,
|
| 191 |
+
"input_size": INPUT_SIZE,
|
| 192 |
+
"weights_loaded": weights_loaded,
|
| 193 |
}
|
| 194 |
|
| 195 |
# ── VIDEO ────────────────────────────────────────────────────────────────
|
|
|
|
| 205 |
|
| 206 |
try:
|
| 207 |
cap = cv2.VideoCapture(tmp_path)
|
| 208 |
+
fps = cap.get(cv2.CAP_PROP_FPS) or 30
|
| 209 |
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
|
| 210 |
|
|
|
|
| 211 |
sample_every = max(1, int(fps))
|
| 212 |
max_samples = 30
|
| 213 |
|
| 214 |
frame_results: list[dict] = []
|
| 215 |
+
frames_with_face = 0
|
| 216 |
frame_idx = 0
|
| 217 |
|
| 218 |
detector.eval()
|
|
|
|
| 221 |
ret, frame = cap.read()
|
| 222 |
if not ret:
|
| 223 |
break
|
| 224 |
+
|
| 225 |
if frame_idx % sample_every == 0:
|
| 226 |
+
img = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
|
| 227 |
+
face_img, _ = crop_largest_face(img)
|
| 228 |
+
if face_img is not None:
|
| 229 |
+
frames_with_face += 1
|
| 230 |
+
|
| 231 |
+
tensor = preprocess_image(face_img if face_img is not None else img)
|
| 232 |
with torch.no_grad():
|
| 233 |
prob = torch.sigmoid(detector(tensor.to(device))).item()
|
| 234 |
|
|
|
|
| 237 |
"timestamp": round(frame_idx / fps, 2),
|
| 238 |
"score": round(prob, 4),
|
| 239 |
"label": "FAKE" if prob > 0.5 else "REAL",
|
| 240 |
+
"face": face_img is not None,
|
| 241 |
})
|
| 242 |
+
|
| 243 |
frame_idx += 1
|
| 244 |
|
| 245 |
cap.release()
|
|
|
|
| 248 |
os.unlink(tmp_path)
|
| 249 |
|
| 250 |
if not frame_results:
|
| 251 |
+
raise HTTPException(400, "Could not extract frames from that video.")
|
| 252 |
|
| 253 |
scores = [r["score"] for r in frame_results]
|
| 254 |
avg_prob = float(np.mean(scores))
|
|
|
|
| 256 |
|
| 257 |
label = "FAKE" if avg_prob > 0.5 else "REAL"
|
| 258 |
confidence = avg_prob * 100 if label == "FAKE" else (1 - avg_prob) * 100
|
|
|
|
| 259 |
|
| 260 |
return {
|
| 261 |
+
"label": label,
|
| 262 |
+
"confidence": round(confidence, 2),
|
| 263 |
+
"model": "EfficientNet-B4",
|
| 264 |
+
"latency_ms": round((time.time() - start) * 1000, 1),
|
| 265 |
+
"regions": [],
|
| 266 |
+
"gradcam_b64": None,
|
| 267 |
+
"face_detected": frames_with_face > 0,
|
| 268 |
+
"input_size": INPUT_SIZE,
|
| 269 |
+
"weights_loaded": weights_loaded,
|
| 270 |
"video": {
|
| 271 |
+
"fps": round(fps, 1),
|
| 272 |
+
"total_frames": total_frames,
|
| 273 |
+
"frames_analyzed": len(frame_results),
|
| 274 |
+
"frames_with_face": frames_with_face,
|
| 275 |
+
"fake_ratio": round(fake_ratio * 100, 1),
|
| 276 |
+
"frames": frame_results,
|
| 277 |
},
|
|
|
|
| 278 |
}
|
routers/health.py
CHANGED
|
@@ -1,11 +1,28 @@
|
|
| 1 |
-
from fastapi import APIRouter
|
| 2 |
|
| 3 |
router = APIRouter()
|
| 4 |
|
|
|
|
| 5 |
@router.get("/")
|
| 6 |
async def root():
|
| 7 |
-
return {"status": "ok", "service": "DeepGuard
|
|
|
|
| 8 |
|
| 9 |
@router.get("/health")
|
| 10 |
-
async def health():
|
| 11 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
from fastapi import APIRouter, Request
|
| 2 |
|
| 3 |
router = APIRouter()
|
| 4 |
|
| 5 |
+
|
| 6 |
@router.get("/")
|
| 7 |
async def root():
|
| 8 |
+
return {"status": "ok", "service": "DeepGuard", "version": "2.0.0"}
|
| 9 |
+
|
| 10 |
|
| 11 |
@router.get("/health")
|
| 12 |
+
async def health(request: Request):
|
| 13 |
+
"""
|
| 14 |
+
Reports whether the fine-tuned checkpoint actually loaded. If it didn't,
|
| 15 |
+
the detector is running on ImageNet weights and its scores mean nothing —
|
| 16 |
+
worth knowing before trusting any result the API returns.
|
| 17 |
+
"""
|
| 18 |
+
weights_loaded = getattr(request.app.state, "detector_weights_loaded", False)
|
| 19 |
+
return {
|
| 20 |
+
"status": "healthy",
|
| 21 |
+
"detector": {
|
| 22 |
+
"loaded": True,
|
| 23 |
+
"fine_tuned_weights": weights_loaded,
|
| 24 |
+
"input_size": 380,
|
| 25 |
+
"note": None if weights_loaded
|
| 26 |
+
else "No fine-tuned checkpoint found — predictions are not meaningful.",
|
| 27 |
+
},
|
| 28 |
+
}
|
utils/__init__.py
ADDED
|
File without changes
|
utils/face.py
ADDED
|
@@ -0,0 +1,89 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
"""
|
| 2 |
+
Face detection and cropping.
|
| 3 |
+
|
| 4 |
+
The detector was fine-tuned on face crops, so feeding it a full photograph
|
| 5 |
+
puts it out of distribution — most of the input becomes background, and the
|
| 6 |
+
prediction degrades badly. Everything here exists to make the inference input
|
| 7 |
+
look like the training input.
|
| 8 |
+
"""
|
| 9 |
+
|
| 10 |
+
import os
|
| 11 |
+
import numpy as np
|
| 12 |
+
from PIL import Image
|
| 13 |
+
|
| 14 |
+
_face_app = None
|
| 15 |
+
_load_failed = False
|
| 16 |
+
|
| 17 |
+
|
| 18 |
+
def get_face_detector():
|
| 19 |
+
"""
|
| 20 |
+
Lazy-load InsightFace's buffalo_l detector.
|
| 21 |
+
|
| 22 |
+
Returns None if InsightFace is unavailable or the model can't be fetched —
|
| 23 |
+
callers fall back to using the whole image and flag it in the response.
|
| 24 |
+
"""
|
| 25 |
+
global _face_app, _load_failed
|
| 26 |
+
|
| 27 |
+
if _face_app is not None:
|
| 28 |
+
return _face_app
|
| 29 |
+
if _load_failed:
|
| 30 |
+
return None
|
| 31 |
+
|
| 32 |
+
try:
|
| 33 |
+
from insightface.app import FaceAnalysis
|
| 34 |
+
|
| 35 |
+
model_dir = os.path.join(os.path.dirname(__file__), "..", "models")
|
| 36 |
+
root = os.path.abspath(os.path.join(model_dir, "..", ".insightface"))
|
| 37 |
+
|
| 38 |
+
app = FaceAnalysis(name="buffalo_l", root=root, providers=["CPUExecutionProvider"])
|
| 39 |
+
app.prepare(ctx_id=-1, det_size=(320, 320))
|
| 40 |
+
|
| 41 |
+
_face_app = app
|
| 42 |
+
return _face_app
|
| 43 |
+
|
| 44 |
+
except Exception as e:
|
| 45 |
+
print(f" Face detector unavailable ({e}) — falling back to full-image inference")
|
| 46 |
+
_load_failed = True
|
| 47 |
+
return None
|
| 48 |
+
|
| 49 |
+
|
| 50 |
+
def crop_largest_face(img: Image.Image, margin: float = 0.25):
|
| 51 |
+
"""
|
| 52 |
+
Find the largest face and return a cropped copy.
|
| 53 |
+
|
| 54 |
+
margin expands the box beyond the detected bounds, because swap artefacts
|
| 55 |
+
concentrate at the boundary — jaw, hairline, ears. A tight crop cuts off
|
| 56 |
+
exactly the evidence the model needs.
|
| 57 |
+
|
| 58 |
+
Returns (cropped_image, bbox) or (None, None) when no face is found.
|
| 59 |
+
"""
|
| 60 |
+
detector = get_face_detector()
|
| 61 |
+
if detector is None:
|
| 62 |
+
return None, None
|
| 63 |
+
|
| 64 |
+
rgb = np.array(img.convert("RGB"))
|
| 65 |
+
bgr = rgb[:, :, ::-1]
|
| 66 |
+
|
| 67 |
+
try:
|
| 68 |
+
faces = detector.get(bgr)
|
| 69 |
+
except Exception as e:
|
| 70 |
+
print(f" Face detection failed: {e}")
|
| 71 |
+
return None, None
|
| 72 |
+
|
| 73 |
+
if not faces:
|
| 74 |
+
return None, None
|
| 75 |
+
|
| 76 |
+
# Largest by area — the subject, not a bystander in the background
|
| 77 |
+
face = max(faces, key=lambda f: (f.bbox[2] - f.bbox[0]) * (f.bbox[3] - f.bbox[1]))
|
| 78 |
+
x1, y1, x2, y2 = face.bbox
|
| 79 |
+
|
| 80 |
+
w, h = x2 - x1, y2 - y1
|
| 81 |
+
x1 = max(0, int(x1 - w * margin))
|
| 82 |
+
y1 = max(0, int(y1 - h * margin))
|
| 83 |
+
x2 = min(img.width, int(x2 + w * margin))
|
| 84 |
+
y2 = min(img.height, int(y2 + h * margin))
|
| 85 |
+
|
| 86 |
+
if x2 <= x1 or y2 <= y1:
|
| 87 |
+
return None, None
|
| 88 |
+
|
| 89 |
+
return img.convert("RGB").crop((x1, y1, x2, y2)), (x1, y1, x2, y2)
|