app.py

import html
import os
import time
from collections.abc import Callable
from pathlib import Path

import gradio as gr
import requests
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.requests import Request
from starlette.responses import Response

BASE_URL = "https://api.dataspike.io"
IMAGE_ENDPOINT = "/api/v4/deepfake/image/analyze"
VIDEO_ENDPOINT = "/api/v4/deepfake/video/analyze"
AUDIO_ENDPOINT = "/api/v4/deepfake/audio/analyze"
JOB_ENDPOINT = "/api/v4/deepfake/job/{job_id}"
TERMINAL_STATUSES = {"completed", "done", "error", "failed"}
REQUEST_TIMEOUT = (5, 30)  # (connect, read) seconds

GENUINE_COLOR = "#2ECC71"
DEEPFAKE_COLOR = "#E74C3C"
NEUTRAL_COLOR = "#95A5A6"

DEEPFAKE_SCORE_THRESHOLD = 0.5

FAKE_MARKERS = ("fake", "deepfake", "spoof", "manipulat", "suspic")
REAL_MARKERS = ("real", "genuine", "authentic", "live")

FAILURE_STATUSES = {"error", "failed", "timeout"}

PRESETS_DIR = Path(__file__).resolve().parent / "presets"
FACE_MEDIA_HINT_HTML = """
<div class="face-media-hint">
    <div class="face-media-icon" aria-hidden="true">
        <svg viewBox="0 0 24 24" fill="none" stroke="currentColor"
             stroke-width="2" stroke-linecap="round" stroke-linejoin="round">
            <path d="M3 7V5a2 2 0 0 1 2-2h2" />
            <path d="M17 3h2a2 2 0 0 1 2 2v2" />
            <path d="M21 17v2a2 2 0 0 1-2 2h-2" />
            <path d="M7 21H5a2 2 0 0 1-2-2v-2" />
            <path d="M8 14s1.5 2 4 2 4-2 4-2" />
            <path d="M9 9h.01" />
            <path d="M15 9h.01" />
        </svg>
    </div>
    <div>
        <strong>Human faces only</strong>
        <span>Use selfie-style media with a clearly visible face.</span>
    </div>
</div>
"""
IMAGE_EXAMPLES = [
    str(PRESETS_DIR / "image" / "ai-generated-selfie.png"),
    str(PRESETS_DIR / "image" / "genuine-face.jpg"),
]
VIDEO_EXAMPLES = [
    str(PRESETS_DIR / "video" / "manipulated-face.mp4"),
    str(PRESETS_DIR / "video" / "genuine-face.mp4"),
]
AUDIO_EXAMPLES = [
    str(PRESETS_DIR / "audio" / "ai-generated-voice.wav"),
    str(PRESETS_DIR / "audio" / "genuine-voice.wav"),
]


def failure_message(result: dict) -> str:
    message = result.get("message")
    if isinstance(message, str) and message:
        return message
    errors = result.get("errors")
    if isinstance(errors, list) and errors:
        return "; ".join(str(e) for e in errors)
    error = result.get("error")
    if isinstance(error, str) and error:
        return error
    return "analysis failed"


def score_of(result: dict) -> float | None:
    """The deepfake score as a 0..1 float, or None when the API returned none.

    Booleans are rejected even though `bool` is an `int`, and out-of-range
    values are clamped so the scale marker cannot escape its track.
    """

    if not isinstance(result, dict):
        return None
    score = result.get("score")
    if isinstance(score, bool) or not isinstance(score, (int, float)):
        return None
    return min(1.0, max(0.0, float(score)))


def classify(result: dict) -> tuple[str, str]:
    """Best-effort verdict word and colour from a deepfake API response.

    The score is deliberately left out of the text: it is the likelihood of the
    media being a deepfake, so printing it next to "Genuine" read as confidence
    in the verdict when it means the opposite. `verdict_html` renders it on a
    scale instead.
    """

    if not isinstance(result, dict):
        return "Unknown", NEUTRAL_COLOR

    if (
        result.get("status") in FAILURE_STATUSES
        or result.get("overallStatus") == "Failure"
    ):
        return f"Failure: {failure_message(result)}", NEUTRAL_COLOR

    score = score_of(result)
    if score is not None:
        if score >= DEEPFAKE_SCORE_THRESHOLD:
            return "Deepfake", DEEPFAKE_COLOR
        return "Genuine", GENUINE_COLOR

    verdict = result.get("verdict")
    if isinstance(verdict, str) and verdict:
        v = verdict.lower()
        if any(m in v for m in FAKE_MARKERS):
            return "Deepfake", DEEPFAKE_COLOR
        if any(m in v for m in REAL_MARKERS):
            return "Genuine", GENUINE_COLOR
        return verdict, NEUTRAL_COLOR

    error = result.get("error")
    if isinstance(error, str) and error and error.lower() != "ok":
        return f"Failure: {error}", NEUTRAL_COLOR

    return "Unknown", NEUTRAL_COLOR


def auth_headers() -> dict[str, str | None]:
    return {"ds-api-token": os.getenv("API_KEY")}


def http_error_message(exc: requests.HTTPError) -> str:
    response = exc.response
    if response is None:
        return str(exc)
    try:
        body = response.json()
    except ValueError:
        body = None
    if isinstance(body, dict):
        msg = body.get("message") or body.get("error")
        if msg:
            return f"{response.status_code}: {msg}"
    return f"{response.status_code}: {response.reason}"


def post_file(endpoint: str, file_path: str) -> dict:
    try:
        with open(file_path, "rb") as f:
            response = requests.post(
                BASE_URL + endpoint,
                headers=auth_headers(),
                files={"file": f},
                timeout=REQUEST_TIMEOUT,
            )
        response.raise_for_status()
        return response.json()
    except requests.HTTPError as exc:
        return {"status": "error", "message": http_error_message(exc)}
    except (requests.RequestException, ValueError) as exc:
        return {"status": "error", "message": str(exc)}


def poll_job(job_id: str, interval: float = 2.0, max_retries: int = 90) -> dict:
    url = BASE_URL + JOB_ENDPOINT.format(job_id=job_id)
    for _ in range(max_retries):
        try:
            response = requests.get(
                url, headers=auth_headers(), timeout=REQUEST_TIMEOUT
            )
            response.raise_for_status()
            result = response.json()
        except requests.HTTPError as exc:
            return {"status": "error", "message": http_error_message(exc)}
        except (requests.RequestException, ValueError) as exc:
            return {"status": "error", "message": str(exc)}
        if result.get("status") in TERMINAL_STATUSES:
            return result
        time.sleep(interval)
    return {"status": "timeout", "message": f"Job {job_id} did not complete in time"}


LANDING_URL = "https://dataspike.io/deepfake-detection"
SUBTITLE = (
    "Deepfakes, AI-generated and manipulated human faces. "
    "Selfie-style photo, video or voice. Built for KYC liveness."
)

HTML_HEADER = f"""
<header style="text-align: center; padding: 20px; border-bottom: 2px solid #cc3300;">
    <h1>Demo of Deepfake Detection</h1>
    <p style="font-size: 18px; margin: 8px auto; max-width: 760px;">
        {SUBTITLE}
    </p>
    <p style="font-size: 18px;">
        To learn more, visit our website: <a href="{LANDING_URL}" target="_blank" style="font-size: 20px; text-decoration: none;">
        {LANDING_URL} </a>
    </p>
</header>
"""

HTML_EXPLANATION = """
<ul style="margin: 0; padding-left: 20px;">
    <li><strong>Score</strong>: how likely the media is a deepfake, from 0 to 1.
        The lower it sits on the scale, the more authentic the media looks.</li>
    <li><strong>Genuine</strong>: score below 0.5.</li>
    <li><strong>Deepfake</strong>: score 0.5 or above, the production threshold
        marked on the scale.</li>
</ul>
"""


VERDICT_INK = "#0B0F19"
VERDICT_CAPTION = "Verdict"
AUTHENTIC_END_LABEL = "Authentic"
DEEPFAKE_END_LABEL = "Deepfake"
THRESHOLD_LABEL = f"Threshold {DEEPFAKE_SCORE_THRESHOLD:.2f}"

EMPTY_VERDICT_HTML = (
    '<div class="verdict-panel"><div class="verdict-headline">'
    f'<span class="verdict-cap">{VERDICT_CAPTION}</span>'
    '<span class="verdict-empty">Run an analysis to see the result.</span>'
    "</div></div>"
)


def verdict_html(result: dict) -> str:
    """The verdict word, plus a scale placing the score against the threshold.

    Text coming from the API is escaped: unlike `gr.Label`, `gr.HTML` renders
    its value as markup.
    """

    text, color = classify(result)
    headline = (
        '<div class="verdict-headline">'
        f'<span class="verdict-cap">{VERDICT_CAPTION}</span>'
        f'<span class="verdict-word" style="color: {color}">{html.escape(text)}</span>'
        "</div>"
    )
    score = score_of(result)
    if score is None:
        return f'<div class="verdict-panel">{headline}</div>'
    return (
        '<div class="verdict-panel">'
        f'<div class="verdict-top">{headline}'
        f'<span class="verdict-score" style="background: {color};'
        f' color: {VERDICT_INK}">Score {score:.2f}</span>'
        "</div>"
        '<div class="verdict-scale">'
        '<div class="verdict-track">'
        '<span class="verdict-threshold"'
        f' style="left: {DEEPFAKE_SCORE_THRESHOLD:.1%}"></span>'
        f'<span class="verdict-marker" style="left: {score:.1%};'
        f' background: {color}"></span>'
        "</div>"
        '<div class="verdict-ends">'
        f"<span>{AUTHENTIC_END_LABEL}</span>"
        f'<span class="verdict-mid">{THRESHOLD_LABEL}</span>'
        f"<span>{DEEPFAKE_END_LABEL}</span>"
        "</div>"
        "</div>"
        "</div>"
    )


def no_file_failure(message: str) -> tuple[dict, str]:
    result = {"overallStatus": "Failure", "errors": [message]}
    return result, verdict_html(result)


def analyze_image(file_path: str | None) -> tuple[dict, str]:
    if not file_path:
        return no_file_failure("Please submit an image first.")
    result = post_file(IMAGE_ENDPOINT, file_path)
    return result, verdict_html(result)


def analyze_video(
    file_path: str | None, progress: gr.Progress = gr.Progress()
) -> tuple[dict, str]:
    if not file_path:
        return no_file_failure("Please submit a video first.")
    submitted = post_file(VIDEO_ENDPOINT, file_path)
    job_id = submitted.get("id")
    if not job_id:
        return submitted, verdict_html(submitted)
    progress(0.5, desc="Analyzing video...")
    result = poll_job(job_id)
    return result, verdict_html(result)


def analyze_audio(
    file_path: str | None, progress: gr.Progress = gr.Progress()
) -> tuple[dict, str]:
    if not file_path:
        return no_file_failure("Please submit an audio file first.")
    submitted = post_file(AUDIO_ENDPOINT, file_path)
    job_id = submitted.get("id")
    if not job_id:
        return submitted, verdict_html(submitted)
    progress(0.5, desc="Analyzing audio...")
    result = poll_job(job_id)
    return result, verdict_html(result)


tabs_css = """
button[role="tab"] {
    font-size: 14px !important;
    font-family: 'Montserrat', sans-serif !important;
    font-weight: 600 !important;
    padding: 12px 24px !important;
    margin: 0 6px !important;
    background-color: #0B0F19 !important;
    color: #F3F4F6 !important;
    border-radius: 8px !important;
    border: 1px solid #1a1a1a !important;
    box-shadow: none !important;
    transition: all 0.2s ease !important;
}
button[role="tab"].selected {
    background-color: #635bff !important;
    color: white !important;
    box-shadow: 0 0 6px rgba(99, 91, 255, 0.5) !important;
}
button[role="tab"]:not(.selected) {
    background-color: #9D2C53 !important;
    color: #F3F4F6 !important;
}
button[role="tab"]:hover {
    background-color: #1a1a2b !important;
    color: white !important;
}
.face-media-hint {
    display: flex;
    align-items: center;
    gap: 12px;
    margin-bottom: 8px;
    padding: 10px 12px;
    border: 1px solid rgba(99, 91, 255, 0.35);
    border-radius: 12px;
    background: rgba(99, 91, 255, 0.08);
}
.face-media-icon {
    display: grid;
    width: 52px;
    height: 52px;
    flex: 0 0 52px;
    place-items: center;
    border: 1px solid rgba(139, 140, 255, 0.2);
    border-radius: 14px;
    background: #202127;
    color: #8b8cff;
}
.face-media-icon svg {
    width: 32px;
    height: 32px;
}
.face-media-hint strong,
.face-media-hint span {
    display: block;
}
.face-media-hint span {
    margin-top: 2px;
    opacity: 0.78;
}
.verdict-panel {
    display: flex;
    flex-direction: column;
    gap: 16px;
    padding: 18px;
    border: 1px solid #262b3d;
    border-radius: 12px;
    background: #171b28;
}
.verdict-top {
    display: flex;
    align-items: center;
    justify-content: space-between;
    gap: 12px;
}
.verdict-headline {
    display: flex;
    flex-direction: column;
    gap: 2px;
}
.verdict-cap {
    font-size: 10.5px;
    font-weight: 700;
    letter-spacing: 0.12em;
    text-transform: uppercase;
    color: #9aa0b4;
}
.verdict-empty {
    font-size: 14px;
    color: #9aa0b4;
}
.verdict-word {
    font-size: 23px;
    font-weight: 700;
    letter-spacing: -0.01em;
}
.verdict-score {
    padding: 4px 10px;
    border-radius: 999px;
    font-size: 11px;
    font-weight: 700;
    letter-spacing: 0.08em;
    text-transform: uppercase;
    font-variant-numeric: tabular-nums;
    white-space: nowrap;
}
.verdict-scale {
    display: flex;
    flex-direction: column;
    gap: 7px;
}
.verdict-track {
    position: relative;
    height: 9px;
    border-radius: 999px;
}
.verdict-track::before {
    content: "";
    position: absolute;
    inset: 0;
    border-radius: inherit;
    background: linear-gradient(90deg, #2ecc71 0%, #b8c832 46%, #e74c3c 100%);
    opacity: 0.35;
}
.verdict-threshold {
    position: absolute;
    z-index: 1;
    top: -5px;
    bottom: -5px;
    width: 2px;
    background: #f3f4f6;
    opacity: 0.5;
}
.verdict-marker {
    position: absolute;
    z-index: 1;
    top: 50%;
    width: 16px;
    height: 16px;
    margin-left: -8px;
    border: 3px solid #171b28;
    border-radius: 50%;
    transform: translateY(-50%);
}
.verdict-ends {
    display: flex;
    justify-content: space-between;
    font-size: 10.5px;
    font-weight: 600;
    letter-spacing: 0.07em;
    text-transform: uppercase;
    color: #9aa0b4;
}
.verdict-mid {
    opacity: 0.8;
}
"""


ENGLISH_LOCALE_SCRIPT = b"""<script>
Object.defineProperty(navigator, "language", {get: () => "en"});
Object.defineProperty(navigator, "languages", {get: () => ["en"]});
</script>"""


class EnglishLocaleMiddleware(BaseHTTPMiddleware):
    """Pins the Gradio UI to English, which otherwise follows the browser locale."""

    async def dispatch(
        self, request: Request, call_next: RequestResponseEndpoint
    ) -> Response:
        response = await call_next(request)
        if not response.headers.get("content-type", "").startswith("text/html"):
            return response
        body = b"".join([chunk async for chunk in response.body_iterator])
        headers = dict(response.headers)
        headers.pop("content-length", None)
        return Response(
            body.replace(b"<head>", b"<head>" + ENGLISH_LOCALE_SCRIPT, 1),
            status_code=response.status_code,
            headers=headers,
            media_type=response.media_type,
        )


def build_tab(
    input_component: gr.components.Component,
    button_text: str,
    handler: Callable[..., tuple[dict, str]],
    examples: list[str],
    show_face_hint: bool = False,
):
    with gr.Row():
        with gr.Column(scale=1):
            if show_face_hint:
                gr.HTML(FACE_MEDIA_HINT_HTML)
            media = input_component
            gr.Examples(examples=examples, inputs=media)
            button = gr.Button(button_text, variant="primary")
        with gr.Column(scale=1):
            verdict_view = gr.HTML(value=EMPTY_VERDICT_HTML)
            with gr.Accordion("Raw API response", open=False):
                result_json = gr.JSON()
    button.click(handler, inputs=media, outputs=[result_json, verdict_view])


with gr.Blocks(title="Deepfake Detection Demo") as Demo:
    gr.HTML(HTML_HEADER)
    with gr.Tabs():
        with gr.Tab("Image"):
            build_tab(
                gr.Image(label="Input Image", type="filepath", height=300),
                "Analyze Image!",
                analyze_image,
                IMAGE_EXAMPLES,
                show_face_hint=True,
            )
        with gr.Tab("Video"):
            build_tab(
                gr.Video(label="Input Video", height=300),
                "Analyze Video!",
                analyze_video,
                VIDEO_EXAMPLES,
                show_face_hint=True,
            )
        with gr.Tab("Audio"):
            build_tab(
                gr.Audio(label="Input Audio", type="filepath"),
                "Analyze Audio!",
                analyze_audio,
                AUDIO_EXAMPLES,
            )
    with gr.Accordion("How to read results", open=False):
        gr.HTML(HTML_EXPLANATION)


if __name__ == "__main__":
    Demo.launch(
        theme=gr.themes.Soft(),
        css=tabs_css,
        app_kwargs={"middleware": [Middleware(EnglishLocaleMiddleware)]},
    )