src/deepfake_scanner/api/v1.py

"""
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()