app.py

"""
DEEPFYND Intelligence Plane — Gradio Space
- Web UI at /            (demo interface, useful for thesis screenshots)
- REST API at /analyze   (called by the Make.com scenarios)
- Heatmaps at /heatmaps/{name}

Contract for POST /analyze:
  request : { "file_url": "...", "media_type": "image|audio|video", "scan_id": "rec..." }
  response: { "verdict", "confidence", "insights", "heatmap_url", "file_hash", "raw_scores" }

Zero fabrication: if analysis cannot run, an HTTP error is returned. No verdict is invented.
"""
import os

# Gradio 6 runs a Node/SvelteKit SSR layer in front of the Python app, which
# intercepts POST requests to custom routes. Disable it before Gradio loads.
os.environ.setdefault("GRADIO_SSR_MODE", "false")

import io
import hashlib
import tempfile
import traceback

import numpy as np
import torch
from PIL import Image
import cv2
import requests
import gradio as gr
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse
from pydantic import BaseModel
from transformers import (
    AutoImageProcessor, AutoModelForImageClassification,
    AutoFeatureExtractor, AutoModelForAudioClassification,
)
import librosa

# ---------------- Config ----------------
IMAGE_MODEL_NAME = "prithivMLmods/Deep-Fake-Detector-v2-Model"
AUDIO_MODEL_NAME = "Hemgg/Deepfake-audio-detection"
VIDEO_FRAMES_TO_SAMPLE = 3   # CPU-basic friendly: 8 frames was too slow (>120s)
MAX_MEDIA_BYTES = 40 * 1024 * 1024   # 40 MB guard for free-tier memory

HEATMAP_DIR = "/tmp/heatmaps"
os.makedirs(HEATMAP_DIR, exist_ok=True)

# Set this in Space Settings → Variables, e.g.
#   https://YOURNAME-deepfynd-intelligence.hf.space
SPACE_URL = os.environ.get("SPACE_URL", "").rstrip("/")

# ---------------- Lazy model loading ----------------
_image_processor = None
_image_model = None
_audio_extractor = None
_audio_model = None


def get_image_model():
    global _image_processor, _image_model
    if _image_model is None:
        _image_model = AutoModelForImageClassification.from_pretrained(IMAGE_MODEL_NAME).eval()
        try:
            _image_processor = AutoImageProcessor.from_pretrained(IMAGE_MODEL_NAME)
        except Exception:
            try:
                _image_processor = AutoImageProcessor.from_pretrained("google/vit-base-patch16-224-in21k")
            except Exception:
                from transformers import ViTImageProcessor
                _image_processor = ViTImageProcessor(
                    size={"height": 224, "width": 224},
                    image_mean=[0.5, 0.5, 0.5], image_std=[0.5, 0.5, 0.5],
                    do_resize=True, do_normalize=True,
                )
    return _image_processor, _image_model


def get_audio_model():
    global _audio_extractor, _audio_model
    if _audio_model is None:
        _audio_model = AutoModelForAudioClassification.from_pretrained(AUDIO_MODEL_NAME).eval()
        try:
            _audio_extractor = AutoFeatureExtractor.from_pretrained(AUDIO_MODEL_NAME)
        except Exception:
            _audio_extractor = AutoFeatureExtractor.from_pretrained("facebook/wav2vec2-base")
    return _audio_extractor, _audio_model


# ---------------- Layer 3: risk scoring and fusion ----------------
# Decision-level fusion weights. The learned detector (Layer 2) is the primary
# signal; classical forensics (Layer 1) is corroborative, consistent with the
# thesis position that metadata is "corroborative rather than decisive"
# (Chapter 3, section 3.5.1). These two weights and the band thresholds below
# are the values to tune on validation data and report in Table 4.3.
W_MODEL = 0.85       # weight on Layer 2 model fake-probability
W_FORENSIC = 0.15    # weight on Layer 1 forensic score
BAND_SUSPICIOUS = 0.35   # fused score at/above this is at least "suspicious"
BAND_DEEPFAKE = 0.65     # fused score at/above this is "likely_deepfake"
# Fail-honest guardrail: forensics may raise a flag, but only the learned model
# may make a hard accusation. A "likely_deepfake" verdict therefore requires the
# model itself to be at least this confident, regardless of the fused score.
MODEL_ACCUSATION_FLOOR = 0.50


def fuse_scores(model_fake_prob: float, forensic_score: float) -> float:
    """
    Combine the Layer 2 model probability with the Layer 1 forensic score into a
    single fused likelihood in [0, 1] using a transparent weighted sum.

    This is real decision-level fusion: the forensic score can shift the outcome,
    so corroborating metadata (e.g. no EXIF on a generator-typical square image)
    can lift a borderline case. It never fabricates a verdict — it only adjusts a
    likelihood the model has already produced.
    """
    fused = (W_MODEL * float(model_fake_prob)) + (W_FORENSIC * float(forensic_score))
    return max(0.0, min(1.0, fused))


def score_to_band(fused_score: float, model_fake_prob: float = None) -> str:
    """
    Map a fused likelihood to one of the three verdict bands.

    If model_fake_prob is supplied, the fail-honest guardrail applies: the score
    cannot be banded as "likely_deepfake" unless the learned model itself is at
    least MODEL_ACCUSATION_FLOOR confident. Forensics alone can lift a case to
    "suspicious" but cannot, on its own, produce a hard accusation.
    """
    if fused_score < BAND_SUSPICIOUS:
        return "authentic"
    if fused_score < BAND_DEEPFAKE:
        return "suspicious"
    # Fused score is in the deepfake range — check the model actually agrees.
    if model_fake_prob is not None and model_fake_prob < MODEL_ACCUSATION_FLOOR:
        return "suspicious"
    return "likely_deepfake"


def split_probs(probs, id2label):
    """Map a model's label set onto (fake_prob, real_prob)."""
    fake_prob, real_prob = 0.0, 0.0
    for idx, label in id2label.items():
        lname = str(label).lower()
        p = float(probs[int(idx)])
        if any(k in lname for k in ["fake", "deepfake", "synthetic", "manipulated", "spoof", "ai"]):
            fake_prob = max(fake_prob, p)
        if any(k in lname for k in ["real", "authentic", "genuine", "realism", "bonafide", "human"]):
            real_prob = max(real_prob, p)
    if fake_prob == 0.0 and real_prob == 0.0:
        fake_prob = float(probs[0])
    return fake_prob, real_prob


# ---------------- Layer 1: media forensics ----------------
def image_forensic_analysis(pil_image: Image.Image):
    """
    Deterministic, interpretable forensic checks (Layer 1).

    Returns (forensic_score, flags):
      - forensic_score: a normalised value in [0, 1] where 0 means "no
        corroborating signs of manipulation" and higher means "more forensic
        signals consistent with synthetic or edited media". This score feeds the
        Layer 3 fusion (it is the missing link the earlier version did not wire
        up: previously the flags were shown to the user but never influenced the
        verdict).
      - flags: the same human-readable strings as before, for the insights text
        and the explainability layer.

    The score is built additively from independent weak signals and then clamped.
    None of these signals is decisive on its own; that is by design — Layer 1 is
    corroborative, and the fusion weight (W_FORENSIC) keeps it in proportion.
    """
    flags = []
    score = 0.0

    # Signal 1: missing EXIF. Common in AI-generated images, but also in
    # screenshots and platform-re-saved media, so it is a weak signal only.
    try:
        exif = pil_image.getexif()
    except Exception:
        exif = None
    if not exif or len(exif) == 0:
        flags.append("No EXIF metadata (common in AI-generated or re-saved images)")
        score += 0.35
    else:
        # Signal 2: editing/generation software tag. A stronger signal when present.
        software = exif.get(305) or exif.get(0x0131)
        if software and any(k in str(software).lower()
                            for k in ["photoshop", "gimp", "affinity", "midjourney", "dall", "stable"]):
            flags.append(f"Editing/generation software tag detected: {software}")
            score += 0.60

    # Signal 3: generator-typical square dimensions.
    w, h = pil_image.size
    if w == h and w in (512, 768, 1024, 2048):
        flags.append(f"Square {w}x{h} dimensions (typical of AI image generators)")
        score += 0.30

    forensic_score = max(0.0, min(1.0, score))
    return forensic_score, flags


# ---------------- Face detection + crop (preprocessing) ----------------
# Uses the Haar cascades that ship inside opencv-python-headless, so there is no
# new dependency to conflict with Hugging Face's injected packages. The detector
# is hardened against the variations introduced by the fetch/decode path (a CDN
# may serve a rotated, recompressed or colour-shifted copy of the image): it
# honours EXIF orientation, equalises contrast, and tries a second cascade with
# looser settings before giving up. Every outcome is logged so the Container log
# shows exactly which strategy matched, or that a genuine no-face fallback
# occurred. If no face is found the original image is returned and the caller
# discloses a whole-image scan — a legitimate, disclosed degradation (NFR6).
# A learned face detector (e.g. MediaPipe) remains documented future work.
_face_cascade = None
_face_cascade_alt = None
_face_detection_available = None   # None = untested, True/False once probed


def _haarcascade_path(filename):
    """
    Locate a bundled Haar cascade without assuming cv2.data exists.
    Some OpenCV builds on managed platforms omit cv2.data; fall back to the
    package directory, then to any readable copy under site-packages.
    """
    # Preferred: cv2.data.haarcascades (present in standard opencv-python builds).
    data = getattr(cv2, "data", None)
    if data is not None and getattr(data, "haarcascades", None):
        p = os.path.join(data.haarcascades, filename)
        if os.path.isfile(p):
            return p
    # Fallback: <cv2 package dir>/data/<filename>.
    try:
        pkg_dir = os.path.dirname(os.path.abspath(cv2.__file__))
        p = os.path.join(pkg_dir, "data", filename)
        if os.path.isfile(p):
            return p
    except Exception:
        pass
    return None


def get_face_cascades():
    """
    Lazily build the cascades. Sets _face_detection_available to False (and logs
    once) if this OpenCV build lacks CascadeClassifier or the cascade files, so
    the rest of the app degrades to whole-image analysis instead of crashing on
    every request.
    """
    global _face_cascade, _face_cascade_alt, _face_detection_available
    if _face_detection_available is False:
        return None, None
    if _face_cascade is not None or _face_cascade_alt is not None:
        return _face_cascade, _face_cascade_alt

    if not hasattr(cv2, "CascadeClassifier"):
        _face_detection_available = False
        print("[face] this OpenCV build has no CascadeClassifier — "
              "face cropping disabled, analysing whole images", flush=True)
        return None, None

    default_path = _haarcascade_path("haarcascade_frontalface_default.xml")
    alt_path = _haarcascade_path("haarcascade_frontalface_alt2.xml")
    try:
        if default_path:
            _face_cascade = cv2.CascadeClassifier(default_path)
        if alt_path:
            _face_cascade_alt = cv2.CascadeClassifier(alt_path)
    except Exception as e:
        print(f"[face] cascade load failed ({e}) — face cropping disabled", flush=True)
        _face_detection_available = False
        return None, None

    have_any = (_face_cascade is not None and not _face_cascade.empty()) or \
               (_face_cascade_alt is not None and not _face_cascade_alt.empty())
    if not have_any:
        _face_detection_available = False
        print("[face] no usable cascade files found — face cropping disabled", flush=True)
        return None, None

    _face_detection_available = True
    return _face_cascade, _face_cascade_alt


def _detect_largest_face(pil_image):
    """
    Robustly detect the largest face. Returns (box_or_None, debug_note).
    Tries orientation-corrected, contrast-equalised grayscale against two cascades
    at several sensitivities before reporting no face.
    """
    try:
        from PIL import ImageOps
        rgb = ImageOps.exif_transpose(pil_image).convert("RGB")
    except Exception:
        rgb = pil_image.convert("RGB")

    arr = np.array(rgb)
    gray = cv2.cvtColor(arr, cv2.COLOR_RGB2GRAY)
    try:
        gray = cv2.equalizeHist(gray)
    except Exception:
        pass

    default_c, alt_c = get_face_cascades()
    if default_c is None and alt_c is None:
        return None, "face-detection-unavailable"
    for cname, casc in [("default", default_c), ("alt2", alt_c)]:
        if casc is None or casc.empty():
            continue
        for sf, mn in [(1.1, 5), (1.05, 4), (1.2, 3)]:
            faces = casc.detectMultiScale(gray, scaleFactor=sf, minNeighbors=mn, minSize=(40, 40))
            if len(faces) > 0:
                box = max(faces, key=lambda f: f[2] * f[3])
                return box, f"{cname}/sf{sf}/mn{mn}"
    return None, "no-face-after-all-attempts"


def crop_to_face(pil_image: Image.Image, margin: float = 0.20):
    """
    Detect the largest face and crop to it with a margin.

    Returns (cropped_or_original_pil, face_found: bool). On any failure or when no
    face is detected, returns the original image and False so analysis proceeds on
    the whole image with disclosure.
    """
    try:
        rgb = pil_image.convert("RGB")
        box, note = _detect_largest_face(rgb)
        if box is None:
            print(f"[face] {note} — using whole image", flush=True)
            return pil_image, False

        x, y, w, h = box
        mx, my = int(w * margin), int(h * margin)
        left = max(0, int(x) - mx)
        top = max(0, int(y) - my)
        right = min(rgb.width, int(x) + int(w) + mx)
        bottom = min(rgb.height, int(y) + int(h) + my)
        print(f"[face] detected via {note}, box=({x},{y},{w},{h})", flush=True)
        return rgb.crop((left, top, right, bottom)), True
    except Exception as e:
        print(f"[face] detection failed, using whole image: {e}", flush=True)
        return pil_image, False


# ---------------- Layer 4: saliency heatmap ----------------
def make_saliency_heatmap(pil_image, model, processor, target_class, scan_id):
    try:
        rgb = pil_image.convert("RGB")
        original_size = rgb.size
        inputs = processor(images=rgb, return_tensors="pt")
        pixel_values = inputs["pixel_values"].clone().detach().requires_grad_(True)

        model.zero_grad()
        outputs = model(pixel_values=pixel_values)
        outputs.logits[0, target_class].backward()

        grads = pixel_values.grad[0].abs().mean(dim=0).cpu().numpy()
        gmin, gmax = grads.min(), grads.max()
        if gmax - gmin < 1e-8:
            return ""
        heat = ((grads - gmin) / (gmax - gmin) * 255).astype(np.uint8)
        heat_resized = cv2.resize(heat, original_size, interpolation=cv2.INTER_CUBIC)
        heat_colour = cv2.applyColorMap(heat_resized, cv2.COLORMAP_JET)
        orig_bgr = cv2.cvtColor(np.array(rgb), cv2.COLOR_RGB2BGR)
        overlay = cv2.addWeighted(orig_bgr, 0.55, heat_colour, 0.45, 0)

        safe_id = "".join(c for c in str(scan_id) if c.isalnum() or c in "-_")[:64] or "scan"
        out_path = os.path.join(HEATMAP_DIR, f"{safe_id}.jpg")
        cv2.imwrite(out_path, overlay, [int(cv2.IMWRITE_JPEG_QUALITY), 85])

        if SPACE_URL:
            return f"{SPACE_URL}/heatmaps/{safe_id}.jpg"
        return ""
    except Exception as e:
        print(f"[heatmap] failed: {e}")
        return ""


def fake_class_index(id2label):
    for idx, label in id2label.items():
        if any(k in str(label).lower() for k in ["fake", "deepfake", "synthetic", "manipulated"]):
            return int(idx)
    return None


# ---------------- Analysis: image ----------------
def analyze_image_bytes(image_bytes: bytes, scan_id: str) -> dict:
    processor, model = get_image_model()
    original = Image.open(io.BytesIO(image_bytes)).convert("RGB")

    # Layer 1 forensics run on the ORIGINAL image (EXIF/dimensions belong to the
    # file as submitted, not to a crop).
    forensic_score, flags = image_forensic_analysis(original)

    # Preprocessing: crop to the detected face before inference. Fall back to the
    # whole image (disclosed) when no face is found.
    face_img, face_found = crop_to_face(original)

    # Layer 2: learned detector.
    inputs = processor(images=face_img, return_tensors="pt")
    with torch.no_grad():
        logits = model(**inputs).logits
    probs = torch.softmax(logits, dim=-1)[0].tolist()
    fake_prob, real_prob = split_probs(probs, model.config.id2label)

    # Layer 3: fuse the model probability with the forensic score, then band.
    fused = fuse_scores(fake_prob, forensic_score)
    band = score_to_band(fused, model_fake_prob=fake_prob)
    confidence = round((fused if band != "authentic" else (1.0 - fused)) * 100, 1)

    # Layer 4: heatmap when the fused score reaches the suspicious threshold.
    heatmap_url = ""
    if fused >= BAND_SUSPICIOUS:
        tc = fake_class_index(model.config.id2label)
        if tc is not None:
            heatmap_url = make_saliency_heatmap(face_img, model, processor, tc, scan_id)

    lines = []
    if band == "authentic":
        lines.append(f"• Detection model: {round(real_prob*100,1)}% consistent with a real photograph")
    elif band == "suspicious":
        lines.append(f"• Assessment uncertain (combined manipulation score {round(fused*100,1)}%)")
    else:
        lines.append(f"• Detection model: {round(fake_prob*100,1)}% likely AI-generated or manipulated")
    lines.append(
        "• Analysis focused on the detected face"
        if face_found else
        "• No face detected — whole image analysed (result may be less precise)"
    )
    for f in flags:
        lines.append(f"• {f}")
    if forensic_score > 0 and band != "authentic":
        lines.append(f"• Forensic signals contributed to this verdict (Layer 1 score {round(forensic_score*100)}%)")
    if heatmap_url:
        lines.append("• Heatmap shows the regions that most influenced this verdict")

    return {
        "verdict": band,
        "confidence": confidence,
        "insights": "\n".join(lines),
        "heatmap_url": heatmap_url,
        "raw_scores": {
            "layer1_forensic": round(forensic_score, 4),
            "layer2_model_fake": round(fake_prob, 4),
            "layer2_model_real": round(real_prob, 4),
            "layer3_fused": round(fused, 4),
            "face_detected": face_found,
        },
    }


# ---------------- Analysis: audio ----------------
def analyze_audio_bytes(audio_bytes: bytes) -> dict:
    extractor, model = get_audio_model()

    with tempfile.NamedTemporaryFile(delete=False, suffix=".audio") as f:
        f.write(audio_bytes)
        tmp_path = f.name
    try:
        waveform, _ = librosa.load(tmp_path, sr=16000, mono=True)
    finally:
        try:
            os.unlink(tmp_path)
        except Exception:
            pass

    max_samples = 30 * 16000  # cap at 30s
    truncated = len(waveform) > max_samples
    waveform = waveform[:max_samples]

    inputs = extractor(waveform, sampling_rate=16000, return_tensors="pt")
    with torch.no_grad():
        logits = model(**inputs).logits
    probs = torch.softmax(logits, dim=-1)[0].tolist()

    fake_prob, real_prob = split_probs(probs, model.config.id2label)
    # Audio has no EXIF/dimension forensics, so Layer 1 contributes no signal here
    # and the fused score equals the model probability. This is disclosed rather
    # than papered over with an invented forensic score.
    forensic_score = 0.0
    fused = fuse_scores(fake_prob, forensic_score)
    band = score_to_band(fused, model_fake_prob=fake_prob)
    confidence = round((fused if band != "authentic" else (1.0 - fused)) * 100, 1)

    lines = []
    if band == "authentic":
        lines.append(f"• Voice appears human ({round(real_prob*100,1)}% confidence)")
    elif band == "suspicious":
        lines.append(f"• Uncertain — borderline synthetic characteristics ({round(fake_prob*100,1)}%)")
    else:
        lines.append(f"• Voice appears AI-generated or cloned ({round(fake_prob*100,1)}% confidence)")
    secs = int(len(waveform) / 16000)
    lines.append(f"• Analysed {secs}s of audio at 16 kHz" + (" (clip truncated to 30s)" if truncated else ""))

    return {
        "verdict": band,
        "confidence": confidence,
        "insights": "\n".join(lines),
        "heatmap_url": "",
        "raw_scores": {
            "layer1_forensic": 0.0,
            "layer2_model_fake": round(fake_prob, 4),
            "layer2_model_real": round(real_prob, 4),
            "layer3_fused": round(fused, 4),
        },
    }


# ---------------- Analysis: video ----------------
def analyze_video_bytes(video_bytes: bytes, scan_id: str) -> dict:
    import time
    t0 = time.time()
    print(f"[video] received {len(video_bytes)} bytes", flush=True)

    with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as f:
        f.write(video_bytes)
        tmp_path = f.name
    try:
        cap = cv2.VideoCapture(tmp_path)
        if not cap.isOpened():
            raise ValueError("Video could not be opened (unsupported codec or corrupt file)")

        total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
        print(f"[video] reported frame count: {total}", flush=True)

        processor, model = get_image_model()
        id2label = model.config.id2label
        fake_probs, real_probs = [], []
        best_frame, best_fake = None, -1.0
        faces_found = 0

        def _score(pil):
            # Crop to the detected face before inference; fall back to whole frame.
            nonlocal faces_found
            face_img, face_ok = crop_to_face(pil)
            if face_ok:
                faces_found += 1
            inputs = processor(images=face_img, return_tensors="pt")
            with torch.no_grad():
                logits = model(**inputs).logits
            probs = torch.softmax(logits, dim=-1)[0].tolist()
            fp, rp = split_probs(probs, id2label)
            return fp, rp, face_img

        if total > 0:
            # Seek to evenly-spaced frames
            indices = np.linspace(0, total - 1, num=min(VIDEO_FRAMES_TO_SAMPLE, total), dtype=int)
            for idx in indices:
                cap.set(cv2.CAP_PROP_POS_FRAMES, int(idx))
                ok, frame = cap.read()
                if not ok:
                    continue
                pil = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
                fp, rp, face_img = _score(pil)
                fake_probs.append(fp)
                real_probs.append(rp)
                if fp > best_fake:
                    best_fake, best_frame = fp, face_img
        else:
            # Some containers report 0 frames: fall back to sequential reading
            print("[video] frame count unknown, reading sequentially", flush=True)
            step, read, kept = 10, 0, 0
            while kept < VIDEO_FRAMES_TO_SAMPLE and read < 600:
                ok, frame = cap.read()
                if not ok:
                    break
                read += 1
                if read % step:
                    continue
                pil = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
                fp, rp, face_img = _score(pil)
                fake_probs.append(fp)
                real_probs.append(rp)
                if fp > best_fake:
                    best_fake, best_frame = fp, face_img
                kept += 1
        cap.release()
    finally:
        try:
            os.unlink(tmp_path)
        except Exception:
            pass

    if not fake_probs:
        raise ValueError("Could not extract any frames from the video")

    print(f"[video] scored {len(fake_probs)} frames in {time.time()-t0:.1f}s", flush=True)

    mean_fake = float(np.mean(fake_probs))
    mean_real = float(np.mean(real_probs))
    # Decoded video frames carry no reliable per-file forensic metadata, so the
    # forensic contribution is honestly zero and the fused score is model-driven.
    # It is still routed through the same fusion/banding path for consistency and
    # to keep the fail-honest accusation guardrail.
    forensic_score = 0.0
    fused = fuse_scores(mean_fake, forensic_score)
    band = score_to_band(fused, model_fake_prob=mean_fake)
    confidence = round((fused if band != "authentic" else (1.0 - fused)) * 100, 1)

    heatmap_url = ""
    if best_frame is not None and fused >= BAND_SUSPICIOUS:
        tc = fake_class_index(id2label)
        if tc is not None:
            heatmap_url = make_saliency_heatmap(best_frame, model, processor, tc, scan_id)

    lines = [f"• Sampled {len(fake_probs)} frames evenly across the video"]
    if band == "authentic":
        lines.append(f"• Frames consistent with real footage (mean {round(mean_real*100,1)}% real)")
    elif band == "suspicious":
        lines.append(f"• Some frames flagged; mean manipulation score {round(mean_fake*100,1)}%")
    else:
        lines.append(f"• Manipulation likely; mean manipulation score {round(mean_fake*100,1)}%")
    lines.append(
        f"• Face detected and analysed in {faces_found} of {len(fake_probs)} sampled frames"
        if faces_found else
        "• No face detected in sampled frames — whole frames analysed (result may be less precise)"
    )
    if heatmap_url:
        lines.append("• Heatmap shows the most suspicious frame")

    return {
        "verdict": band,
        "confidence": confidence,
        "insights": "\n".join(lines),
        "heatmap_url": heatmap_url,
        "raw_scores": {
            "layer2_model_fake": round(mean_fake, 4),
            "layer3_fused": round(fused, 4),
            "frames_sampled": len(fake_probs),
            "frames_with_face": faces_found,
        },
    }


# ---------------- Core dispatch ----------------
def run_analysis(raw: bytes, media_type: str, scan_id: str) -> dict:
    mt = (media_type or "").lower().strip()
    if mt == "image":
        result = analyze_image_bytes(raw, scan_id)
    elif mt == "audio":
        result = analyze_audio_bytes(raw)
    elif mt == "video":
        result = analyze_video_bytes(raw, scan_id)
    else:
        raise ValueError(f"Unknown media_type: {media_type}")
    result["file_hash"] = "sha256:" + hashlib.sha256(raw).hexdigest()
    return result


# ---------------- FastAPI endpoints mounted into Gradio ----------------
api = FastAPI()


class AnalyzeRequest(BaseModel):
    file_url: str
    media_type: str
    scan_id: str = "scan"


@api.get("/health")
def health():
    return {"status": "ok", "service": "DEEPFYND Intelligence Plane"}


@api.get("/heatmaps/{name}")
def get_heatmap(name: str):
    safe = os.path.basename(name)
    path = os.path.join(HEATMAP_DIR, safe)
    if not os.path.isfile(path):
        raise HTTPException(status_code=404, detail="heatmap not found")
    return FileResponse(path, media_type="image/jpeg")


def fetch_media(url: str) -> bytes:
    """
    Download the media to analyse.

    Hugging Face's outbound connections (notably to api.telegram.org) can be slow,
    so we use a generous read timeout, stream the body, and retry transient failures.
    """
    # Defensive cleaning: upstream channels (e.g. a Make.com HTTP body with a stray
    # space after the URL pill) can append whitespace, which becomes %20 and causes
    # a 404. Strip surrounding whitespace and any literal/encoded trailing spaces so
    # the fetch is robust to that class of mistake.
    if url:
        url = url.strip()
        while url.endswith("%20") or url.endswith("%09"):
            url = url[:-3].strip()
    last_err = None
    for attempt in range(3):
        try:
            print(f"[fetch] attempt {attempt + 1}: {url[:80]}...", flush=True)
            # Single value applies to BOTH connect and read phases.
            r = requests.get(url, timeout=120, stream=True)
            r.raise_for_status()
            chunks = []
            total = 0
            for chunk in r.iter_content(chunk_size=65536):
                if not chunk:
                    continue
                chunks.append(chunk)
                total += len(chunk)
                if total > MAX_MEDIA_BYTES:
                    raise ValueError(
                        f"Media exceeds {MAX_MEDIA_BYTES // (1024*1024)} MB limit"
                    )
            return b"".join(chunks)
        except ValueError:
            raise
        except Exception as e:
            last_err = e
            print(f"[fetch] attempt {attempt + 1} failed: {e}", flush=True)
    raise RuntimeError(f"Could not fetch media after 3 attempts: {last_err}")


@api.post("/analyze")
def analyze(req: AnalyzeRequest):
    try:
        raw = fetch_media(req.file_url)
    except ValueError as e:
        raise HTTPException(status_code=413, detail=str(e))
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"Could not fetch media: {e}")

    try:
        return run_analysis(raw, req.media_type, req.scan_id)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=f"Analysis failed: {e}")


# ---------------- Gradio demo UI (DEEPFYND branded) ----------------
BAND_META = {
    "authentic":       {"label": "AUTHENTIC",       "colour": "#1a7f37", "icon": "\u2713"},
    "suspicious":      {"label": "SUSPICIOUS",      "colour": "#b58900", "icon": "!"},
    "likely_deepfake": {"label": "LIKELY DEEPFAKE", "colour": "#D11A1A", "icon": "\u2715"},
}

BRAND_CSS = """
@import url('https://fonts.googleapis.com/css2?family=Manrope:wght@400;600;700;800&display=swap');

.gradio-container, .gradio-container * {
    font-family:'Manrope', -apple-system, 'Segoe UI', Roboto, sans-serif !important;
    -webkit-font-smoothing:antialiased; -moz-osx-font-smoothing:grayscale;
}
.gradio-container { max-width: 1120px !important; margin: 0 auto !important; }
footer { display:none !important; }

/* ---- Header ---- */
#dfy-head {
    display:flex; flex-direction:column; align-items:center; justify-content:center;
    text-align:center; padding: 30px 0 4px; width:100%;
}
#dfy-head img { width: 320px; max-width:74vw; height:auto; display:block; margin:0 auto; }
#dfy-rule { width:72px; height:3px; background:#D11A1A; border-radius:2px; margin:18px auto 14px; }
#dfy-sub {
    text-align:center; color:#333; font-size:1.18rem; font-weight:600;
    margin:0 auto 6px; max-width:680px; line-height:1.5;
}
#dfy-api {
    text-align:center; font-size:.92rem; color:#6b6b6b; margin:0 auto 26px;
    max-width:720px; line-height:1.55;
}
#dfy-api code {
    background:#f5f5f5; padding:2px 7px; border-radius:5px; color:#D11A1A;
    font-size:.88rem; font-weight:600;
}

/* ---- Controls ---- */
.gradio-container label, .gradio-container .label-wrap span { font-size:.98rem !important; font-weight:600 !important; }
#dfy-note { font-size:.90rem; color:#777; margin-top:10px; line-height:1.6; }

/* ---- Cards ---- */
#dfy-card {
    border:1px solid #e8e8e8; border-radius:16px; padding:24px 26px; background:#fff;
    box-shadow:0 2px 18px rgba(0,0,0,.07);
}
.dfy-verdict {
    display:flex; align-items:center; gap:16px;
    border-radius:12px; padding:20px 22px; color:#fff; margin-bottom:18px;
}
.dfy-verdict .badge {
    width:50px; height:50px; border-radius:50%; background:rgba(255,255,255,.20);
    display:flex; align-items:center; justify-content:center;
    font-size:1.6rem; font-weight:800; flex:none;
}
.dfy-verdict .vtxt { font-size:1.52rem; font-weight:800; letter-spacing:.02em; line-height:1.15; }
.dfy-verdict .vconf { font-size:.98rem; opacity:.94; margin-top:3px; font-weight:600; }
.dfy-bar-wrap { background:#ececec; border-radius:99px; height:10px; overflow:hidden; margin:0 0 20px; }
.dfy-bar { height:100%; border-radius:99px; }
.dfy-h {
    font-size:.78rem; letter-spacing:.16em; text-transform:uppercase;
    color:#666; font-weight:800; margin:20px 0 9px;
}
.dfy-reasons { line-height:1.8; font-size:1.04rem; color:#141414; font-weight:400; }
.dfy-hash {
    font-family:ui-monospace,Menlo,monospace !important; font-size:.78rem; color:#7a7a7a;
    word-break:break-all; background:#fafafa; padding:11px 13px; border-radius:8px;
}
.dfy-disc {
    border-left:4px solid #D11A1A; background:#fff6f6; padding:13px 15px;
    border-radius:7px; font-size:.96rem; color:#6f1414; margin-top:18px; line-height:1.6;
}
.dfy-err { border-left:4px solid #D11A1A; background:#fff6f6; padding:20px 22px; border-radius:11px; }
.dfy-err h3 { margin:0 0 8px; color:#D11A1A; font-size:1.2rem; font-weight:800; }
"""


def _placeholder_html() -> str:
    return (
        "<div id='dfy-card' style='text-align:center;color:#888;padding:44px 20px;'>"
        "<div style='font-size:2rem;margin-bottom:8px;'>\u25CE</div>"
        "<div style='font-weight:600;color:#444;'>No analysis yet</div>"
        "<div style='font-size:.88rem;margin-top:4px;'>"
        "Upload an image, audio clip or video, then press Analyse.</div></div>"
    )


def _result_html(res: dict) -> str:
    meta = BAND_META.get(res["verdict"], BAND_META["suspicious"])
    conf = res["confidence"]
    reasons = "".join(
        f"<div>{line.strip()}</div>"
        for line in res["insights"].split("\n") if line.strip()
    )
    return f"""
    <div id='dfy-card'>
      <div class='dfy-verdict' style='background:{meta["colour"]};'>
        <div class='badge'>{meta["icon"]}</div>
        <div>
          <div class='vtxt'>{meta["label"]}</div>
          <div class='vconf'>Confidence {conf}%</div>
        </div>
      </div>
      <div class='dfy-bar-wrap'>
        <div class='dfy-bar' style='width:{conf}%;background:{meta["colour"]};'></div>
      </div>
      <div class='dfy-h'>Why this verdict</div>
      <div class='dfy-reasons'>{reasons}</div>
      <div class='dfy-h'>File hash (chain of custody)</div>
      <div class='dfy-hash'>{res["file_hash"]}</div>
      <div class='dfy-disc'>
        <b>Decision support, not proof.</b> DEEPFYND explains what it found so you can
        judge. Verify important content with a professional fact-checker.
      </div>
    </div>
    """


def _error_html(msg: str) -> str:
    return (
        "<div id='dfy-card'><div class='dfy-err'>"
        "<h3>Analysis failed</h3>"
        "<div style='color:#444;font-size:.92rem;'>Nothing was assessed. "
        "DEEPFYND never produces a verdict when analysis cannot run.</div>"
        f"<div class='dfy-hash' style='margin-top:10px;'>{msg}</div>"
        "</div></div>"
    )


def ui_analyze(file_obj, media_type):
    if file_obj is None:
        return _error_html("No file supplied. Please upload media first."), None
    try:
        with open(file_obj, "rb") as f:
            raw = f.read()
        res = run_analysis(raw, media_type, "uiscan")
        heat_path = None
        local = os.path.join(HEATMAP_DIR, "uiscan.jpg")
        if res.get("heatmap_url") and os.path.isfile(local):
            heat_path = local
        return _result_html(res), heat_path
    except Exception as e:
        return _error_html(str(e)), None


LOGO_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "logo.png")


def _logo_html() -> str:
    """Inline the logo as base64 so it renders regardless of static file routing."""
    import base64
    try:
        with open(LOGO_FILE, "rb") as f:
            b64 = base64.b64encode(f.read()).decode()
        return (
            "<div id='dfy-head'>"
            f"<img src='data:image/png;base64,{b64}' alt='DEEPFYND — Detect. Analyze. Report.'/>"
            "</div>"
        )
    except Exception:
        return (
            "<div id='dfy-head'><h1 style='margin:0;font-size:2.4rem;letter-spacing:-.02em;'>"
            "deep<span style='color:#D11A1A'>fynd</span></h1>"
            "<div style='letter-spacing:.28em;font-size:.7rem;font-weight:700;margin-top:4px;'>"
            "DETECT. ANALYZE. <span style='color:#D11A1A'>REPORT.</span></div></div>"
        )


with gr.Blocks(title="DEEPFYND — Intelligence Plane") as demo:
    gr.HTML(
        _logo_html()
        + "<div id='dfy-rule'></div>"
        + "<div id='dfy-sub'>Explainable deepfake detection for images, audio and video.</div>"
        + "<div id='dfy-api'>This Space also serves the <code>POST /analyze</code> API used by "
          "the DEEPFYND web, Android and Telegram channels.</div>"
    )
    with gr.Row():
        with gr.Column(scale=4):
            file_in = gr.File(label="Upload media", type="filepath")
            type_in = gr.Radio(
                ["image", "audio", "video"], value="image", label="Media type"
            )
            btn = gr.Button("Analyse", variant="stop")
            gr.HTML(
                "<div id='dfy-note'>"
                "Images return in seconds. Video is sampled across "
                f"{VIDEO_FRAMES_TO_SAMPLE} frames and takes longer on free CPU."
                "</div>"
            )
        with gr.Column(scale=6):
            out_html = gr.HTML(_placeholder_html())
            out_img = gr.Image(label="Explanation heatmap", type="filepath")

    btn.click(ui_analyze, inputs=[file_in, type_in], outputs=[out_html, out_img])


# ---------------- Launch ----------------
# On a Gradio Space, HF runs `python app.py`.
#
# Gradio 6 registers a catch-all GET route. A POST to /analyze matches that
# path but not its method, so Starlette answers 405 "Method Not Allowed".
# We therefore build our routes explicitly and PREPEND them to the route table
# so they are matched before Gradio's catch-all.
if __name__ == "__main__":
    from fastapi.routing import APIRoute

    demo.queue()
    demo.launch(
        server_name="0.0.0.0",
        server_port=int(os.environ.get("GRADIO_SERVER_PORT", 7860)),
        prevent_thread_lock=True,
        css=BRAND_CSS,
        # Gradio 6 puts a Node/SvelteKit SSR server in front of Python. It answers
        # POST /analyze with "Method Not Allowed" before FastAPI ever sees it.
        # Disabling SSR makes the Python FastAPI app serve every request.
        ssr_mode=False,
    )

    platform_routes = [
        APIRoute("/analyze", analyze, methods=["POST"]),
        APIRoute("/health", health, methods=["GET"]),
        APIRoute("/heatmaps/{name}", get_heatmap, methods=["GET"]),
    ]
    demo.app.router.routes[0:0] = platform_routes

    print(
        ">>> DEEPFYND REST routes mounted with priority: "
        + ", ".join(r.path for r in platform_routes),
        flush=True,
    )

    # Warm the image model in the background so the first real request does not
    # pay the ~40-60s model-load cost (which caused Make.com timeouts).
    import threading

    def _warm():
        try:
            get_image_model()
            print(">>> image model warmed", flush=True)
        except Exception as e:
            print(f">>> warm-up failed (non-fatal): {e}", flush=True)

    threading.Thread(target=_warm, daemon=True).start()

    threading.Event().wait()