feat: enhance face-lock service with improved upload handling and response structure
python / test (push) Failing after 8s

- Updated README.md to reflect new features and API changes.
- Introduced versioning in app initialization.
- Enhanced configuration management in app/config.py with new validation functions.
- Refactored main.py to improve request handling and response generation.
- Added new models in app/models.py for structured API responses.
- Implemented a dedicated UI rendering function in app/ui.py.
- Improved Docker configuration for better security and health checks.
- Updated tests to cover new validation rules and response formats.
- Added CLAUDE.md for project guidelines and working rules.
This commit is contained in:
Space-Banane
2026-06-20 11:23:12 +02:00
parent f1072cb7b0
commit 3cf4a9a40a
17 changed files with 759 additions and 222 deletions
+3
View File
@@ -0,0 +1,3 @@
__all__ = ["__version__"]
__version__ = "1.0.0"
+62 -3
View File
@@ -5,6 +5,14 @@ from dotenv import load_dotenv
load_dotenv()
DEFAULT_ALLOWED_MIME_TYPES = (
"image/jpeg",
"image/png",
"image/webp",
"image/gif",
)
VALID_LOG_LEVELS = {"critical", "error", "warning", "info", "debug"}
def _env_bool(name: str, default: bool = False) -> bool:
value = os.getenv(name)
@@ -13,16 +21,67 @@ def _env_bool(name: str, default: bool = False) -> bool:
return value.strip().lower() in {"1", "true", "yes", "on"}
def _env_int(name: str, default: int, *, minimum: int | None = None, maximum: int | None = None) -> int:
raw_value = os.getenv(name)
value = default if raw_value is None else int(raw_value.strip())
if minimum is not None and value < minimum:
raise ValueError(f"{name} must be >= {minimum}")
if maximum is not None and value > maximum:
raise ValueError(f"{name} must be <= {maximum}")
return value
def _env_csv(name: str, default: tuple[str, ...]) -> tuple[str, ...]:
raw_value = os.getenv(name, "").strip()
if not raw_value:
return default
values = tuple(part.strip().lower() for part in raw_value.split(",") if part.strip())
if not values:
raise ValueError(f"{name} must contain at least one value when set")
return values
def _normalized_env() -> str:
value = os.getenv("FACE_LOCK_ENV") or os.getenv("ENV") or "production"
return value.strip().lower()
def _validated_log_level() -> str:
value = os.getenv("LOG_LEVEL", "info").strip().lower()
if value not in VALID_LOG_LEVELS:
raise ValueError(f"LOG_LEVEL must be one of: {', '.join(sorted(VALID_LOG_LEVELS))}")
return value
def _validated_header_name() -> str:
value = os.getenv("FACE_LOCK_AUTH_HEADER", "X-API-Key").strip()
if not value:
raise ValueError("FACE_LOCK_AUTH_HEADER cannot be empty")
if any(char in value for char in ("\r", "\n", ":")):
raise ValueError("FACE_LOCK_AUTH_HEADER contains invalid characters")
return value
@dataclass(frozen=True)
class Settings:
env: str = os.getenv("ENV", "prod").strip().lower()
test_ui_enabled: bool = _env_bool("FACE_LOCK_TEST_UI", True)
env: str = _normalized_env()
host: str = os.getenv("FACE_LOCK_HOST", "0.0.0.0").strip() or "0.0.0.0"
port: int = _env_int("PORT", 8000, minimum=1, maximum=65535)
log_level: str = _validated_log_level()
test_ui_enabled: bool = _env_bool("FACE_LOCK_TEST_UI", False)
docs_enabled: bool = _env_bool("FACE_LOCK_DOCS", True)
auth_token: str = os.getenv("FACE_LOCK_AUTH_TOKEN", "").strip()
auth_header_name: str = os.getenv("FACE_LOCK_AUTH_HEADER", "X-API-Key").strip()
auth_header_name: str = _validated_header_name()
max_upload_bytes: int = _env_int("FACE_LOCK_MAX_UPLOAD_BYTES", 8 * 1024 * 1024, minimum=1024, maximum=25 * 1024 * 1024)
allowed_mime_types: tuple[str, ...] = _env_csv("FACE_LOCK_ALLOWED_MIME_TYPES", DEFAULT_ALLOWED_MIME_TYPES)
@property
def auth_enabled(self) -> bool:
return bool(self.auth_token)
@property
def is_production(self) -> bool:
return self.env in {"prod", "production"}
settings = Settings()
+140 -158
View File
@@ -1,185 +1,167 @@
from io import BytesIO
from __future__ import annotations
from fastapi import Depends, FastAPI, File, Form, HTTPException, Request, UploadFile
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
app = FastAPI(title="face-lock", version="0.3.0")
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
header_name = settings.auth_header_name.lower()
provided = request.headers.get(header_name)
if not provided and request.headers.get("authorization", "").lower().startswith("bearer "):
provided = request.headers.get("authorization", "")[7:].strip()
if provided != settings.auth_token:
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")
@app.get("/health")
def health():
return {
"ok": True,
"env": settings.env,
"test_ui_enabled": settings.test_ui_enabled,
"auth_enabled": settings.auth_enabled,
"auth_header": settings.auth_header_name if settings.auth_enabled else None,
}
def _safe_filename(filename: str | None) -> str:
name = Path(filename or "upload").name.strip()
return name or "upload"
@app.get("/", response_class=HTMLResponse)
def index():
if not settings.test_ui_enabled:
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><p><a href='/docs' style='color:#67e8f9'>Open API docs</a></p></body></html>"
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)",
)
return HTMLResponse(
"""
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<script src="https://cdn.tailwindcss.com"></script>
<title>face-lock</title>
</head>
<body class="bg-slate-950 text-slate-100 min-h-screen">
<main class="mx-auto max-w-6xl p-6">
<div class="mb-6 flex items-center justify-between gap-4">
<div>
<h1 class="text-3xl font-bold">face-lock</h1>
<p class="text-slate-400">Square the subject, crop it, and keep the raw blobs out of sight.</p>
</div>
<a class="rounded-lg border border-slate-700 px-3 py-2 text-sm text-cyan-300 hover:bg-slate-900" href="/docs" target="_blank">Docs</a>
</div>
<div class="grid gap-6 md:grid-cols-2">
<section class="rounded-2xl border border-slate-800 bg-slate-900 p-4">
<label class="block text-sm text-slate-400">Image</label>
<input id="file" type="file" accept="image/*" class="mt-2 block w-full rounded-lg border border-slate-700 bg-slate-950 p-3" />
<div class="mt-4 grid gap-4 sm:grid-cols-2">
<div>
<label class="block text-sm text-slate-400">Detector</label>
<select id="detector" class="mt-2 block w-full rounded-lg border border-slate-700 bg-slate-950 p-3">
<option value="face">Face</option>
<option value="animal">Animal</option>
<option value="person">Person</option>
<option value="subject" selected>Subject</option>
</select>
</div>
<div>
<label class="block text-sm text-slate-400">Buffer ratio</label>
<input id="buffer_ratio" type="number" step="0.05" min="0" max="0.6" value="0.20" class="mt-2 block w-full rounded-lg border border-slate-700 bg-slate-950 p-3" />
</div>
</div>
<div class="mt-4">
<label class="block text-sm text-slate-400">Auth token (only if enabled)</label>
<input id="auth_token" type="password" placeholder="paste token here" class="mt-2 block w-full rounded-lg border border-slate-700 bg-slate-950 p-3" />
</div>
<button id="go" class="mt-4 rounded-lg bg-cyan-500 px-4 py-2 font-semibold text-slate-950">Process</button>
<pre id="meta" class="mt-4 whitespace-pre-wrap rounded-lg bg-slate-950 p-3 text-xs text-slate-300"></pre>
</section>
<section class="rounded-2xl border border-slate-800 bg-slate-900 p-4">
<div class="mb-3 text-sm font-semibold text-slate-400">Result</div>
<div class="grid gap-4">
<div>
<div class="mb-2 text-xs uppercase tracking-wide text-slate-500">Crop</div>
<img id="crop" class="hidden w-full rounded-xl border border-slate-800" />
</div>
<div>
<div class="mb-2 text-xs uppercase tracking-wide text-slate-500">Annotated source</div>
<img id="annotated" class="hidden w-full rounded-xl border border-slate-800" />
</div>
</div>
</section>
</div>
</main>
<script>
const file = document.getElementById('file');
const go = document.getElementById('go');
const crop = document.getElementById('crop');
const annotated = document.getElementById('annotated');
const meta = document.getElementById('meta');
go.onclick = async () => {
if (!file.files.length) return;
const form = new FormData();
form.append('file', file.files[0]);
form.append('detector', document.getElementById('detector').value);
form.append('buffer_ratio', document.getElementById('buffer_ratio').value);
meta.textContent = 'Working...';
const headers = {};
const token = document.getElementById('auth_token').value.trim();
if (token) headers['__AUTH_HEADER_NAME__'] = token;
const resp = await fetch('/api/focus', { method: 'POST', body: form, headers });
const data = await resp.json();
if (!resp.ok) {
meta.textContent = JSON.stringify(data, null, 2);
return;
}
meta.textContent = JSON.stringify({
filename: data.filename,
detector: data.detector,
method: data.method,
buffer_ratio: data.buffer_ratio,
detected_bbox: data.detected_bbox,
square_bbox: data.square_bbox,
source_size: data.source_size,
}, null, 2);
crop.src = data.crop_data_url;
annotated.src = data.annotated_data_url;
crop.classList.remove('hidden');
annotated.classList.remove('hidden');
};
</script>
</body>
</html>
""".replace("__AUTH_HEADER_NAME__", settings.auth_header_name)
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.post("/api/focus")
@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(
request: Request,
file: UploadFile = File(...),
buffer_ratio: float = Form(0.15),
detector: str = Form("subject"),
buffer_ratio: Annotated[float, Form(ge=0.0, le=0.6)] = 0.15,
detector: Annotated[Detector, Form()] = Detector.SUBJECT,
_auth: None = Depends(require_auth),
):
from app.vision import process_image
try:
payload = await file.read()
result = process_image(payload, file.filename or "upload", 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"],
"crop_data_url": result["crop_data_url"],
"annotated_data_url": result["annotated_data_url"],
}
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
) -> 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")
@app.post("/api/focus/image", tags=["focus"])
async def focus_image(
request: Request,
file: UploadFile = File(...),
buffer_ratio: float = Form(0.15),
detector: str = Form("subject"),
buffer_ratio: Annotated[float, Form(ge=0.0, le=0.6)] = 0.15,
detector: Annotated[Detector, Form()] = Detector.SUBJECT,
_auth: None = Depends(require_auth),
):
from app.vision import process_image
try:
payload = await file.read()
result = process_image(payload, file.filename or "upload", buffer_ratio=buffer_ratio, detector=detector)
return StreamingResponse(BytesIO(result["crop_bytes"]), media_type="image/jpeg")
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
) -> 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"'},
)
+49
View File
@@ -0,0 +1,49 @@
from __future__ import annotations
from enum import StrEnum
from pydantic import BaseModel, Field
class Detector(StrEnum):
FACE = "face"
ANIMAL = "animal"
PERSON = "person"
SUBJECT = "subject"
class BoundingBoxResponse(BaseModel):
x: int = Field(ge=0)
y: int = Field(ge=0)
w: int = Field(gt=0)
h: int = Field(gt=0)
class SourceSizeResponse(BaseModel):
width: int = Field(gt=0)
height: int = Field(gt=0)
class FocusResponse(BaseModel):
filename: str
detector: Detector
method: str
buffer_ratio: float = Field(ge=0.0, le=0.6)
detected_bbox: BoundingBoxResponse
square_bbox: BoundingBoxResponse
source_size: SourceSizeResponse
mime_type: str
crop_data_url: str
annotated_data_url: str
class HealthResponse(BaseModel):
ok: bool = True
env: str
version: str
docs_enabled: bool
test_ui_enabled: bool
auth_enabled: bool
auth_header: str | None
max_upload_bytes: int
allowed_mime_types: list[str]
+103
View File
@@ -0,0 +1,103 @@
from __future__ import annotations
def render_index_html(*, auth_header_name: str) -> str:
return """
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<script src="https://cdn.tailwindcss.com"></script>
<title>face-lock</title>
</head>
<body class="bg-slate-950 text-slate-100 min-h-screen">
<main class="mx-auto max-w-6xl p-6">
<div class="mb-6 flex items-center justify-between gap-4">
<div>
<h1 class="text-3xl font-bold">face-lock</h1>
<p class="text-slate-400">Square the subject, crop it, and keep the raw blobs out of sight.</p>
</div>
<a class="rounded-lg border border-slate-700 px-3 py-2 text-sm text-cyan-300 hover:bg-slate-900" href="/docs" target="_blank" rel="noreferrer">Docs</a>
</div>
<div class="grid gap-6 md:grid-cols-2">
<section class="rounded-2xl border border-slate-800 bg-slate-900 p-4">
<label class="block text-sm text-slate-400">Image</label>
<input id="file" type="file" accept="image/*" class="mt-2 block w-full rounded-lg border border-slate-700 bg-slate-950 p-3" />
<div class="mt-4 grid gap-4 sm:grid-cols-2">
<div>
<label class="block text-sm text-slate-400">Detector</label>
<select id="detector" class="mt-2 block w-full rounded-lg border border-slate-700 bg-slate-950 p-3">
<option value="face">Face</option>
<option value="animal">Animal</option>
<option value="person">Person</option>
<option value="subject" selected>Subject</option>
</select>
</div>
<div>
<label class="block text-sm text-slate-400">Buffer ratio</label>
<input id="buffer_ratio" type="number" step="0.05" min="0" max="0.6" value="0.20" class="mt-2 block w-full rounded-lg border border-slate-700 bg-slate-950 p-3" />
</div>
</div>
<div class="mt-4">
<label class="block text-sm text-slate-400">Auth token (only if enabled)</label>
<input id="auth_token" type="password" placeholder="paste token here" class="mt-2 block w-full rounded-lg border border-slate-700 bg-slate-950 p-3" />
</div>
<button id="go" class="mt-4 rounded-lg bg-cyan-500 px-4 py-2 font-semibold text-slate-950">Process</button>
<pre id="meta" class="mt-4 whitespace-pre-wrap rounded-lg bg-slate-950 p-3 text-xs text-slate-300"></pre>
</section>
<section class="rounded-2xl border border-slate-800 bg-slate-900 p-4">
<div class="mb-3 text-sm font-semibold text-slate-400">Result</div>
<div class="grid gap-4">
<div>
<div class="mb-2 text-xs uppercase tracking-wide text-slate-500">Crop</div>
<img id="crop" class="hidden w-full rounded-xl border border-slate-800" />
</div>
<div>
<div class="mb-2 text-xs uppercase tracking-wide text-slate-500">Annotated source</div>
<img id="annotated" class="hidden w-full rounded-xl border border-slate-800" />
</div>
</div>
</section>
</div>
</main>
<script>
const file = document.getElementById('file');
const go = document.getElementById('go');
const crop = document.getElementById('crop');
const annotated = document.getElementById('annotated');
const meta = document.getElementById('meta');
go.onclick = async () => {
if (!file.files.length) return;
const form = new FormData();
form.append('file', file.files[0]);
form.append('detector', document.getElementById('detector').value);
form.append('buffer_ratio', document.getElementById('buffer_ratio').value);
meta.textContent = 'Working...';
const headers = {};
const token = document.getElementById('auth_token').value.trim();
if (token) headers['__AUTH_HEADER_NAME__'] = token;
const resp = await fetch('/api/focus', { method: 'POST', body: form, headers });
const data = await resp.json();
if (!resp.ok) {
meta.textContent = JSON.stringify(data, null, 2);
return;
}
meta.textContent = JSON.stringify({
filename: data.filename,
detector: data.detector,
method: data.method,
buffer_ratio: data.buffer_ratio,
detected_bbox: data.detected_bbox,
square_bbox: data.square_bbox,
source_size: data.source_size,
}, null, 2);
crop.src = data.crop_data_url;
annotated.src = data.annotated_data_url;
crop.classList.remove('hidden');
annotated.classList.remove('hidden');
};
</script>
</body>
</html>
""".replace("__AUTH_HEADER_NAME__", auth_header_name)
+18 -2
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from typing import TypedDict
import cv2
import numpy as np
@@ -31,6 +31,20 @@ HOG = cv2.HOGDescriptor()
HOG.setSVMDetector(cv2.HOGDescriptor_getDefaultPeopleDetector())
class ProcessedImage(TypedDict):
filename: str
detector: str
method: str
buffer_ratio: float
detected_bbox: dict[str, int]
square_bbox: dict[str, int]
source_size: dict[str, int]
crop_data_url: str
annotated_data_url: str
mime_type: str
crop_bytes: bytes
def decode_image(image_bytes: bytes) -> np.ndarray:
data = np.frombuffer(image_bytes, dtype=np.uint8)
image = cv2.imdecode(data, cv2.IMREAD_COLOR)
@@ -68,6 +82,8 @@ def fallback_bbox(image: np.ndarray) -> BBox:
def detect_face(image: np.ndarray) -> BBox | None:
if FACE_CASCADE.empty():
return None
gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
gray = cv2.equalizeHist(gray)
faces = FACE_CASCADE.detectMultiScale(gray, scaleFactor=1.08, minNeighbors=5, minSize=(24, 24))
@@ -173,7 +189,7 @@ def _data_url(image_bytes: bytes, mime_type: str) -> str:
return f"data:{mime_type};base64,{base64.b64encode(image_bytes).decode('ascii')}"
def process_image(image_bytes: bytes, filename: str, buffer_ratio: float = 0.15, detector: str = "subject") -> dict[str, Any]:
def process_image(image_bytes: bytes, filename: str, buffer_ratio: float = 0.15, detector: str = "subject") -> ProcessedImage:
image = decode_image(image_bytes)
bbox, method = select_primary_bbox(image, detector=detector)
square = square_bbox(bbox, image.shape, buffer_ratio=buffer_ratio)