Upload 7 files

.huggingface.yml

@@ -0,0 +1,6 @@
+sdk: docker
+app_port: 7860
+title: DeepGuard - Deepfake Detection API
+emoji: 🧠
+colorFrom: blue
+colorTo: indigo

Dockerfile

@@ -0,0 +1,20 @@
+# -----------------------------
+# 🧠 DeepGuard - Python ML Backend (Docker for Hugging Face)
+# -----------------------------
+
+FROM python:3.10-slim
+
+# Install dependencies
+RUN apt-get update && apt-get install -y \
+    ffmpeg libsm6 libxext6 git && \
+    rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+COPY . .
+
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Hugging Face Spaces uses port 7860
+EXPOSE 7860
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

image_model_core.py

@@ -0,0 +1,494 @@
+# # image_model_core.py
+# import os
+# import logging
+# import warnings
+# import numpy as np
+# from PIL import Image
+# import cv2
+# import torch
+# from transformers import AutoImageProcessor, AutoModelForImageClassification
+
+# warnings.filterwarnings("ignore")
+# logger = logging.getLogger(__name__)
+# logger.setLevel(logging.INFO)
+
+# device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+# logger.info(f"Using device: {device}")
+
+# # --- Face detector: prefer RetinaFace if installed, otherwise fallback to MTCNN ---
+# USE_RETINA = False
+# try:
+#     # retina-face package (pip install retina-face)
+#     from retinaface import RetinaFace
+#     USE_RETINA = True
+#     logger.info("Using RetinaFace for face detection (retina-face).")
+# except Exception:
+#     try:
+#         # alternative retinaface implementation
+#         from retinaface_pytorch import RetinaFaceDetector  # optional naming
+#         USE_RETINA = True
+#         logger.info("Using retinaface-pytorch for face detection.")
+#     except Exception:
+#         USE_RETINA = False
+
+# if not USE_RETINA:
+#     try:
+#         from facenet_pytorch import MTCNN
+#         mtcnn = MTCNN(keep_all=False, device=device)
+#         logger.info("RetinaFace not available — falling back to MTCNN.")
+#     except Exception:
+#         mtcnn = None
+#         logger.warning("No RetinaFace or MTCNN available — face detection will be very basic.")
+
+
+# # ---------- Models ----------
+# # Replace the invalid non-existing model id with a working prithiv model or other public deepfake models.
+# MODEL_PATHS = [
+#     # balanced ensemble: CNN-style deepfake (prithiv), ViT-based and BEiT-based
+#     "prithivMLmods/deepfake-detector-model-v1",   # (public prithiv variant) — fallback to a valid prithiv model
+#     "Wvolf/ViT_Deepfake_Detection",
+#     "microsoft/beit-large-patch16-224-pt22k-ft22k"
+# ]
+
+# models = []
+# processors = []
+# for mid in MODEL_PATHS:
+#     try:
+#         proc = AutoImageProcessor.from_pretrained(mid)
+#         model = AutoModelForImageClassification.from_pretrained(mid).to(device)
+#         model.eval()
+#         models.append(model)
+#         processors.append(proc)
+#         logger.info(f"✅ Loaded image model: {mid}")
+#     except Exception as e:
+#         logger.warning(f"⚠️ Failed to load model {mid}: {e}")
+
+# if len(models) == 0:
+#     logger.error("No models could be loaded. Please check MODEL_PATHS and internet / HF auth.")
+
+
+# # ---------- Heuristics (optimized) ----------
+# def _frequency_artifact_score(face_bgr):
+#     # faster but stable frequency heuristic
+#     gray = cv2.cvtColor(face_bgr, cv2.COLOR_BGR2GRAY)
+#     # downsample to small size for FFT to speed up
+#     small = cv2.resize(gray, (64, 64), interpolation=cv2.INTER_LINEAR)
+#     f = np.fft.fft2(small)
+#     fshift = np.fft.fftshift(f)
+#     mag = np.log(np.abs(fshift) + 1)
+#     high_freq = np.mean(mag[32:, 32:])
+#     return float(np.clip(high_freq / 6.0, 0, 1))
+
+
+# def _illumination_consistency(face_bgr):
+#     lab = cv2.cvtColor(face_bgr, cv2.COLOR_BGR2LAB)
+#     l_std = np.std(lab[:, :, 0])
+#     return float(np.clip(l_std / 64.0, 0, 1))
+
+
+# def _edge_density(face_bgr):
+#     gray = cv2.cvtColor(face_bgr, cv2.COLOR_BGR2GRAY)
+#     edges = cv2.Canny(gray, 80, 160)
+#     return float(np.clip(np.mean(edges) / 255.0 * 2.0, 0, 1))
+
+
+# def aggregate_heuristics(face_bgr):
+#     # compute all using the same precomputed gray if needed
+#     try:
+#         return float(np.mean([
+#             _frequency_artifact_score(face_bgr),
+#             _illumination_consistency(face_bgr),
+#             _edge_density(face_bgr)
+#         ]))
+#     except Exception as e:
+#         logger.warning(f"Heuristic error: {e}")
+#         return 0.0
+
+
+# # ---------- Face extraction (robust) ----------
+# def _detect_face_boxes(img_bgr):
+#     """
+#     Return list of bounding boxes in x1,y1,x2,y2 format.
+#     """
+#     h, w = img_bgr.shape[:2]
+#     boxes = []
+#     if USE_RETINA:
+#         try:
+#             # retinaface returns dict or list depending on implementation
+#             dets = RetinaFace.detect_faces(img_bgr, align=False)
+#             # for many retinaface wrappers dets is dict with keys being faceIDs
+#             if isinstance(dets, dict):
+#                 for k, v in dets.items():
+#                     bb = v.get("facial_area") or v.get("bbox") or None
+#                     if bb:
+#                         x1, y1, x2, y2 = bb
+#                         boxes.append([max(0, int(x1)), max(0, int(y1)), min(w, int(x2)), min(h, int(y2))])
+#             elif isinstance(dets, list):
+#                 for d in dets:
+#                     if len(d) >= 4:
+#                         x1, y1, x2, y2 = d[:4]
+#                         boxes.append([max(0, int(x1)), max(0, int(y1)), min(w, int(x2)), min(h, int(y2))])
+#         except Exception:
+#             # some retina wrappers expect RGB; attempt conversion fallback
+#             try:
+#                 dets = RetinaFace.detect_faces(cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB), align=False)
+#                 if isinstance(dets, dict):
+#                     for k, v in dets.items():
+#                         bb = v.get("facial_area") or v.get("bbox") or None
+#                         if bb:
+#                             x1, y1, x2, y2 = bb
+#                             boxes.append([max(0, int(x1)), max(0, int(y1)), min(w, int(x2)), min(h, int(y2))])
+#             except Exception as ex:
+#                 logger.warning(f"RetinaFace detect exception: {ex}")
+#     else:
+#         if mtcnn is not None:
+#             try:
+#                 # mtcnn.detect expects RGB
+#                 boxes_mt, _ = mtcnn.detect(cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB))
+#                 if boxes_mt is not None:
+#                     for b in boxes_mt:
+#                         x1, y1, x2, y2 = [int(max(0, val)) for val in b]
+#                         boxes.append([x1, y1, x2, y2])
+#             except Exception as e:
+#                 logger.warning(f"MTCNN detect failure: {e}")
+#     # clamp and filter
+#     clean_boxes = []
+#     for (x1, y1, x2, y2) in boxes:
+#         if x2 - x1 < 10 or y2 - y1 < 10:  # tiny box
+#             continue
+#         if x1 < 0 or y1 < 0 or x2 <= x1 or y2 <= y1:
+#             continue
+#         clean_boxes.append([x1, y1, x2, y2])
+#     return clean_boxes
+
+
+# def _extract_face_region(img_bgr):
+#     boxes = _detect_face_boxes(img_bgr)
+#     if not boxes:
+#         return None
+#     # pick the largest box
+#     boxes = sorted(boxes, key=lambda b: (b[2] - b[0]) * (b[3] - b[1]), reverse=True)
+#     x1, y1, x2, y2 = boxes[0]
+#     # safe clamp
+#     h, w = img_bgr.shape[:2]
+#     x1, y1, x2, y2 = max(0, x1), max(0, y1), min(w, x2), min(h, y2)
+#     face = img_bgr[y1:y2, x1:x2]
+#     if face is None or face.size == 0:
+#         return None
+#     face = cv2.resize(face, (224, 224), interpolation=cv2.INTER_AREA)
+#     return face
+
+
+# # ---------- Batched inference helper ----------
+# def _batched_model_predict(pil_images, batch_size=8):
+#     """
+#     pil_images: list[PIL.Image]
+#     returns: list of per-image composite scores between 0..1 where higher means more "fake"
+#     """
+#     if len(models) == 0:
+#         return [0.0] * len(pil_images)
+
+#     # For each model, produce per-image probabilities; then ensemble across models
+#     all_model_scores = []  # shape: (n_models, n_images)
+#     for model, proc in zip(models, processors):
+#         try:
+#             inputs = proc(images=pil_images, return_tensors="pt", padding=True).to(device)
+#             # If inputs are large, split by batch
+#             logits = None
+#             with torch.no_grad():
+#                 logits = model(**inputs).logits  # (batch, classes)
+#                 probs = torch.nn.functional.softmax(logits, dim=-1).cpu().numpy()
+#             # pick predicted class confidence mapped to "fakeness"
+#             id2label = model.config.id2label
+#             per_img_scores = []
+#             for p in probs:
+#                 label_idx = int(np.argmax(p))
+#                 label = str(id2label.get(str(label_idx), id2label.get(label_idx, "unknown"))).lower()
+#                 is_fake = any(k in label for k in ["fake", "manipulated", "forged", "edited"])
+#                 conf = float(p[label_idx])
+#                 score = conf if is_fake else 1.0 - conf
+#                 per_img_scores.append(score)
+#             all_model_scores.append(per_img_scores)
+#         except Exception as e:
+#             logger.warning(f"Model batch predict failed: {e}")
+#             # fallback: zeros
+#             all_model_scores.append([0.0] * len(pil_images))
+
+#     # ensemble across models
+#     all_model_scores = np.array(all_model_scores)  # shape (m, n)
+#     # weights proportional to number of models loaded (keep default relative weights)
+#     base_weights = np.array([0.4, 0.35, 0.25])[:all_model_scores.shape[0]]
+#     if base_weights.sum() == 0:
+#         base_weights = np.ones(all_model_scores.shape[0]) / all_model_scores.shape[0]
+#     else:
+#         base_weights = base_weights / base_weights.sum()
+#     weighted = np.dot(base_weights, all_model_scores)  # size n
+#     return weighted.tolist()
+
+
+# # ---------- Public API ----------
+# def predict_image(image_path):
+#     """
+#     Main image-level API (synchronous)
+#     Returns dict compatible with your existing responses:
+#       { "top": {"label": "fake"/"real", "score": 0.xx}, "model_score": ..., "heuristic_score": ..., "source":"image" }
+#     """
+#     try:
+#         img_bgr = cv2.imread(image_path)
+#         if img_bgr is None:
+#             return {"error": "cannot_read_image"}
+
+#         face = _extract_face_region(img_bgr)
+#         if face is None:
+#             # fallback: resize whole image
+#             try:
+#                 face = cv2.resize(img_bgr, (224, 224), interpolation=cv2.INTER_AREA)
+#             except Exception:
+#                 return {"error": "no_face_detected"}
+
+#         pil = Image.fromarray(cv2.cvtColor(face, cv2.COLOR_BGR2RGB))
+#         model_scores = _batched_model_predict([pil])  # returns list len 1
+#         model_score = float(model_scores[0])
+#         heuristic_score = aggregate_heuristics(face)
+#         final = float(np.clip(0.85 * model_score + 0.15 * heuristic_score, 0, 1))
+#         label = "fake" if final > 0.55 else "real"
+
+#         return {
+#             "top": {"label": label, "score": round(final, 4)},
+#             "model_score": round(model_score, 4),
+#             "heuristic_score": round(heuristic_score, 4),
+#             "source": "image"
+#         }
+#     except Exception as e:
+#         logger.exception("predict_image failed")
+#         return {"error": str(e)}
+
+
+# ############################3333333333333333333333333333333333333333333333333#############################################################################################################################################333333333333333333
+
+
+# image_model_core.py
+"""
+Image detection core (accuracy-first).
+- Uses RetinaFace preferred, otherwise MTCNN.
+- Runs batched inference (but for single image it's small).
+- Uses more model weight on model outputs (0.85) and heuristics 0.15.
+"""
+
+import os
+import logging
+import warnings
+import numpy as np
+from PIL import Image
+import cv2
+import torch
+from dotenv import load_dotenv
+from transformers import AutoImageProcessor, AutoModelForImageClassification
+
+
+warnings.filterwarnings("ignore")
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+load_dotenv()
+
+
+device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+logger.info(f"Device for image_model_core: {device}")
+
+# Prefer RetinaFace, else MTCNN (same approach as model_helper)
+USE_RETINA = False
+try:
+    from retinaface import RetinaFace
+    USE_RETINA = True
+    logger.info("Using RetinaFace for image face detection.")
+except Exception:
+    try:
+        from facenet_pytorch import MTCNN
+        mtcnn = MTCNN(keep_all=False, device=device)
+        logger.info("RetinaFace not available — falling back to MTCNN for image pipeline.")
+    except Exception:
+        mtcnn = None
+        logger.warning("No RetinaFace or MTCNN available — image face detection will be basic.")
+
+# models (same ensemble)
+MODEL_PATHS = [
+
+    os.getenv("IMAGE_MODEL_1"),
+    os.getenv("IMAGE_MODEL_2"),
+    os.getenv("IMAGE_MODEL_3")
+]
+
+models = []
+processors = []
+def load_image_models():
+    global models, processors
+    models = []
+    processors = []
+    for mid in MODEL_PATHS:
+        try:
+            proc = AutoImageProcessor.from_pretrained(mid, trust_remote_code=False)
+            model = AutoModelForImageClassification.from_pretrained(mid).to(device)
+            model.eval()
+            models.append(model)
+            processors.append(proc)
+            logger.info(f"✅ Loaded image model: {mid.split('/')[-1]}")
+        except Exception as e:
+            logger.warning(f"⚠️ Failed to load model {mid}: {e}")
+
+load_image_models()
+if len(models) == 0:
+    logger.error("No image models loaded. Image detection disabled until models are present.")
+
+
+# --------------- heuristics ----------------
+def _frequency_artifact_score(face_bgr):
+    gray = cv2.cvtColor(face_bgr, cv2.COLOR_BGR2GRAY)
+    small = cv2.resize(gray, (64,64), interpolation=cv2.INTER_LINEAR)
+    f = np.fft.fft2(small)
+    fshift = np.fft.fftshift(f)
+    mag = np.log(np.abs(fshift) + 1)
+    high_freq = np.mean(mag[32:, 32:]) if mag.shape[0] > 32 else np.mean(mag)
+    return float(np.clip(high_freq / 6.0, 0, 1))
+
+
+def _illumination_consistency(face_bgr):
+    lab = cv2.cvtColor(face_bgr, cv2.COLOR_BGR2LAB)
+    l_std = np.std(lab[:,:,0])
+    return float(np.clip(l_std / 64.0, 0, 1))
+
+
+def _edge_density(face_bgr):
+    gray = cv2.cvtColor(face_bgr, cv2.COLOR_BGR2GRAY)
+    edges = cv2.Canny(gray, 80, 160)
+    return float(np.clip(np.mean(edges) / 255.0 * 2.0, 0, 1))
+
+
+def aggregate_heuristics(face_bgr):
+    try:
+        return float(np.mean([_frequency_artifact_score(face_bgr),
+                              _illumination_consistency(face_bgr),
+                              _edge_density(face_bgr)]))
+    except Exception as e:
+        logger.warning(f"Heuristic error: {e}")
+        return 0.0
+
+
+# ---------------- face extraction -------------
+def _detect_face_boxes(img_bgr):
+    h,w = img_bgr.shape[:2]
+    boxes = []
+    if USE_RETINA:
+        try:
+            dets = RetinaFace.detect_faces(img_bgr, align=False)
+            if isinstance(dets, dict):
+                for k,v in dets.items():
+                    bb = v.get("facial_area") or v.get("bbox")
+                    if bb:
+                        x1,y1,x2,y2 = bb
+                        boxes.append([max(0,int(x1)), max(0,int(y1)), min(w,int(x2)), min(h,int(y2))])
+            elif isinstance(dets, list):
+                for d in dets:
+                    if len(d) >= 4:
+                        x1,y1,x2,y2 = d[:4]
+                        boxes.append([max(0,int(x1)), max(0,int(y1)), min(w,int(x2)), min(h,int(y2))])
+        except Exception as e:
+            logger.debug(f"RetinaFace detection error (image): {e}")
+    elif 'mtcnn' in globals() and mtcnn is not None:
+        try:
+            boxes_mt, _ = mtcnn.detect(cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB))
+            if boxes_mt is not None:
+                for b in boxes_mt:
+                    x1,y1,x2,y2 = [int(max(0,val)) for val in b]
+                    boxes.append([x1,y1,x2,y2])
+        except Exception as e:
+            logger.debug(f"MTCNN detection error (image): {e}")
+    cleaned = []
+    for x1,y1,x2,y2 in boxes:
+        if x2-x1 < 12 or y2-y1 < 12: continue
+        if x1<0 or y1<0 or x2<=x1 or y2<=y1: continue
+        cleaned.append([x1,y1,x2,y2])
+    return cleaned
+
+
+def _extract_face_region(img_bgr):
+    boxes = _detect_face_boxes(img_bgr)
+    if not boxes:
+        return None
+    boxes = sorted(boxes, key=lambda b: (b[2]-b[0])*(b[3]-b[1]), reverse=True)
+    x1,y1,x2,y2 = boxes[0]
+    h,w = img_bgr.shape[:2]
+    x1,y1,x2,y2 = max(0,x1), max(0,y1), min(w,x2), min(h,y2)
+    face = img_bgr[y1:y2, x1:x2]
+    if face is None or face.size == 0: return None
+    face = cv2.resize(face, (224,224), interpolation=cv2.INTER_AREA)
+    return face
+
+
+# ---------------- batched inference helper -------------
+def _batched_model_predict(pil_images):
+    if len(models) == 0:
+        return [0.0] * len(pil_images)
+    per_model_outputs = []
+    for model, proc in zip(models, processors):
+        try:
+            inputs = proc(images=pil_images, return_tensors="pt", padding=True).to(device)
+            with torch.no_grad():
+                if device.type == "cuda":
+                    with torch.cuda.amp.autocast():
+                        logits = model(**inputs).logits
+                else:
+                    logits = model(**inputs).logits
+                probs = torch.nn.functional.softmax(logits, dim=-1).cpu().numpy()
+            id2label = getattr(model.config, "id2label", {}) or {}
+            out_scores = []
+            for p in probs:
+                idx = int(np.argmax(p))
+                label = str(id2label.get(str(idx), id2label.get(idx, "unknown"))).lower()
+                is_fake = any(k in label for k in ["fake","manipulated","forged","edited"])
+                conf = float(p[idx])
+                out_scores.append(conf if is_fake else 1.0 - conf)
+            per_model_outputs.append(out_scores)
+        except Exception as e:
+            logger.warning(f"Model batch predict failed (image): {e}")
+            per_model_outputs.append([0.0]*len(pil_images))
+
+    all_scores = np.array(per_model_outputs)
+    base_weights = np.array([0.4, 0.35, 0.25])[:all_scores.shape[0]]
+    if base_weights.sum() == 0:
+        base_weights = np.ones(all_scores.shape[0]) / all_scores.shape[0]
+    else:
+        base_weights = base_weights / base_weights.sum()
+    weighted = np.dot(base_weights, all_scores)
+    return weighted.tolist()
+
+
+# ---------------- public API ----------------
+def predict_image(image_path):
+    try:
+        img_bgr = cv2.imread(image_path)
+        if img_bgr is None:
+            return {"error": "cannot_read_image"}
+        face = _extract_face_region(img_bgr)
+        if face is None:
+            # fallback: whole image attempted
+            try:
+                face = cv2.resize(img_bgr, (224,224), interpolation=cv2.INTER_AREA)
+            except Exception:
+                return {"error": "no_face_detected"}
+
+        pil = Image.fromarray(cv2.cvtColor(face, cv2.COLOR_BGR2RGB))
+        model_scores = _batched_model_predict([pil])
+        model_score = float(model_scores[0])
+        heuristic_score = aggregate_heuristics(face)
+        final = float(np.clip(0.85 * model_score + 0.15 * heuristic_score, 0, 1))
+        label = "fake" if final > 0.55 else "real"
+        return {
+            "top": {"label": label, "score": round(final, 4)},
+            "model_score": round(model_score, 4),
+            "heuristic_score": round(heuristic_score, 4),
+            "source": "image"
+        }
+    except Exception as e:
+        logger.exception("predict_image failed")
+        return {"error": str(e)}

main.py

@@ -0,0 +1,157 @@
+
+
+#***********************************************************************************************************************************************
+
+
+from fastapi import FastAPI, UploadFile, File
+from fastapi.middleware.cors import CORSMiddleware
+import uvicorn, tempfile, cv2, numpy as np, logging
+from model_helper import ensemble_predict_from_path  
+from image_model_core import predict_image            
+
+# ------------------------------
+# ⚙️ App Setup
+# ------------------------------
+app = FastAPI(title="Deepfake Detection API", version="2.0")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"], 
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# ------------------------------
+# 🪵 Logging
+# ------------------------------
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+# ------------------------------
+# 🧩 Heuristic functions (for videos)
+# ------------------------------
+def compute_fft_artifact_score(frame):
+    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+    f = np.fft.fft2(gray)
+    fshift = np.fft.fftshift(f)
+    magnitude = 20 * np.log(np.abs(fshift) + 1)
+    high_freq = np.mean(magnitude[-20:, -20:])
+    return float(min(high_freq / 255.0, 1.0))
+
+def color_inconsistency_score(frame):
+    hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV)
+    h_std = np.std(hsv[:, :, 0])
+    return float(min(h_std / 90.0, 1.0))
+
+def edge_warp_score(frame):
+    edges = cv2.Canny(frame, 100, 200)
+    return float(min(np.mean(edges) / 255.0, 1.0))
+
+def aggregate_heuristics(frame):
+    fft_score = compute_fft_artifact_score(frame)
+    color_score = color_inconsistency_score(frame)
+    warp_score = edge_warp_score(frame)
+    return float(np.mean([fft_score, color_score, warp_score]))
+
+# ------------------------------
+# 🎥 Video Analysis Endpoint
+# ------------------------------
+@app.post("/analyze")
+async def analyze_video(file: UploadFile = File(...)):
+    logger.info(f"🎞️ Received video file: {file.filename}")
+
+    # Save uploaded video temporarily
+    with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmp:
+        tmp.write(await file.read())
+        video_path = tmp.name
+
+    cap = cv2.VideoCapture(video_path)
+    frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+    if frame_count == 0:
+        return {"error": "Unable to read video"}
+
+    sample_frames = max(1, frame_count // 10)
+    model_scores, heuristic_scores = [], []
+
+    for i in range(0, frame_count, sample_frames):
+        cap.set(cv2.CAP_PROP_POS_FRAMES, i)
+        ret, frame = cap.read()
+        if not ret:
+            continue
+
+        # --- Heuristic ---
+        h_score = aggregate_heuristics(frame)
+        heuristic_scores.append(h_score)
+
+        # --- Model ensemble prediction ---
+        temp_img_path = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False).name
+        cv2.imwrite(temp_img_path, frame)
+        preds = ensemble_predict_from_path(temp_img_path)
+        fake_score = preds["top"]["label"].lower() == "fake"
+        model_scores.append(float(preds["top"]["score"] if fake_score else 1 - preds["top"]["score"]))
+
+    cap.release()
+
+    final_model_score = float(np.mean(model_scores) if model_scores else 0.0)
+    final_heuristic_score = float(np.mean(heuristic_scores) if heuristic_scores else 0.0)
+    final_score = 0.7 * final_model_score + 0.3 * final_heuristic_score
+    is_fake = bool(final_score > 0.5)
+
+    logger.info(f"✅ Video analyzed: score={final_score:.4f}, fake={is_fake}")
+
+    return {
+        "source": "video",
+        "model_score": round(final_model_score, 4),
+        "heuristic_score": round(final_heuristic_score, 4),
+        "final_score": round(final_score, 4),
+        "is_deepfake": is_fake
+    }
+
+# ------------------------------
+# 🖼️ Image Analysis Endpoint
+# ------------------------------
+@app.post("/predict/image")
+async def analyze_image(file: UploadFile = File(...)):
+    logger.info(f"🖼️ Received image file: {file.filename}")
+    try:
+        with tempfile.NamedTemporaryFile(delete=False, suffix=".jpg") as tmp:
+            tmp.write(await file.read())
+            image_path = tmp.name
+
+        # 🔍 Run prediction
+        preds = predict_image(image_path)
+        if "error" in preds:
+            return {"error": preds["error"]}
+
+        model_score = preds.get("model_score", 0.0)
+        heuristic_score = preds.get("heuristic_score", 0.0)
+        final_score = preds["top"]["score"]
+        is_fake = preds["top"]["label"].lower() == "fake"
+
+        logger.info(f"✅ Image analyzed: score={final_score:.4f}, fake={is_fake}")
+
+        return {
+            "source": "image",
+            "model_score": round(model_score, 4),
+            "heuristic_score": round(heuristic_score, 4),
+            "final_score": round(final_score, 4),
+            "is_deepfake": is_fake
+        }
+
+    except Exception as e:
+        logger.exception("❌ Error during image analysis")
+        return {"error": str(e)}
+
+# ------------------------------
+# 🚀 Run Server
+# ------------------------------
+if __name__ == "__main__":
+    import os
+    port = int(os.environ.get("PORT", 8000))
+    uvicorn.run(app, host="0.0.0.0", port=port)
+
+

model_helper.py

@@ -0,0 +1,179 @@
+import torch, cv2, numpy as np
+from transformers import AutoImageProcessor, AutoModelForImageClassification
+from PIL import Image
+from facenet_pytorch import MTCNN
+from temporal_model import TemporalConsistencyModel
+import warnings, logging
+import os
+from dotenv import load_dotenv
+
+warnings.filterwarnings("ignore")
+
+# ---------- Logger Setup ----------
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    handlers=[logging.StreamHandler()]
+)
+logger = logging.getLogger(__name__)
+
+load_dotenv()
+
+device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+
+# ---------- Face Detector ----------
+face_detector = MTCNN(keep_all=False, device=device)
+
+# ---------- Temporal Model ----------
+temporal_model = TemporalConsistencyModel(window=7, alpha=0.75)
+
+# ---------- Model Definitions ----------
+MODEL_PATHS = [
+
+
+    os.getenv("VIDEO_MODEL_1"),
+    os.getenv("VIDEO_MODEL_2"),
+    os.getenv("VIDEO_MODEL_3")
+
+
+]
+
+models, processors = [], []
+for mid in MODEL_PATHS:
+    try:
+        proc = AutoImageProcessor.from_pretrained(mid)
+        model = AutoModelForImageClassification.from_pretrained(mid).to(device)
+        model.eval()
+        models.append(model)
+        processors.append(proc)
+        logger.info(f"✅ Loaded model: {mid}")
+    except Exception as e:
+        logger.warning(f"⚠️ Failed to load {mid}: {e}")
+
+# ---------- Heuristic ----------
+def heuristic_texture_analysis(frame):
+    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+    freq = np.fft.fft2(gray)
+    freq_shift = np.fft.fftshift(freq)
+    mag = np.log(np.abs(freq_shift) + 1)
+    edge_var = np.var(cv2.Laplacian(gray, cv2.CV_64F))
+    texture_score = np.mean(mag) / (edge_var + 1e-5)
+    norm_score = np.clip(np.tanh(texture_score / 60), 0, 1)
+    return float(norm_score)
+
+# ---------- Face Cropper (Fixed) ----------
+def extract_face(frame):
+    boxes, _ = face_detector.detect(frame)
+    if boxes is not None and len(boxes) > 0:
+        x1, y1, x2, y2 = [int(b) for b in boxes[0]]
+        face = frame[y1:y2, x1:x2]
+
+        if face is None or face.size == 0:
+            logger.warning("⚠️ Detected invalid face region; skipping frame.")
+            return None
+
+        return cv2.resize(face, (224, 224))
+    else:
+        logger.info("ℹ️ No face detected in this frame; skipping.")
+        return None
+
+# ---------- Prediction ----------
+def predict_frame(frame):
+    face_img = extract_face(frame)
+    if face_img is None:
+        return None  # skip frame gracefully
+
+    frame_img = Image.fromarray(cv2.cvtColor(face_img, cv2.COLOR_BGR2RGB))
+    preds = []
+
+    for model, proc in zip(models, processors):
+        try:
+            inputs = proc(images=frame_img, return_tensors="pt").to(device)
+            with torch.no_grad():
+                logits = model(**inputs).logits
+                probs = torch.nn.functional.softmax(logits, dim=-1)[0].cpu().numpy()
+
+            id2label = model.config.id2label
+            label_idx = np.argmax(probs)
+
+            if str(label_idx) in id2label:
+                label = id2label[str(label_idx)].lower()
+            elif label_idx in id2label:
+                label = id2label[label_idx].lower()
+            else:
+                label = "unknown"
+
+            is_fake = any(k in label for k in ["fake", "forged", "manipulated", "edited"])
+            confidence = float(probs[label_idx])
+
+            score = confidence if is_fake else 1 - confidence
+            preds.append(score)
+
+        except Exception as e:
+            logger.warning(f"⚠️ Model prediction failed for {model.__class__.__name__}: {e}")
+
+    if not preds:
+        logger.warning("⚠️ No valid model predictions; skipping frame.")
+        return None
+
+    # Weighted average (CNN:0.4, ViT:0.35, BEiT:0.25)
+    weights = np.array([0.4, 0.35, 0.25])[:len(preds)]
+    weights /= weights.sum()
+    weighted_score = np.dot(preds, weights)
+    return float(np.clip(weighted_score, 0, 1))
+
+# ---------- Main Pipeline ----------
+def ensemble_predict_video(video_path, frame_interval=10):
+    cap = cv2.VideoCapture(video_path)
+    frame_preds, heuristics = [], []
+    frame_count = 0
+
+    while True:
+        ret, frame = cap.read()
+        if not ret:
+            break
+
+        if frame_count % frame_interval == 0:
+            model_score = predict_frame(frame)
+            if model_score is None:
+                frame_count += 1
+                continue
+
+            heuristic_score = heuristic_texture_analysis(frame)
+            combined_score = 0.8 * model_score + 0.2 * heuristic_score
+            temporal_score = temporal_model.update(combined_score)
+
+            frame_preds.append(temporal_score)
+            heuristics.append(heuristic_score)
+
+        frame_count += 1
+
+    cap.release()
+
+    if not frame_preds:
+        logger.error("❌ No valid frames processed. Returning unknown result.")
+        return {"top": {"label": "unknown", "score": 0.0}}
+
+    model_score = float(np.mean(frame_preds))
+    heuristic_score = float(np.mean(heuristics))
+    final_score = float(np.clip(model_score, 0, 1))
+
+    logger.info(f"✅ Video processed | Final Score: {final_score:.4f}")
+
+    return {
+        "top": {
+            "label": "fake" if final_score > 0.55 else "real",
+            "score": round(final_score, 4)
+        },
+        "model_score": round(model_score, 4),
+        "heuristic_score": round(heuristic_score, 4),
+    }
+
+# ---------- Compatibility Wrapper ----------
+def ensemble_predict_from_path(video_path):
+    """Compatibility wrapper for main.py"""
+    return ensemble_predict_video(video_path)
+
+
+#***********************************************************************************************************************************************************************************************************************
+

requirements.txt

@@ -0,0 +1,18 @@
+fastapi
+uvicorn[standard]
+pillow
+opencv-python
+numpy
+torch
+torchvision
+timm
+transformers
+facenet-pytorch
+scipy
+python-multipart
+aiofiles
+ffmpeg-python
+imageio
+matplotlib
+scikit-image
+retina-face

temporal_model.py

@@ -0,0 +1,25 @@
+import numpy as np
+
+class TemporalConsistencyModel:
+    """
+    Simple temporal smoothing model to capture flicker and irregular changes.
+    Works as a moving average + penalty for inconsistent transitions.
+    """
+    def __init__(self, window=5, alpha=0.7):
+        self.window = window
+        self.alpha = alpha
+        self.history = []
+
+    def update(self, score):
+        self.history.append(score)
+        if len(self.history) > self.window:
+            self.history.pop(0)
+        smoothed = np.mean(self.history)
+        # penalize high oscillations
+        flicker_penalty = np.std(self.history)
+        final = (self.alpha * smoothed) - (0.5 * flicker_penalty)
+        return np.clip(final, 0, 1)
+
+
+####################################################################################################################################################3
+