Spaces:
Sleeping
Sleeping
File size: 5,642 Bytes
bd2c5ca |
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 183 184 |
"""
KYC POC API - Main Application Entry Point
This is a FastAPI application for KYC (Know Your Customer) verification
using face matching (AuraFace) and liveness detection (Silent-Face-Anti-Spoofing).
Run with:
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
"""
import logging
from contextlib import asynccontextmanager
from concurrent.futures import ThreadPoolExecutor
import asyncio
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from .config import settings
from .api.routes import health, kyc, kyc_base64, ocr
from .services.face_recognition import face_recognition_service
from .services.liveness_detection import liveness_detection_service
from .services.ktp_ocr import ktp_ocr_service
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
# Thread pool for ML model initialization
executor = ThreadPoolExecutor(max_workers=3)
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
Application lifespan manager.
Initializes ML models on startup and cleans up on shutdown.
"""
logger.info("Starting KYC POC API...")
# Initialize ML models in background threads
loop = asyncio.get_event_loop()
try:
# Initialize face recognition service
logger.info("Initializing face recognition service...")
await loop.run_in_executor(executor, face_recognition_service.initialize)
logger.info("Face recognition service ready")
except Exception as e:
logger.error(f"Failed to initialize face recognition: {e}")
try:
# Initialize liveness detection service
logger.info("Initializing liveness detection service...")
await loop.run_in_executor(executor, liveness_detection_service.initialize)
logger.info("Liveness detection service ready")
except Exception as e:
logger.error(f"Failed to initialize liveness detection: {e}")
try:
# Initialize KTP OCR service
logger.info("Initializing KTP OCR service...")
await loop.run_in_executor(executor, ktp_ocr_service.initialize)
logger.info("KTP OCR service ready")
except Exception as e:
logger.error(f"Failed to initialize KTP OCR: {e}")
logger.info("KYC POC API started successfully")
yield
# Cleanup on shutdown
logger.info("Shutting down KYC POC API...")
executor.shutdown(wait=True)
logger.info("Shutdown complete")
# Create FastAPI application
app = FastAPI(
title=settings.APP_NAME,
version=settings.APP_VERSION,
description="""
## KYC POC API
A proof-of-concept API for KYC (Know Your Customer) verification using:
- **AuraFace** for face recognition and matching
- **Silent-Face-Anti-Spoofing** for liveness detection
- **EasyOCR** for KTP text extraction
### Features
- Face matching between KTP (ID card) and selfie
- Liveness detection to prevent spoofing
- Face quality analysis (blur, brightness, pose)
- Age and gender estimation
- **KTP OCR**: Extract and parse Indonesian ID card data (NIK, name, address, etc.)
- **NIK Validation**: Validate and decode NIK information
### Endpoints
- **File Upload**: `/api/v1/kyc/*` - Accepts multipart/form-data
- **Base64**: `/api/v1/kyc/base64/*` - Accepts JSON with base64 images
- **OCR**: `/api/v1/kyc/ocr/*` - KTP text extraction and NIK validation
""",
docs_url="/docs",
redoc_url="/redoc",
lifespan=lifespan
)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# ============================================================================
# Exception Handlers
# ============================================================================
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
"""Handle request validation errors."""
errors = exc.errors()
return JSONResponse(
status_code=422,
content={
"error_code": "VALIDATION_ERROR",
"message": "Request validation failed",
"detail": errors
}
)
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
"""Handle unexpected errors."""
logger.error(f"Unexpected error: {exc}", exc_info=True)
return JSONResponse(
status_code=500,
content={
"error_code": "INTERNAL_ERROR",
"message": "An unexpected error occurred",
"detail": str(exc) if settings.DEBUG else None
}
)
# ============================================================================
# Register Routes
# ============================================================================
# Health check routes (no prefix)
app.include_router(health.router)
# KYC routes (file upload)
app.include_router(kyc.router, prefix="/api/v1")
# KYC routes (base64)
app.include_router(kyc_base64.router, prefix="/api/v1")
# OCR routes
app.include_router(ocr.router, prefix="/api/v1")
# ============================================================================
# Main Entry Point
# ============================================================================
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"app.main:app",
host="0.0.0.0",
port=8000,
reload=settings.DEBUG
)
|