import base64 import json import logging import os import tempfile # DeepFace/RetinaFace can break on newer TF/Keras combinations without this flag. os.environ.setdefault("TF_USE_LEGACY_KERAS", "1") import cv2 import numpy as np import requests from deepface import DeepFace from fastapi import FastAPI from fastapi.responses import JSONResponse from pydantic import BaseModel class Verify(BaseModel): image: str images: list[str] app = FastAPI() logger = logging.getLogger("plugg_verification") if not logger.handlers: logging.basicConfig(level=logging.INFO) def log_event(event_type: str, **fields): payload = {"event": event_type, **fields} logger.error(json.dumps(payload, default=str)) def log_info_event(event_type: str, **fields): payload = {"event": event_type, **fields} logger.info(json.dumps(payload, default=str)) def extract_root_cause(exc: Exception) -> str: cause = getattr(exc, "__cause__", None) if cause: return str(cause) return str(exc) def safe_remove(path: str): try: os.remove(path) except OSError: pass def summarize_result(result: dict): return { "verified": result.get("verified"), "distance": result.get("distance"), "threshold": result.get("threshold"), "model": result.get("model"), "detector_backend": result.get("detector_backend"), "facial_areas": result.get("facial_areas"), } def prepare_image_for_deepface(source: str): """ Return a filesystem path DeepFace can consume reliably. For URLs / base64 we materialize a temp file and return (path, True). For local paths we return (path, False). """ if not isinstance(source, str): raise TypeError("Unsupported image source type") if source.startswith("http://") or source.startswith("https://"): resp = requests.get(source, headers={"User-Agent": "Mozilla/5.0"}, timeout=20) resp.raise_for_status() binary = resp.content elif source.startswith("data:image"): b64_payload = source.split(",", 1)[1] if "," in source else source binary = base64.b64decode(b64_payload) else: img = cv2.imread(source) if img is None: raise ValueError(f"Failed to load local image: {source}") return source, False data = np.frombuffer(binary, dtype=np.uint8) img = cv2.imdecode(data, cv2.IMREAD_COLOR) if img is None: raise ValueError(f"Failed to decode image: {source}") tmp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".jpg") tmp_file_path = tmp_file.name tmp_file.close() wrote = cv2.imwrite(tmp_file_path, img) if not wrote: safe_remove(tmp_file_path) raise ValueError(f"Failed to write temp image: {source}") return tmp_file_path, True @app.get("/") def greet_json(): return {"Hello": "World!"} @app.post("/verify") def verify(v: Verify): data = v.model_dump() selfie = data["image"] gallery = data["images"] true_count = 0 print(selfie) selfie_path = None selfie_is_temp = False try: selfie_path, selfie_is_temp = prepare_image_for_deepface(selfie) except Exception as e: print(f"Failed to load selfie image: {e}") log_event("load_error", target="selfie", source=selfie, error=str(e)) return JSONResponse(content={"verified": False, "image": None, "error": "failed_to_load_selfie"}) log_info_event("verify_started", selfie=selfie, gallery_count=len(gallery)) try: for image in gallery: print(image) gallery_path = None gallery_is_temp = False try: gallery_path, gallery_is_temp = prepare_image_for_deepface(image) except Exception as e: print(f"Failed to load gallery image {image}: {e}") log_event("load_error", target="gallery", source=image, error=str(e)) continue try: result = DeepFace.verify( img1_path=selfie_path, img2_path=gallery_path, model_name="Facenet512", detector_backend="opencv", enforce_detection=False ) log_info_event( "verify_attempt", stage="primary", gallery_image=image, result=summarize_result(result), ) if result.get("verified", False): true_count += 1 log_info_event("verify_match_count", gallery_image=image, true_count=true_count) if true_count >= 2: log_info_event("verify_response", verified=True, matched_image=image, true_count=true_count) return JSONResponse(content={"verified": True, "image": image}) except Exception as e: msg = str(e) root_cause = extract_root_cause(e) print(f"DeepFace verification error for {image}: {msg}") if "img1_path" in msg: log_event("img1_path_error", gallery_image=image, error=msg, root_cause=root_cause) if "Face could not be detected" in msg or "No face" in msg: log_event("face_not_detected", gallery_image=image, error=msg, root_cause=root_cause) # Fallback path on generic processing or face-detection errors. if "img1_path" in msg or "Face could not be detected" in msg or "No face" in msg: try: result = DeepFace.verify( img1_path=selfie_path, img2_path=gallery_path, model_name="VGG-Face", detector_backend="opencv", enforce_detection=False ) log_info_event( "verify_attempt", stage="fallback", gallery_image=image, result=summarize_result(result), ) if result.get("verified", False): true_count += 1 log_info_event("verify_match_count", gallery_image=image, true_count=true_count) if true_count >= 1: log_info_event("verify_response", verified=True, matched_image=image, true_count=true_count) return JSONResponse(content={"verified": True, "image": image}) except Exception as e2: print(f"DeepFace fallback error for {image}: {e2}") log_event("fallback_error", gallery_image=image, error=str(e2), root_cause=extract_root_cause(e2)) finally: if gallery_is_temp and gallery_path: safe_remove(gallery_path) log_info_event("verify_response", verified=False, matched_image=None, true_count=true_count) return JSONResponse(content={"verified": False, "image": None}) finally: if selfie_is_temp and selfie_path: safe_remove(selfie_path)