deepface_fastapi/app.py

from fastapi import FastAPI, File, UploadFile, HTTPException, Form
from deepface import DeepFace
from manticoresearch import ApiClient, Configuration
from manticoresearch.api import IndexApi, SearchApi
from manticoresearch.model import SearchRequest
from manticoresearch.model import InsertDocumentRequest
import os
import shutil
from numpy import dot
from numpy.linalg import norm

# Configurar el cliente API
config = Configuration()
config.host = "http://localhost:9308"
client = ApiClient(configuration=config)

# Inicializar las APIs
index_api = IndexApi(client)
search_api = SearchApi(client)

app = FastAPI()

# Función para calcular similitud de coseno
def cosine_similarity(vec1, vec2):
    return dot(vec1, vec2) / (norm(vec1) * norm(vec2))

# Ruta para registrar un nuevo usuario
@app.post("/register_user")
async def register_user(user_id: str = Form(...), image: UploadFile = File(...)):
    try:
        # Guardar la imagen temporalmente
        img_path = f"/tmp/{image.filename}"
        with open(img_path, "wb") as buffer:
            shutil.copyfileobj(image.file, buffer)

        # Obtener los embeddings faciales
        embedding_result = DeepFace.represent(img_path, model_name="ArcFace")

        # Extraer los embeddings
        embeddings = embedding_result[0]['embedding']  # Extrae el vector de embeddings

        # Verifica si embeddings tiene 512 valores
        if len(embeddings) != 512:
            raise HTTPException(status_code=400, detail=f"Los embeddings generados no tienen 512 valores: {len(embeddings)}")

        # Almacenar los embeddings en Manticore Search
        doc = {
            "user_id": user_id,
            "embeddings": embeddings
        }

        # Crea la solicitud de inserción de documento
        insert_request = InsertDocumentRequest(index="users", doc=doc)

        # Inserta el documento en Manticore Search
        index_api.insert(insert_document_request=insert_request)

        # Eliminar el archivo temporal
        os.remove(img_path)

        return {"status": "user registered successfully"}

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error registering user: {str(e)}")

# Ruta para verificar una imagen
@app.post("/verify_user")
async def verify_user(image: UploadFile = File(...)):
    try:
        # Guardar la imagen temporalmente
        img_path = f"/tmp/{image.filename}"
        with open(img_path, "wb") as buffer:
            shutil.copyfileobj(image.file, buffer)

        # Obtener los embeddings faciales de la imagen
        query_embeddings = DeepFace.represent(img_path, model_name="ArcFace")[0]['embedding']

        # Consulta en Manticore Search para obtener los usuarios registrados
        query = {
            "size": 100,  # Ajusta el tamaño si tienes muchos usuarios
            "query": {
                "match_all": {}
            }
        }

        # Crear la solicitud de búsqueda
        search_request = SearchRequest(index="users", query=query)

        # Realizar la búsqueda
        result = search_api.search(search_request)

        # Recuperar todos los documentos (usuarios y sus embeddings)
        hits = result.hits.hits
        if not hits:
            raise HTTPException(status_code=404, detail="No registered users found.")

        # Calcular similitud de coseno con cada usuario registrado
        max_similarity = -1
        matched_user = None
        for hit in hits:
            stored_embeddings = hit['_source']['embeddings']  # Embeddings del usuario en la base de datos
            similarity = cosine_similarity(query_embeddings, stored_embeddings)

            # Imprimir la similitud de cada usuario para depuración
            print(f"Similitud con {hit['_source']['user_id']}: {similarity}")

            # Encontrar la mayor similitud
            if similarity > max_similarity:
                max_similarity = similarity
                matched_user = hit['_source']['user_id']

        # Establecer un umbral para decidir si hay coincidencia
        similarity_threshold = 0.5  # Ajusta este valor según tus necesidades
        if max_similarity < similarity_threshold:
            raise HTTPException(status_code=404, detail="No matching user found")

        # Eliminar el archivo temporal
        os.remove(img_path)

        return {"status": "success", "matched_user": matched_user, "similarity": max_similarity}

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error verifying user: {str(e)}")