Spaces:
Running
Running
| """Sightengine AI image detector with multi-key failover support. | |
| Docs: https://sightengine.com/docs/detect-ai-generated-images | |
| Uses models: genai (AI-generated images) + deepfake (face swaps) | |
| Supports multiple API keys for automatic failover on rate limits (HTTP 429). | |
| Keys rotate round-robin to distribute load. | |
| Retry policy: up to MAX_PASSES full sweeps through all keys before giving up. | |
| Between passes a short back-off delay is applied so transient rate limits clear. | |
| Response shape: | |
| { | |
| "type": {"ai_generated": 0.95}, | |
| "media": {...} | |
| } | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| import logging | |
| import httpx | |
| from backend.core.config import settings | |
| from backend.core.schema import DetectionResult, Verdict | |
| from .base import BaseDetector | |
| log = logging.getLogger(__name__) | |
| _ENDPOINT = "https://api.sightengine.com/1.0/check.json" | |
| _MAX_PASSES = 3 # full sweeps through all keys | |
| _PASS_DELAY = 2.0 # seconds to wait between passes | |
| class SightengineDetector(BaseDetector): | |
| name = "sightengine" | |
| _current_key_index = 0 # Class-level state for round-robin | |
| _lock = asyncio.Lock() # Protect index from concurrent access | |
| async def _try_key(self, client: httpx.AsyncClient, image_path: str, user: str, secret: str): | |
| """Attempt one API call. Returns parsed data dict or raises.""" | |
| with open(image_path, "rb") as f: | |
| resp = await client.post( | |
| _ENDPOINT, | |
| data={"models": "genai,deepfake", "api_user": user, "api_secret": secret}, | |
| files={"media": f}, | |
| ) | |
| if resp.status_code == 429: | |
| raise Exception("rate_limited") | |
| if resp.status_code != 200: | |
| raise Exception(f"HTTP {resp.status_code}: {resp.text[:120]}") | |
| return resp.json() | |
| async def detect(self, image_path: str) -> DetectionResult: | |
| pairs = settings.sightengine_pairs | |
| if not pairs: | |
| return self._error_result("SIGHTENGINE credentials not set") | |
| last_error = "unknown" | |
| async with httpx.AsyncClient(timeout=settings.request_timeout) as client: | |
| for pass_num in range(_MAX_PASSES): | |
| if pass_num > 0: | |
| log.warning( | |
| "Sightengine pass %d/%d after %.1fs delay (last error: %s)", | |
| pass_num + 1, _MAX_PASSES, _PASS_DELAY, last_error, | |
| ) | |
| await asyncio.sleep(_PASS_DELAY) | |
| for _ in range(len(pairs)): | |
| async with self._lock: | |
| current_idx = self._current_key_index % len(pairs) | |
| self._current_key_index = (current_idx + 1) % len(pairs) | |
| user, secret = pairs[current_idx] | |
| try: | |
| data = await self._try_key(client, image_path, user, secret) | |
| except Exception as exc: | |
| last_error = str(exc) | |
| log.debug("Sightengine key[%d] failed: %s", current_idx, last_error) | |
| continue | |
| # Success — parse and return | |
| genai_score = float(data.get("type", {}).get("ai_generated", 0.5)) | |
| deepfake_score = ( | |
| float(data.get("faces", [{}])[0].get("deepfake", 0.0)) | |
| if data.get("faces") else 0.0 | |
| ) | |
| p_fake = max(genai_score, deepfake_score) | |
| verdict = self._verdict_from_p_fake(p_fake) | |
| evidence = [f"Sightengine genai score: {genai_score:.1%}"] | |
| if deepfake_score > 0.1: | |
| evidence.append(f"Sightengine deepfake score: {deepfake_score:.1%}") | |
| generator = None | |
| if genai_score > 0.5: | |
| generator = data.get("type", {}).get("generator") | |
| if pass_num > 0: | |
| log.info("Sightengine succeeded on pass %d", pass_num + 1) | |
| return DetectionResult( | |
| detector=self.name, | |
| p_fake=p_fake, | |
| verdict=verdict, | |
| confidence=abs(p_fake - 0.5) * 2, | |
| evidence=evidence, | |
| raw=data, | |
| generator=generator, | |
| ) | |
| return self._error_result( | |
| f"Sightengine: all {len(pairs)} keys × {_MAX_PASSES} passes failed — last: {last_error}" | |
| ) | |