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 = """
"""
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"""
"""
HTML_EXPLANATION = """
- Score: how likely the media is a deepfake, from 0 to 1.
The lower it sits on the scale, the more authentic the media looks.
- Genuine: score below 0.5.
- Deepfake: score 0.5 or above, the production threshold
marked on the scale.
"""
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 = (
''
f'{VERDICT_CAPTION}'
'Run an analysis to see the result.'
"
'
f'{VERDICT_CAPTION}'
f'{html.escape(text)}'
"
"
)
score = score_of(result)
if score is None:
return f'{headline}
'
return (
''
f'
{headline}'
f'Score {score:.2f}'
"
"
'
'
'
'
''
f''
"
"
'
'
f"{AUTHENTIC_END_LABEL}"
f'{THRESHOLD_LABEL}'
f"{DEEPFAKE_END_LABEL}"
"
"
"
"
"