"""
Signature Verification — Stateless REST API (FastAPI)
=====================================================
Stateless compute service for the TÜBİTAK 2209-A signature verification web app.
There is NO database and NO enrollment storage here. Reference embeddings are
passed in per request. All state (enrolled persons, their embeddings) lives in
Supabase and is orchestrated by the Next.js server, which is the ONLY caller of
this API (server-to-server, authenticated with X-API-Key).
Endpoints (all require header X-API-Key except /health):
POST /embed — compute reference embeddings from N signature images
POST /verify — verify a signature image against reference embeddings
POST /verify-pdf — extract signature from a PDF, verify against embeddings
GET /health — liveness (no auth)
Model + preprocessing + decision logic are UNCHANGED (inference.py / pdf_extractor.py).
Run:
uvicorn api:app --host 0.0.0.0 --port 7860
"""
import io
import os
import json
from pathlib import Path
from typing import Optional
import numpy as np
from fastapi import (
FastAPI, File, UploadFile, Form, HTTPException, Request, Header, Depends,
)
from fastapi.responses import JSONResponse, HTMLResponse
from fastapi.middleware.cors import CORSMiddleware
from PIL import Image
from inference import SignatureVerifier
from pdf_extractor import extract_signature_from_pdf, image_to_base64
# ── Configuration (env-driven) ───────────────────────────────────────────
MODEL_PATH = os.environ.get(
"MODEL_PATH", str(Path(__file__).parent / "best_model.pth")
)
API_SHARED_SECRET = os.environ.get("API_SHARED_SECRET", "")
ALLOWED_ORIGIN = os.environ.get("ALLOWED_ORIGIN", "*")
# Tier 1: raised from model EER (0.8147) to 0.88 to reduce false positives.
DEFAULT_THRESHOLD = float(os.environ.get("DEFAULT_THRESHOLD", "0.88"))
# ── App Setup ────────────────────────────────────────────────────────────
app = FastAPI(
title="Signature Verification API (stateless)",
description="TÜBİTAK 2209-A — İmza Doğrulama (stateless compute)",
version="2.0.0",
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"] if ALLOWED_ORIGIN == "*" else [ALLOWED_ORIGIN],
allow_credentials=False,
allow_methods=["*"],
allow_headers=["*"],
)
@app.exception_handler(Exception)
async def unhandled_exception_handler(request: Request, exc: Exception):
return JSONResponse(
status_code=500,
content={"detail": f"Internal Server Error: {str(exc)}"},
)
# ── Auth dependency ──────────────────────────────────────────────────────
def require_key(x_api_key: Optional[str] = Header(None)):
"""Reject requests without the shared secret. If no secret is configured
(local dev), auth is skipped. In production the secret is always set."""
if not API_SHARED_SECRET:
return
if x_api_key != API_SHARED_SECRET:
raise HTTPException(status_code=401, detail="unauthorized")
# ── Model bootstrap ──────────────────────────────────────────────────────
verifier: Optional[SignatureVerifier] = None
def _ensure_model_path() -> str:
"""Return a local path to the model weights, downloading from a private
Hugging Face model repo if the file is not present (used on HF Spaces where
the 148MB weight is not committed to the Space repo)."""
if os.path.exists(MODEL_PATH):
return MODEL_PATH
repo = os.environ.get("HF_MODEL_REPO")
if repo:
from huggingface_hub import hf_hub_download
return hf_hub_download(
repo_id=repo,
filename=os.environ.get("HF_MODEL_FILE", "best_model.pth"),
token=os.environ.get("HF_TOKEN"),
)
raise RuntimeError(
f"Model file not found at {MODEL_PATH} and HF_MODEL_REPO not set."
)
@app.on_event("startup")
async def startup():
global verifier
path = _ensure_model_path()
verifier = SignatureVerifier(path)
print(f"Stateless API ready. Default threshold={DEFAULT_THRESHOLD} "
f"(model EER: {verifier.threshold:.6f})")
# ── Helpers ──────────────────────────────────────────────────────────────
async def read_image(file: UploadFile) -> np.ndarray:
"""Read an uploaded image file into a numpy RGB array."""
contents = await file.read()
image = Image.open(io.BytesIO(contents)).convert("RGB")
return np.array(image)
def _parse_reference_embeddings(reference_embeddings: str) -> list[np.ndarray]:
"""Parse the JSON list-of-lists sent by the Next.js server into a list of
256-dim float32 numpy vectors."""
try:
raw = json.loads(reference_embeddings)
except Exception:
raise HTTPException(
status_code=400, detail="reference_embeddings geçerli JSON değil."
)
if not isinstance(raw, list) or len(raw) == 0:
raise HTTPException(
status_code=400, detail="reference_embeddings boş veya hatalı."
)
return [np.asarray(v, dtype=np.float32) for v in raw]
def _decide(query_embedding: np.ndarray, refs: list[np.ndarray], threshold: float) -> dict:
"""Run the (unchanged) verification decision. Multi-reference when >=2 refs."""
if len(refs) >= 2:
return verifier.verify_multi_reference(query_embedding, refs, threshold=threshold)
return verifier.verify(query_embedding, refs[0], threshold=threshold)
# ── Endpoints ────────────────────────────────────────────────────────────
@app.post("/embed", dependencies=[Depends(require_key)])
async def embed(
signatures: list[UploadFile] = File(
..., description="1 veya daha fazla imza görseli"
),
):
"""Compute reference embeddings for a person from 1+ signature images.
Returns the centroid + per-image embeddings (to be stored in Supabase)."""
if len(signatures) == 0:
raise HTTPException(status_code=400, detail="En az 1 imza görseli yükleyin.")
images = [await read_image(s) for s in signatures]
reference_embedding = verifier.compute_reference_embedding(images)
individual_embeddings = verifier.compute_reference_embeddings_list(images)
# Quality check: warn if any reference is far from the centroid.
warnings_list = []
for i in range(len(individual_embeddings)):
sim_to_centroid = verifier.cosine_similarity(
individual_embeddings[i], reference_embedding
)
if sim_to_centroid < 0.70:
warnings_list.append(
f"İmza #{i+1} referans setinden düşük benzerlik gösteriyor "
f"({sim_to_centroid:.3f}). Farklı kalitede veya yanlış imza olabilir."
)
return {
"reference_embedding": reference_embedding.tolist(),
"individual_embeddings": [e.tolist() for e in individual_embeddings],
"num_signatures": len(signatures),
"warnings": warnings_list,
}
@app.post("/verify", dependencies=[Depends(require_key)])
async def verify(
signature: UploadFile = File(..., description="Doğrulanacak imza görseli"),
reference_embeddings: str = Form(..., description="JSON [[256], ...]"),
threshold: Optional[float] = Form(None),
):
"""Verify a signature image against a person's reference embeddings."""
refs = _parse_reference_embeddings(reference_embeddings)
img = await read_image(signature)
query_embedding = verifier.extract_embedding(img)
thr = float(threshold) if threshold is not None else DEFAULT_THRESHOLD
result = _decide(query_embedding, refs, thr)
# Previews: original + what the model sees after Otsu.
original_b64 = image_to_base64(img, max_dim=600)
import cv2
gray = cv2.cvtColor(img, cv2.COLOR_RGB2GRAY) if len(img.shape) == 3 else img
_, otsu = cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
otsu_rgb = cv2.cvtColor(otsu, cv2.COLOR_GRAY2RGB)
otsu_b64 = image_to_base64(otsu_rgb, max_dim=600)
return {
"mode": "image",
"verified": result["verified"],
"confidence": result["confidence"],
"similarity": result["similarity"],
"threshold": result["threshold"],
"verdict": "GERCEK IMZA" if result["verified"] else "SAHTE IMZA",
"min_similarity": result.get("min_similarity"),
"max_similarity": result.get("max_similarity"),
"std_similarity": result.get("std_similarity"),
"consistency": result.get("consistency"),
"uploaded_signature_base64": original_b64,
"otsu_processed_base64": otsu_b64,
}
@app.post("/verify-pdf", dependencies=[Depends(require_key)])
async def verify_pdf(
pdf_file: UploadFile = File(..., description="İmza içeren PDF belgesi"),
reference_embeddings: str = Form(..., description="JSON [[256], ...]"),
threshold: Optional[float] = Form(None),
):
"""Extract a signature from a PDF and verify it against reference embeddings."""
refs = _parse_reference_embeddings(reference_embeddings)
pdf_bytes = await pdf_file.read()
if len(pdf_bytes) < 100:
raise HTTPException(status_code=400, detail="Geçersiz veya boş PDF dosyası.")
extraction = extract_signature_from_pdf(pdf_bytes)
if not extraction["extraction_success"] or extraction["signature_image"] is None:
return JSONResponse(status_code=422, content={
"detail": "PDF'den imza çıkarılamadı.",
"extraction": {
"extraction_success": False,
"scenario_detected": extraction["scenario_detected"],
"steps": extraction["steps"],
"blue_ink_pixel_count": extraction["blue_ink_pixel_count"],
},
})
sig_img = extraction["signature_image"]
query_embedding = verifier.extract_embedding(sig_img)
thr = float(threshold) if threshold is not None else DEFAULT_THRESHOLD
result = _decide(query_embedding, refs, thr)
sig_clean = extraction.get("signature_clean", sig_img)
extracted_b64 = image_to_base64(sig_clean, max_dim=600)
page_b64 = image_to_base64(extraction["page_image_rgb"], max_dim=400)
return {
"mode": "pdf",
"verified": result["verified"],
"confidence": result["confidence"],
"similarity": result["similarity"],
"threshold": result["threshold"],
"verdict": "GERCEK IMZA" if result["verified"] else "SAHTE IMZA",
"min_similarity": result.get("min_similarity"),
"max_similarity": result.get("max_similarity"),
"std_similarity": result.get("std_similarity"),
"consistency": result.get("consistency"),
"extraction": {
"extraction_success": True,
"scenario_detected": extraction["scenario_detected"],
"steps": extraction["steps"],
"signature_bbox": extraction["signature_bbox"],
"blue_ink_pixel_count": extraction["blue_ink_pixel_count"],
},
"extracted_signature_base64": extracted_b64,
"page_preview_base64": page_b64,
}
LANDING_HTML = """
Signature Verification API
TÜBİTAK 2209-A · stateless
✍️ Signature Verification API
Offline handwritten-signature verification. ● live — a stateless compute service; reference embeddings travel with each request, no data is stored here.
POST/embedcompute reference embeddings from signature imagesX-API-Key
POST/verifyverify a signature image against reference embeddingsX-API-Key
POST/verify-pdfextract a signature from a PDF and verify itX-API-Key