app.py

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)