Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion app/core/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ class Settings(BaseSettings):
DEBUG: bool = False

# For OpenFake
OPENFAKE_API_URL: str = "https://complexdatalab-openfakedemo.hf.space/api/predict"
OPENFAKE_API_URL: str = "https://deepfake-detector.ai4.institute/api/predict"
MEDIA_VERIFICATION_ENABLED: bool = True

def __init__(self, **kwargs):
Expand Down
65 changes: 50 additions & 15 deletions app/services/openfake_service.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
import os
from typing import List, Optional

import httpx
from fastapi import HTTPException, UploadFile

OPENFAKE_API_URL = os.getenv(
"OPENFAKE_API_URL",
"https://complexdatalab-openfakedemo.hf.space/api/predict",
"https://deepfake-detector.ai4.institute/api/predict",
)

ALLOWED_TYPES = {
Expand All @@ -20,6 +21,33 @@
}


def _as_float(value: object, default: float = 0.0) -> float:
try:
return float(value)
except (TypeError, ValueError):
return default


def _verdict_from_score(p_fake: float) -> str:
if p_fake >= 0.75:
return "Likely fake"
if p_fake >= 0.45:
return "Uncertain"
return "Likely real"


def _frame_probs_from_frames(frames: object) -> Optional[List[float]]:
if not isinstance(frames, list):
return None

frame_probs = []
for frame in frames:
if isinstance(frame, dict) and "p_fake" in frame:
frame_probs.append(_as_float(frame.get("p_fake")))

return frame_probs or None


async def verify_media_with_openfake(file: UploadFile) -> dict:
if file.content_type not in ALLOWED_TYPES:
raise HTTPException(
Expand All @@ -29,7 +57,6 @@ async def verify_media_with_openfake(file: UploadFile) -> dict:

await file.seek(0)

# This forwards the uploaded file object to OpenFake
files = {
"file": (
file.filename,
Expand All @@ -45,30 +72,38 @@ async def verify_media_with_openfake(file: UploadFile) -> dict:
)

if response.status_code != 200:
detail = "OpenFake detector failed"
try:
error_data = response.json()
detail = error_data.get("detail") or error_data.get("error") or detail
except ValueError:
pass

raise HTTPException(
status_code=502,
detail="OpenFake detector failed",
detail=detail,
)

data = response.json()

p_fake = float(data.get("p_fake", 0))
reliability = float(data.get("reliability", 1 - p_fake))

if p_fake >= 0.75:
verdict = "Likely fake"
elif p_fake >= 0.45:
verdict = "Uncertain"
else:
verdict = "Likely real"
p_fake = _as_float(data.get("p_fake"))
reliability = _as_float(data.get("reliability"), 1 - p_fake)
frame_probs = data.get("frame_probs") or _frame_probs_from_frames(data.get("frames"))

return {
"media_type": data.get("media_type"),
"p_fake": p_fake,
"p_real": data.get("p_real"),
"p_localized": data.get("p_localized"),
"p_full_synthetic": data.get("p_full_synthetic"),
"p_fake_max": data.get("p_fake_max"),
"generators": data.get("generators", []),
"reliability": reliability,
"reliability_score": round(reliability * 100),
"verdict": verdict,
"verdict": _verdict_from_score(p_fake),
"n_frames": data.get("n_frames"),
"frame_probs": data.get("frame_probs"),
"explanation": (f"The detector estimates a {round(p_fake * 100)}% probability " f"that this media is fake."),
"frame_probs": frame_probs,
"frames": data.get("frames"),
"mask": data.get("mask"),
"explanation": (f"The detector estimates a {round(p_fake * 100)}% probability that this media is fake."),
}
Loading