Update app.py

app.py

@@ -1,341 +1,495 @@
-"""
-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)}")
-
-
-@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")
+"""
+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")