Initial deployment to Hugging Face Spaces

.gitignore

@@ -0,0 +1,39 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# Virtual environment
+venv/
+.venv/
+ENV/
+
+# Environment files
+.env
+.env.local
+
+# IDE
+.idea/
+.vscode/
+*.swp
+*.swo
+
+# OS
+.DS_Store
+Thumbs.db
+
+# Models (downloaded during build)
+models/
+Silent-Face-Anti-Spoofing/
+
+# Logs
+*.log
+
+# Test files
+test_images/
+*.jpg
+*.png
+*.jpeg
+
+# HuggingFace cache
+.cache/

Dockerfile

@@ -0,0 +1,49 @@
+# Hugging Face Spaces Dockerfile for KYC POC Backend
+FROM python:3.10-slim
+
+# Set environment variables
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+ENV DEBIAN_FRONTEND=noninteractive
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    git \
+    libgl1-mesa-glx \
+    libglib2.0-0 \
+    libsm6 \
+    libxext6 \
+    libxrender-dev \
+    libgomp1 \
+    && rm -rf /var/lib/apt/lists/*
+
+# Create app user for HF Spaces (required)
+RUN useradd -m -u 1000 user
+WORKDIR /home/user/app
+
+# Copy requirements first for better caching
+COPY --chown=user:user requirements.txt .
+
+# Install Python dependencies
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir -r requirements.txt
+
+# Copy application code
+COPY --chown=user:user . .
+
+# Download models during build
+RUN python setup_models.py
+
+# Switch to non-root user (required for HF Spaces)
+USER user
+
+# Expose port (HF Spaces uses 7860 by default)
+EXPOSE 7860
+
+# Set environment variables for production
+ENV DEBUG=False
+ENV USE_GPU=False
+ENV DEVICE_ID=-1
+
+# Run the application
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -0,0 +1,321 @@
+---
+title: KYC POC Backend
+emoji: 🔐
+colorFrom: blue
+colorTo: purple
+sdk: docker
+app_port: 7860
+pinned: false
+license: mit
+---
+
+# 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
+
+## Features
+
+- Face matching between KTP (ID card) and selfie
+- Liveness detection to prevent spoofing attacks
+- Face quality analysis (blur, brightness, pose)
+- Age and gender estimation
+- Automatic face extraction from KTP images
+- Multiple faces rejection
+
+## Requirements
+
+- Python 3.9+
+- Git (for cloning Silent-Face-Anti-Spoofing)
+
+## Installation
+
+### 1. Create Virtual Environment
+
+```bash
+# Windows
+python -m venv venv
+venv\Scripts\activate
+
+# Linux/Mac
+python -m venv venv
+source venv/bin/activate
+```
+
+### 2. Install Dependencies
+
+```bash
+pip install -r requirements.txt
+```
+
+### 3. Download ML Models
+
+Run the setup script to download the required models:
+
+```bash
+python setup_models.py
+```
+
+This will:
+- Download AuraFace model from HuggingFace
+- Clone Silent-Face-Anti-Spoofing repository
+- Copy model files to the correct locations
+
+### 4. Run the Application
+
+```bash
+uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
+```
+
+The API will be available at: http://localhost:8000
+
+## API Documentation
+
+- **Swagger UI**: http://localhost:8000/docs
+- **ReDoc**: http://localhost:8000/redoc
+
+## API Endpoints
+
+### Health Check
+
+```
+GET /health
+```
+
+### File Upload Endpoints
+
+These endpoints accept `multipart/form-data`:
+
+| Endpoint | Method | Description |
+|----------|--------|-------------|
+| `/api/v1/kyc/verify` | POST | Full KYC verification |
+| `/api/v1/kyc/face-match` | POST | Face matching only |
+| `/api/v1/kyc/liveness` | POST | Liveness detection only |
+| `/api/v1/kyc/quality` | POST | Face quality check only |
+
+### Base64 Endpoints
+
+These endpoints accept `application/json` with base64 encoded images:
+
+| Endpoint | Method | Description |
+|----------|--------|-------------|
+| `/api/v1/kyc/base64/verify` | POST | Full KYC verification |
+| `/api/v1/kyc/base64/face-match` | POST | Face matching only |
+| `/api/v1/kyc/base64/liveness` | POST | Liveness detection only |
+| `/api/v1/kyc/base64/quality` | POST | Face quality check only |
+
+## Usage Examples
+
+### Using curl (File Upload)
+
+**Full KYC Verification:**
+```bash
+curl -X POST "http://localhost:8000/api/v1/kyc/verify" \
+  -F "ktp_image=@/path/to/ktp.jpg" \
+  -F "selfie_image=@/path/to/selfie.jpg" \
+  -F "threshold=0.5"
+```
+
+**Face Match Only:**
+```bash
+curl -X POST "http://localhost:8000/api/v1/kyc/face-match" \
+  -F "ktp_image=@/path/to/ktp.jpg" \
+  -F "selfie_image=@/path/to/selfie.jpg"
+```
+
+**Liveness Check:**
+```bash
+curl -X POST "http://localhost:8000/api/v1/kyc/liveness" \
+  -F "image=@/path/to/selfie.jpg"
+```
+
+### Using Insomnia/Postman (Base64)
+
+**Full KYC Verification:**
+
+```http
+POST /api/v1/kyc/base64/verify
+Content-Type: application/json
+
+{
+  "ktp_image": "base64_encoded_ktp_image_here...",
+  "selfie_image": "base64_encoded_selfie_image_here...",
+  "threshold": 0.5
+}
+```
+
+**Face Match Only:**
+
+```http
+POST /api/v1/kyc/base64/face-match
+Content-Type: application/json
+
+{
+  "image1": "base64_encoded_image1_here...",
+  "image2": "base64_encoded_image2_here...",
+  "threshold": 0.5
+}
+```
+
+**Liveness Check:**
+
+```http
+POST /api/v1/kyc/base64/liveness
+Content-Type: application/json
+
+{
+  "image": "base64_encoded_image_here..."
+}
+```
+
+**Quality Check:**
+
+```http
+POST /api/v1/kyc/base64/quality
+Content-Type: application/json
+
+{
+  "image": "base64_encoded_image_here..."
+}
+```
+
+## Response Examples
+
+### Successful Verification
+
+```json
+{
+  "success": true,
+  "face_match": {
+    "is_match": true,
+    "similarity_score": 0.87,
+    "threshold": 0.5
+  },
+  "liveness": {
+    "is_real": true,
+    "confidence": 0.95,
+    "label": "Real Face",
+    "prediction_class": 1,
+    "models_used": 2
+  },
+  "quality": {
+    "ktp": {
+      "blur_score": 125.5,
+      "is_blurry": false,
+      "brightness": 0.65,
+      "is_too_dark": false,
+      "is_too_bright": false,
+      "is_good_quality": true
+    },
+    "selfie": {
+      "blur_score": 200.3,
+      "is_blurry": false,
+      "brightness": 0.58,
+      "is_too_dark": false,
+      "is_too_bright": false,
+      "pose": {
+        "yaw": 5.2,
+        "pitch": -3.1,
+        "roll": 1.5,
+        "is_frontal": true
+      },
+      "is_good_quality": true
+    }
+  },
+  "demographics": {
+    "ktp": { "age": 28, "gender": "Male" },
+    "selfie": { "age": 29, "gender": "Male" }
+  },
+  "face_boxes": {
+    "ktp": { "x": 120, "y": 80, "width": 150, "height": 180 },
+    "selfie": { "x": 200, "y": 100, "width": 250, "height": 300 }
+  },
+  "message": "KYC verification successful"
+}
+```
+
+### Error Response
+
+```json
+{
+  "error_code": "FACE_NOT_DETECTED",
+  "message": "No face detected in image"
+}
+```
+
+## Error Codes
+
+| Code | HTTP | Description |
+|------|------|-------------|
+| `FACE_NOT_DETECTED` | 400 | No face found in uploaded image |
+| `MULTIPLE_FACES_DETECTED` | 400 | Multiple faces detected - rejected |
+| `LIVENESS_FAILED` | 400 | Spoofing attempt detected |
+| `IMAGE_INVALID` | 400 | Invalid or corrupt image file |
+| `IMAGE_TOO_LARGE` | 413 | Image exceeds size limit |
+| `UNSUPPORTED_FORMAT` | 415 | Image format not JPEG/PNG |
+| `MODEL_NOT_LOADED` | 503 | ML models not initialized |
+
+## Configuration
+
+Configuration can be set via environment variables or `.env` file:
+
+| Variable | Default | Description |
+|----------|---------|-------------|
+| `DEBUG` | `true` | Enable debug mode |
+| `FACE_MATCH_THRESHOLD` | `0.5` | Face similarity threshold |
+| `LIVENESS_THRESHOLD` | `0.5` | Liveness confidence threshold |
+| `BLUR_THRESHOLD` | `100.0` | Blur detection threshold |
+| `BRIGHTNESS_MIN` | `0.2` | Minimum brightness |
+| `BRIGHTNESS_MAX` | `0.8` | Maximum brightness |
+| `USE_GPU` | `false` | Enable GPU acceleration |
+| `MAX_IMAGE_SIZE_MB` | `10.0` | Maximum upload size |
+
+## Project Structure
+
+```
+sentinel/
+├── app/
+│   ├── __init__.py
+│   ├── main.py                    # FastAPI application entry point
+│   ├── config.py                  # Configuration settings
+│   ├── api/
+│   │   ├── __init__.py
+│   │   ├── dependencies.py        # Shared dependencies
+│   │   └── routes/
+│   │       ├── __init__.py
+│   │       ├── health.py          # Health check endpoint
+│   │       ├── kyc.py             # File upload endpoints
+│   │       └── kyc_base64.py      # Base64 endpoints
+│   ├── services/
+│   │   ├── __init__.py
+│   │   ├── face_recognition.py    # AuraFace service
+│   │   ├── face_quality.py        # Quality analysis service
+│   │   └── liveness_detection.py  # Anti-spoofing service
+│   ├── models/
+│   │   ├── __init__.py
+│   │   └── schemas.py             # Pydantic models
+│   └── utils/
+│       ├── __init__.py
+│       ├── image_utils.py         # Image processing
+│       └── ktp_extractor.py       # KTP face extraction
+├── models/                        # ML model files
+│   ├── auraface/                  # AuraFace model
+│   └── anti_spoof/                # Anti-spoofing models
+├── Silent-Face-Anti-Spoofing/     # Cloned repository
+├── requirements.txt
+├── setup_models.py                # Model download script
+└── README.md
+```
+
+## Notes
+
+- AuraFace produces 512-dimensional face embeddings
+- Similarity threshold of 0.5 is balanced; increase for higher security
+- Silent-Face-Anti-Spoofing uses 2 models for fusion (MiniFASNetV1SE + MiniFASNetV2)
+- First request may be slow due to model warm-up
+- CPU mode is used by default; set `USE_GPU=true` for GPU acceleration
+
+## License
+
+This is a proof-of-concept for educational purposes.

app/__init__.py

@@ -0,0 +1 @@
+# KYC POC Application

app/api/__init__.py

@@ -0,0 +1 @@
+# API Package

app/api/dependencies.py

@@ -0,0 +1,100 @@
+"""
+Shared dependencies for API routes.
+"""
+
+from fastapi import UploadFile, HTTPException
+from typing import Tuple
+import numpy as np
+
+from ..config import settings
+from ..utils.image_utils import (
+    read_image_from_upload,
+    validate_content_type,
+    decode_base64_image
+)
+from ..services.face_recognition import face_recognition_service
+from ..services.liveness_detection import liveness_detection_service
+from ..services.face_quality import face_quality_service
+from ..services.ktp_ocr import ktp_ocr_service
+
+
+async def get_validated_image(file: UploadFile) -> np.ndarray:
+    """
+    Validate and read an uploaded image file.
+
+    Args:
+        file: Uploaded file
+
+    Returns:
+        Image as numpy array (BGR format)
+    """
+    # Validate content type
+    validate_content_type(file.content_type, settings.ALLOWED_IMAGE_TYPES)
+
+    # Read and decode image
+    image = await read_image_from_upload(file)
+
+    return image
+
+
+async def get_validated_images(
+    file1: UploadFile,
+    file2: UploadFile
+) -> Tuple[np.ndarray, np.ndarray]:
+    """
+    Validate and read two uploaded image files.
+
+    Args:
+        file1: First uploaded file
+        file2: Second uploaded file
+
+    Returns:
+        Tuple of images as numpy arrays (BGR format)
+    """
+    image1 = await get_validated_image(file1)
+    image2 = await get_validated_image(file2)
+    return image1, image2
+
+
+def get_face_recognition_service():
+    """Get the face recognition service instance."""
+    if not face_recognition_service.initialized:
+        raise HTTPException(
+            status_code=503,
+            detail={
+                "error_code": "MODEL_NOT_LOADED",
+                "message": "Face recognition model not loaded. Please wait for initialization."
+            }
+        )
+    return face_recognition_service
+
+
+def get_liveness_service():
+    """Get the liveness detection service instance."""
+    if not liveness_detection_service.initialized:
+        raise HTTPException(
+            status_code=503,
+            detail={
+                "error_code": "MODEL_NOT_LOADED",
+                "message": "Liveness detection model not loaded. Please wait for initialization."
+            }
+        )
+    return liveness_detection_service
+
+
+def get_quality_service():
+    """Get the face quality service instance."""
+    return face_quality_service
+
+
+def get_ocr_service():
+    """Get the KTP OCR service instance."""
+    if not ktp_ocr_service.initialized:
+        raise HTTPException(
+            status_code=503,
+            detail={
+                "error_code": "MODEL_NOT_LOADED",
+                "message": "OCR model not loaded. Please wait for initialization."
+            }
+        )
+    return ktp_ocr_service

app/api/routes/__init__.py

@@ -0,0 +1 @@
+# Routes Package

app/api/routes/health.py

@@ -0,0 +1,57 @@
+"""
+Health check endpoints.
+"""
+
+from fastapi import APIRouter
+
+from ...config import settings
+from ...models.schemas import HealthResponse
+from ...services.face_recognition import face_recognition_service
+from ...services.liveness_detection import liveness_detection_service
+from ...services.ktp_ocr import ktp_ocr_service
+
+router = APIRouter()
+
+
+@router.get(
+    "/health",
+    response_model=HealthResponse,
+    summary="Health Check",
+    description="Check the health status of the API and its models."
+)
+async def health_check() -> HealthResponse:
+    """
+    Check the health status of the API.
+
+    Returns:
+        Health status including model loading states.
+    """
+    models_loaded = {
+        "face_recognition": face_recognition_service.initialized,
+        "liveness_detection": liveness_detection_service.initialized,
+        "ktp_ocr": ktp_ocr_service.initialized
+    }
+
+    # Determine overall status
+    all_loaded = all(models_loaded.values())
+    status = "healthy" if all_loaded else "degraded"
+
+    return HealthResponse(
+        status=status,
+        models_loaded=models_loaded,
+        version=settings.APP_VERSION
+    )
+
+
+@router.get(
+    "/",
+    summary="Root",
+    description="API root endpoint."
+)
+async def root():
+    """API root endpoint."""
+    return {
+        "name": settings.APP_NAME,
+        "version": settings.APP_VERSION,
+        "docs": "/docs"
+    }

app/api/routes/kyc.py

@@ -0,0 +1,440 @@
+"""
+KYC verification endpoints (file upload).
+
+These endpoints accept multipart/form-data file uploads.
+"""
+
+from fastapi import APIRouter, File, UploadFile, Form, HTTPException
+from typing import Optional
+import logging
+
+from ...config import settings
+from ...models.schemas import (
+    VerifyResponse,
+    FaceMatchResponse,
+    LivenessResponse,
+    QualityResponse,
+    FaceMatchResult,
+    LivenessResult,
+    QualityAnalysis,
+    BoundingBox,
+    Demographics,
+    FaceInfo,
+    FacePose
+)
+from ..dependencies import (
+    get_validated_image,
+    get_validated_images,
+    get_face_recognition_service,
+    get_liveness_service,
+    get_quality_service
+)
+from ...utils.ktp_extractor import KTPFaceExtractor
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/kyc", tags=["KYC - File Upload"])
+
+
+@router.post(
+    "/verify",
+    response_model=VerifyResponse,
+    summary="Full KYC Verification",
+    description="Perform complete KYC verification: face matching, liveness detection, and quality analysis."
+)
+async def verify_kyc(
+    ktp_image: UploadFile = File(..., description="KTP/ID card photo"),
+    selfie_image: UploadFile = File(..., description="Selfie photo"),
+    threshold: float = Form(default=0.5, ge=0.0, le=1.0, description="Face match threshold")
+) -> VerifyResponse:
+    """
+    Perform complete KYC verification.
+
+    This endpoint:
+    1. Extracts face from KTP image
+    2. Checks liveness of selfie
+    3. Compares faces between KTP and selfie
+    4. Analyzes image quality
+    5. Extracts demographics (age, gender)
+
+    Args:
+        ktp_image: KTP/ID card image file
+        selfie_image: Selfie image file
+        threshold: Similarity threshold for face matching
+
+    Returns:
+        Complete verification results
+    """
+    # Get services
+    face_service = get_face_recognition_service()
+    liveness_service = get_liveness_service()
+    quality_service = get_quality_service()
+
+    # Read and validate images
+    ktp_img, selfie_img = await get_validated_images(ktp_image, selfie_image)
+
+    # Setup KTP extractor
+    ktp_extractor = KTPFaceExtractor()
+    ktp_extractor.set_detector(face_service.face_app)
+
+    try:
+        # Extract face from KTP
+        try:
+            ktp_face_img, ktp_face_info = ktp_extractor.extract_face(ktp_img, padding=0.3)
+        except ValueError as e:
+            raise HTTPException(
+                status_code=400,
+                detail={
+                    "error_code": "FACE_NOT_DETECTED",
+                    "message": f"KTP image: {str(e)}"
+                }
+            )
+
+        # Extract face info from selfie
+        try:
+            selfie_face_info = face_service.extract_face_info(selfie_img, allow_multiple=False)
+        except ValueError as e:
+            if "Multiple faces" in str(e):
+                raise HTTPException(
+                    status_code=400,
+                    detail={
+                        "error_code": "MULTIPLE_FACES_DETECTED",
+                        "message": f"Selfie: {str(e)}"
+                    }
+                )
+            raise HTTPException(
+                status_code=400,
+                detail={
+                    "error_code": "FACE_NOT_DETECTED",
+                    "message": f"Selfie: {str(e)}"
+                }
+            )
+
+        # Extract face info from KTP (cropped face)
+        try:
+            ktp_embedding_info = face_service.extract_face_info(ktp_face_img, allow_multiple=False)
+        except ValueError as e:
+            raise HTTPException(
+                status_code=400,
+                detail={
+                    "error_code": "FACE_NOT_DETECTED",
+                    "message": f"Could not extract embedding from KTP face: {str(e)}"
+                }
+            )
+
+        # Compare faces
+        face_match = face_service.compare_faces(
+            ktp_embedding_info["embedding"],
+            selfie_face_info["embedding"],
+            threshold
+        )
+
+        # Check liveness on selfie
+        liveness = liveness_service.check_liveness(selfie_img)
+
+        # Quality analysis
+        ktp_quality = quality_service.analyze_quality(ktp_face_img, ktp_embedding_info)
+        selfie_quality = quality_service.analyze_quality(selfie_img, selfie_face_info)
+
+        # Build response
+        return VerifyResponse(
+            success=face_match["is_match"] and liveness.get("is_real", False),
+            face_match=FaceMatchResult(**face_match),
+            liveness=LivenessResult(**liveness),
+            quality={
+                "ktp": _build_quality_analysis(ktp_quality),
+                "selfie": _build_quality_analysis(selfie_quality)
+            },
+            demographics={
+                "ktp": Demographics(
+                    age=ktp_embedding_info.get("age"),
+                    gender=ktp_embedding_info.get("gender")
+                ),
+                "selfie": Demographics(
+                    age=selfie_face_info.get("age"),
+                    gender=selfie_face_info.get("gender")
+                )
+            },
+            face_boxes={
+                "ktp": BoundingBox(**ktp_face_info["bbox"]),
+                "selfie": BoundingBox(**selfie_face_info["bbox"])
+            },
+            message=_build_verification_message(face_match, liveness)
+        )
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Verification error: {e}", exc_info=True)
+        raise HTTPException(
+            status_code=500,
+            detail={
+                "error_code": "VERIFICATION_ERROR",
+                "message": f"Verification failed: {str(e)}"
+            }
+        )
+
+
+@router.post(
+    "/face-match",
+    response_model=FaceMatchResponse,
+    summary="Face Matching Only",
+    description="Compare faces between two images without liveness check."
+)
+async def face_match(
+    ktp_image: UploadFile = File(..., description="KTP/ID card photo"),
+    selfie_image: UploadFile = File(..., description="Selfie photo"),
+    threshold: float = Form(default=0.5, ge=0.0, le=1.0, description="Face match threshold")
+) -> FaceMatchResponse:
+    """
+    Compare faces between KTP and selfie images.
+
+    Args:
+        ktp_image: KTP/ID card image file
+        selfie_image: Selfie image file
+        threshold: Similarity threshold for face matching
+
+    Returns:
+        Face matching results
+    """
+    face_service = get_face_recognition_service()
+
+    # Read and validate images
+    ktp_img, selfie_img = await get_validated_images(ktp_image, selfie_image)
+
+    # Setup KTP extractor
+    ktp_extractor = KTPFaceExtractor()
+    ktp_extractor.set_detector(face_service.face_app)
+
+    try:
+        # Extract face from KTP
+        try:
+            ktp_face_img, ktp_face_info = ktp_extractor.extract_face(ktp_img, padding=0.3)
+            ktp_embedding_info = face_service.extract_face_info(ktp_face_img, allow_multiple=False)
+        except ValueError as e:
+            raise HTTPException(
+                status_code=400,
+                detail={
+                    "error_code": "FACE_NOT_DETECTED",
+                    "message": f"KTP image: {str(e)}"
+                }
+            )
+
+        # Extract face from selfie
+        try:
+            selfie_face_info = face_service.extract_face_info(selfie_img, allow_multiple=False)
+        except ValueError as e:
+            if "Multiple faces" in str(e):
+                raise HTTPException(
+                    status_code=400,
+                    detail={
+                        "error_code": "MULTIPLE_FACES_DETECTED",
+                        "message": f"Selfie: {str(e)}"
+                    }
+                )
+            raise HTTPException(
+                status_code=400,
+                detail={
+                    "error_code": "FACE_NOT_DETECTED",
+                    "message": f"Selfie: {str(e)}"
+                }
+            )
+
+        # Compare faces
+        face_match_result = face_service.compare_faces(
+            ktp_embedding_info["embedding"],
+            selfie_face_info["embedding"],
+            threshold
+        )
+
+        return FaceMatchResponse(
+            success=face_match_result["is_match"],
+            face_match=FaceMatchResult(**face_match_result),
+            face1=FaceInfo(
+                bbox=BoundingBox(**ktp_face_info["bbox"]),
+                demographics=Demographics(
+                    age=ktp_embedding_info.get("age"),
+                    gender=ktp_embedding_info.get("gender")
+                ),
+                det_score=ktp_embedding_info.get("det_score")
+            ),
+            face2=FaceInfo(
+                bbox=BoundingBox(**selfie_face_info["bbox"]),
+                demographics=Demographics(
+                    age=selfie_face_info.get("age"),
+                    gender=selfie_face_info.get("gender")
+                ),
+                det_score=selfie_face_info.get("det_score")
+            ),
+            message="Faces match" if face_match_result["is_match"] else "Faces do not match"
+        )
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Face match error: {e}", exc_info=True)
+        raise HTTPException(
+            status_code=500,
+            detail={
+                "error_code": "FACE_MATCH_ERROR",
+                "message": f"Face matching failed: {str(e)}"
+            }
+        )
+
+
+@router.post(
+    "/liveness",
+    response_model=LivenessResponse,
+    summary="Liveness Detection Only",
+    description="Check if a face image is from a real person."
+)
+async def check_liveness(
+    image: UploadFile = File(..., description="Face image to check")
+) -> LivenessResponse:
+    """
+    Check liveness of a face image.
+
+    Args:
+        image: Face image file
+
+    Returns:
+        Liveness detection results
+    """
+    liveness_service = get_liveness_service()
+
+    # Read and validate image
+    img = await get_validated_image(image)
+
+    try:
+        liveness = liveness_service.check_liveness(img)
+
+        return LivenessResponse(
+            success=liveness.get("is_real", False),
+            liveness=LivenessResult(**liveness),
+            message="Real face detected" if liveness.get("is_real") else "Possible spoofing detected"
+        )
+
+    except Exception as e:
+        logger.error(f"Liveness check error: {e}", exc_info=True)
+        raise HTTPException(
+            status_code=500,
+            detail={
+                "error_code": "LIVENESS_ERROR",
+                "message": f"Liveness check failed: {str(e)}"
+            }
+        )
+
+
+@router.post(
+    "/quality",
+    response_model=QualityResponse,
+    summary="Face Quality Check Only",
+    description="Analyze the quality of a face image."
+)
+async def check_quality(
+    image: UploadFile = File(..., description="Face image to analyze")
+) -> QualityResponse:
+    """
+    Analyze the quality of a face image.
+
+    Args:
+        image: Face image file
+
+    Returns:
+        Quality analysis results
+    """
+    face_service = get_face_recognition_service()
+    quality_service = get_quality_service()
+
+    # Read and validate image
+    img = await get_validated_image(image)
+
+    try:
+        # Extract face info
+        try:
+            face_info = face_service.extract_face_info(img, allow_multiple=False)
+        except ValueError as e:
+            if "Multiple faces" in str(e):
+                raise HTTPException(
+                    status_code=400,
+                    detail={
+                        "error_code": "MULTIPLE_FACES_DETECTED",
+                        "message": str(e)
+                    }
+                )
+            raise HTTPException(
+                status_code=400,
+                detail={
+                    "error_code": "FACE_NOT_DETECTED",
+                    "message": str(e)
+                }
+            )
+
+        # Analyze quality
+        quality = quality_service.analyze_quality(img, face_info)
+
+        return QualityResponse(
+            success=quality.get("is_good_quality", False),
+            quality=_build_quality_analysis(quality),
+            face_box=BoundingBox(**face_info["bbox"]),
+            demographics=Demographics(
+                age=face_info.get("age"),
+                gender=face_info.get("gender")
+            ),
+            message="Good quality" if quality.get("is_good_quality") else "Quality issues detected"
+        )
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Quality check error: {e}", exc_info=True)
+        raise HTTPException(
+            status_code=500,
+            detail={
+                "error_code": "QUALITY_ERROR",
+                "message": f"Quality check failed: {str(e)}"
+            }
+        )
+
+
+# ============================================================================
+# Helper Functions
+# ============================================================================
+
+def _build_quality_analysis(quality: dict) -> QualityAnalysis:
+    """Build QualityAnalysis from quality dict."""
+    pose = None
+    if "pose" in quality:
+        pose = FacePose(
+            yaw=quality["pose"].get("yaw", 0),
+            pitch=quality["pose"].get("pitch", 0),
+            roll=quality["pose"].get("roll", 0),
+            is_frontal=quality["pose"].get("is_frontal", True)
+        )
+
+    return QualityAnalysis(
+        blur_score=quality.get("blur_score", 0),
+        blur_threshold=quality.get("blur_threshold", settings.BLUR_THRESHOLD),
+        is_blurry=quality.get("is_blurry", False),
+        brightness=quality.get("brightness", 0.5),
+        brightness_min=quality.get("brightness_min", settings.BRIGHTNESS_MIN),
+        brightness_max=quality.get("brightness_max", settings.BRIGHTNESS_MAX),
+        is_too_dark=quality.get("is_too_dark", False),
+        is_too_bright=quality.get("is_too_bright", False),
+        pose=pose,
+        is_good_quality=quality.get("is_good_quality", True)
+    )
+
+
+def _build_verification_message(face_match: dict, liveness: dict) -> str:
+    """Build verification result message."""
+    is_match = face_match.get("is_match", False)
+    is_real = liveness.get("is_real", False)
+
+    if is_match and is_real:
+        return "KYC verification successful"
+    elif not is_real:
+        return "Liveness check failed - possible spoofing attempt"
+    elif not is_match:
+        return "Face matching failed - faces do not match"
+    else:
+        return "Verification failed"

app/api/routes/kyc_base64.py

@@ -0,0 +1,465 @@
+"""
+KYC verification endpoints (Base64 input).
+
+These endpoints accept JSON with base64 encoded images.
+Useful for testing with Insomnia, Postman, or similar tools.
+"""
+
+from fastapi import APIRouter, HTTPException
+import logging
+
+from ...config import settings
+from ...models.schemas import (
+    VerifyResponse,
+    FaceMatchResponse,
+    LivenessResponse,
+    QualityResponse,
+    Base64VerifyRequest,
+    Base64FaceMatchRequest,
+    Base64SingleImageRequest,
+    FaceMatchResult,
+    LivenessResult,
+    QualityAnalysis,
+    BoundingBox,
+    Demographics,
+    FaceInfo,
+    FacePose
+)
+from ..dependencies import (
+    get_face_recognition_service,
+    get_liveness_service,
+    get_quality_service
+)
+from ...utils.image_utils import decode_base64_image
+from ...utils.ktp_extractor import KTPFaceExtractor
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/kyc/base64", tags=["KYC - Base64"])
+
+
+@router.post(
+    "/verify",
+    response_model=VerifyResponse,
+    summary="Full KYC Verification (Base64)",
+    description="Perform complete KYC verification with base64 encoded images."
+)
+async def verify_kyc_base64(request: Base64VerifyRequest) -> VerifyResponse:
+    """
+    Perform complete KYC verification with base64 images.
+
+    Args:
+        request: Request containing base64 encoded KTP and selfie images
+
+    Returns:
+        Complete verification results
+    """
+    # Get services
+    face_service = get_face_recognition_service()
+    liveness_service = get_liveness_service()
+    quality_service = get_quality_service()
+
+    # Decode images
+    try:
+        ktp_img = decode_base64_image(request.ktp_image)
+        selfie_img = decode_base64_image(request.selfie_image)
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "error_code": "IMAGE_INVALID",
+                "message": f"Failed to decode base64 image: {str(e)}"
+            }
+        )
+
+    # Setup KTP extractor
+    ktp_extractor = KTPFaceExtractor()
+    ktp_extractor.set_detector(face_service.face_app)
+
+    try:
+        # Extract face from KTP
+        try:
+            ktp_face_img, ktp_face_info = ktp_extractor.extract_face(ktp_img, padding=0.3)
+        except ValueError as e:
+            raise HTTPException(
+                status_code=400,
+                detail={
+                    "error_code": "FACE_NOT_DETECTED",
+                    "message": f"KTP image: {str(e)}"
+                }
+            )
+
+        # Extract face info from selfie
+        try:
+            selfie_face_info = face_service.extract_face_info(selfie_img, allow_multiple=False)
+        except ValueError as e:
+            if "Multiple faces" in str(e):
+                raise HTTPException(
+                    status_code=400,
+                    detail={
+                        "error_code": "MULTIPLE_FACES_DETECTED",
+                        "message": f"Selfie: {str(e)}"
+                    }
+                )
+            raise HTTPException(
+                status_code=400,
+                detail={
+                    "error_code": "FACE_NOT_DETECTED",
+                    "message": f"Selfie: {str(e)}"
+                }
+            )
+
+        # Extract face info from KTP (cropped face)
+        try:
+            ktp_embedding_info = face_service.extract_face_info(ktp_face_img, allow_multiple=False)
+        except ValueError as e:
+            raise HTTPException(
+                status_code=400,
+                detail={
+                    "error_code": "FACE_NOT_DETECTED",
+                    "message": f"Could not extract embedding from KTP face: {str(e)}"
+                }
+            )
+
+        # Compare faces
+        face_match = face_service.compare_faces(
+            ktp_embedding_info["embedding"],
+            selfie_face_info["embedding"],
+            request.threshold
+        )
+
+        # Check liveness on selfie
+        liveness = liveness_service.check_liveness(selfie_img)
+
+        # Quality analysis
+        ktp_quality = quality_service.analyze_quality(ktp_face_img, ktp_embedding_info)
+        selfie_quality = quality_service.analyze_quality(selfie_img, selfie_face_info)
+
+        # Build response
+        return VerifyResponse(
+            success=face_match["is_match"] and liveness.get("is_real", False),
+            face_match=FaceMatchResult(**face_match),
+            liveness=LivenessResult(**liveness),
+            quality={
+                "ktp": _build_quality_analysis(ktp_quality),
+                "selfie": _build_quality_analysis(selfie_quality)
+            },
+            demographics={
+                "ktp": Demographics(
+                    age=ktp_embedding_info.get("age"),
+                    gender=ktp_embedding_info.get("gender")
+                ),
+                "selfie": Demographics(
+                    age=selfie_face_info.get("age"),
+                    gender=selfie_face_info.get("gender")
+                )
+            },
+            face_boxes={
+                "ktp": BoundingBox(**ktp_face_info["bbox"]),
+                "selfie": BoundingBox(**selfie_face_info["bbox"])
+            },
+            message=_build_verification_message(face_match, liveness)
+        )
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Verification error: {e}", exc_info=True)
+        raise HTTPException(
+            status_code=500,
+            detail={
+                "error_code": "VERIFICATION_ERROR",
+                "message": f"Verification failed: {str(e)}"
+            }
+        )
+
+
+@router.post(
+    "/face-match",
+    response_model=FaceMatchResponse,
+    summary="Face Matching Only (Base64)",
+    description="Compare faces between two base64 encoded images."
+)
+async def face_match_base64(request: Base64FaceMatchRequest) -> FaceMatchResponse:
+    """
+    Compare faces between two base64 encoded images.
+
+    Args:
+        request: Request containing base64 encoded images
+
+    Returns:
+        Face matching results
+    """
+    face_service = get_face_recognition_service()
+
+    # Decode images
+    try:
+        img1 = decode_base64_image(request.image1)
+        img2 = decode_base64_image(request.image2)
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "error_code": "IMAGE_INVALID",
+                "message": f"Failed to decode base64 image: {str(e)}"
+            }
+        )
+
+    # Setup KTP extractor for first image
+    ktp_extractor = KTPFaceExtractor()
+    ktp_extractor.set_detector(face_service.face_app)
+
+    try:
+        # Extract face from first image (treated as KTP)
+        try:
+            img1_face, img1_face_info = ktp_extractor.extract_face(img1, padding=0.3)
+            img1_embedding_info = face_service.extract_face_info(img1_face, allow_multiple=False)
+        except ValueError as e:
+            raise HTTPException(
+                status_code=400,
+                detail={
+                    "error_code": "FACE_NOT_DETECTED",
+                    "message": f"Image 1: {str(e)}"
+                }
+            )
+
+        # Extract face from second image
+        try:
+            img2_face_info = face_service.extract_face_info(img2, allow_multiple=False)
+        except ValueError as e:
+            if "Multiple faces" in str(e):
+                raise HTTPException(
+                    status_code=400,
+                    detail={
+                        "error_code": "MULTIPLE_FACES_DETECTED",
+                        "message": f"Image 2: {str(e)}"
+                    }
+                )
+            raise HTTPException(
+                status_code=400,
+                detail={
+                    "error_code": "FACE_NOT_DETECTED",
+                    "message": f"Image 2: {str(e)}"
+                }
+            )
+
+        # Compare faces
+        face_match_result = face_service.compare_faces(
+            img1_embedding_info["embedding"],
+            img2_face_info["embedding"],
+            request.threshold
+        )
+
+        return FaceMatchResponse(
+            success=face_match_result["is_match"],
+            face_match=FaceMatchResult(**face_match_result),
+            face1=FaceInfo(
+                bbox=BoundingBox(**img1_face_info["bbox"]),
+                demographics=Demographics(
+                    age=img1_embedding_info.get("age"),
+                    gender=img1_embedding_info.get("gender")
+                ),
+                det_score=img1_embedding_info.get("det_score")
+            ),
+            face2=FaceInfo(
+                bbox=BoundingBox(**img2_face_info["bbox"]),
+                demographics=Demographics(
+                    age=img2_face_info.get("age"),
+                    gender=img2_face_info.get("gender")
+                ),
+                det_score=img2_face_info.get("det_score")
+            ),
+            message="Faces match" if face_match_result["is_match"] else "Faces do not match"
+        )
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Face match error: {e}", exc_info=True)
+        raise HTTPException(
+            status_code=500,
+            detail={
+                "error_code": "FACE_MATCH_ERROR",
+                "message": f"Face matching failed: {str(e)}"
+            }
+        )
+
+
+@router.post(
+    "/liveness",
+    response_model=LivenessResponse,
+    summary="Liveness Detection Only (Base64)",
+    description="Check if a base64 encoded face image is from a real person."
+)
+async def check_liveness_base64(request: Base64SingleImageRequest) -> LivenessResponse:
+    """
+    Check liveness of a base64 encoded face image.
+
+    Args:
+        request: Request containing base64 encoded image
+
+    Returns:
+        Liveness detection results
+    """
+    liveness_service = get_liveness_service()
+
+    # Decode image
+    try:
+        img = decode_base64_image(request.image)
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "error_code": "IMAGE_INVALID",
+                "message": f"Failed to decode base64 image: {str(e)}"
+            }
+        )
+
+    try:
+        liveness = liveness_service.check_liveness(img)
+
+        return LivenessResponse(
+            success=liveness.get("is_real", False),
+            liveness=LivenessResult(**liveness),
+            message="Real face detected" if liveness.get("is_real") else "Possible spoofing detected"
+        )
+
+    except Exception as e:
+        logger.error(f"Liveness check error: {e}", exc_info=True)
+        raise HTTPException(
+            status_code=500,
+            detail={
+                "error_code": "LIVENESS_ERROR",
+                "message": f"Liveness check failed: {str(e)}"
+            }
+        )
+
+
+@router.post(
+    "/quality",
+    response_model=QualityResponse,
+    summary="Face Quality Check Only (Base64)",
+    description="Analyze the quality of a base64 encoded face image."
+)
+async def check_quality_base64(request: Base64SingleImageRequest) -> QualityResponse:
+    """
+    Analyze the quality of a base64 encoded face image.
+
+    Args:
+        request: Request containing base64 encoded image
+
+    Returns:
+        Quality analysis results
+    """
+    face_service = get_face_recognition_service()
+    quality_service = get_quality_service()
+
+    # Decode image
+    try:
+        img = decode_base64_image(request.image)
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "error_code": "IMAGE_INVALID",
+                "message": f"Failed to decode base64 image: {str(e)}"
+            }
+        )
+
+    try:
+        # Extract face info
+        try:
+            face_info = face_service.extract_face_info(img, allow_multiple=False)
+        except ValueError as e:
+            if "Multiple faces" in str(e):
+                raise HTTPException(
+                    status_code=400,
+                    detail={
+                        "error_code": "MULTIPLE_FACES_DETECTED",
+                        "message": str(e)
+                    }
+                )
+            raise HTTPException(
+                status_code=400,
+                detail={
+                    "error_code": "FACE_NOT_DETECTED",
+                    "message": str(e)
+                }
+            )
+
+        # Analyze quality
+        quality = quality_service.analyze_quality(img, face_info)
+
+        return QualityResponse(
+            success=quality.get("is_good_quality", False),
+            quality=_build_quality_analysis(quality),
+            face_box=BoundingBox(**face_info["bbox"]),
+            demographics=Demographics(
+                age=face_info.get("age"),
+                gender=face_info.get("gender")
+            ),
+            message="Good quality" if quality.get("is_good_quality") else "Quality issues detected"
+        )
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Quality check error: {e}", exc_info=True)
+        raise HTTPException(
+            status_code=500,
+            detail={
+                "error_code": "QUALITY_ERROR",
+                "message": f"Quality check failed: {str(e)}"
+            }
+        )
+
+
+# ============================================================================
+# Helper Functions
+# ============================================================================
+
+def _build_quality_analysis(quality: dict) -> QualityAnalysis:
+    """Build QualityAnalysis from quality dict."""
+    pose = None
+    if "pose" in quality:
+        pose = FacePose(
+            yaw=quality["pose"].get("yaw", 0),
+            pitch=quality["pose"].get("pitch", 0),
+            roll=quality["pose"].get("roll", 0),
+            is_frontal=quality["pose"].get("is_frontal", True)
+        )
+
+    return QualityAnalysis(
+        blur_score=quality.get("blur_score", 0),
+        blur_threshold=quality.get("blur_threshold", settings.BLUR_THRESHOLD),
+        is_blurry=quality.get("is_blurry", False),
+        brightness=quality.get("brightness", 0.5),
+        brightness_min=quality.get("brightness_min", settings.BRIGHTNESS_MIN),
+        brightness_max=quality.get("brightness_max", settings.BRIGHTNESS_MAX),
+        is_too_dark=quality.get("is_too_dark", False),
+        is_too_bright=quality.get("is_too_bright", False),
+        pose=pose,
+        is_good_quality=quality.get("is_good_quality", True)
+    )
+
+
+def _build_verification_message(face_match: dict, liveness: dict) -> str:
+    """Build verification result message."""
+    is_match = face_match.get("is_match", False)
+    is_real = liveness.get("is_real", False)
+
+    if is_match and is_real:
+        return "KYC verification successful"
+    elif not is_real:
+        return "Liveness check failed - possible spoofing attempt"
+    elif not is_match:
+        return "Face matching failed - faces do not match"
+    else:
+        return "Verification failed"

app/api/routes/ocr.py

@@ -0,0 +1,272 @@
+"""
+KTP OCR endpoints (File upload and Base64).
+
+These endpoints extract text from Indonesian KTP (ID card) images
+and return structured, sanitized data.
+"""
+
+from fastapi import APIRouter, HTTPException, UploadFile, File, Query
+import logging
+
+from ...models.schemas import (
+    OCRResponse,
+    Base64OCRRequest,
+    KTPOCRData,
+    OCRFieldResult,
+    OCRTextBlock,
+    KTPValidation,
+    NIKValidation
+)
+from ..dependencies import get_ocr_service, get_validated_image
+from ...utils.image_utils import decode_base64_image
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/kyc/ocr", tags=["KTP OCR"])
+
+
+def _build_ocr_response(result: dict) -> OCRResponse:
+    """Build OCRResponse from service result."""
+    # Build KTPOCRData from extracted data
+    data_dict = result.get('data', {})
+    ktp_data = KTPOCRData(
+        provinsi=_build_field_result(data_dict.get('provinsi')),
+        kabupaten_kota=_build_field_result(data_dict.get('kabupaten_kota')),
+        nik=_build_field_result(data_dict.get('nik')),
+        nama=_build_field_result(data_dict.get('nama')),
+        tempat_lahir=_build_field_result(data_dict.get('tempat_lahir')),
+        tanggal_lahir=_build_field_result(data_dict.get('tanggal_lahir')),
+        jenis_kelamin=_build_field_result(data_dict.get('jenis_kelamin')),
+        golongan_darah=_build_field_result(data_dict.get('golongan_darah')),
+        alamat=_build_field_result(data_dict.get('alamat')),
+        rt_rw=_build_field_result(data_dict.get('rt_rw')),
+        kelurahan_desa=_build_field_result(data_dict.get('kelurahan_desa')),
+        kecamatan=_build_field_result(data_dict.get('kecamatan')),
+        agama=_build_field_result(data_dict.get('agama')),
+        status_perkawinan=_build_field_result(data_dict.get('status_perkawinan')),
+        pekerjaan=_build_field_result(data_dict.get('pekerjaan')),
+        kewarganegaraan=_build_field_result(data_dict.get('kewarganegaraan')),
+        berlaku_hingga=_build_field_result(data_dict.get('berlaku_hingga'))
+    )
+
+    # Build raw text blocks (convert numpy types to native Python types)
+    raw_text = []
+    for item in result.get('raw_text', []):
+        bbox = item.get('bbox', [])
+        # Convert numpy arrays/values to native Python lists/ints
+        if bbox:
+            bbox = [[int(coord) for coord in point] for point in bbox]
+        raw_text.append(
+            OCRTextBlock(
+                text=item.get('text', ''),
+                confidence=float(item.get('confidence', 0.0)),
+                bbox=bbox
+            )
+        )
+
+    # Build validation result
+    validation = None
+    if result.get('validation'):
+        nik_validation = result['validation'].get('nik')
+        if nik_validation:
+            validation = KTPValidation(
+                nik=NIKValidation(
+                    is_valid=nik_validation.get('is_valid', False),
+                    errors=nik_validation.get('errors', []),
+                    extracted=nik_validation.get('extracted', {})
+                )
+            )
+
+    # Determine success based on whether any fields were extracted
+    fields_extracted = sum(1 for v in data_dict.values() if v is not None)
+    success = fields_extracted > 0
+
+    return OCRResponse(
+        success=success,
+        data=ktp_data,
+        raw_text=raw_text,
+        validation=validation,
+        message=f"Extracted {fields_extracted} fields from KTP" if success else "No fields could be extracted"
+    )
+
+
+def _build_field_result(field_data: dict | None) -> OCRFieldResult | None:
+    """Build OCRFieldResult from field data dict."""
+    if not field_data:
+        return None
+    return OCRFieldResult(
+        value=field_data.get('value', ''),
+        confidence=field_data.get('confidence', 0.0),
+        raw_value=field_data.get('raw_value', '')
+    )
+
+
+# ============================================================================
+# File Upload Endpoints
+# ============================================================================
+
+@router.post(
+    "/extract",
+    response_model=OCRResponse,
+    summary="Extract KTP Data (File Upload)",
+    description="""
+    Extract and parse data from a KTP (Indonesian ID card) image.
+
+    This endpoint performs OCR on the uploaded KTP image and returns:
+    - Structured data (NIK, name, address, birth date, etc.)
+    - Raw OCR text with confidence scores and bounding boxes
+    - NIK validation (optional)
+
+    Supported image formats: JPEG, PNG
+    Max file size: 10MB
+    """
+)
+async def extract_ktp_data(
+    ktp_image: UploadFile = File(..., description="KTP image file"),
+    validate: bool = Query(default=True, description="Validate extracted data (e.g., NIK)")
+) -> OCRResponse:
+    """
+    Extract data from KTP image (file upload).
+
+    Args:
+        ktp_image: Uploaded KTP image file
+        validate: Whether to validate extracted data
+
+    Returns:
+        Structured KTP data with validation results
+    """
+    ocr_service = get_ocr_service()
+
+    # Validate and read image
+    try:
+        image = await get_validated_image(ktp_image)
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "error_code": "IMAGE_INVALID",
+                "message": f"Failed to read image: {str(e)}"
+            }
+        )
+
+    try:
+        # Extract KTP data
+        result = ocr_service.extract_ktp_data(image, validate=validate)
+        return _build_ocr_response(result)
+
+    except Exception as e:
+        logger.error(f"OCR extraction error: {e}", exc_info=True)
+        raise HTTPException(
+            status_code=500,
+            detail={
+                "error_code": "OCR_ERROR",
+                "message": f"OCR extraction failed: {str(e)}"
+            }
+        )
+
+
+# ============================================================================
+# Base64 Endpoints
+# ============================================================================
+
+@router.post(
+    "/base64/extract",
+    response_model=OCRResponse,
+    summary="Extract KTP Data (Base64)",
+    description="""
+    Extract and parse data from a base64-encoded KTP image.
+
+    This endpoint performs OCR on the KTP image and returns:
+    - Structured data (NIK, name, address, birth date, etc.)
+    - Raw OCR text with confidence scores and bounding boxes
+    - NIK validation (optional)
+    """
+)
+async def extract_ktp_data_base64(request: Base64OCRRequest) -> OCRResponse:
+    """
+    Extract data from KTP image (base64).
+
+    Args:
+        request: Request containing base64 encoded KTP image
+
+    Returns:
+        Structured KTP data with validation results
+    """
+    ocr_service = get_ocr_service()
+
+    # Decode base64 image
+    try:
+        image = decode_base64_image(request.image)
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "error_code": "IMAGE_INVALID",
+                "message": f"Failed to decode base64 image: {str(e)}"
+            }
+        )
+
+    try:
+        # Extract KTP data
+        result = ocr_service.extract_ktp_data(image, validate=request.validate)
+        return _build_ocr_response(result)
+
+    except Exception as e:
+        logger.error(f"OCR extraction error: {e}", exc_info=True)
+        raise HTTPException(
+            status_code=500,
+            detail={
+                "error_code": "OCR_ERROR",
+                "message": f"OCR extraction failed: {str(e)}"
+            }
+        )
+
+
+@router.post(
+    "/validate-nik",
+    summary="Validate NIK",
+    description="""
+    Validate a 16-digit Indonesian NIK (Nomor Induk Kependudukan).
+
+    Returns validation status and extracted information:
+    - Province code
+    - City/Regency code
+    - District code
+    - Birth date
+    - Gender
+    - Sequence number
+    """
+)
+async def validate_nik(
+    nik: str = Query(..., description="16-digit NIK to validate", min_length=16, max_length=16)
+) -> NIKValidation:
+    """
+    Validate a NIK string.
+
+    Args:
+        nik: 16-digit NIK string
+
+    Returns:
+        Validation result with extracted information
+    """
+    ocr_service = get_ocr_service()
+
+    try:
+        result = ocr_service.validate_nik(nik)
+        return NIKValidation(
+            is_valid=result.get('is_valid', False),
+            errors=result.get('errors', []),
+            extracted=result.get('extracted', {})
+        )
+    except Exception as e:
+        logger.error(f"NIK validation error: {e}", exc_info=True)
+        raise HTTPException(
+            status_code=500,
+            detail={
+                "error_code": "VALIDATION_ERROR",
+                "message": f"NIK validation failed: {str(e)}"
+            }
+        )

app/config.py

@@ -0,0 +1,69 @@
+"""
+Configuration settings for KYC POC application.
+"""
+
+from pydantic_settings import BaseSettings
+from typing import List
+from pathlib import Path
+
+
+class Settings(BaseSettings):
+    """Application settings."""
+
+    # Application
+    APP_NAME: str = "KYC POC API"
+    APP_VERSION: str = "1.0.0"
+    DEBUG: bool = True
+
+    # Model paths
+    AURAFACE_MODEL_DIR: str = "models/auraface"
+    ANTISPOOF_MODEL_DIR: str = "models/anti_spoof"
+    SILENT_FACE_REPO_DIR: str = "Silent-Face-Anti-Spoofing"
+
+    # Face matching
+    FACE_MATCH_THRESHOLD: float = 0.5
+
+    # Liveness detection
+    LIVENESS_THRESHOLD: float = 0.5
+
+    # Face quality thresholds
+    BLUR_THRESHOLD: float = 100.0  # Below this = blurry
+    BRIGHTNESS_MIN: float = 0.2  # Below this = too dark
+    BRIGHTNESS_MAX: float = 0.8  # Above this = too bright
+    POSE_MAX_YAW: float = 30.0  # Max yaw angle for frontal face
+    POSE_MAX_PITCH: float = 30.0  # Max pitch angle for frontal face
+    POSE_MAX_ROLL: float = 30.0  # Max roll angle for frontal face
+
+    # Device settings
+    USE_GPU: bool = False  # CPU mode for POC
+    DEVICE_ID: int = -1  # -1 for CPU, 0+ for GPU
+
+    # API settings
+    MAX_IMAGE_SIZE_MB: float = 10.0
+    ALLOWED_IMAGE_TYPES: List[str] = ["image/jpeg", "image/png", "image/jpg"]
+
+    # Face detection settings
+    DET_SIZE: tuple = (640, 640)  # Detection input size
+
+    class Config:
+        env_file = ".env"
+        env_file_encoding = "utf-8"
+
+    @property
+    def max_image_size_bytes(self) -> int:
+        """Get max image size in bytes."""
+        return int(self.MAX_IMAGE_SIZE_MB * 1024 * 1024)
+
+    @property
+    def auraface_path(self) -> Path:
+        """Get AuraFace model path."""
+        return Path(self.AURAFACE_MODEL_DIR)
+
+    @property
+    def antispoof_path(self) -> Path:
+        """Get anti-spoof model path."""
+        return Path(self.ANTISPOOF_MODEL_DIR)
+
+
+# Global settings instance
+settings = Settings()

app/main.py

@@ -0,0 +1,183 @@
+"""
+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
+    )

app/services/__init__.py

@@ -0,0 +1 @@
+# Services Package

app/services/face_quality.py

@@ -0,0 +1,233 @@
+"""
+Face Quality Analysis Service.
+
+This service provides face quality assessment including:
+- Blur detection (Laplacian variance)
+- Brightness analysis
+- Face pose estimation
+"""
+
+import cv2
+import numpy as np
+from typing import Dict, Any, Optional
+import logging
+
+from ..config import settings
+
+logger = logging.getLogger(__name__)
+
+
+class FaceQualityService:
+    """Service for analyzing face image quality."""
+
+    def __init__(self):
+        """Initialize the face quality service."""
+        pass
+
+    def analyze_quality(
+        self,
+        image: np.ndarray,
+        face_info: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
+        """
+        Analyze the quality of a face image.
+
+        Args:
+            image: Input image (BGR format)
+            face_info: Optional face info dict containing pose data from face detection
+
+        Returns:
+            Dictionary containing quality metrics
+        """
+        result = {}
+
+        # Analyze blur
+        blur_result = self.analyze_blur(image)
+        result.update(blur_result)
+
+        # Analyze brightness
+        brightness_result = self.analyze_brightness(image)
+        result.update(brightness_result)
+
+        # Add pose analysis if face_info provided
+        if face_info and "pose" in face_info:
+            pose_result = self.analyze_pose(face_info["pose"])
+            result["pose"] = pose_result
+
+        # Overall quality assessment
+        result["is_good_quality"] = self._assess_overall_quality(result)
+
+        return result
+
+    def analyze_blur(self, image: np.ndarray) -> Dict[str, Any]:
+        """
+        Analyze image blur using Laplacian variance method.
+
+        Higher variance = sharper image
+        Lower variance = blurrier image
+
+        Args:
+            image: Input image (BGR format)
+
+        Returns:
+            Dictionary with blur metrics
+        """
+        # Convert to grayscale
+        if len(image.shape) == 3:
+            gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
+        else:
+            gray = image
+
+        # Calculate Laplacian variance
+        laplacian = cv2.Laplacian(gray, cv2.CV_64F)
+        variance = laplacian.var()
+
+        is_blurry = variance < settings.BLUR_THRESHOLD
+
+        return {
+            "blur_score": round(float(variance), 2),
+            "blur_threshold": settings.BLUR_THRESHOLD,
+            "is_blurry": is_blurry
+        }
+
+    def analyze_brightness(self, image: np.ndarray) -> Dict[str, Any]:
+        """
+        Analyze image brightness.
+
+        Args:
+            image: Input image (BGR format)
+
+        Returns:
+            Dictionary with brightness metrics
+        """
+        # Convert to grayscale
+        if len(image.shape) == 3:
+            gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
+        else:
+            gray = image
+
+        # Calculate mean brightness (normalized to 0-1)
+        mean_brightness = np.mean(gray) / 255.0
+
+        is_too_dark = mean_brightness < settings.BRIGHTNESS_MIN
+        is_too_bright = mean_brightness > settings.BRIGHTNESS_MAX
+
+        return {
+            "brightness": round(float(mean_brightness), 3),
+            "brightness_min": settings.BRIGHTNESS_MIN,
+            "brightness_max": settings.BRIGHTNESS_MAX,
+            "is_too_dark": is_too_dark,
+            "is_too_bright": is_too_bright
+        }
+
+    def analyze_pose(self, pose: Dict[str, float]) -> Dict[str, Any]:
+        """
+        Analyze face pose angles.
+
+        Args:
+            pose: Dictionary with yaw, pitch, roll angles
+
+        Returns:
+            Dictionary with pose analysis
+        """
+        yaw = abs(pose.get("yaw", 0))
+        pitch = abs(pose.get("pitch", 0))
+        roll = abs(pose.get("roll", 0))
+
+        is_frontal = (
+            yaw <= settings.POSE_MAX_YAW and
+            pitch <= settings.POSE_MAX_PITCH and
+            roll <= settings.POSE_MAX_ROLL
+        )
+
+        return {
+            "yaw": round(pose.get("yaw", 0), 2),
+            "pitch": round(pose.get("pitch", 0), 2),
+            "roll": round(pose.get("roll", 0), 2),
+            "max_yaw": settings.POSE_MAX_YAW,
+            "max_pitch": settings.POSE_MAX_PITCH,
+            "max_roll": settings.POSE_MAX_ROLL,
+            "is_frontal": is_frontal
+        }
+
+    def analyze_contrast(self, image: np.ndarray) -> Dict[str, Any]:
+        """
+        Analyze image contrast.
+
+        Args:
+            image: Input image (BGR format)
+
+        Returns:
+            Dictionary with contrast metrics
+        """
+        # Convert to grayscale
+        if len(image.shape) == 3:
+            gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
+        else:
+            gray = image
+
+        # Calculate standard deviation as contrast measure
+        contrast = np.std(gray) / 255.0
+
+        return {
+            "contrast": round(float(contrast), 3),
+            "is_low_contrast": contrast < 0.1
+        }
+
+    def analyze_face_size(
+        self,
+        image: np.ndarray,
+        bbox: Dict[str, int],
+        min_face_ratio: float = 0.1
+    ) -> Dict[str, Any]:
+        """
+        Analyze face size relative to image.
+
+        Args:
+            image: Input image
+            bbox: Face bounding box
+            min_face_ratio: Minimum acceptable face to image ratio
+
+        Returns:
+            Dictionary with face size metrics
+        """
+        img_height, img_width = image.shape[:2]
+        img_area = img_height * img_width
+
+        face_area = bbox["width"] * bbox["height"]
+        face_ratio = face_area / img_area
+
+        return {
+            "face_area": face_area,
+            "image_area": img_area,
+            "face_ratio": round(face_ratio, 4),
+            "is_face_too_small": face_ratio < min_face_ratio
+        }
+
+    def _assess_overall_quality(self, metrics: Dict[str, Any]) -> bool:
+        """
+        Assess overall image quality based on metrics.
+
+        Args:
+            metrics: Dictionary of quality metrics
+
+        Returns:
+            True if image passes quality checks
+        """
+        # Check blur
+        if metrics.get("is_blurry", False):
+            return False
+
+        # Check brightness
+        if metrics.get("is_too_dark", False) or metrics.get("is_too_bright", False):
+            return False
+
+        # Check pose if available
+        if "pose" in metrics and not metrics["pose"].get("is_frontal", True):
+            return False
+
+        return True
+
+
+# Global service instance
+face_quality_service = FaceQualityService()

app/services/face_recognition.py

@@ -0,0 +1,228 @@
+"""
+Face Recognition Service using AuraFace.
+
+This service provides face detection, embedding extraction,
+and face comparison functionality using InsightFace with AuraFace model.
+"""
+
+import numpy as np
+from typing import Dict, Any, Optional, List, Tuple
+from pathlib import Path
+import logging
+
+from ..config import settings
+
+logger = logging.getLogger(__name__)
+
+
+class FaceRecognitionService:
+    """Service for face recognition using AuraFace model."""
+
+    def __init__(self):
+        """Initialize the face recognition service."""
+        self.face_app = None
+        self.initialized = False
+
+    def initialize(self) -> None:
+        """
+        Initialize the face recognition model.
+        Should be called on application startup.
+        """
+        if self.initialized:
+            logger.info("Face recognition service already initialized")
+            return
+
+        try:
+            from insightface.app import FaceAnalysis
+
+            logger.info("Initializing AuraFace model...")
+
+            # Determine provider based on GPU setting
+            if settings.USE_GPU:
+                providers = ["CUDAExecutionProvider", "CPUExecutionProvider"]
+            else:
+                providers = ["CPUExecutionProvider"]
+
+            # Initialize FaceAnalysis with AuraFace
+            self.face_app = FaceAnalysis(
+                name="auraface",
+                root=str(Path(settings.AURAFACE_MODEL_DIR).parent),
+                providers=providers
+            )
+
+            # Prepare the model
+            ctx_id = settings.DEVICE_ID if settings.USE_GPU else -1
+            self.face_app.prepare(ctx_id=ctx_id, det_size=settings.DET_SIZE)
+
+            self.initialized = True
+            logger.info("Face recognition service initialized successfully")
+
+        except Exception as e:
+            logger.error(f"Failed to initialize face recognition service: {e}")
+            raise RuntimeError(f"Face recognition initialization failed: {e}")
+
+    def get_faces(self, image: np.ndarray) -> List[Any]:
+        """
+        Detect faces in image and return face objects.
+
+        Args:
+            image: Input image (BGR format)
+
+        Returns:
+            List of detected face objects
+        """
+        self._ensure_initialized()
+        return self.face_app.get(image)
+
+    def extract_face_info(
+        self,
+        image: np.ndarray,
+        allow_multiple: bool = False
+    ) -> Dict[str, Any]:
+        """
+        Extract face information from image.
+
+        Args:
+            image: Input image (BGR format)
+            allow_multiple: If False, raises error when multiple faces detected
+
+        Returns:
+            Dictionary containing face information
+
+        Raises:
+            ValueError: If no face detected or multiple faces detected (when not allowed)
+        """
+        self._ensure_initialized()
+
+        faces = self.face_app.get(image)
+
+        if not faces:
+            raise ValueError("No face detected in image")
+
+        if len(faces) > 1 and not allow_multiple:
+            raise ValueError(f"Multiple faces detected ({len(faces)}). Expected single face.")
+
+        face = faces[0]
+
+        # Extract bounding box
+        bbox = face.bbox.astype(int)
+        x1, y1, x2, y2 = bbox
+
+        # Build result dictionary
+        result = {
+            "embedding": face.normed_embedding,
+            "bbox": {
+                "x": int(x1),
+                "y": int(y1),
+                "width": int(x2 - x1),
+                "height": int(y2 - y1)
+            },
+            "det_score": float(face.det_score) if hasattr(face, 'det_score') else None,
+            "face_count": len(faces)
+        }
+
+        # Add age if available
+        if hasattr(face, 'age') and face.age is not None:
+            result["age"] = int(face.age)
+
+        # Add gender if available
+        if hasattr(face, 'gender') and face.gender is not None:
+            # Gender: 0 = Female, 1 = Male
+            result["gender"] = "Male" if face.gender == 1 else "Female"
+
+        # Add pose if available (yaw, pitch, roll)
+        if hasattr(face, 'pose') and face.pose is not None:
+            result["pose"] = {
+                "yaw": float(face.pose[1]) if len(face.pose) > 1 else 0.0,
+                "pitch": float(face.pose[0]) if len(face.pose) > 0 else 0.0,
+                "roll": float(face.pose[2]) if len(face.pose) > 2 else 0.0
+            }
+
+        # Add landmarks if available
+        if hasattr(face, 'landmark_2d_106') and face.landmark_2d_106 is not None:
+            result["has_landmarks"] = True
+        elif hasattr(face, 'kps') and face.kps is not None:
+            result["has_landmarks"] = True
+        else:
+            result["has_landmarks"] = False
+
+        return result
+
+    def compare_faces(
+        self,
+        embedding1: np.ndarray,
+        embedding2: np.ndarray,
+        threshold: Optional[float] = None
+    ) -> Dict[str, Any]:
+        """
+        Compare two face embeddings.
+
+        Args:
+            embedding1: First face embedding
+            embedding2: Second face embedding
+            threshold: Similarity threshold (uses default from config if not provided)
+
+        Returns:
+            Dictionary with comparison results
+        """
+        if threshold is None:
+            threshold = settings.FACE_MATCH_THRESHOLD
+
+        # Calculate cosine similarity (embeddings are already normalized)
+        similarity = float(np.dot(embedding1, embedding2))
+
+        return {
+            "is_match": similarity >= threshold,
+            "similarity_score": round(similarity, 4),
+            "threshold": threshold
+        }
+
+    def verify_faces(
+        self,
+        image1: np.ndarray,
+        image2: np.ndarray,
+        threshold: Optional[float] = None
+    ) -> Dict[str, Any]:
+        """
+        Verify if two images contain the same person.
+
+        Args:
+            image1: First image (BGR format)
+            image2: Second image (BGR format)
+            threshold: Similarity threshold
+
+        Returns:
+            Dictionary with verification results and face info
+        """
+        # Extract face info from both images
+        face1_info = self.extract_face_info(image1, allow_multiple=False)
+        face2_info = self.extract_face_info(image2, allow_multiple=False)
+
+        # Compare embeddings
+        comparison = self.compare_faces(
+            face1_info["embedding"],
+            face2_info["embedding"],
+            threshold
+        )
+
+        # Remove embeddings from result (they're large arrays)
+        face1_info.pop("embedding")
+        face2_info.pop("embedding")
+
+        return {
+            "face_match": comparison,
+            "face1": face1_info,
+            "face2": face2_info
+        }
+
+    def _ensure_initialized(self) -> None:
+        """Ensure the service is initialized."""
+        if not self.initialized:
+            raise RuntimeError(
+                "Face recognition service not initialized. "
+                "Call initialize() first or wait for app startup."
+            )
+
+
+# Global service instance
+face_recognition_service = FaceRecognitionService()

app/services/ktp_ocr.py

@@ -0,0 +1,775 @@
+"""
+KTP OCR Service for extracting and parsing Indonesian ID card data.
+
+This service uses EasyOCR to extract text from KTP images and parses
+the extracted text into structured fields with sanitization.
+"""
+
+import re
+import logging
+from typing import Dict, Any, Optional, List, Tuple
+from dataclasses import dataclass, field
+from datetime import datetime
+
+import cv2
+import numpy as np
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class KTPField:
+    """Represents a single KTP field with confidence score."""
+    value: str
+    confidence: float
+    raw_value: str = ""
+
+
+@dataclass
+class KTPData:
+    """Structured KTP data extracted from OCR."""
+    provinsi: Optional[KTPField] = None
+    kabupaten_kota: Optional[KTPField] = None
+    nik: Optional[KTPField] = None
+    nama: Optional[KTPField] = None
+    tempat_lahir: Optional[KTPField] = None
+    tanggal_lahir: Optional[KTPField] = None
+    jenis_kelamin: Optional[KTPField] = None
+    golongan_darah: Optional[KTPField] = None
+    alamat: Optional[KTPField] = None
+    rt_rw: Optional[KTPField] = None
+    kelurahan_desa: Optional[KTPField] = None
+    kecamatan: Optional[KTPField] = None
+    agama: Optional[KTPField] = None
+    status_perkawinan: Optional[KTPField] = None
+    pekerjaan: Optional[KTPField] = None
+    kewarganegaraan: Optional[KTPField] = None
+    berlaku_hingga: Optional[KTPField] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary for API response."""
+        result = {}
+        for field_name in [
+            'provinsi', 'kabupaten_kota', 'nik', 'nama', 'tempat_lahir',
+            'tanggal_lahir', 'jenis_kelamin', 'golongan_darah', 'alamat',
+            'rt_rw', 'kelurahan_desa', 'kecamatan', 'agama', 'status_perkawinan',
+            'pekerjaan', 'kewarganegaraan', 'berlaku_hingga'
+        ]:
+            field_value = getattr(self, field_name)
+            if field_value:
+                result[field_name] = {
+                    'value': field_value.value,
+                    'confidence': field_value.confidence,
+                    'raw_value': field_value.raw_value
+                }
+            else:
+                result[field_name] = None
+        return result
+
+
+class KTPOCRService:
+    """
+    Service for performing OCR on Indonesian KTP (ID card) images.
+
+    Features:
+    - Text extraction using EasyOCR
+    - Field parsing and validation
+    - NIK validation
+    - Data sanitization
+    """
+
+    def __init__(self):
+        self.reader = None
+        self.initialized = False
+
+        # KTP field labels for matching
+        self.field_labels = {
+            'nik': ['NIK', 'N I K', 'NlK'],
+            'nama': ['Nama', 'NAMA', 'Name'],
+            'tempat_tanggal_lahir': ['Tempat/Tgl Lahir', 'Tempat/TglLahir', 'Tempat / Tgl Lahir', 'Tempat/Tgl.Lahir'],
+            'jenis_kelamin': ['Jenis Kelamin', 'Jenis kelamin', 'JenisKelamin', 'JENIS KELAMIN'],
+            'golongan_darah': ['Gol. Darah', 'Gol.Darah', 'Gol Darah', 'GOL. DARAH'],
+            'alamat': ['Alamat', 'ALAMAT', 'Address'],
+            'rt_rw': ['RT/RW', 'RT / RW', 'RTRW'],
+            'kelurahan_desa': ['Kel/Desa', 'Kel / Desa', 'Kelurahan/Desa', 'KEL/DESA'],
+            'kecamatan': ['Kecamatan', 'KECAMATAN', 'Kec'],
+            'agama': ['Agama', 'AGAMA', 'Religion'],
+            'status_perkawinan': ['Status Perkawinan', 'Status perkawinan', 'STATUS PERKAWINAN'],
+            'pekerjaan': ['Pekerjaan', 'PEKERJAAN', 'Occupation'],
+            'kewarganegaraan': ['Kewarganegaraan', 'KEWARGANEGARAAN', 'Nationality'],
+            'berlaku_hingga': ['Berlaku Hingga', 'Berlaku hingga', 'BERLAKU HINGGA', 'Valid Until']
+        }
+
+        # Valid values for certain fields
+        self.valid_genders = ['LAKI-LAKI', 'PEREMPUAN']
+        self.valid_religions = ['ISLAM', 'KRISTEN', 'KATOLIK', 'HINDU', 'BUDDHA', 'KONGHUCU']
+        self.valid_marital_status = ['BELUM KAWIN', 'KAWIN', 'CERAI HIDUP', 'CERAI MATI']
+        self.valid_blood_types = ['A', 'B', 'AB', 'O', 'A+', 'A-', 'B+', 'B-', 'AB+', 'AB-', 'O+', 'O-', '-']
+        self.valid_nationalities = ['WNI', 'WNA', 'INDONESIA']
+
+    def initialize(self) -> None:
+        """Initialize PaddleOCR reader."""
+        if self.initialized:
+            return
+
+        try:
+            from paddleocr import PaddleOCR
+            logger.info("Initializing PaddleOCR reader...")
+            self.reader = PaddleOCR(
+                lang='en',  # Use English (includes Latin characters for Indonesian KTP)
+            )
+            self.initialized = True
+            logger.info("PaddleOCR reader initialized successfully")
+        except Exception as e:
+            logger.error(f"Failed to initialize PaddleOCR: {e}")
+            raise
+
+    def preprocess_image(self, image: np.ndarray) -> np.ndarray:
+        """
+        Preprocess KTP image for better OCR results.
+
+        Args:
+            image: Input image (BGR format)
+
+        Returns:
+            Preprocessed image
+        """
+        # Convert to grayscale
+        if len(image.shape) == 3:
+            gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
+        else:
+            gray = image.copy()
+
+        # Apply CLAHE for contrast enhancement
+        clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
+        enhanced = clahe.apply(gray)
+
+        # Denoise
+        denoised = cv2.fastNlMeansDenoising(enhanced, None, 10, 7, 21)
+
+        # Adaptive thresholding for better text contrast
+        binary = cv2.adaptiveThreshold(
+            denoised, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
+            cv2.THRESH_BINARY, 11, 2
+        )
+
+        # Convert back to BGR for EasyOCR
+        result = cv2.cvtColor(binary, cv2.COLOR_GRAY2BGR)
+
+        return result
+
+    def extract_text(
+        self,
+        image: np.ndarray,
+        preprocess: bool = True
+    ) -> List[Tuple[List[List[int]], str, float]]:
+        """
+        Extract text from KTP image using PaddleOCR.
+
+        Args:
+            image: Input image (BGR format)
+            preprocess: Whether to preprocess the image
+
+        Returns:
+            List of (bounding_box, text, confidence) tuples
+        """
+        if not self.initialized:
+            raise RuntimeError("KTP OCR service not initialized")
+
+        # Preprocess image if requested
+        if preprocess:
+            processed = self.preprocess_image(image)
+        else:
+            processed = image
+
+        # Run PaddleOCR
+        result = self.reader.ocr(processed)
+
+        # Convert PaddleOCR format to expected format
+        # New PaddleOCR returns: [{'rec_texts': [...], 'rec_scores': [...], 'rec_polys': [...]}]
+        results = []
+        if result and len(result) > 0:
+            ocr_result = result[0]
+            texts = ocr_result.get('rec_texts', [])
+            scores = ocr_result.get('rec_scores', [])
+            polys = ocr_result.get('rec_polys', [])
+
+            for i, text in enumerate(texts):
+                bbox = polys[i].tolist() if i < len(polys) else []
+                confidence = scores[i] if i < len(scores) else 0.0
+                results.append((bbox, text, confidence))
+
+        # Also try on original image and merge results
+        if preprocess:
+            original_result = self.reader.ocr(image)
+            original_results = []
+            if original_result and len(original_result) > 0:
+                ocr_result = original_result[0]
+                texts = ocr_result.get('rec_texts', [])
+                scores = ocr_result.get('rec_scores', [])
+                polys = ocr_result.get('rec_polys', [])
+
+                for i, text in enumerate(texts):
+                    bbox = polys[i].tolist() if i < len(polys) else []
+                    confidence = scores[i] if i < len(scores) else 0.0
+                    original_results.append((bbox, text, confidence))
+            # Merge results, preferring higher confidence
+            results = self._merge_ocr_results(results, original_results)
+
+        return results
+
+    def _merge_ocr_results(
+        self,
+        results1: List[Tuple],
+        results2: List[Tuple]
+    ) -> List[Tuple]:
+        """Merge OCR results from two runs, keeping higher confidence."""
+        all_results = results1 + results2
+
+        # Group by similar text and keep highest confidence
+        text_map = {}
+        for bbox, text, conf in all_results:
+            normalized_text = text.upper().strip()
+            if normalized_text not in text_map or text_map[normalized_text][2] < conf:
+                text_map[normalized_text] = (bbox, text, conf)
+
+        return list(text_map.values())
+
+    def parse_ktp_data(
+        self,
+        ocr_results: List[Tuple[List[List[int]], str, float]]
+    ) -> KTPData:
+        """
+        Parse OCR results into structured KTP data.
+
+        Args:
+            ocr_results: List of (bounding_box, text, confidence) tuples
+
+        Returns:
+            Structured KTP data
+        """
+        ktp_data = KTPData()
+
+        # Sort results by vertical position (y-coordinate)
+        sorted_results = sorted(ocr_results, key=lambda x: x[0][0][1] if x[0] else 0)
+
+        # Extract all text lines
+        lines = [(text.strip(), conf) for _, text, conf in sorted_results if text.strip()]
+
+        # Join all text for regex-based extraction
+        full_text = ' '.join([line[0] for line in lines])
+
+        # Extract NIK (16 digits)
+        ktp_data.nik = self._extract_nik(lines, full_text)
+
+        # Extract province and city from header
+        ktp_data.provinsi, ktp_data.kabupaten_kota = self._extract_location(lines)
+
+        # Extract other fields
+        ktp_data.nama = self._extract_field_value(lines, full_text, 'nama')
+
+        # Extract birth place and date
+        birth_info = self._extract_birth_info(lines, full_text)
+        ktp_data.tempat_lahir = birth_info[0]
+        ktp_data.tanggal_lahir = birth_info[1]
+
+        ktp_data.jenis_kelamin = self._extract_gender(lines, full_text)
+        ktp_data.golongan_darah = self._extract_blood_type(lines, full_text)
+        ktp_data.alamat = self._extract_address(lines, full_text)
+        ktp_data.rt_rw = self._extract_rt_rw(lines, full_text)
+        ktp_data.kelurahan_desa = self._extract_field_value(lines, full_text, 'kelurahan_desa')
+        ktp_data.kecamatan = self._extract_field_value(lines, full_text, 'kecamatan')
+        ktp_data.agama = self._extract_religion(lines, full_text)
+        ktp_data.status_perkawinan = self._extract_marital_status(lines, full_text)
+        ktp_data.pekerjaan = self._extract_field_value(lines, full_text, 'pekerjaan')
+        ktp_data.kewarganegaraan = self._extract_nationality(lines, full_text)
+        ktp_data.berlaku_hingga = self._extract_validity(lines, full_text)
+
+        return ktp_data
+
+    def _extract_nik(
+        self,
+        lines: List[Tuple[str, float]],
+        full_text: str
+    ) -> Optional[KTPField]:
+        """Extract NIK (16-digit ID number)."""
+        # Pattern for NIK: 16 consecutive digits
+        nik_pattern = r'\b(\d{16})\b'
+
+        for line_text, conf in lines:
+            # Clean the text
+            cleaned = re.sub(r'[^\d]', '', line_text)
+            if len(cleaned) == 16:
+                return KTPField(
+                    value=cleaned,
+                    confidence=conf,
+                    raw_value=line_text
+                )
+
+        # Try from full text
+        match = re.search(nik_pattern, re.sub(r'\s', '', full_text))
+        if match:
+            return KTPField(
+                value=match.group(1),
+                confidence=0.7,  # Lower confidence for pattern match
+                raw_value=match.group(1)
+            )
+
+        return None
+
+    def _extract_location(
+        self,
+        lines: List[Tuple[str, float]]
+    ) -> Tuple[Optional[KTPField], Optional[KTPField]]:
+        """Extract province and city from KTP header."""
+        provinsi = None
+        kab_kota = None
+
+        for i, (line_text, conf) in enumerate(lines[:5]):  # Check first 5 lines
+            upper_text = line_text.upper()
+
+            # Look for "PROVINSI" keyword
+            if 'PROVINSI' in upper_text:
+                # Extract province name
+                prov_match = re.search(r'PROVINSI\s*[:\.]?\s*(.+)', upper_text)
+                if prov_match:
+                    provinsi = KTPField(
+                        value=self._sanitize_text(prov_match.group(1)),
+                        confidence=conf,
+                        raw_value=line_text
+                    )
+                elif i + 1 < len(lines):
+                    # Province name might be on next line
+                    provinsi = KTPField(
+                        value=self._sanitize_text(lines[i + 1][0]),
+                        confidence=lines[i + 1][1],
+                        raw_value=lines[i + 1][0]
+                    )
+
+            # Look for "KABUPATEN" or "KOTA"
+            if 'KABUPATEN' in upper_text or 'KOTA' in upper_text:
+                kab_match = re.search(r'(KABUPATEN|KOTA)\s*[:\.]?\s*(.+)', upper_text)
+                if kab_match:
+                    kab_kota = KTPField(
+                        value=self._sanitize_text(kab_match.group(0)),
+                        confidence=conf,
+                        raw_value=line_text
+                    )
+
+        return provinsi, kab_kota
+
+    def _extract_birth_info(
+        self,
+        lines: List[Tuple[str, float]],
+        full_text: str
+    ) -> Tuple[Optional[KTPField], Optional[KTPField]]:
+        """Extract birth place and date."""
+        tempat_lahir = None
+        tanggal_lahir = None
+
+        # Date pattern: DD-MM-YYYY or DD/MM/YYYY
+        date_pattern = r'(\d{2}[-/]\d{2}[-/]\d{4})'
+
+        for line_text, conf in lines:
+            upper_text = line_text.upper()
+
+            # Look for birth info line
+            if any(label.upper() in upper_text for label in self.field_labels.get('tempat_tanggal_lahir', [])):
+                # Extract after the label
+                for label in self.field_labels['tempat_tanggal_lahir']:
+                    if label.upper() in upper_text:
+                        rest = upper_text.split(label.upper())[-1].strip()
+                        rest = re.sub(r'^[:\s]+', '', rest)
+
+                        # Find date in the rest
+                        date_match = re.search(date_pattern, rest)
+                        if date_match:
+                            date_str = date_match.group(1)
+                            place = rest[:date_match.start()].strip().rstrip(',')
+
+                            tempat_lahir = KTPField(
+                                value=self._sanitize_text(place),
+                                confidence=conf,
+                                raw_value=place
+                            )
+                            tanggal_lahir = KTPField(
+                                value=self._sanitize_date(date_str),
+                                confidence=conf,
+                                raw_value=date_str
+                            )
+                        break
+
+            # Also check for standalone date
+            if not tanggal_lahir:
+                date_match = re.search(date_pattern, line_text)
+                if date_match:
+                    tanggal_lahir = KTPField(
+                        value=self._sanitize_date(date_match.group(1)),
+                        confidence=conf,
+                        raw_value=date_match.group(1)
+                    )
+
+        return tempat_lahir, tanggal_lahir
+
+    def _extract_gender(
+        self,
+        lines: List[Tuple[str, float]],
+        full_text: str
+    ) -> Optional[KTPField]:
+        """Extract gender (Jenis Kelamin)."""
+        for line_text, conf in lines:
+            upper_text = line_text.upper()
+
+            for valid_gender in self.valid_genders:
+                if valid_gender in upper_text:
+                    return KTPField(
+                        value=valid_gender,
+                        confidence=conf,
+                        raw_value=line_text
+                    )
+
+        return None
+
+    def _extract_blood_type(
+        self,
+        lines: List[Tuple[str, float]],
+        full_text: str
+    ) -> Optional[KTPField]:
+        """Extract blood type (Golongan Darah)."""
+        for line_text, conf in lines:
+            upper_text = line_text.upper()
+
+            # Look for blood type field
+            if any(label.upper() in upper_text for label in self.field_labels.get('golongan_darah', [])):
+                for blood_type in self.valid_blood_types:
+                    if blood_type in upper_text:
+                        return KTPField(
+                            value=blood_type,
+                            confidence=conf,
+                            raw_value=line_text
+                        )
+
+        return None
+
+    def _extract_address(
+        self,
+        lines: List[Tuple[str, float]],
+        full_text: str
+    ) -> Optional[KTPField]:
+        """Extract address (Alamat)."""
+        for i, (line_text, conf) in enumerate(lines):
+            upper_text = line_text.upper()
+
+            if any(label.upper() in upper_text for label in self.field_labels.get('alamat', [])):
+                # Get the address part after the label
+                for label in self.field_labels['alamat']:
+                    if label.upper() in upper_text:
+                        rest = upper_text.split(label.upper())[-1].strip()
+                        rest = re.sub(r'^[:\s]+', '', rest)
+
+                        if rest:
+                            return KTPField(
+                                value=self._sanitize_text(rest),
+                                confidence=conf,
+                                raw_value=line_text
+                            )
+                        # Address might be on next line
+                        elif i + 1 < len(lines):
+                            next_line = lines[i + 1]
+                            return KTPField(
+                                value=self._sanitize_text(next_line[0]),
+                                confidence=next_line[1],
+                                raw_value=next_line[0]
+                            )
+
+        return None
+
+    def _extract_rt_rw(
+        self,
+        lines: List[Tuple[str, float]],
+        full_text: str
+    ) -> Optional[KTPField]:
+        """Extract RT/RW."""
+        rt_rw_pattern = r'(\d{3})\s*/\s*(\d{3})'
+
+        for line_text, conf in lines:
+            match = re.search(rt_rw_pattern, line_text)
+            if match:
+                value = f"{match.group(1)}/{match.group(2)}"
+                return KTPField(
+                    value=value,
+                    confidence=conf,
+                    raw_value=line_text
+                )
+
+        return None
+
+    def _extract_religion(
+        self,
+        lines: List[Tuple[str, float]],
+        full_text: str
+    ) -> Optional[KTPField]:
+        """Extract religion (Agama)."""
+        for line_text, conf in lines:
+            upper_text = line_text.upper()
+
+            for religion in self.valid_religions:
+                if religion in upper_text:
+                    return KTPField(
+                        value=religion,
+                        confidence=conf,
+                        raw_value=line_text
+                    )
+
+        return None
+
+    def _extract_marital_status(
+        self,
+        lines: List[Tuple[str, float]],
+        full_text: str
+    ) -> Optional[KTPField]:
+        """Extract marital status (Status Perkawinan)."""
+        for line_text, conf in lines:
+            upper_text = line_text.upper()
+
+            for status in self.valid_marital_status:
+                if status in upper_text:
+                    return KTPField(
+                        value=status,
+                        confidence=conf,
+                        raw_value=line_text
+                    )
+
+        return None
+
+    def _extract_nationality(
+        self,
+        lines: List[Tuple[str, float]],
+        full_text: str
+    ) -> Optional[KTPField]:
+        """Extract nationality (Kewarganegaraan)."""
+        for line_text, conf in lines:
+            upper_text = line_text.upper()
+
+            for nationality in self.valid_nationalities:
+                if nationality in upper_text:
+                    return KTPField(
+                        value=nationality if nationality != 'INDONESIA' else 'WNI',
+                        confidence=conf,
+                        raw_value=line_text
+                    )
+
+        return None
+
+    def _extract_validity(
+        self,
+        lines: List[Tuple[str, float]],
+        full_text: str
+    ) -> Optional[KTPField]:
+        """Extract validity period (Berlaku Hingga)."""
+        for line_text, conf in lines:
+            upper_text = line_text.upper()
+
+            if any(label.upper() in upper_text for label in self.field_labels.get('berlaku_hingga', [])):
+                # Check for "SEUMUR HIDUP"
+                if 'SEUMUR HIDUP' in upper_text:
+                    return KTPField(
+                        value='SEUMUR HIDUP',
+                        confidence=conf,
+                        raw_value=line_text
+                    )
+
+                # Check for date
+                date_pattern = r'(\d{2}[-/]\d{2}[-/]\d{4})'
+                date_match = re.search(date_pattern, line_text)
+                if date_match:
+                    return KTPField(
+                        value=self._sanitize_date(date_match.group(1)),
+                        confidence=conf,
+                        raw_value=line_text
+                    )
+
+        return None
+
+    def _extract_field_value(
+        self,
+        lines: List[Tuple[str, float]],
+        full_text: str,
+        field_name: str
+    ) -> Optional[KTPField]:
+        """Generic field value extraction."""
+        labels = self.field_labels.get(field_name, [])
+
+        for i, (line_text, conf) in enumerate(lines):
+            for label in labels:
+                if label.upper() in line_text.upper():
+                    # Get value after label
+                    rest = line_text.upper().split(label.upper())[-1].strip()
+                    rest = re.sub(r'^[:\s]+', '', rest)
+
+                    if rest:
+                        return KTPField(
+                            value=self._sanitize_text(rest),
+                            confidence=conf,
+                            raw_value=line_text
+                        )
+                    # Value might be on next line
+                    elif i + 1 < len(lines):
+                        next_line = lines[i + 1]
+                        return KTPField(
+                            value=self._sanitize_text(next_line[0]),
+                            confidence=next_line[1],
+                            raw_value=next_line[0]
+                        )
+
+        return None
+
+    def _sanitize_text(self, text: str) -> str:
+        """Sanitize extracted text."""
+        if not text:
+            return ""
+
+        # Remove extra whitespace
+        text = ' '.join(text.split())
+
+        # Remove leading/trailing punctuation
+        text = text.strip('.:,;-_')
+
+        # Convert to title case for names
+        text = text.strip()
+
+        return text
+
+    def _sanitize_date(self, date_str: str) -> str:
+        """Sanitize and standardize date format to DD-MM-YYYY."""
+        if not date_str:
+            return ""
+
+        # Replace / with -
+        date_str = date_str.replace('/', '-')
+
+        return date_str
+
+    def validate_nik(self, nik: str) -> Dict[str, Any]:
+        """
+        Validate NIK and extract encoded information.
+
+        NIK Format: PPKKCC-DDMMYY-XXXX
+        - PP: Province code (2 digits)
+        - KK: City/Regency code (2 digits)
+        - CC: District code (2 digits)
+        - DD: Birth date (01-31, add 40 for females)
+        - MM: Birth month (01-12)
+        - YY: Birth year (last 2 digits)
+        - XXXX: Sequence number (4 digits)
+
+        Args:
+            nik: NIK string (16 digits)
+
+        Returns:
+            Validation result with extracted info
+        """
+        result = {
+            'is_valid': False,
+            'errors': [],
+            'extracted': {}
+        }
+
+        # Clean NIK
+        nik = re.sub(r'[^\d]', '', nik)
+
+        # Check length
+        if len(nik) != 16:
+            result['errors'].append(f"Invalid length: {len(nik)} (expected 16)")
+            return result
+
+        try:
+            # Extract components
+            province_code = nik[0:2]
+            city_code = nik[2:4]
+            district_code = nik[4:6]
+            birth_day = int(nik[6:8])
+            birth_month = int(nik[8:10])
+            birth_year = int(nik[10:12])
+            sequence = nik[12:16]
+
+            # Determine gender from birth day
+            gender = 'PEREMPUAN' if birth_day > 40 else 'LAKI-LAKI'
+            actual_day = birth_day - 40 if birth_day > 40 else birth_day
+
+            # Validate birth date
+            if actual_day < 1 or actual_day > 31:
+                result['errors'].append(f"Invalid birth day: {actual_day}")
+
+            if birth_month < 1 or birth_month > 12:
+                result['errors'].append(f"Invalid birth month: {birth_month}")
+
+            # Determine full birth year (assume 19xx for > 30, 20xx for <= 30)
+            current_year = datetime.now().year % 100
+            if birth_year > current_year:
+                full_year = 1900 + birth_year
+            else:
+                full_year = 2000 + birth_year
+
+            result['extracted'] = {
+                'province_code': province_code,
+                'city_code': city_code,
+                'district_code': district_code,
+                'birth_date': f"{actual_day:02d}-{birth_month:02d}-{full_year}",
+                'gender': gender,
+                'sequence': sequence
+            }
+
+            result['is_valid'] = len(result['errors']) == 0
+
+        except Exception as e:
+            result['errors'].append(f"Parsing error: {str(e)}")
+
+        return result
+
+    def extract_ktp_data(
+        self,
+        image: np.ndarray,
+        validate: bool = True
+    ) -> Dict[str, Any]:
+        """
+        Extract and parse all KTP data from image.
+
+        Args:
+            image: Input image (BGR format)
+            validate: Whether to validate extracted data
+
+        Returns:
+            Dictionary with extracted data, raw OCR results, and validation
+        """
+        # Run OCR
+        ocr_results = self.extract_text(image)
+
+        # Parse into structured data
+        ktp_data = self.parse_ktp_data(ocr_results)
+
+        # Build response
+        response = {
+            'data': ktp_data.to_dict(),
+            'raw_text': [
+                {
+                    'text': text,
+                    'confidence': conf,
+                    'bbox': bbox
+                }
+                for bbox, text, conf in ocr_results
+            ],
+            'validation': None
+        }
+
+        # Validate NIK if found and validation requested
+        if validate and ktp_data.nik:
+            response['validation'] = {
+                'nik': self.validate_nik(ktp_data.nik.value)
+            }
+
+        return response
+
+
+# Global service instance
+ktp_ocr_service = KTPOCRService()

app/services/liveness_detection.py

@@ -0,0 +1,204 @@
+"""
+Liveness Detection Service using Silent-Face-Anti-Spoofing.
+
+This service detects whether a face image is from a real person
+or a spoofing attempt (photo, video, mask, etc.).
+"""
+
+import os
+import sys
+import numpy as np
+import cv2
+from typing import Dict, Any, Optional
+from pathlib import Path
+import logging
+
+from ..config import settings
+
+logger = logging.getLogger(__name__)
+
+
+class LivenessDetectionService:
+    """Service for detecting face liveness (anti-spoofing)."""
+
+    def __init__(self):
+        """Initialize the liveness detection service."""
+        self.model = None
+        self.image_cropper = None
+        self.initialized = False
+        self._models_info = []
+
+    def initialize(self) -> None:
+        """
+        Initialize the liveness detection models.
+        Should be called on application startup.
+        """
+        if self.initialized:
+            logger.info("Liveness detection service already initialized")
+            return
+
+        try:
+            # Add Silent-Face-Anti-Spoofing to path
+            repo_path = Path(settings.SILENT_FACE_REPO_DIR)
+            if repo_path.exists():
+                sys.path.insert(0, str(repo_path))
+
+            from src.anti_spoof_predict import AntiSpoofPredict
+            from src.generate_patches import CropImage
+
+            logger.info("Initializing liveness detection models...")
+
+            # Initialize predictor
+            device_id = settings.DEVICE_ID if settings.USE_GPU else 0
+            self.model = AntiSpoofPredict(device_id)
+            self.image_cropper = CropImage()
+
+            # Verify model files exist
+            model_dir = Path(settings.ANTISPOOF_MODEL_DIR) / "anti_spoof_models"
+            if model_dir.exists():
+                self._models_info = list(model_dir.glob("*.pth"))
+                logger.info(f"Found {len(self._models_info)} anti-spoof models")
+            else:
+                logger.warning(f"Anti-spoof models directory not found: {model_dir}")
+
+            self.initialized = True
+            logger.info("Liveness detection service initialized successfully")
+
+        except ImportError as e:
+            logger.error(f"Failed to import Silent-Face-Anti-Spoofing: {e}")
+            logger.warning("Liveness detection will not be available")
+            self.initialized = False
+        except Exception as e:
+            logger.error(f"Failed to initialize liveness detection: {e}")
+            raise RuntimeError(f"Liveness detection initialization failed: {e}")
+
+    def check_liveness(
+        self,
+        image: np.ndarray,
+        threshold: Optional[float] = None
+    ) -> Dict[str, Any]:
+        """
+        Check if the face in the image is real or spoofed.
+
+        Args:
+            image: Input image (BGR format)
+            threshold: Confidence threshold for classification
+
+        Returns:
+            Dictionary with liveness detection results
+        """
+        self._ensure_initialized()
+
+        if threshold is None:
+            threshold = settings.LIVENESS_THRESHOLD
+
+        try:
+            # Import utilities
+            from src.utility import parse_model_name
+
+            # Get face bounding box
+            image_bbox = self.model.get_bbox(image)
+
+            if image_bbox is None:
+                return {
+                    "is_real": False,
+                    "confidence": 0.0,
+                    "label": "No Face Detected",
+                    "error": "No face detected in image"
+                }
+
+            # Get model directory
+            model_dir = Path(settings.ANTISPOOF_MODEL_DIR) / "anti_spoof_models"
+
+            # Accumulate predictions from all models
+            prediction = np.zeros((1, 3))
+            model_count = 0
+
+            for model_name in os.listdir(model_dir):
+                if not model_name.endswith(".pth"):
+                    continue
+
+                try:
+                    # Parse model parameters from filename
+                    h_input, w_input, model_type, scale = parse_model_name(model_name)
+
+                    # Crop face patch according to model requirements
+                    param = {
+                        "org_img": image,
+                        "bbox": image_bbox,
+                        "scale": scale,
+                        "out_w": w_input,
+                        "out_h": h_input,
+                        "crop": True,
+                    }
+
+                    if scale is not None:
+                        img_patch = self.image_cropper.crop(**param)
+                    else:
+                        img_patch = image
+
+                    # Run prediction
+                    model_path = os.path.join(str(model_dir), model_name)
+                    prediction += self.model.predict(img_patch, model_path)
+                    model_count += 1
+
+                except Exception as e:
+                    logger.warning(f"Error processing model {model_name}: {e}")
+                    continue
+
+            if model_count == 0:
+                return {
+                    "is_real": False,
+                    "confidence": 0.0,
+                    "label": "Model Error",
+                    "error": "No models could process the image"
+                }
+
+            # Get final prediction
+            # Label: 1 = Real, 0 or 2 = Fake
+            label = np.argmax(prediction)
+            confidence = float(prediction[0][label] / model_count)
+
+            is_real = label == 1
+
+            return {
+                "is_real": is_real,
+                "confidence": round(confidence, 4),
+                "label": "Real Face" if is_real else "Fake Face",
+                "prediction_class": int(label),
+                "models_used": model_count
+            }
+
+        except Exception as e:
+            logger.error(f"Liveness detection error: {e}")
+            return {
+                "is_real": False,
+                "confidence": 0.0,
+                "label": "Error",
+                "error": str(e)
+            }
+
+    def check_liveness_simple(self, image: np.ndarray) -> bool:
+        """
+        Simple liveness check returning only boolean.
+
+        Args:
+            image: Input image (BGR format)
+
+        Returns:
+            True if face is real, False otherwise
+        """
+        result = self.check_liveness(image)
+        return result.get("is_real", False)
+
+    def _ensure_initialized(self) -> None:
+        """Ensure the service is initialized."""
+        if not self.initialized:
+            raise RuntimeError(
+                "Liveness detection service not initialized. "
+                "Call initialize() first or wait for app startup."
+            )
+
+
+# Global service instance
+liveness_detection_service = LivenessDetectionService()

app/utils/__init__.py

@@ -0,0 +1 @@
+# Utils Package

app/utils/image_utils.py

@@ -0,0 +1,214 @@
+"""
+Image processing utilities for KYC POC.
+"""
+
+import cv2
+import numpy as np
+import base64
+from typing import Optional, Tuple
+from fastapi import UploadFile, HTTPException
+
+
+async def read_image_from_upload(file: UploadFile) -> np.ndarray:
+    """
+    Read uploaded image file into numpy array (OpenCV BGR format).
+
+    Args:
+        file: FastAPI UploadFile object
+
+    Returns:
+        numpy array in BGR format (OpenCV)
+
+    Raises:
+        HTTPException: If image is invalid or cannot be decoded
+    """
+    contents = await file.read()
+    return decode_image_bytes(contents)
+
+
+def decode_image_bytes(image_bytes: bytes) -> np.ndarray:
+    """
+    Decode image bytes to numpy array.
+
+    Args:
+        image_bytes: Raw image bytes
+
+    Returns:
+        numpy array in BGR format (OpenCV)
+
+    Raises:
+        HTTPException: If image cannot be decoded
+    """
+    nparr = np.frombuffer(image_bytes, np.uint8)
+    image = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
+
+    if image is None:
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "error_code": "IMAGE_INVALID",
+                "message": "Failed to decode image. Please ensure the file is a valid image."
+            }
+        )
+
+    return image
+
+
+def decode_base64_image(base64_string: str) -> np.ndarray:
+    """
+    Decode base64 encoded image string to numpy array.
+
+    Args:
+        base64_string: Base64 encoded image string (with or without data URI prefix)
+
+    Returns:
+        numpy array in BGR format (OpenCV)
+
+    Raises:
+        HTTPException: If base64 string is invalid or image cannot be decoded
+    """
+    try:
+        # Remove data URI prefix if present
+        if "," in base64_string:
+            base64_string = base64_string.split(",")[1]
+
+        # Decode base64
+        image_bytes = base64.b64decode(base64_string)
+        return decode_image_bytes(image_bytes)
+
+    except Exception as e:
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "error_code": "IMAGE_INVALID",
+                "message": f"Failed to decode base64 image: {str(e)}"
+            }
+        )
+
+
+def encode_image_to_base64(image: np.ndarray, format: str = ".jpg") -> str:
+    """
+    Encode numpy array image to base64 string.
+
+    Args:
+        image: numpy array in BGR format
+        format: Image format (.jpg, .png)
+
+    Returns:
+        Base64 encoded string
+    """
+    _, buffer = cv2.imencode(format, image)
+    return base64.b64encode(buffer).decode("utf-8")
+
+
+def resize_image(
+    image: np.ndarray,
+    max_size: int = 1024,
+    keep_aspect: bool = True
+) -> np.ndarray:
+    """
+    Resize image if it exceeds max size.
+
+    Args:
+        image: Input image
+        max_size: Maximum dimension size
+        keep_aspect: Whether to keep aspect ratio
+
+    Returns:
+        Resized image
+    """
+    height, width = image.shape[:2]
+
+    if max(height, width) <= max_size:
+        return image
+
+    if keep_aspect:
+        if width > height:
+            new_width = max_size
+            new_height = int(height * max_size / width)
+        else:
+            new_height = max_size
+            new_width = int(width * max_size / height)
+    else:
+        new_width = max_size
+        new_height = max_size
+
+    return cv2.resize(image, (new_width, new_height), interpolation=cv2.INTER_AREA)
+
+
+def crop_face_region(
+    image: np.ndarray,
+    bbox: Tuple[int, int, int, int],
+    padding: float = 0.2
+) -> np.ndarray:
+    """
+    Crop face region from image with padding.
+
+    Args:
+        image: Input image
+        bbox: Face bounding box (x1, y1, x2, y2)
+        padding: Padding ratio to add around face
+
+    Returns:
+        Cropped face image
+    """
+    height, width = image.shape[:2]
+    x1, y1, x2, y2 = bbox
+
+    # Calculate padding
+    face_width = x2 - x1
+    face_height = y2 - y1
+    pad_x = int(face_width * padding)
+    pad_y = int(face_height * padding)
+
+    # Apply padding with bounds checking
+    x1 = max(0, x1 - pad_x)
+    y1 = max(0, y1 - pad_y)
+    x2 = min(width, x2 + pad_x)
+    y2 = min(height, y2 + pad_y)
+
+    return image[y1:y2, x1:x2]
+
+
+def validate_image_size(image_bytes: bytes, max_size_bytes: int) -> None:
+    """
+    Validate image size doesn't exceed maximum.
+
+    Args:
+        image_bytes: Image bytes
+        max_size_bytes: Maximum allowed size in bytes
+
+    Raises:
+        HTTPException: If image exceeds size limit
+    """
+    if len(image_bytes) > max_size_bytes:
+        max_mb = max_size_bytes / (1024 * 1024)
+        actual_mb = len(image_bytes) / (1024 * 1024)
+        raise HTTPException(
+            status_code=413,
+            detail={
+                "error_code": "IMAGE_TOO_LARGE",
+                "message": f"Image size ({actual_mb:.2f}MB) exceeds maximum allowed ({max_mb:.2f}MB)"
+            }
+        )
+
+
+def validate_content_type(content_type: Optional[str], allowed_types: list) -> None:
+    """
+    Validate image content type.
+
+    Args:
+        content_type: MIME type of the file
+        allowed_types: List of allowed MIME types
+
+    Raises:
+        HTTPException: If content type is not allowed
+    """
+    if content_type not in allowed_types:
+        raise HTTPException(
+            status_code=415,
+            detail={
+                "error_code": "UNSUPPORTED_FORMAT",
+                "message": f"Unsupported image format: {content_type}. Allowed: {allowed_types}"
+            }
+        )

app/utils/ktp_extractor.py

@@ -0,0 +1,154 @@
+"""
+KTP (Indonesian ID Card) face extraction utility.
+
+This module provides functionality to detect and extract the face
+photo from a KTP card image.
+"""
+
+import cv2
+import numpy as np
+from typing import Optional, Tuple, Dict, Any
+
+
+class KTPFaceExtractor:
+    """Extracts face from KTP (Indonesian ID card) images."""
+
+    def __init__(self, face_detector=None):
+        """
+        Initialize KTP face extractor.
+
+        Args:
+            face_detector: Face detector instance (from face_recognition service)
+        """
+        self.face_detector = face_detector
+
+    def set_detector(self, face_detector):
+        """Set the face detector instance."""
+        self.face_detector = face_detector
+
+    def extract_face(
+        self,
+        ktp_image: np.ndarray,
+        padding: float = 0.3
+    ) -> Tuple[np.ndarray, Dict[str, Any]]:
+        """
+        Extract face from KTP image.
+
+        Args:
+            ktp_image: KTP image as numpy array (BGR)
+            padding: Padding ratio around detected face
+
+        Returns:
+            Tuple of (cropped_face_image, face_info_dict)
+
+        Raises:
+            ValueError: If no face detected or multiple faces found
+        """
+        if self.face_detector is None:
+            raise RuntimeError("Face detector not initialized. Call set_detector first.")
+
+        # Detect faces in KTP image
+        faces = self.face_detector.get(ktp_image)
+
+        if not faces:
+            raise ValueError("No face detected in KTP image")
+
+        if len(faces) > 1:
+            raise ValueError(f"Multiple faces ({len(faces)}) detected in KTP image")
+
+        face = faces[0]
+
+        # Get bounding box
+        bbox = face.bbox.astype(int)
+        x1, y1, x2, y2 = bbox
+
+        # Apply padding
+        height, width = ktp_image.shape[:2]
+        face_width = x2 - x1
+        face_height = y2 - y1
+        pad_x = int(face_width * padding)
+        pad_y = int(face_height * padding)
+
+        # Expand bounding box with padding (with bounds checking)
+        x1_padded = max(0, x1 - pad_x)
+        y1_padded = max(0, y1 - pad_y)
+        x2_padded = min(width, x2 + pad_x)
+        y2_padded = min(height, y2 + pad_y)
+
+        # Crop face region
+        cropped_face = ktp_image[y1_padded:y2_padded, x1_padded:x2_padded]
+
+        # Build face info
+        face_info = {
+            "bbox": {
+                "x": int(x1),
+                "y": int(y1),
+                "width": int(x2 - x1),
+                "height": int(y2 - y1)
+            },
+            "bbox_padded": {
+                "x": int(x1_padded),
+                "y": int(y1_padded),
+                "width": int(x2_padded - x1_padded),
+                "height": int(y2_padded - y1_padded)
+            },
+            "det_score": float(face.det_score) if hasattr(face, 'det_score') else None
+        }
+
+        return cropped_face, face_info
+
+    def extract_face_with_fallback(
+        self,
+        ktp_image: np.ndarray,
+        padding: float = 0.3
+    ) -> Tuple[np.ndarray, Dict[str, Any], bool]:
+        """
+        Extract face from KTP with fallback to full image if detection fails.
+
+        Args:
+            ktp_image: KTP image as numpy array (BGR)
+            padding: Padding ratio around detected face
+
+        Returns:
+            Tuple of (image, face_info, is_face_detected)
+        """
+        try:
+            cropped, info = self.extract_face(ktp_image, padding)
+            return cropped, info, True
+        except ValueError:
+            # Fallback: return the whole image
+            height, width = ktp_image.shape[:2]
+            info = {
+                "bbox": {"x": 0, "y": 0, "width": width, "height": height},
+                "bbox_padded": {"x": 0, "y": 0, "width": width, "height": height},
+                "det_score": None,
+                "warning": "Face not detected, using full image"
+            }
+            return ktp_image, info, False
+
+
+def preprocess_ktp_image(image: np.ndarray) -> np.ndarray:
+    """
+    Preprocess KTP image for better face detection.
+
+    Args:
+        image: Input KTP image
+
+    Returns:
+        Preprocessed image
+    """
+    # Convert to grayscale for processing
+    if len(image.shape) == 3:
+        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
+    else:
+        gray = image
+
+    # Apply CLAHE (Contrast Limited Adaptive Histogram Equalization)
+    clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
+    enhanced = clahe.apply(gray)
+
+    # Convert back to BGR if original was color
+    if len(image.shape) == 3:
+        enhanced = cv2.cvtColor(enhanced, cv2.COLOR_GRAY2BGR)
+
+    return enhanced

requirements.txt

@@ -0,0 +1,16 @@
+fastapi>=0.104.0
+uvicorn[standard]>=0.24.0
+python-multipart>=0.0.6
+opencv-python>=4.8.0
+numpy>=1.26.0,<2.0.0
+insightface>=0.7.3
+huggingface-hub>=0.19.0
+onnxruntime>=1.18.0
+torch>=2.2.0
+torchvision>=0.17.0
+scikit-learn>=1.3.0
+pydantic>=2.0.0
+pydantic-settings>=2.0.0
+paddlepaddle>=2.6.0
+paddleocr>=2.7.0
+rapidfuzz>=3.0.0

setup_models.py

@@ -0,0 +1,168 @@
+"""
+Model Setup Script for KYC POC
+
+This script downloads and sets up the required ML models:
+1. AuraFace - Face recognition model from HuggingFace
+2. Silent-Face-Anti-Spoofing - Liveness detection models from GitHub
+
+Run this script before starting the application:
+    python setup_models.py
+"""
+
+import os
+import sys
+import shutil
+import subprocess
+from pathlib import Path
+
+
+def setup_auraface():
+    """Download AuraFace model from HuggingFace."""
+    print("=" * 50)
+    print("Setting up AuraFace model...")
+    print("=" * 50)
+
+    try:
+        from huggingface_hub import snapshot_download
+
+        model_dir = Path("models/auraface")
+        model_dir.mkdir(parents=True, exist_ok=True)
+
+        print("Downloading AuraFace-v1 from HuggingFace...")
+        snapshot_download(
+            repo_id="fal/AuraFace-v1",
+            local_dir=str(model_dir),
+            local_dir_use_symlinks=False
+        )
+        print(f"AuraFace model downloaded to: {model_dir}")
+        return True
+
+    except ImportError:
+        print("ERROR: huggingface_hub not installed. Run: pip install huggingface-hub")
+        return False
+    except Exception as e:
+        print(f"ERROR downloading AuraFace: {e}")
+        return False
+
+
+def setup_silent_face_anti_spoofing():
+    """Clone Silent-Face-Anti-Spoofing repository and copy models."""
+    print("\n" + "=" * 50)
+    print("Setting up Silent-Face-Anti-Spoofing...")
+    print("=" * 50)
+
+    repo_dir = Path("Silent-Face-Anti-Spoofing")
+    models_dir = Path("models/anti_spoof")
+
+    # Clone repository if not exists
+    if not repo_dir.exists():
+        print("Cloning Silent-Face-Anti-Spoofing repository...")
+        try:
+            result = subprocess.run(
+                ["git", "clone", "https://github.com/minivision-ai/Silent-Face-Anti-Spoofing.git"],
+                capture_output=True,
+                text=True
+            )
+            if result.returncode != 0:
+                print(f"ERROR cloning repository: {result.stderr}")
+                return False
+            print("Repository cloned successfully.")
+        except FileNotFoundError:
+            print("ERROR: git not found. Please install git and try again.")
+            return False
+    else:
+        print("Repository already exists, skipping clone.")
+
+    # Copy model files
+    models_dir.mkdir(parents=True, exist_ok=True)
+
+    # Copy anti_spoof_models
+    src_anti_spoof = repo_dir / "resources" / "anti_spoof_models"
+    dst_anti_spoof = models_dir / "anti_spoof_models"
+
+    if src_anti_spoof.exists():
+        if dst_anti_spoof.exists():
+            shutil.rmtree(dst_anti_spoof)
+        shutil.copytree(src_anti_spoof, dst_anti_spoof)
+        print(f"Copied anti_spoof_models to: {dst_anti_spoof}")
+    else:
+        print(f"WARNING: {src_anti_spoof} not found")
+
+    # Copy detection_model
+    src_detection = repo_dir / "resources" / "detection_model"
+    dst_detection = models_dir / "detection_model"
+
+    if src_detection.exists():
+        if dst_detection.exists():
+            shutil.rmtree(dst_detection)
+        shutil.copytree(src_detection, dst_detection)
+        print(f"Copied detection_model to: {dst_detection}")
+    else:
+        print(f"WARNING: {src_detection} not found")
+
+    return True
+
+
+def verify_models():
+    """Verify all required model files exist."""
+    print("\n" + "=" * 50)
+    print("Verifying model files...")
+    print("=" * 50)
+
+    required_files = [
+        # AuraFace models (these are in the auraface directory after download)
+        "models/auraface",
+        # Anti-spoofing models
+        "models/anti_spoof/anti_spoof_models",
+        "models/anti_spoof/detection_model",
+    ]
+
+    all_exist = True
+    for file_path in required_files:
+        path = Path(file_path)
+        exists = path.exists()
+        status = "OK" if exists else "MISSING"
+        print(f"  [{status}] {file_path}")
+        if not exists:
+            all_exist = False
+
+    return all_exist
+
+
+def main():
+    """Main setup function."""
+    print("\n" + "#" * 60)
+    print("# KYC POC - Model Setup")
+    print("#" * 60 + "\n")
+
+    # Change to script directory
+    script_dir = Path(__file__).parent
+    os.chdir(script_dir)
+
+    success = True
+
+    # Setup AuraFace
+    if not setup_auraface():
+        success = False
+
+    # Setup Silent-Face-Anti-Spoofing
+    if not setup_silent_face_anti_spoofing():
+        success = False
+
+    # Verify all models
+    if not verify_models():
+        success = False
+
+    print("\n" + "#" * 60)
+    if success:
+        print("# Setup completed successfully!")
+        print("# You can now run the application with: uvicorn app.main:app --reload")
+    else:
+        print("# Setup completed with errors. Please check the messages above.")
+    print("#" * 60 + "\n")
+
+    return 0 if success else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())