app.py

"""
Face Verification System - Main Application
Supports user registration with profile picture upload and live face verification
"""

import os
import cv2
import numpy as np
import base64
import logging
from datetime import datetime
from typing import Optional, Dict, Any
from pathlib import Path

from fastapi import FastAPI, File, UploadFile, HTTPException, Form
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import uvicorn

# Import our modules
from face_detector import FaceDetector
from face_verifier import FaceVerifier
from liveness_detector import LivenessDetector
from database_manager import DatabaseManager

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Initialize FastAPI app
app = FastAPI(
    title="Face Verification System",
    version="1.0.0",
    description="Real-time face verification with anti-spoofing"
)

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Initialize components
face_detector = FaceDetector()
face_verifier = FaceVerifier()
liveness_detector = LivenessDetector()
db_manager = DatabaseManager()

# Pydantic models
class VerificationRequest(BaseModel):
    user_id: str
    live_image_base64: str
    check_liveness: bool = True

class VerificationResponse(BaseModel):
    success: bool
    match: bool
    confidence: float
    is_live: Optional[bool] = None
    message: str
    timestamp: str

class RegistrationResponse(BaseModel):
    success: bool
    user_id: str
    message: str
    face_detected: bool
    face_quality_score: float


@app.on_event("startup")
async def startup_event():
    """Initialize system on startup"""
    logger.info("=" * 60)
    logger.info("🚀 Face Verification System Starting")
    logger.info("=" * 60)
    
    # Create necessary directories
    Path("uploads").mkdir(exist_ok=True)
    Path("temp").mkdir(exist_ok=True)
    
    # Initialize database
    db_manager.initialize()
    
    logger.info("✓ System initialized successfully")


@app.get("/")
async def root():
    """API root endpoint with documentation links"""
    return {
        "service": "Face Verification API",
        "version": "1.0.0",
        "status": "running",
        "documentation": "/docs",
        "endpoints": {
            "register": "POST /register - Register a new user with profile picture",
            "verify": "POST /verify - Verify face against registered profile",
            "stats": "GET /stats - Get system statistics",
            "health": "GET /health - Health check"
        },
        "usage": {
            "register": {
                "method": "POST",
                "content_type": "multipart/form-data",
                "fields": {
                    "user_id": "string (required)",
                    "profile_picture": "file (required)"
                }
            },
            "verify": {
                "method": "POST",
                "content_type": "application/json",
                "body": {
                    "user_id": "string (required)",
                    "live_image_base64": "string (required, base64 encoded image)",
                    "check_liveness": "boolean (optional, default: true)"
                }
            }
        }
    }


@app.post("/register", response_model=RegistrationResponse)
async def register_user(
    user_id: str = Form(...),
    profile_picture: UploadFile = File(...)
):
    """
    Register a new user with their profile picture
    """
    try:
        logger.info(f"Registration request for user: {user_id}")
        
        # Check if user already exists
        if db_manager.user_exists(user_id):
            raise HTTPException(400, f"User {user_id} already registered")
        
        # Read image
        image_bytes = await profile_picture.read()
        nparr = np.frombuffer(image_bytes, np.uint8)
        image = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
        
        if image is None:
            raise HTTPException(400, "Invalid image format")
        
        # Detect face
        faces = face_detector.detect_faces(image)
        
        if len(faces) == 0:
            return RegistrationResponse(
                success=False,
                user_id=user_id,
                message="No face detected in the image",
                face_detected=False,
                face_quality_score=0.0
            )
        
        if len(faces) > 1:
            return RegistrationResponse(
                success=False,
                user_id=user_id,
                message="Multiple faces detected. Please upload image with single face",
                face_detected=True,
                face_quality_score=0.0
            )
        
        # Get face quality score
        face = faces[0]
        quality_score = face_detector.assess_face_quality(image, face)
        
        if quality_score < 0.5:
            return RegistrationResponse(
                success=False,
                user_id=user_id,
                message=f"Face quality too low ({quality_score:.2f}). Please use a clearer image",
                face_detected=True,
                face_quality_score=quality_score
            )
        
        # Extract face embedding
        embedding = face_verifier.extract_embedding(image, face)
        
        if embedding is None:
            raise HTTPException(500, "Failed to extract face embedding")
        
        # Save to database
        image_path = f"uploads/{user_id}.jpg"
        cv2.imwrite(image_path, image)
        
        db_manager.register_user(user_id, embedding, image_path)
        
        logger.info(f"✓ User {user_id} registered successfully")
        
        return RegistrationResponse(
            success=True,
            user_id=user_id,
            message="User registered successfully",
            face_detected=True,
            face_quality_score=quality_score
        )
        
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Registration error: {e}")
        raise HTTPException(500, f"Registration failed: {str(e)}")


@app.post("/verify", response_model=VerificationResponse)
async def verify_face(request: VerificationRequest):
    """
    Verify a live face capture against registered profile
    """
    try:
        logger.info(f"Verification request for user: {request.user_id}")
        
        # Check if user exists
        user_data = db_manager.get_user(request.user_id)
        if user_data is None:
            return VerificationResponse(
                success=False,
                match=False,
                confidence=0.0,
                message=f"User {request.user_id} not found",
                timestamp=datetime.now().isoformat()
            )
        
        # Decode live image
        image_bytes = base64.b64decode(request.live_image_base64)
        nparr = np.frombuffer(image_bytes, np.uint8)
        live_image = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
        
        if live_image is None:
            raise HTTPException(400, "Invalid image format")
        
        # Detect face in live image
        faces = face_detector.detect_faces(live_image)
        
        if len(faces) == 0:
            return VerificationResponse(
                success=True,
                match=False,
                confidence=0.0,
                message="No face detected in live image",
                timestamp=datetime.now().isoformat()
            )
        
        if len(faces) > 1:
            return VerificationResponse(
                success=True,
                match=False,
                confidence=0.0,
                message="Multiple faces detected. Please ensure only one face is visible",
                timestamp=datetime.now().isoformat()
            )
        
        face = faces[0]
        
        # Check liveness if requested
        is_live = None
        if request.check_liveness:
            is_live = liveness_detector.detect_liveness(live_image, face)
            if not is_live:
                logger.warning(f"Liveness check failed for user {request.user_id}")
                # Continue with verification but flag the result
        
        # Extract embedding from live image
        live_embedding = face_verifier.extract_embedding(live_image, face)
        
        if live_embedding is None:
            raise HTTPException(500, "Failed to extract face embedding from live image")
        
        # Compare with stored embedding
        stored_embedding = np.array(user_data['embedding'])
        similarity = face_verifier.compare_embeddings(stored_embedding, live_embedding)
        
        # Determine match (threshold: 0.6)
        threshold = 0.6
        is_match = similarity >= threshold
        
        # Record verification attempt
        db_manager.record_verification(request.user_id, is_match, similarity, is_live)
        
        message = "Face verified successfully" if is_match else "Face does not match"
        if is_live is False:
            message += " (Warning: Possible spoofing attempt detected)"
        
        logger.info(f"✓ Verification complete for {request.user_id}: match={is_match}, confidence={similarity:.3f}")
        
        return VerificationResponse(
            success=True,
            match=is_match,
            confidence=similarity,
            is_live=is_live,
            message=message,
            timestamp=datetime.now().isoformat()
        )
        
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Verification error: {e}")
        raise HTTPException(500, f"Verification failed: {str(e)}")


# Pydantic model for direct comparison
class CompareRequest(BaseModel):
    image1_base64: str
    image2_base64: str
    check_liveness: bool = False

class CompareResponse(BaseModel):
    success: bool
    match: bool
    similarity: float
    confidence: float
    is_live_image1: Optional[bool] = None
    is_live_image2: Optional[bool] = None
    message: str
    details: Optional[Dict[str, Any]] = None


@app.post("/compare", response_model=CompareResponse)
async def compare_faces(request: CompareRequest):
    """
    Direct face comparison - Compare two images to check if they are the same person
    No registration required - just send two images
    
    This is the main endpoint for Flutter app integration
    """
    try:
        logger.info("Direct face comparison request received")
        
        # Decode first image
        image1_bytes = base64.b64decode(request.image1_base64)
        nparr1 = np.frombuffer(image1_bytes, np.uint8)
        image1 = cv2.imdecode(nparr1, cv2.IMREAD_COLOR)
        
        if image1 is None:
            raise HTTPException(400, "Invalid format for first image")
        
        # Decode second image
        image2_bytes = base64.b64decode(request.image2_base64)
        nparr2 = np.frombuffer(image2_bytes, np.uint8)
        image2 = cv2.imdecode(nparr2, cv2.IMREAD_COLOR)
        
        if image2 is None:
            raise HTTPException(400, "Invalid format for second image")
        
        # Detect faces in first image
        faces1 = face_detector.detect_faces(image1)
        if len(faces1) == 0:
            return CompareResponse(
                success=True,
                match=False,
                similarity=0.0,
                confidence=0.0,
                message="No face detected in first image",
                details={"faces_in_image1": 0, "faces_in_image2": "not_checked"}
            )
        
        if len(faces1) > 1:
            return CompareResponse(
                success=True,
                match=False,
                similarity=0.0,
                confidence=0.0,
                message="Multiple faces detected in first image. Please use image with single face",
                details={"faces_in_image1": len(faces1), "faces_in_image2": "not_checked"}
            )
        
        # Detect faces in second image
        faces2 = face_detector.detect_faces(image2)
        if len(faces2) == 0:
            return CompareResponse(
                success=True,
                match=False,
                similarity=0.0,
                confidence=0.0,
                message="No face detected in second image",
                details={"faces_in_image1": 1, "faces_in_image2": 0}
            )
        
        if len(faces2) > 1:
            return CompareResponse(
                success=True,
                match=False,
                similarity=0.0,
                confidence=0.0,
                message="Multiple faces detected in second image. Please use image with single face",
                details={"faces_in_image1": 1, "faces_in_image2": len(faces2)}
            )
        
        face1 = faces1[0]
        face2 = faces2[0]
        
        # Check liveness if requested
        is_live1 = None
        is_live2 = None
        if request.check_liveness:
            is_live1 = liveness_detector.detect_liveness(image1, face1)
            is_live2 = liveness_detector.detect_liveness(image2, face2)
            
            if not is_live1 or not is_live2:
                logger.warning("Liveness check failed for one or both images")
        
        # Extract embeddings
        embedding1 = face_verifier.extract_embedding(image1, face1)
        if embedding1 is None:
            raise HTTPException(500, "Failed to extract face embedding from first image")
        
        embedding2 = face_verifier.extract_embedding(image2, face2)
        if embedding2 is None:
            raise HTTPException(500, "Failed to extract face embedding from second image")
        
        # Compare embeddings
        similarity = face_verifier.compare_embeddings(embedding1, embedding2)
        
        # Determine match (threshold: 0.6)
        threshold = 0.6
        is_match = similarity >= threshold
        
        # Build message
        if is_match:
            message = "✓ MATCH - Both images are of the same person"
        else:
            message = "✗ NOT MATCH - Images are of different persons"
        
        if request.check_liveness:
            if not is_live1:
                message += " (Warning: First image may be a spoof)"
            if not is_live2:
                message += " (Warning: Second image may be a spoof)"
        
        logger.info(f"✓ Comparison complete: match={is_match}, similarity={similarity:.3f}")
        
        return CompareResponse(
            success=True,
            match=is_match,
            similarity=similarity,
            confidence=similarity,
            is_live_image1=is_live1,
            is_live_image2=is_live2,
            message=message,
            details={
                "faces_in_image1": 1,
                "faces_in_image2": 1,
                "threshold_used": threshold,
                "similarity_percentage": round(similarity * 100, 2)
            }
        )
        
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Comparison error: {e}")
        raise HTTPException(500, f"Face comparison failed: {str(e)}")


@app.get("/stats")
async def get_statistics():
    """Get system statistics"""
    try:
        stats = db_manager.get_statistics()
        return stats
    except Exception as e:
        logger.error(f"Stats error: {e}")
        raise HTTPException(500, f"Failed to get statistics: {str(e)}")


@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {
        "status": "healthy",
        "service": "Face Verification System",
        "version": "1.0.0",
        "timestamp": datetime.now().isoformat()
    }


if __name__ == "__main__":
    port = int(os.getenv("PORT", 7860))
    host = os.getenv("HOST", "0.0.0.0")
    
    logger.info(f"Starting Face Verification System on {host}:{port}")
    uvicorn.run(app, host=host, port=port, log_level="info")