face-lock/app/main.py

from __future__ import annotations

from io import BytesIO
import logging
from pathlib import Path
import secrets
from typing import Annotated

from fastapi import Depends, FastAPI, File, Form, HTTPException, Request, Response, UploadFile
from fastapi.responses import HTMLResponse, StreamingResponse

from app import __version__
from app.config import settings
from app.models import Detector, FocusResponse, HealthResponse
from app.ui import render_index_html

logging.basicConfig(
    level=getattr(logging, settings.log_level.upper(), logging.INFO),
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
logger = logging.getLogger("face_lock")

app = FastAPI(
    title="face-lock",
    version=__version__,
    docs_url="/docs" if settings.docs_enabled else None,
    redoc_url=None,
    openapi_url="/openapi.json" if settings.docs_enabled else None,
)


def require_auth(request: Request) -> None:
    if not settings.auth_enabled:
        return

    provided = (request.headers.get(settings.auth_header_name) or "").strip()
    if settings.auth_header_name.lower() == "authorization" and provided.lower().startswith("bearer "):
        provided = provided[7:].strip()

    if not provided:
        authorization = request.headers.get("authorization", "").strip()
        if authorization.lower().startswith("bearer "):
            provided = authorization[7:].strip()

    if not provided or not secrets.compare_digest(provided, settings.auth_token):
        raise HTTPException(status_code=401, detail="unauthorized")


def _safe_filename(filename: str | None) -> str:
    name = Path(filename or "upload").name.strip()
    return name or "upload"


def _validate_upload(file: UploadFile, payload: bytes) -> None:
    if not payload:
        raise HTTPException(status_code=400, detail="empty upload")

    if len(payload) > settings.max_upload_bytes:
        raise HTTPException(
            status_code=413,
            detail=f"upload exceeds FACE_LOCK_MAX_UPLOAD_BYTES ({settings.max_upload_bytes} bytes)",
        )

    content_type = (file.content_type or "").strip().lower()
    if content_type not in settings.allowed_mime_types:
        raise HTTPException(
            status_code=415,
            detail=f"unsupported content type: {content_type or 'unknown'}",
        )


async def _process_upload(file: UploadFile, *, buffer_ratio: float, detector: Detector) -> dict[str, object]:
    from app.vision import process_image

    payload = await file.read()
    _validate_upload(file, payload)

    try:
        return process_image(
            payload,
            _safe_filename(file.filename),
            buffer_ratio=buffer_ratio,
            detector=detector.value,
        )
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc


@app.middleware("http")
async def add_response_headers(request: Request, call_next) -> Response:
    response = await call_next(request)
    response.headers.setdefault("X-Content-Type-Options", "nosniff")
    response.headers.setdefault("X-Frame-Options", "DENY")
    response.headers.setdefault("Referrer-Policy", "no-referrer")
    response.headers.setdefault("Permissions-Policy", "camera=(), geolocation=(), microphone=()")
    return response


@app.get("/health", response_model=HealthResponse, tags=["meta"])
def health() -> HealthResponse:
    return HealthResponse(
        env=settings.env,
        version=__version__,
        docs_enabled=settings.docs_enabled,
        test_ui_enabled=settings.test_ui_enabled,
        auth_enabled=settings.auth_enabled,
        auth_header=settings.auth_header_name if settings.auth_enabled else None,
        max_upload_bytes=settings.max_upload_bytes,
        allowed_mime_types=list(settings.allowed_mime_types),
    )


@app.get("/", response_class=HTMLResponse, tags=["meta"])
def index() -> HTMLResponse:
    if not settings.test_ui_enabled:
        docs_link = "<p><a href='/docs' style='color:#67e8f9'>Open API docs</a></p>" if settings.docs_enabled else ""
        return HTMLResponse(
            (
                "<!doctype html><html><body style='font-family:sans-serif;background:#0f172a;color:#e2e8f0;padding:2rem'>"
                "<h1>face-lock</h1><p>Test UI is disabled.</p>"
                f"{docs_link}</body></html>"
            ),
            headers={"Cache-Control": "no-store"},
        )

    return HTMLResponse(
        render_index_html(auth_header_name=settings.auth_header_name),
        headers={"Cache-Control": "no-store"},
    )


@app.post("/api/focus", response_model=FocusResponse, tags=["focus"])
async def focus(
    file: UploadFile = File(...),
    buffer_ratio: Annotated[float, Form(ge=0.0, le=0.6)] = 0.15,
    detector: Annotated[Detector, Form()] = Detector.SUBJECT,
    _auth: None = Depends(require_auth),
) -> dict[str, object]:
    result = await _process_upload(file, buffer_ratio=buffer_ratio, detector=detector)
    return {
        "filename": result["filename"],
        "detector": result["detector"],
        "method": result["method"],
        "buffer_ratio": result["buffer_ratio"],
        "detected_bbox": result["detected_bbox"],
        "square_bbox": result["square_bbox"],
        "source_size": result["source_size"],
        "mime_type": result["mime_type"],
        "crop_data_url": result["crop_data_url"],
        "annotated_data_url": result["annotated_data_url"],
    }


@app.post("/api/focus/image", tags=["focus"])
async def focus_image(
    file: UploadFile = File(...),
    buffer_ratio: Annotated[float, Form(ge=0.0, le=0.6)] = 0.15,
    detector: Annotated[Detector, Form()] = Detector.SUBJECT,
    _auth: None = Depends(require_auth),
) -> StreamingResponse:
    result = await _process_upload(file, buffer_ratio=buffer_ratio, detector=detector)
    filename_stem = Path(str(result["filename"])).stem or "focus"
    return StreamingResponse(
        BytesIO(result["crop_bytes"]),
        media_type=str(result["mime_type"]),
        headers={"Content-Disposition": f'inline; filename="{filename_stem}-crop.jpg"'},
    )