""" 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")