Update app.py

app.py

@@ -125,6 +125,226 @@ async def health_check():
     """
     return {"status": "healthy", "tools_initialized": TOOLS_INITIALIZED_SUCCESSFULLY}
 
+def process_kyc_verification(user_image: Image.Image, id_card_image: Image.Image):
+    """
+    Main KYC verification function that processes both images and returns comprehensive results
+    
+    Args:
+        user_image: PIL Image of the user's face photo
+        id_card_image: PIL Image of the ID card/document
+    
+    Returns:
+        dict: Comprehensive verification results with scores and analysis
+    """
+    start_time = time.time()
+    
+    try:
+        # Validate inputs
+        if user_image is None:
+            return {
+                "error": "User image is required",
+                "liveness_score": 0.0,
+                "face_similarity": 0.0,
+                "deepfake_score": 1.0,
+                "overall_decision": "REJECTED",
+                "processing_time": 0.0
+            }
+        
+        if id_card_image is None:
+            return {
+                "error": "ID card image is required", 
+                "liveness_score": 0.0,
+                "face_similarity": 0.0,
+                "deepfake_score": 1.0,
+                "overall_decision": "REJECTED",
+                "processing_time": 0.0
+            }
+        
+        if not TOOLS_INITIALIZED_SUCCESSFULLY:
+            return {
+                "error": "KYC tools failed to initialize",
+                "liveness_score": 0.0,
+                "face_similarity": 0.0,
+                "deepfake_score": 1.0,
+                "overall_decision": "ERROR",
+                "processing_time": 0.0
+            }
+        
+        print("Starting KYC verification process...")
+        
+        # Initialize scores
+        liveness_score = 0.0
+        face_similarity = 0.0
+        deepfake_score = 1.0
+        
+        # Save images to temporary files for tool processing
+        with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as temp_user:
+            user_image.save(temp_user.name, format='JPEG')
+            user_image_path = temp_user.name
+        
+        with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as temp_id:
+            id_card_image.save(temp_id.name, format='JPEG')
+            id_card_path = temp_id.name
+        
+        temp_files_to_clean = [user_image_path, id_card_path]
+        
+        try:
+            # 1. ID Document Analysis
+            print("Analyzing ID document...")
+            try:
+                id_analysis = image_description_tool_instance.apply(id_card_path)
+            except Exception as e:
+                id_analysis = f"Error analyzing ID document: {str(e)}"
+            
+            # 2. Liveness Detection
+            print("Performing liveness detection...")
+            try:
+                liveness_result_str = liveness_detection_tool_instance.apply(user_image_path)
+                confidence_match = re.search(r"Confidence:\s*([0-9.]+)", liveness_result_str, re.IGNORECASE)
+                if confidence_match:
+                    liveness_score = float(confidence_match.group(1))
+                else:
+                    liveness_score = 0.85 if "real" in liveness_result_str.lower() or "live" in liveness_result_str.lower() else 0.25
+                liveness_details = liveness_result_str
+            except Exception as e:
+                liveness_score = 0.0
+                liveness_details = f"Error in liveness detection: {str(e)}"
+            
+            # 3. Face Comparison
+            print("Comparing faces...")
+            try:
+                face_sim_result_str = face_similarity_tool_instance.apply(user_image_path, id_card_path)
+                score_match = re.search(r"similarity score:\s*([0-9.]+)", face_sim_result_str, re.IGNORECASE)
+                if score_match:
+                    face_similarity = float(score_match.group(1))
+                else:
+                    try:
+                        potential_score_str = face_sim_result_str.split(":")[-1].strip().split(" ")[0].replace(')', '').replace('(', '')
+                        face_similarity = float(re.sub(r"[^0-9.]", "", potential_score_str))
+                    except:
+                        face_similarity = 0.75 if "match" in face_sim_result_str.lower() or "true" in face_sim_result_str.lower() else 0.35
+                face_details = face_sim_result_str
+            except Exception as e:
+                face_similarity = 0.0
+                face_details = f"Error in face similarity: {str(e)}"
+            
+            # 4. Deepfake Detection
+            print("Detecting deepfakes...")
+            try:
+                img_df_result = image_deepfake_tool_instance.apply(user_image_path)
+                if img_df_result.get("label", "").lower() == "error":
+                    deepfake_score = 1.0
+                    deepfake_details = img_df_result.get('error_message', 'Unknown error during deepfake detection.')
+                else:
+                    label = img_df_result.get("label", "AI").lower()
+                    confidence = img_df_result.get("confidence", 0.0)
+                    if label == "human":
+                        deepfake_score = 1.0 - confidence
+                    elif label == "ai":
+                        deepfake_score = confidence
+                    else:
+                        deepfake_score = 1.0
+                    deepfake_details = f"Prediction: '{label.capitalize()}' with confidence {confidence:.4f}"
+            except Exception as e:
+                deepfake_score = 1.0
+                deepfake_details = f"Error in deepfake detection: {str(e)}"
+            
+            # 5. Overall Decision Logic
+            processing_time = time.time() - start_time
+            
+            # Decision criteria
+            liveness_pass = liveness_score >= LIVENESS_THRESHOLD
+            face_match_pass = face_similarity >= FACE_MATCH_THRESHOLD
+            deepfake_pass = deepfake_score <= DEEPFAKE_THRESHOLD
+            
+            overall_decision = "APPROVED" if (liveness_pass and face_match_pass and deepfake_pass) else "REJECTED"
+            confidence_overall = min(liveness_score, face_similarity, 1 - deepfake_score)
+            
+            # Generate recommendations
+            recommendations = []
+            if liveness_pass:
+                recommendations.append("✅ Liveness check passed - Live person detected")
+            else:
+                recommendations.append("❌ Liveness check failed - Manual review required")
+                
+            if face_match_pass:
+                recommendations.append("✅ Face similarity acceptable - Identity match confirmed")
+            else:
+                recommendations.append("❌ Face similarity below threshold - Identity mismatch")
+                
+            if deepfake_pass:
+                recommendations.append("✅ No deepfake detected - Image appears authentic")
+            else:
+                recommendations.append("❌ Potential manipulation detected - Manual verification needed")
+            
+            # Compile comprehensive results
+            result = {
+                # Main scores
+                "liveness_score": round(float(liveness_score), 3),
+                "face_similarity": round(float(face_similarity), 3),
+                "deepfake_score": round(float(deepfake_score), 3),
+                
+                # Overall assessment
+                "overall_decision": overall_decision,
+                "confidence_overall": round(float(confidence_overall), 3),
+                
+                # Individual check results
+                "liveness_passed": liveness_pass,
+                "face_match_passed": face_match_pass,
+                "deepfake_passed": deepfake_pass,
+                
+                # Detailed analysis
+                "id_analysis": id_analysis,
+                "analysis_details": {
+                    "liveness_analysis": liveness_details,
+                    "face_comparison": face_details,
+                    "deepfake_analysis": deepfake_details,
+                    "document_analysis": id_analysis
+                },
+                
+                # Recommendations and metadata
+                "recommendations": recommendations,
+                "processing_time": round(processing_time, 2),
+                "model_version": MODEL_VERSION,
+                "timestamp": time.strftime("%Y-%m-%d %H:%M:%S UTC", time.gmtime()),
+                
+                # Thresholds used
+                "thresholds": {
+                    "liveness_threshold": LIVENESS_THRESHOLD,
+                    "face_similarity_threshold": FACE_MATCH_THRESHOLD,
+                    "deepfake_threshold": DEEPFAKE_THRESHOLD
+                }
+            }
+            
+            print(f"KYC verification completed in {processing_time:.2f}s - Decision: {overall_decision}")
+            return result
+            
+        finally:
+            # Clean up temporary files
+            for temp_path in temp_files_to_clean:
+                try:
+                    if os.path.exists(temp_path):
+                        os.remove(temp_path)
+                except Exception as e_clean:
+                    print(f"Error cleaning up temp file {temp_path}: {e_clean}")
+        
+    except Exception as e:
+        error_msg = f"KYC verification failed: {str(e)}"
+        print(error_msg)
+        
+        return {
+            "error": error_msg,
+            "liveness_score": 0.0,
+            "face_similarity": 0.0,
+            "deepfake_score": 1.0,
+            "overall_decision": "ERROR",
+            "confidence_overall": 0.0,
+            "analysis_details": {"error": str(e)},
+            "recommendations": ["Manual review required due to processing error"],
+            "processing_time": time.time() - start_time,
+            "model_version": MODEL_VERSION,
+            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S UTC", time.gmtime())
+        }
 
 # Create the Gradio interface
 def create_interface():