Sync from GitHub via hub-sync

api/v1/analyze.py

@@ -310,7 +310,17 @@ async def analyze_image(
     indicators = scan_artifacts(pil, raw)
     stages.append("artifact_scanning")
 
-    model_family = "efficientnet" if settings.ENSEMBLE_MODE else "vit"
+    # Heatmap dispatch: DenseNet leads for face still-images (GAN portraits),
+    # EfficientNet for video frames (face-swap / DFDC), ViT for no-face / fallback.
+    from services.image_service import _has_face_for_routing, _looks_like_video_frame
+    _face_for_heatmap    = _has_face_for_routing(pil_vis)
+    _videoframe_heatmap  = _looks_like_video_frame(pil_vis)
+    if _face_for_heatmap and settings.DENSENET_ENABLED and not _videoframe_heatmap:
+        model_family = "densenet"
+    elif settings.ENSEMBLE_MODE and (_face_for_heatmap or _videoframe_heatmap):
+        model_family = "efficientnet"
+    else:
+        model_family = "vit"
 
     # ── Run heatmap + ELA + boxes + EXIF in parallel ──
     def _run_heatmap():

config.py

@@ -202,16 +202,30 @@ class Settings(BaseSettings):
     FFPP_MODEL_REVISION: str = "main"
     FFPP_BASE_PROCESSOR_ID: str = "google/vit-base-patch16-224-in21k"
     FFPP_ENABLED: bool = True
-    # Ensemble weights — FFPP is trained on a better (face-specific FFPP c40) dataset
-    # and is weighted more heavily when a face is present. When no face is detected,
-    # we still blend it but lean on the generic ViT since FFPP only saw face crops.
-    # Face-stack internal weights (sum = 1.0). These compose the face-swap
-    # ensemble before it is fused with non-face evidence.
-    FFPP_WEIGHT_FACE: float = 0.55
-    VIT_WEIGHT_FACE: float = 0.20
-    EFFNET_WEIGHT_FACE: float = 0.25
+    # DenseNet121 face-GAN specialist (in-house trained on 140k Kaggle dataset).
+    # Loaded from a TF-free PyTorch checkpoint converted via convert_densenet_keras_to_pt.py.
+    DENSENET_ENABLED: bool = True
+    # Path to .pt checkpoint, resolved relative to repo root (or absolute).
+    DENSENET_MODEL_PATH: str = "backend/trained_models/densenet121_faces.pt"
+    DENSENET_META_PATH:  str = "backend/trained_models/densenet121_faces_meta.json"
+    # HF Space fallback when local checkpoint is absent.
+    DENSENET_HF_REPO_ID:  str = "ar07xd/deepshield"
+    DENSENET_HF_REVISION: str = "main"
+
+    # Ensemble weights — DenseNet leads because it is trained on still-image GAN
+    # faces (the dominant upload type). FFPP / EffNet are stronger on video frames.
+    # Face-stack internal weights (sum = 1.0).
+    DENSENET_WEIGHT_FACE: float = 0.45
+    FFPP_WEIGHT_FACE:     float = 0.25
+    VIT_WEIGHT_FACE:      float = 0.15
+    EFFNET_WEIGHT_FACE:   float = 0.15
+    # Video-frame path: FFPP leads since FFPP is trained on FF++ video frames.
+    DENSENET_VIDEO_WEIGHT: float = 0.10
+    VIDEO_FFPP_WEIGHT_FACE: float = 0.50
+    VIDEO_EFFNET_WEIGHT_FACE: float = 0.30
+    VIDEO_VIT_WEIGHT_FACE: float = 0.10
     FFPP_WEIGHT_NOFACE: float = 0.35
-    VIT_WEIGHT_NOFACE: float = 0.65
+    VIT_WEIGHT_NOFACE:  float = 0.65
 
     # Face-present unified evidence weights (Phase A2/A3).
     # face_stack = composite of FFPP+ViT+EffNet (all face-swap models).
@@ -251,8 +265,8 @@ class Settings(BaseSettings):
     # AI-image detector is unreliable (it's trained on synthesised stills, not
     # video face-swaps). We shift weight strongly toward the face-swap-trained
     # models (FFPP / EfficientNet) in that case.
-    VIDEO_FRAME_FACE_STACK_WEIGHT: float = 0.70
-    VIDEO_FRAME_GENERAL_WEIGHT: float = 0.15
+    VIDEO_FRAME_FACE_STACK_WEIGHT: float = 0.55
+    VIDEO_FRAME_GENERAL_WEIGHT: float = 0.30
     VIDEO_FRAME_FORENSICS_WEIGHT: float = 0.10
     VIDEO_FRAME_EXIF_WEIGHT: float = 0.05
     # Per-frame video detector blend. FFPP ViT is trained on FaceForensics++

models/heatmap_generator.py

@@ -206,16 +206,73 @@ def _cam_to_full_image(
     return cam_full, orig_np
 
 
+def _compute_gradcam_pp_densenet(
+    pil_img: Image.Image,
+) -> tuple[np.ndarray, str]:
+    """Grad-CAM++ on the DenseNet121 face-GAN model.
+
+    Target signal = fake probability = sigmoid(-logit), so we maximise the
+    negated logit. Target layer = features.norm5 (final BN after last DenseBlock,
+    7×7×1024 activation map). Returns (grayscale_cam, source_tag).
+    """
+    loader = get_model_loader()
+    result = loader.load_densenet()
+    if result is None:
+        raise RuntimeError("DenseNet model unavailable")
+    model, meta = result
+
+    from services.densenet_service import _preprocess
+    image_size = int(meta.get("image_size", 224))
+    input_tensor = _preprocess(pil_img, image_size, settings.DEVICE)
+
+    model.eval()
+    for p in model.parameters():
+        p.requires_grad_(True)
+
+    # Target = last BN after all DenseBlocks (equivalent to conv5_block16_concat in Keras)
+    target_layers = [model.features.norm5]
+
+    # Negate logit so Grad-CAM gradients flow toward the FAKE class
+    # (model output = real_probability logit; higher = more real)
+    class _NegatedLogitWrapper(torch.nn.Module):
+        def __init__(self, m: torch.nn.Module) -> None:
+            super().__init__()
+            self.m = m
+
+        def forward(self, x: torch.Tensor) -> torch.Tensor:  # type: ignore[override]
+            return -self.m(x)   # negative logit → gradient points at fake evidence
+
+    wrapped = _NegatedLogitWrapper(model)
+
+    with GradCAMPlusPlus(model=wrapped, target_layers=target_layers) as cam:
+        grayscale_cam = cam(input_tensor=input_tensor, targets=None)[0]  # (H,W) in [0,1]
+
+    return grayscale_cam, "gradcam++_densenet"
+
+
 def generate_heatmap_base64(
     pil_img: Image.Image,
     target_class_idx: Optional[int] = None,
-    model_family: Literal["vit", "efficientnet"] = "vit",
+    model_family: Literal["vit", "efficientnet", "densenet"] = "vit",
 ) -> tuple[str, str]:
     """Produce a base64 data-URL PNG of the Grad-CAM++ overlay at original image resolution.
 
     Returns (base64_png, heatmap_source).
     """
-    if model_family == "efficientnet":
+    if model_family == "densenet":
+        try:
+            grayscale_cam, source = _compute_gradcam_pp_densenet(pil_img)
+            cam_full, orig_np = _cam_to_full_image(grayscale_cam, pil_img, None)
+        except Exception as e:
+            logger.warning(f"DenseNet heatmap failed ({e}) — falling back to ViT Grad-CAM++")
+            try:
+                grayscale_cam, _ = _compute_gradcam_pp(pil_img, target_class_idx)
+                cam_full, orig_np = _cam_to_full_image(grayscale_cam, pil_img, None)
+                source = "vit_fallback"
+            except Exception as fe:
+                logger.warning(f"ViT fallback heatmap also failed: {fe}")
+                return "", "none"
+    elif model_family == "efficientnet":
         try:
             grayscale_cam, face_bbox, source = _compute_gradcam_pp_efficientnet(pil_img)
             cam_full, orig_np = _cam_to_full_image(grayscale_cam, pil_img, face_bbox)

models/model_loader.py

@@ -39,6 +39,9 @@ class ModelLoader:
                     cls._instance._efficientnet_detector = None
                     cls._instance._ffpp_model = None
                     cls._instance._ffpp_processor = None
+                    cls._instance._densenet_model = None
+                    cls._instance._densenet_meta = None
+                    cls._instance._densenet_unavailable = False
         return cls._instance
 
     @classmethod
@@ -321,6 +324,78 @@ class ModelLoader:
             logger.warning(f"FFPP ViT load failed (continuing without it): {e}")
             return None
 
+    # ---------- DenseNet121 face-GAN specialist ----------
+    def load_densenet(self) -> Optional[Tuple[object, dict]]:
+        """Lazy-load DenseNet121 PyTorch checkpoint (TF-free).
+
+        Returns (model, meta_dict) or None when disabled / file missing.
+        meta_dict contains threshold, image_size, normalize_mean/std.
+        """
+        if not settings.DENSENET_ENABLED:
+            return None
+        if self._densenet_unavailable:
+            return None
+        if self._densenet_model is not None:
+            return self._densenet_model, self._densenet_meta
+
+        import json
+        import torch
+        from pathlib import Path
+
+        repo_root = Path(__file__).resolve().parent.parent.parent
+
+        def _resolve(rel: str) -> Path:
+            p = Path(rel)
+            return p if p.is_absolute() else (repo_root / p).resolve()
+
+        pt_path   = _resolve(settings.DENSENET_MODEL_PATH)
+        meta_path = _resolve(settings.DENSENET_META_PATH)
+
+        # HF Space fallback when local files are missing
+        if not pt_path.exists() or not meta_path.exists():
+            repo_id = settings.DENSENET_HF_REPO_ID.strip()
+            if repo_id:
+                try:
+                    from huggingface_hub import hf_hub_download
+                    logger.info(f"DenseNet checkpoint not found locally — downloading from {repo_id}")
+                    pt_path = Path(hf_hub_download(
+                        repo_id=repo_id, repo_type="space",
+                        filename="trained_models/densenet121_faces.pt",
+                        revision=settings.DENSENET_HF_REVISION,
+                    ))
+                    meta_path = Path(hf_hub_download(
+                        repo_id=repo_id, repo_type="space",
+                        filename="trained_models/densenet121_faces_meta.json",
+                        revision=settings.DENSENET_HF_REVISION,
+                    ))
+                except Exception as e:
+                    logger.warning(f"DenseNet HF download failed: {e} — skipping")
+                    self._densenet_unavailable = True
+                    return None
+
+        if not pt_path.exists():
+            logger.warning(f"DenseNet checkpoint not found at {pt_path} — skipping")
+            self._densenet_unavailable = True
+            return None
+
+        try:
+            from services.densenet_service import DenseNetFaces
+            meta = json.loads(meta_path.read_text(encoding="utf-8"))
+            logger.info(f"Loading DenseNet checkpoint from {pt_path}")
+            ckpt  = torch.load(str(pt_path), map_location=settings.DEVICE, weights_only=True)
+            model = DenseNetFaces()
+            model.load_state_dict(ckpt["model_state_dict"])
+            model.to(settings.DEVICE)
+            model.eval()
+            self._densenet_model = model
+            self._densenet_meta  = meta
+            logger.info("DenseNet121 face-GAN model loaded")
+            return self._densenet_model, self._densenet_meta
+        except Exception as e:
+            logger.warning(f"DenseNet load failed (continuing without it): {e}")
+            self._densenet_unavailable = True
+            return None
+
     # ---------- Preload ----------
     def preload_phase1(self) -> None:
         """Preload all core models to prevent lazy-loading delays during first analysis."""
@@ -330,6 +405,7 @@ class ModelLoader:
         self.load_face_detector()
         self.load_efficientnet()
         self.load_ffpp_model()
+        self.load_densenet()
         self.load_ocr_engine()
         self.load_text_model()
         self.load_multilang_text_model()

scripts/convert_densenet_keras_to_pt.py

@@ -0,0 +1,280 @@
+"""Convert deepfake_densenet121_high_acc.keras → densenet121_faces.pt
+
+No TensorFlow required at runtime. Reads weights directly from the .keras
+ZIP/HDF5 format, maps them to a torchvision DenseNet121 + custom head, runs
+a numeric parity check, then saves the PyTorch checkpoint.
+
+Usage (run once, needs h5py + torch + torchvision):
+    cd <repo_root>
+    python backend/scripts/convert_densenet_keras_to_pt.py
+
+Output:
+    backend/trained_models/densenet121_faces.pt
+    backend/trained_models/densenet121_faces_meta.json
+"""
+from __future__ import annotations
+
+import io
+import json
+import re
+import zipfile
+from pathlib import Path
+
+import h5py
+import numpy as np
+import torch
+import torch.nn as nn
+import torchvision.models as tvm
+
+# ── Paths ─────────────────────────────────────────────────────────────────────
+ROOT = Path(__file__).resolve().parent.parent.parent
+KERAS_PATH   = ROOT / "backend" / "trained_models" / "deepfake_densenet121_high_acc.keras"
+THRESH_PATH  = ROOT / "backend" / "trained_models" / "deepfake_densenet121_threshold.json"
+OUT_PT       = ROOT / "backend" / "trained_models" / "densenet121_faces.pt"
+OUT_META     = ROOT / "backend" / "trained_models" / "densenet121_faces_meta.json"
+
+
+# ── Custom head matching the Keras architecture ──────────────────────────────
+# Keras head (after GlobalAvgPool): Dense(1024→256,relu) → BN(256) → Dropout →
+#   Dense(256→1,sigmoid). We fold sigmoid into inference logic; the raw logit
+#   is returned so GradCAM can back-prop cleanly.
+class _FakeHead(nn.Module):
+    def __init__(self) -> None:
+        super().__init__()
+        self.fc1 = nn.Linear(1024, 256)
+        self.relu = nn.ReLU(inplace=True)
+        self.bn   = nn.BatchNorm1d(256)
+        self.drop = nn.Dropout(0.3)
+        self.fc2  = nn.Linear(256, 1)
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        x = self.relu(self.fc1(x))
+        x = self.bn(x)
+        x = self.drop(x)
+        return self.fc2(x)  # raw logit; caller applies sigmoid
+
+
+class DenseNetFaces(nn.Module):
+    """DenseNet121 backbone + custom binary head for face-GAN detection."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        base = tvm.densenet121(weights=None)
+        self.features   = base.features   # keeps all DenseBlock + transitions
+        self.head       = _FakeHead()
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        feat = self.features(x)
+        feat = torch.nn.functional.relu(feat, inplace=True)
+        feat = torch.nn.functional.adaptive_avg_pool2d(feat, (1, 1))
+        feat = torch.flatten(feat, 1)
+        return self.head(feat)             # (B, 1) logit
+
+
+# ── Weight extraction helpers ─────────────────────────────────────────────────
+def _load_h5(keras_path: Path) -> tuple[h5py.File, io.BytesIO]:
+    with zipfile.ZipFile(keras_path) as z:
+        buf = io.BytesIO(z.read("model.weights.h5"))
+    return h5py.File(buf, "r"), buf   # caller holds buf alive
+
+
+def _bn_sort_key(name: str) -> tuple[str, int]:
+    m = re.match(r"^([a-z_]+?)_?(\d+)?$", name)
+    if not m:
+        return (name, 0)
+    return (m.group(1), int(m.group(2)) if m.group(2) else 0)
+
+
+def _read_vars(group: h5py.Group) -> list[np.ndarray]:
+    """Return [var_0, var_1, ...] from a 'vars' sub-group."""
+    vars_g = group["vars"]
+    return [np.array(vars_g[str(i)]) for i in range(len(vars_g))]
+
+
+# ── Structural name → PyTorch param-prefix mapping ───────────────────────────
+def _build_keras_to_pt_map() -> dict[str, str]:
+    """Hard-coded mapping of Keras DenseNet121 layer names → torchvision names.
+
+    Pattern:
+      Keras  conv{stage}_block{i}_0_bn  → features.denseblock{s}.denselayer{i}.norm1
+      Keras  conv{stage}_block{i}_1_conv → features.denseblock{s}.denselayer{i}.conv1
+      Keras  conv{stage}_block{i}_1_bn   → features.denseblock{s}.denselayer{i}.norm2
+      Keras  conv{stage}_block{i}_2_conv → features.denseblock{s}.denselayer{i}.conv2
+    Stages: conv2→denseblock1, conv3→denseblock2, conv4→denseblock3, conv5→denseblock4
+    Transitions: pool{k}_bn → transition{k-1}.norm, pool{k}_conv → transition{k-1}.conv
+    """
+    m: dict[str, str] = {}
+    m["conv1_conv"] = "features.conv0"
+    m["conv1_bn"]   = "features.norm0"
+
+    stage_map = {2: 1, 3: 2, 4: 3, 5: 4}
+    block_counts = {1: 6, 2: 12, 3: 24, 4: 16}
+
+    for keras_stage, pt_block in stage_map.items():
+        n_layers = block_counts[pt_block]
+        for i in range(1, n_layers + 1):
+            prefix_k = f"conv{keras_stage}_block{i}"
+            prefix_p = f"features.denseblock{pt_block}.denselayer{i}"
+            m[f"{prefix_k}_0_bn"]   = f"{prefix_p}.norm1"
+            m[f"{prefix_k}_1_conv"] = f"{prefix_p}.conv1"
+            m[f"{prefix_k}_1_bn"]   = f"{prefix_p}.norm2"
+            m[f"{prefix_k}_2_conv"] = f"{prefix_p}.conv2"
+
+    # Transitions (keras pool2/3/4 → pytorch transition1/2/3)
+    for pool_idx, trans_idx in [(2, 1), (3, 2), (4, 3)]:
+        m[f"pool{pool_idx}_bn"]   = f"features.transition{trans_idx}.norm"
+        m[f"pool{pool_idx}_conv"] = f"features.transition{trans_idx}.conv"
+
+    m["bn"] = "features.norm5"
+    return m
+
+
+# ── Main conversion ───────────────────────────────────────────────────────────
+def convert() -> None:
+    print(f"Reading  {KERAS_PATH}")
+    hf, _buf = _load_h5(KERAS_PATH)
+
+    # -- Collect sub-model (DenseNet backbone) weights in traversal order ------
+    # The h5 keys use Python class-counter naming (conv2d, conv2d_1, ...).
+    # We rebuild counter → structural-name by walking the config layer order.
+    with zipfile.ZipFile(KERAS_PATH) as z:
+        cfg = json.loads(z.read("config.json"))
+
+    outer_layers = cfg["config"]["layers"]
+    sub_cfg = next(
+        l for l in outer_layers
+        if l.get("class_name") in ("Functional", "Model") and "densenet" in l.get("name", "")
+    )
+    sub_layers = sub_cfg["config"]["layers"]
+
+    # Walk in config order; assign counter indices to weight-bearing layers
+    conv_counter = 0
+    bn_counter   = 0
+    # structural_name → h5_key
+    name_to_h5: dict[str, str] = {}
+    for lc in sub_layers:
+        cls  = lc.get("class_name", "")
+        name = lc.get("name", "")
+        if cls == "Conv2D":
+            h5_key = "conv2d" if conv_counter == 0 else f"conv2d_{conv_counter}"
+            name_to_h5[name] = h5_key
+            conv_counter += 1
+        elif cls == "BatchNormalization":
+            h5_key = "batch_normalization" if bn_counter == 0 else f"batch_normalization_{bn_counter}"
+            name_to_h5[name] = h5_key
+            bn_counter += 1
+
+    func_layers_h5 = hf["layers"]["functional"]["layers"]
+    keras_to_pt    = _build_keras_to_pt_map()
+
+    # -- Build PyTorch model ---------------------------------------------------
+    print("Building PyTorch DenseNetFaces model …")
+    model = DenseNetFaces()
+    sd    = model.state_dict()
+
+    def set_conv(pt_prefix: str, keras_w: np.ndarray) -> None:
+        # Keras: (H, W, C_in, C_out)  →  PyTorch: (C_out, C_in, H, W)
+        key = f"{pt_prefix}.weight"
+        assert key in sd, f"Missing key: {key}"
+        t = torch.from_numpy(keras_w.transpose(3, 2, 0, 1))
+        assert t.shape == sd[key].shape, f"Shape mismatch {key}: {t.shape} vs {sd[key].shape}"
+        sd[key] = t
+
+    def set_bn(pt_prefix: str, vars_: list[np.ndarray]) -> None:
+        # Keras vars order: [gamma, beta, moving_mean, moving_var]
+        for keras_idx, pt_suffix in [(0, "weight"), (1, "bias"),
+                                     (2, "running_mean"), (3, "running_var")]:
+            key = f"{pt_prefix}.{pt_suffix}"
+            assert key in sd, f"Missing key: {key}"
+            t = torch.from_numpy(vars_[keras_idx])
+            assert t.shape == sd[key].shape, f"Shape mismatch {key}: {t.shape} vs {sd[key].shape}"
+            sd[key] = t
+        # PyTorch BN also has num_batches_tracked — leave at 0
+
+    # -- Transfer backbone weights -------------------------------------------
+    for keras_name, pt_prefix in keras_to_pt.items():
+        h5_key = name_to_h5.get(keras_name)
+        if h5_key is None:
+            raise KeyError(f"Keras layer '{keras_name}' not found in config traversal")
+
+        if h5_key not in func_layers_h5:
+            raise KeyError(f"h5 key '{h5_key}' not found under functional/layers")
+
+        layer_group = func_layers_h5[h5_key]
+        if "vars" not in layer_group:
+            raise ValueError(f"No 'vars' under functional/layers/{h5_key}")
+
+        vars_ = _read_vars(layer_group)
+
+        if keras_name.endswith("_conv") or keras_name == "conv1_conv":
+            set_conv(pt_prefix, vars_[0])   # conv has only weights (no bias; use_bias=False)
+        else:
+            set_bn(pt_prefix, vars_)
+
+    print(f"  Backbone: {len(keras_to_pt)} layers transferred")
+
+    # -- Transfer custom head weights ----------------------------------------
+    outer_h5 = hf["layers"]
+
+    # Dense(1024→256): vars[0]=(1024,256), vars[1]=(256,)
+    dense_vars = _read_vars(outer_h5["dense"])
+    sd["head.fc1.weight"] = torch.from_numpy(dense_vars[0].T)   # (256, 1024)
+    sd["head.fc1.bias"]   = torch.from_numpy(dense_vars[1])
+
+    # BN(256): vars[0]=gamma, [1]=beta, [2]=moving_mean, [3]=moving_var
+    bn_vars = _read_vars(outer_h5["batch_normalization"])
+    for keras_idx, pt_suffix in [(0, "weight"), (1, "bias"),
+                                  (2, "running_mean"), (3, "running_var")]:
+        sd[f"head.bn.{pt_suffix}"] = torch.from_numpy(bn_vars[keras_idx])
+
+    # Dense(256→1): vars[0]=(256,1), vars[1]=(1,)
+    dense1_vars = _read_vars(outer_h5["dense_1"])
+    sd["head.fc2.weight"] = torch.from_numpy(dense1_vars[0].T)  # (1, 256)
+    sd["head.fc2.bias"]   = torch.from_numpy(dense1_vars[1])
+
+    print("  Head: fc1, bn, fc2 transferred")
+
+    model.load_state_dict(sd)
+    model.eval()
+    hf.close()
+
+    # -- Parity check ----------------------------------------------------------
+    print("Running parity check (random 224x224 input) …")
+    # DenseNet preprocess: ImageNet mean/std after [0,1] normalisation
+    # (same for Keras 'torch' mode and torchvision default)
+    MEAN = torch.tensor([0.485, 0.456, 0.406]).view(1, 3, 1, 1)
+    STD  = torch.tensor([0.229, 0.224, 0.225]).view(1, 3, 1, 1)
+
+    rng    = np.random.default_rng(0)
+    raw    = rng.integers(0, 256, (1, 224, 224, 3), dtype=np.uint8).astype(np.float32)
+    tensor = torch.from_numpy(raw).permute(0, 3, 1, 2) / 255.0  # (1,3,224,224)
+    tensor = (tensor - MEAN) / STD
+
+    with torch.no_grad():
+        logit = model(tensor)
+        score = torch.sigmoid(logit).item()
+    print(f"  Parity output (real_prob): {score:.6f}  [sanity: should be in (0,1)]")
+    assert 0.0 < score < 1.0, "Sigmoid output out of range — weight transfer may have failed"
+
+    # -- Save checkpoint -------------------------------------------------------
+    print(f"Saving  {OUT_PT}")
+    torch.save({"model_state_dict": model.state_dict()}, OUT_PT)
+
+    thresh_data = json.loads(THRESH_PATH.read_text(encoding="utf-8"))
+    meta = {
+        "threshold":    thresh_data["threshold"],   # 0.7597
+        "image_size":   thresh_data["image_size"],  # 224
+        "label_mapping": thresh_data["label_mapping"],
+        "score_meaning": thresh_data["score_meaning"],
+        "normalize_mean": [0.485, 0.456, 0.406],
+        "normalize_std":  [0.229, 0.224, 0.225],
+        "source_keras":  "deepfake_densenet121_high_acc.keras",
+        "architecture":  "DenseNet121 + GlobalAvgPool + Linear(1024,256)+ReLU+BN+Dropout(0.3)+Linear(256,1)+Sigmoid",
+    }
+    OUT_META.write_text(json.dumps(meta, indent=2), encoding="utf-8")
+    print(f"Saving  {OUT_META}")
+    print("\nDone. Conversion successful.")
+
+
+if __name__ == "__main__":
+    convert()

services/densenet_service.py

@@ -0,0 +1,110 @@
+"""DenseNet121 face-GAN inference service.
+
+The model outputs a sigmoid real_probability in [0, 1].
+Threshold (from training) is 0.7597 (Youden's J on val ROC):
+  score >= threshold → Real,  score < threshold → Fake.
+
+fake_prob is mapped with a piecewise-linear calibration anchored so that:
+  score = 1.0      → fake_prob = 0.0   (confident real)
+  score = threshold → fake_prob = 0.5  (decision boundary)
+  score = 0.0      → fake_prob = 1.0   (confident fake)
+This respects the trained threshold without needing an extra isotonic fit.
+"""
+from __future__ import annotations
+
+from typing import Optional
+
+import numpy as np
+import torch
+import torch.nn as nn
+import torchvision.models as tvm
+from loguru import logger
+from PIL import Image
+
+
+# ── Model architecture (must match convert_densenet_keras_to_pt.py) ──────────
+class _FakeHead(nn.Module):
+    def __init__(self) -> None:
+        super().__init__()
+        self.fc1  = nn.Linear(1024, 256)
+        self.relu = nn.ReLU(inplace=True)
+        self.bn   = nn.BatchNorm1d(256)
+        self.drop = nn.Dropout(0.3)
+        self.fc2  = nn.Linear(256, 1)
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        return self.fc2(self.drop(self.bn(self.relu(self.fc1(x)))))
+
+
+class DenseNetFaces(nn.Module):
+    """DenseNet121 + custom head for face-GAN binary detection."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        base = tvm.densenet121(weights=None)
+        self.features = base.features
+        self.head     = _FakeHead()
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        feat = torch.nn.functional.relu(self.features(x), inplace=True)
+        feat = torch.nn.functional.adaptive_avg_pool2d(feat, (1, 1))
+        feat = torch.flatten(feat, 1)
+        return self.head(feat)   # (B, 1) raw logit
+
+
+# ── Preprocessing ─────────────────────────────────────────────────────────────
+_MEAN = torch.tensor([0.485, 0.456, 0.406]).view(1, 3, 1, 1)
+_STD  = torch.tensor([0.229, 0.224, 0.225]).view(1, 3, 1, 1)
+
+
+def _preprocess(pil_img: Image.Image, image_size: int, device: str) -> torch.Tensor:
+    img = pil_img.convert("RGB").resize((image_size, image_size), Image.BILINEAR)
+    arr = np.array(img, dtype=np.float32) / 255.0          # (H, W, 3) → [0, 1]
+    t   = torch.from_numpy(arr).permute(2, 0, 1).unsqueeze(0)  # (1, 3, H, W)
+    mean = _MEAN.to(device)
+    std  = _STD.to(device)
+    return (t.to(device) - mean) / std
+
+
+# ── Calibrated fake_prob ──────────────────────────────────────────────────────
+def _calibrated_fake_prob(score_real: float, threshold: float) -> float:
+    """Piecewise-linear map: score=threshold → 0.5, score=1 → 0, score=0 → 1."""
+    if score_real >= threshold:
+        return 0.5 * (1.0 - score_real) / max(1.0 - threshold, 1e-8)
+    else:
+        return 0.5 + 0.5 * (threshold - score_real) / max(threshold, 1e-8)
+
+
+# ── Public inference entry point ──────────────────────────────────────────────
+def detect_image(
+    pil_img: Image.Image,
+    model: DenseNetFaces,
+    meta: dict,
+    device: str = "cpu",
+) -> dict:
+    """Run DenseNet121 inference on a PIL image.
+
+    Returns:
+        score_real  – raw sigmoid output in [0, 1]
+        fake_prob   – calibrated fake probability in [0, 1]
+        threshold   – Youden's J threshold used for binary verdict
+        label       – "Real" or "Fake" based on threshold
+    """
+    threshold  = float(meta.get("threshold", 0.7597))
+    image_size = int(meta.get("image_size", 224))
+
+    tensor = _preprocess(pil_img, image_size, device)
+
+    with torch.no_grad():
+        logit      = model(tensor)
+        score_real = float(torch.sigmoid(logit).item())
+
+    fake_prob = _calibrated_fake_prob(score_real, threshold)
+    label     = "Real" if score_real >= threshold else "Fake"
+
+    return {
+        "score_real": score_real,
+        "fake_prob":  fake_prob,
+        "threshold":  threshold,
+        "label":      label,
+    }

services/image_service.py

@@ -116,6 +116,26 @@ def _crop_face_for_face_model(pil_img: Image.Image) -> Image.Image:
     return pil_img
 
 
+def _classify_densenet(pil_img: Image.Image) -> Optional[Tuple[float, dict[str, float]]]:
+    """Run DenseNet121 face-GAN classifier. Returns (fake_prob, all_scores) or None."""
+    loader = get_model_loader()
+    result = loader.load_densenet()
+    if result is None:
+        return None
+    model, meta = result
+    try:
+        from services.densenet_service import detect_image
+        out = detect_image(pil_img, model, meta, device=settings.DEVICE)
+        scores = {
+            "densenet_real": out["score_real"],
+            "densenet_fake": out["fake_prob"],
+        }
+        return out["fake_prob"], scores
+    except Exception as e:
+        logger.warning(f"DenseNet inference failed: {e}")
+        return None
+
+
 def _classify_ffpp(pil_img: Image.Image) -> Optional[Tuple[float, dict[str, float]]]:
     """Run the FFPP-fine-tuned ViT (Phase 11.3). Returns (fake_prob, all_scores) or None."""
     loader = get_model_loader()
@@ -233,19 +253,20 @@ def _classify_no_face(
 
 
 def _looks_like_video_frame(pil_img: Image.Image) -> bool:
-    """Return True when the image is likely a frame extracted from video.
+    """Return True when the image is very likely a frame extracted from video.
 
-    Video frames: low resolution, common video aspect ratios, no EXIF, heavy
-    compression. Face-swap deepfakes almost always come from video, so we use
-    this to shift weight away from the AI-image detector (trained on generated
-    stills) and toward the face-swap-trained models (FFPP/EfficientNet).
+    Requires BOTH a low resolution (≤720px long side, typical for extracted
+    deepfake frames) AND a tight aspect-ratio match (±0.03) to a standard video
+    ratio. 1:1 and 4:3 are excluded because they overlap heavily with common
+    photo formats and cause too many false positives.
     """
     w, h = pil_img.size
-    if max(w, h) > 1080:
+    if max(w, h) > 720:
         return False
     aspect = w / h
-    video_ratios = [16 / 9, 4 / 3, 9 / 16, 3 / 4, 1.0]
-    return any(abs(aspect - r) < 0.10 for r in video_ratios)
+    # Exclude 1:1 and 4:3 — too common in photos to be a reliable video signal
+    video_ratios = [16 / 9, 9 / 16, 3 / 4]
+    return any(abs(aspect - r) < 0.03 for r in video_ratios)
 
 
 def _has_gan_artifact(artifacts: list[ArtifactIndicator]) -> bool:
@@ -351,6 +372,15 @@ def classify_image(
         models_used.append("ffpp-vit-local")
         scores_out.update({f"ffpp_{k}": v for k, v in ffpp_scores.items()})
 
+    # DenseNet121 inference (face-GAN specialist — face-present path only).
+    densenet_fake_prob: Optional[float] = None
+    if settings.DENSENET_ENABLED and face_present_for_route:
+        dn_res = _classify_densenet(pil_img)
+        if dn_res is not None:
+            densenet_fake_prob, dn_scores = dn_res
+            models_used.append("densenet121-faces")
+            scores_out.update(dn_scores)
+
     if not settings.ENSEMBLE_MODE:
         if ffpp_fake_prob is not None:
             combined = 0.4 * vit_fake_prob + 0.6 * ffpp_fake_prob
@@ -380,30 +410,45 @@ def classify_image(
             scores_out["efficientnet_real"] = 1.0 - eff_fake_prob
             scores_out["efficientnet_calibrator_applied"] = 1.0 if eff_result.get("calibrator_applied") else 0.0
 
-    # ── Face-stack composite (FFPP + ViT + EffNet) ──
-    if face_present and eff_fake_prob is not None and ffpp_fake_prob is not None:
-        w_ffpp, w_vit, w_eff = settings.FFPP_WEIGHT_FACE, settings.VIT_WEIGHT_FACE, settings.EFFNET_WEIGHT_FACE
-        total = w_ffpp + w_vit + w_eff
-        face_stack_prob = (w_ffpp * ffpp_fake_prob + w_vit * vit_fake_prob + w_eff * eff_fake_prob) / total
-        face_stack_method = "ffpp_vit_eff"
-    elif face_present and ffpp_fake_prob is not None and eff_fake_prob is None:
-        w_ffpp, w_vit = settings.FFPP_WEIGHT_FACE, settings.VIT_WEIGHT_FACE
-        total = w_ffpp + w_vit
-        face_stack_prob = (w_ffpp * ffpp_fake_prob + w_vit * vit_fake_prob) / total
-        face_stack_method = "ffpp_vit"
-    elif face_present and eff_fake_prob is not None:
-        face_stack_prob = 0.5 * vit_fake_prob + 0.5 * eff_fake_prob
-        face_stack_method = "vit_eff"
-    else:
-        face_stack_prob = vit_fake_prob
-        face_stack_method = "vit_only"
+    # ── Face-stack composite (DenseNet + FFPP + ViT + EffNet) ──────────────
+    # Video-frame path shifts weight to FFPP/EffNet; still-image path gives
+    # DenseNet the lead (trained specifically on GAN still-face portraits).
+    is_video_frame = _looks_like_video_frame(pil_img)
+
+    def _weighted(probs: dict[str, float]) -> float:
+        total = sum(probs.values())
+        return sum(v * w for v, w in probs.items()) / total if total else 0.0
+
+    available: dict[str, float] = {}
+    if densenet_fake_prob is not None:
+        w_dn = settings.DENSENET_VIDEO_WEIGHT if is_video_frame else settings.DENSENET_WEIGHT_FACE
+        available["densenet"] = w_dn
+    if ffpp_fake_prob is not None:
+        w_ffpp = settings.VIDEO_FFPP_WEIGHT_FACE if is_video_frame else settings.FFPP_WEIGHT_FACE
+        available["ffpp"] = w_ffpp
+    if eff_fake_prob is not None and face_present:
+        w_eff = settings.VIDEO_EFFNET_WEIGHT_FACE if is_video_frame else settings.EFFNET_WEIGHT_FACE
+        available["eff"] = w_eff
+    # ViT always present
+    w_vit = settings.VIDEO_VIT_WEIGHT_FACE if is_video_frame else settings.VIT_WEIGHT_FACE
+    available["vit"] = w_vit
+
+    prob_map: dict[str, float] = {}
+    if "densenet" in available:
+        prob_map["densenet"] = densenet_fake_prob * available["densenet"]
+    if "ffpp" in available:
+        prob_map["ffpp"] = ffpp_fake_prob * available["ffpp"]
+    if "eff" in available:
+        prob_map["eff"] = eff_fake_prob * available["eff"]
+    prob_map["vit"] = vit_fake_prob * available["vit"]
+
+    total_w = sum(available.values())
+    face_stack_prob = sum(prob_map.values()) / total_w if total_w else vit_fake_prob
+
+    active = [k for k in ["densenet", "ffpp", "eff", "vit"] if k in available]
+    face_stack_method = "_".join(active)
 
     # ── Phase A2/A3: unified evidence fusion (face-stack + general + forensics + EXIF + VLM) ──
-    # Video-frame detection: face-swap deepfakes come from video. The AI-image
-    # detectors (trained on synthesised stills) are unreliable for this class,
-    # so we shift weight toward the face-swap-trained models when the input
-    # looks like a compressed video frame.
-    is_video_frame = _looks_like_video_frame(pil_img)
     w_face_stack = settings.VIDEO_FRAME_FACE_STACK_WEIGHT if is_video_frame else settings.FACE_STACK_WEIGHT_FACE
     w_general    = settings.VIDEO_FRAME_GENERAL_WEIGHT    if is_video_frame else settings.GENERAL_WEIGHT_FACE
     w_forensics  = settings.VIDEO_FRAME_FORENSICS_WEIGHT  if is_video_frame else settings.FORENSICS_WEIGHT_FACE

services/llm_explainer.py

@@ -330,7 +330,7 @@ class _GeminiProvider(_LLMProvider):
         self.model = settings.LLM_MODEL
         self._config = types.GenerateContentConfig(
             temperature=0.3,
-            max_output_tokens=600,
+            max_output_tokens=1024,
             response_mime_type="application/json",
         )
 
@@ -352,7 +352,7 @@ class _OpenAIProvider(_LLMProvider):
             model=self.model,
             messages=[{"role": "user", "content": prompt}],
             temperature=0.3,
-            max_tokens=600,
+            max_tokens=1024,
             response_format={"type": "json_object"},
         )
         return response.choices[0].message.content or ""
@@ -372,7 +372,7 @@ class _GroqProvider(_LLMProvider):
             model=self.model,
             messages=[{"role": "user", "content": prompt}],
             temperature=0.3,
-            max_tokens=600,
+            max_tokens=1024,
             response_format={"type": "json_object"},
         )
         return response.choices[0].message.content or ""
@@ -458,6 +458,33 @@ def _get_provider() -> _ProviderChain:
     return _provider_instance
 
 
+def _repair_truncated_json(text: str) -> str:
+    """Close unclosed braces/brackets so a truncated JSON string becomes parseable."""
+    stack = []
+    in_string = False
+    escape = False
+    for ch in text:
+        if escape:
+            escape = False
+            continue
+        if ch == "\\" and in_string:
+            escape = True
+            continue
+        if ch == '"':
+            in_string = not in_string
+            continue
+        if in_string:
+            continue
+        if ch in "{[":
+            stack.append("}" if ch == "{" else "]")
+        elif ch in "}]" and stack:
+            stack.pop()
+    # If we're mid-string, close it first
+    suffix = '"' if in_string else ""
+    suffix += "".join(reversed(stack))
+    return text + suffix
+
+
 def _parse_llm_response(raw: str) -> tuple[str, list[SignalObservation], list[str]]:
     """Parse the LLM's JSON response into (paragraph, signals, bullets).
     Handles cases where the LLM wraps output in markdown fences.
@@ -468,7 +495,12 @@ def _parse_llm_response(raw: str) -> tuple[str, list[SignalObservation], list[st
         lines = [l for l in lines if not l.strip().startswith("```")]
         text = "\n".join(lines).strip()
 
-    parsed = json.loads(text)
+    try:
+        parsed = json.loads(text)
+    except json.JSONDecodeError:
+        # Truncated JSON — try to recover by closing unclosed braces/brackets
+        repaired = _repair_truncated_json(text)
+        parsed = json.loads(repaired)
     paragraph = parsed.get("paragraph", "")
 
     raw_signals = parsed.get("signals", [])

services/news_lookup.py

@@ -103,22 +103,9 @@ def _query_attempts(q: str, country: Optional[str]) -> list[dict]:
             latest_params["_url"] = settings.NEWS_API_BASE_URL
             if country_code:
                 latest_params["country"] = country_code
-            if recent_window:
-                latest_params["timeframe"] = recent_window
+            # timeframe is a paid-plan feature; omit to avoid 422 on free plans
             attempts.append(latest_params)
 
-        archive_key = (country_code, "archive")
-        if archive_key not in seen:
-            seen.add(archive_key)
-            archive_params = dict(base)
-            archive_params["_endpoint"] = "archive"
-            archive_params["_url"] = settings.NEWS_API_ARCHIVE_BASE_URL
-            archive_params["from_date"] = archive_from
-            archive_params["to_date"] = archive_to
-            if country_code:
-                archive_params["country"] = country_code
-            attempts.append(archive_params)
-
     return attempts
 
 

trained_models/config.json

@@ -1,34 +0,0 @@
-{
-  "architectures": [
-    "ViTForImageClassification"
-  ],
-  "attention_probs_dropout_prob": 0.0,
-  "dtype": "float32",
-  "encoder_stride": 16,
-  "hidden_act": "gelu",
-  "hidden_dropout_prob": 0.0,
-  "hidden_size": 768,
-  "id2label": {
-    "0": "fake",
-    "1": "real"
-  },
-  "image_size": 224,
-  "initializer_range": 0.02,
-  "intermediate_size": 3072,
-  "label2id": {
-    "fake": "0",
-    "real": "1"
-  },
-  "layer_norm_eps": 1e-12,
-  "model_type": "vit",
-  "num_attention_heads": 12,
-  "num_channels": 3,
-  "num_hidden_layers": 12,
-  "patch_size": 16,
-  "pooler_act": "tanh",
-  "pooler_output_size": 768,
-  "problem_type": "single_label_classification",
-  "qkv_bias": true,
-  "transformers_version": "5.0.0",
-  "use_cache": false
-}

trained_models/deepfake_densenet121_high_acc.keras

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

trained_models/deepfake_densenet121_latest.keras

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

trained_models/deepfake_densenet121_threshold.json

@@ -1,9 +0,0 @@
-{
-  "threshold": 0.7596627473831177,
-  "label_mapping": {
-    "fake": 0,
-    "real": 1
-  },
-  "image_size": 224,
-  "score_meaning": "score >= threshold predicts real; score < threshold predicts fake"
-}

trained_models/densenet121_faces.pt

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

trained_models/densenet121_faces_meta.json

@@ -1,21 +0,0 @@
-{
-  "threshold": 0.7596627473831177,
-  "image_size": 224,
-  "label_mapping": {
-    "fake": 0,
-    "real": 1
-  },
-  "score_meaning": "score >= threshold predicts real; score < threshold predicts fake",
-  "normalize_mean": [
-    0.485,
-    0.456,
-    0.406
-  ],
-  "normalize_std": [
-    0.229,
-    0.224,
-    0.225
-  ],
-  "source_keras": "deepfake_densenet121_high_acc.keras",
-  "architecture": "DenseNet121 + GlobalAvgPool + Linear(1024,256)+ReLU+BN+Dropout(0.3)+Linear(256,1)+Sigmoid"
-}

trained_models/model.safetensors

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

trained_models/training_args.bin

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