""" FastAPI endpoint for identity validation service. This module provides the main API endpoint for identity validation, accepting ID photos, user videos, and gesture requirements to perform comprehensive identity verification. """ import os import json import tempfile import time import logging from typing import Optional from datetime import datetime, timezone from fastapi import FastAPI, UploadFile, File, Form, HTTPException, Depends from fastapi.responses import ORJSONResponse from .models import ValidationRequest, ValidationResponse, ValidationStatus from .facial_validator import FacialValidator from .gesture_validator import GestureValidator from .config import config logger = logging.getLogger(__name__) # Create FastAPI app app = FastAPI( title="Identity Validation API", description="API for identity verification using facial recognition and gesture validation", version="1.0.0", default_response_class=ORJSONResponse ) # Initialize validators facial_validator = FacialValidator() gesture_validator = GestureValidator() def get_validation_request( gestures: str = Form(...), # Gesture validation parameters (optional, fallback to env vars) error_margin: str = Form("default"), min_gesture_duration: str = Form("default"), require_all_gestures: str = Form("default"), confidence_threshold: str = Form("default"), # Facial recognition parameters (optional, fallback to env vars) similarity_threshold: str = Form("default"), frame_sample_rate: str = Form("default"), # Response parameters include_details: bool = Form(False) ) -> ValidationRequest: """ Parse and validate the validation request from form data. All parameters are optional and will fall back to environment variable defaults if not provided. This allows for flexible configuration at both the request level and server level. Parameters ---------- gestures : str JSON string containing the list of required gestures error_margin : Optional[float] Error margin for gesture validation (0.0-1.0). Uses DEFAULT_ERROR_MARGIN env var if None min_gesture_duration : Optional[int] Minimum duration for gesture detection. Uses MIN_GESTURE_DURATION env var if None require_all_gestures : Optional[bool] Whether all gestures must be present. Uses REQUIRE_ALL_GESTURES env var if None confidence_threshold : Optional[float] Minimum confidence threshold for gesture detection. Uses CONFIDENCE_THRESHOLD env var if None similarity_threshold : Optional[float] Minimum similarity threshold for facial matching. Uses SIMILARITY_THRESHOLD env var if None frame_sample_rate : Optional[int] Rate for sampling video frames for face detection. Uses FRAME_SAMPLE_RATE env var if None include_details : Optional[bool] Whether to include detailed results in response Returns ------- ValidationRequest Parsed and validated request object with environment fallbacks Raises ------ HTTPException If request validation fails """ try: # Parse gestures JSON gesture_list = json.loads(gestures) if not isinstance(gesture_list, list): raise ValueError("gestures must be a list") if not gesture_list: raise ValueError("gestures list cannot be empty") # Validate gesture names (basic validation) for gesture in gesture_list: if not isinstance(gesture, str) or not gesture.strip(): raise ValueError(f"Invalid gesture name: {gesture}") except json.JSONDecodeError as e: raise HTTPException( status_code=400, detail=f"Invalid JSON in gestures field: {str(e)}" ) except ValueError as e: raise HTTPException( status_code=400, detail=f"Invalid gestures data: {str(e)}" ) # Parse and convert parameters, using config defaults when "default" is provided def parse_param(value, default_value, value_type): """Parse parameter value, using default if 'default' string is provided.""" # Handle FastAPI Form objects - extract the actual value from .default if hasattr(value, 'default'): actual_value = value.default else: actual_value = value if actual_value == "default" or actual_value is None: return default_value try: if value_type == float: return float(actual_value) elif value_type == int: return int(actual_value) elif value_type == bool: return str(actual_value).lower() in ('true', '1', 'yes', 'on') else: return actual_value except (ValueError, TypeError, AttributeError): raise HTTPException( status_code=400, detail=f"Invalid value for parameter: {actual_value}" ) final_error_margin = parse_param(error_margin, config.default_error_margin, float) final_min_gesture_duration = parse_param(min_gesture_duration, config.min_gesture_duration, int) final_require_all_gestures = parse_param(require_all_gestures, config.require_all_gestures, bool) final_confidence_threshold = parse_param(confidence_threshold, config.confidence_threshold, float) final_similarity_threshold = parse_param(similarity_threshold, config.similarity_threshold, float) final_frame_sample_rate = parse_param(frame_sample_rate, config.frame_sample_rate, int) # Parse include_details parameter final_include_details = parse_param(include_details, False, bool) return ValidationRequest( asked_gestures=gesture_list, error_margin=final_error_margin, min_gesture_duration=final_min_gesture_duration, require_all_gestures=final_require_all_gestures, confidence_threshold=final_confidence_threshold, similarity_threshold=final_similarity_threshold, frame_sample_rate=final_frame_sample_rate, include_details=final_include_details ) @app.post("/", response_model=ValidationResponse) async def validate_identity( photo: UploadFile = File(...), video: UploadFile = File(...), request: ValidationRequest = Depends(get_validation_request) ): """ Validate user identity using facial recognition and gesture validation. This endpoint accepts an ID document photo, a user video containing the person's face and required gestures, and a list of gestures that must be performed. It returns validation results for both facial recognition and gesture compliance. Parameters ---------- photo : UploadFile ID document photo file (image format) video : UploadFile User video file containing face and gestures (video format) request : ValidationRequest Validation configuration and gesture requirements Returns ------- ValidationResponse Validation results with success indicators and optional details Raises ------ HTTPException If validation fails or processing errors occur """ start_time = time.time() logger.info(f"Identity validation request received for {request.asked_gestures}") # Validate file types if not photo.content_type or not photo.content_type.startswith(('image/', 'application/')): raise HTTPException( status_code=400, detail="Photo file must be an image" ) if not video.content_type or not video.content_type.startswith('video/'): raise HTTPException( status_code=400, detail="Video file must be a video" ) # Validate file sizes (basic check) MAX_FILE_SIZE = 100 * 1024 * 1024 # 100MB if photo.size and photo.size > MAX_FILE_SIZE: raise HTTPException( status_code=413, detail="Photo file too large (max 100MB)" ) if video.size and video.size > MAX_FILE_SIZE: raise HTTPException( status_code=413, detail="Video file too large (max 100MB)" ) # Create temporary files for processing temp_photo = None temp_video = None try: # Save uploaded files to temporary location with tempfile.NamedTemporaryFile(delete=False, suffix=f"_photo.{photo.filename.split('.')[-1] if '.' in photo.filename else 'jpg'}") as temp_photo_file: temp_photo = temp_photo_file.name photo_content = await photo.read() temp_photo_file.write(photo_content) with tempfile.NamedTemporaryFile(delete=False, suffix=f"_video.{video.filename.split('.')[-1] if '.' in video.filename else 'mp4'}") as temp_video_file: temp_video = temp_video_file.name video_content = await video.read() temp_video_file.write(video_content) logger.info(f"Files saved: photo={temp_photo}, video={temp_video}") # Perform facial validation logger.info("Starting facial validation") # Update facial validator with request-specific parameters if provided if request.similarity_threshold is not None: facial_validator.similarity_threshold = request.similarity_threshold if request.frame_sample_rate is not None: facial_validator.frame_sample_rate = request.frame_sample_rate face_result = facial_validator.validate_facial_match(temp_photo, temp_video) # Perform gesture validation logger.info("Starting gesture validation") gesture_result = gesture_validator.validate_gestures( temp_video, request.asked_gestures, error_margin=request.error_margin, require_all=request.require_all_gestures ) # Update gesture validator with request-specific parameters if provided if request.confidence_threshold is not None: gesture_validator.confidence_threshold = request.confidence_threshold if request.min_gesture_duration is not None: gesture_validator.min_gesture_duration = request.min_gesture_duration # Determine overall result overall_success = face_result.success and gesture_result.success overall_status = ValidationStatus.SUCCESS if overall_success else ValidationStatus.PARTIAL # Calculate processing time processing_time_ms = int((time.time() - start_time) * 1000) # Build response response = ValidationResponse( face=face_result.success, gestures=gesture_result.success, overall=overall_success, status=overall_status, face_result=face_result if request.include_details else None, gesture_result=gesture_result if request.include_details else None, processing_time_ms=processing_time_ms, timestamp=datetime.now(timezone.utc).isoformat() ) # Log results logger.info( "Identity validation completed", extra={ "face_success": face_result.success, "gesture_success": gesture_result.success, "overall_success": overall_success, "processing_time_ms": processing_time_ms, "requested_gestures": request.asked_gestures } ) return response except Exception as e: logger.error(f"Error during identity validation: {str(e)}", exc_info=True) raise HTTPException( status_code=500, detail=f"Internal server error during validation: {str(e)}" ) finally: # Clean up temporary files for temp_file in [temp_photo, temp_video]: if temp_file and os.path.exists(temp_file): try: os.unlink(temp_file) logger.debug(f"Cleaned up temporary file: {temp_file}") except Exception as e: logger.warning(f"Failed to clean up temporary file {temp_file}: {e}") @app.get("/health") async def health_check(): """ Health check endpoint for the validation service. Returns ------- dict Health status information """ return { "status": "healthy", "service": "identity-validation", "version": "1.0.0", "timestamp": datetime.now(timezone.utc).isoformat(), "components": { "facial_validator": "initialized", "gesture_validator": "initialized" } } @app.get("/") async def root(): """ Root endpoint providing API information. Returns ------- dict API information and usage instructions """ return { "name": "Identity Validation API", "version": "1.0.0", "description": "Identity verification using facial recognition and gesture validation", "endpoints": { "POST /": "Perform identity validation", "GET /health": "Health check", "GET /": "API information" }, "documentation": "/docs" }