Spaces:
Running
Running
File size: 6,153 Bytes
2e175db | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 | """
v1 API.
Endpoints
---------
GET /health → liveness probe
GET /v1/info → model + config metadata
POST /v1/scan/image → scan an uploaded image
Privacy
-------
Image bytes live in-memory only for the duration of the request. They are
never written to disk or to blob storage in Stage 1. After the response is
returned the bytes go out of scope and are garbage-collected.
"""
from __future__ import annotations
import logging
import time
import uuid
from fastapi import FastAPI, File, HTTPException, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from ..config import settings
from ..detectors.ensemble import run_ensemble
from ..preprocess import InvalidImageError, decode
from ..provenance import check_c2pa
from ..storage import record_scan
from ..storage.db import ScanRecord
from .schemas import ErrorResponse, Probabilities, Provenance, ScanResponse
log = logging.getLogger(__name__)
_ALLOWED_TYPES = {"image/jpeg", "image/png", "image/webp"}
def create_app() -> FastAPI:
app = FastAPI(
title=settings.api_title,
version=settings.api_version,
description=(
"Detect AI-generated and manipulated images. "
"Visitor uploads are processed in-memory and never stored."
),
)
# Browser preflight allowlist for the static-HTML frontend. Credentialed
# requests are off (no cookies); the upload is the only state we accept.
app.add_middleware(
CORSMiddleware,
allow_origin_regex=settings.cors_allow_origin_regex,
allow_methods=["GET", "POST", "OPTIONS"],
allow_headers=["*"],
allow_credentials=False,
)
# HEAD is accepted on read-only endpoints so uptime monitors and CDN
# validators can ping cheaply without a body. Per RFC 7231, any URL
# that accepts GET should also accept HEAD.
@app.api_route("/health", methods=["GET", "HEAD"])
def health() -> dict:
return {"status": "ok"}
@app.api_route("/v1/info", methods=["GET", "HEAD"])
def info() -> dict:
return {
"api_version": settings.api_version,
"model_version": settings.model_version,
"detectors": {
"clip_classifier": settings.enable_clip_detector,
"frequency_artifacts": settings.enable_frequency_detector,
"face_swap": settings.enable_face_swap_detector,
},
"c2pa_check_enabled": settings.enable_c2pa_check,
"max_upload_bytes": settings.max_upload_bytes,
"store_uploads": settings.store_uploads,
}
@app.post(
"/v1/scan/image",
response_model=ScanResponse,
responses={
413: {"model": ErrorResponse},
415: {"model": ErrorResponse},
422: {"model": ErrorResponse},
},
)
async def scan_image(file: UploadFile = File(...)) -> ScanResponse:
if file.content_type not in _ALLOWED_TYPES:
raise HTTPException(
status_code=415,
detail=(
f"Unsupported media type '{file.content_type}'. "
"Use JPEG, PNG, or WebP."
),
)
image_bytes = await file.read()
if len(image_bytes) > settings.max_upload_bytes:
raise HTTPException(
status_code=413,
detail=(
f"File exceeds the {settings.max_upload_bytes // (1024*1024)} MB limit."
),
)
t0 = time.perf_counter()
# 1. Provenance check (advisory — does not yet override the model).
provenance_result = check_c2pa(image_bytes)
# 2. Decode image.
try:
image = decode(image_bytes)
except InvalidImageError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
# 3. Run detector ensemble.
ensemble_out = run_ensemble(image)
latency_ms = (time.perf_counter() - t0) * 1000.0
scan_id = f"scn_{uuid.uuid4().hex[:24]}"
response = ScanResponse(
verdict=ensemble_out.verdict,
confidence=round(ensemble_out.confidence, 4),
probabilities=Probabilities(
authentic=round(ensemble_out.probabilities.authentic, 4),
ai_generated=round(ensemble_out.probabilities.ai_generated, 4),
deepfake=round(ensemble_out.probabilities.deepfake, 4),
edited=round(ensemble_out.probabilities.edited, 4),
),
signals=ensemble_out.signals,
provenance=Provenance(
c2pa_present=provenance_result.present,
c2pa_valid=provenance_result.valid,
issuer=provenance_result.issuer,
claim_generator=provenance_result.claim_generator,
),
model_version=settings.model_version,
scan_id=scan_id,
latency_ms=round(latency_ms, 1),
)
# 4. Persist METADATA only — never the image.
record_scan(
ScanRecord(
scan_id=scan_id,
verdict=response.verdict,
confidence=response.confidence,
model_version=response.model_version,
latency_ms=response.latency_ms,
c2pa_present=provenance_result.present,
)
)
# 5. Drop the image bytes ASAP. (Local var goes out of scope when the
# function returns; the explicit del documents the privacy intent.)
del image_bytes
return response
@app.exception_handler(HTTPException)
async def http_exception_handler(_, exc: HTTPException) -> JSONResponse:
return JSONResponse(
status_code=exc.status_code,
content=ErrorResponse(
error=f"HTTP {exc.status_code}",
detail=str(exc.detail),
).model_dump(),
)
return app
# Default app instance for `uvicorn deepfake_scanner.api.v1:app`.
app = create_app()
|