288 lines
9.6 KiB
Python
288 lines
9.6 KiB
Python
import os
|
|
import time
|
|
import json
|
|
import logging
|
|
import numpy as np
|
|
from PIL import Image
|
|
import face_recognition
|
|
from deepface import DeepFace
|
|
from flask import Blueprint, request, jsonify
|
|
|
|
from services.face_service import compare_faces_service
|
|
from services.storage_service import minio_client, BUCKET
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
logging.basicConfig(level=logging.INFO)
|
|
|
|
face_bp = Blueprint('face_bp', __name__)
|
|
|
|
threshold = float(os.getenv("THRESHOLD", 0.6))
|
|
|
|
|
|
# @face_bp.route('/compare_faces', methods=['POST'])
|
|
# def compare_faces():
|
|
# logging.info("🖼️ Recebendo imagens para comparação...")
|
|
|
|
# if 'image1' not in request.files or 'image2' not in request.files:
|
|
# return jsonify({"error": "Missing images. Keys should be 'image1' and 'image2'."}), 400
|
|
|
|
# image1 = request.files['image1']
|
|
# image2 = request.files['image2']
|
|
|
|
# try:
|
|
# result = compare_faces_service(image1, image2)
|
|
# return jsonify(result), 200
|
|
# except ValueError as e:
|
|
# return jsonify({"error": str(e)}), 400
|
|
# except Exception as e:
|
|
# return jsonify({"error": "Internal Server Error", "details": str(e)}), 500
|
|
|
|
|
|
@face_bp.route('/register_face', methods=['POST'])
|
|
def register_face():
|
|
from datetime import datetime
|
|
from io import BytesIO
|
|
|
|
person_id = request.form.get("person_id")
|
|
image_file = request.files.get("image")
|
|
|
|
if not person_id or not image_file:
|
|
return jsonify({"error": "Missing person_id or image"}), 400
|
|
|
|
try:
|
|
# Carrega imagem como array RGB
|
|
image_file.seek(0)
|
|
image = face_recognition.load_image_file(image_file)
|
|
|
|
# Detecta a primeira face
|
|
face_locations = face_recognition.face_locations(image)
|
|
if not face_locations:
|
|
return jsonify({"error": "No face detected in image"}), 400
|
|
|
|
top, right, bottom, left = face_locations[0]
|
|
face_crop = image[top:bottom, left:right]
|
|
|
|
# Converte para PIL
|
|
face_pil = Image.fromarray(face_crop)
|
|
|
|
# Salva em buffer
|
|
buffer = BytesIO()
|
|
face_pil.save(buffer, format="JPEG")
|
|
buffer.seek(0)
|
|
|
|
# Gera nome baseado em timestamp
|
|
timestamp = datetime.utcnow().strftime("%Y%m%d-%H%M%S")
|
|
image_filename = f"registred_faces/{person_id}/{timestamp}.jpg"
|
|
|
|
# Upload para o MinIO
|
|
minio_client.put_object(
|
|
BUCKET,
|
|
image_filename,
|
|
buffer,
|
|
length=buffer.getbuffer().nbytes,
|
|
content_type="image/jpeg"
|
|
)
|
|
|
|
return jsonify({
|
|
"person_id": person_id,
|
|
"image_path": image_filename,
|
|
"timestamp": timestamp,
|
|
"status": "Face cropped and saved"
|
|
}), 200
|
|
|
|
except Exception as e:
|
|
logger.exception("❌ Erro ao registrar face")
|
|
return jsonify({"error": "Failed to register face", "details": str(e)}), 500
|
|
|
|
|
|
@face_bp.route("/checkin", methods=["POST"])
|
|
def checkin():
|
|
from datetime import datetime
|
|
from io import BytesIO
|
|
|
|
logger.info("📥 Início do registro de ponto (/checkin)")
|
|
|
|
person_id = request.form.get("person_id")
|
|
image_file = request.files.get("image")
|
|
ip_address = request.remote_addr
|
|
|
|
if not person_id or not image_file:
|
|
return jsonify({"error": "Missing person_id or image"}), 400
|
|
|
|
try:
|
|
img_probe = np.array(Image.open(image_file).convert("RGB"))
|
|
|
|
objects = list(minio_client.list_objects(BUCKET, prefix=f"registred_faces/{person_id}/", recursive=True))
|
|
image_objects = [obj for obj in objects if obj.object_name.endswith(".jpg")]
|
|
|
|
if not image_objects:
|
|
return jsonify({"error": f"No registered face found for '{person_id}'"}), 404
|
|
|
|
image_objects.sort(key=lambda x: x.object_name, reverse=True)
|
|
target_image_obj = image_objects[0]
|
|
logger.debug(f"🎯 Usando imagem registrada: {target_image_obj.object_name}")
|
|
|
|
response = minio_client.get_object(BUCKET, target_image_obj.object_name)
|
|
img_registered = np.array(Image.open(response).convert("RGB"))
|
|
|
|
t0 = time.time()
|
|
result = DeepFace.verify(
|
|
img_probe,
|
|
img_registered,
|
|
model_name="Dlib",
|
|
enforce_detection=False
|
|
)
|
|
duration = round(time.time() - t0, 4)
|
|
|
|
distance = result["distance"]
|
|
similarity = 1 - distance
|
|
threshold = float(os.getenv("THRESHOLD", 0.85))
|
|
confidence_high = float(os.getenv("CONFIDENCE_HIGH", 0.95))
|
|
confidence_medium = float(os.getenv("CONFIDENCE_MEDIUM", 0.85))
|
|
match = similarity >= threshold
|
|
|
|
if similarity >= confidence_high:
|
|
confidence = "high"
|
|
elif similarity >= confidence_medium:
|
|
confidence = "medium"
|
|
else:
|
|
confidence = "low"
|
|
|
|
if not match:
|
|
return jsonify({
|
|
"match": False,
|
|
"similarity_score": round(similarity, 4),
|
|
"confidence": confidence,
|
|
"message": "Face not recognized with sufficient confidence."
|
|
}), 401
|
|
|
|
# Recorta a face com face_recognition
|
|
image_file.seek(0)
|
|
image_rgb = face_recognition.load_image_file(image_file)
|
|
locations = face_recognition.face_locations(image_rgb)
|
|
if not locations:
|
|
return jsonify({"error": "No face found to crop"}), 400
|
|
top, right, bottom, left = locations[0]
|
|
face_crop = image_rgb[top:bottom, left:right]
|
|
face_pil = Image.fromarray(face_crop)
|
|
|
|
# Organiza por pessoa/data/hora
|
|
now = datetime.utcnow()
|
|
date_str = now.strftime("%Y-%m-%d")
|
|
time_str = now.strftime("%H-%M-%S")
|
|
path_prefix = f"checkins/{person_id}/{date_str}/{time_str}/"
|
|
original_name = f"{path_prefix}original.jpg"
|
|
face_name = f"{path_prefix}face.jpg"
|
|
json_name = f"{path_prefix}metadata.json"
|
|
|
|
# Upload original
|
|
image_file.seek(0)
|
|
minio_client.put_object(
|
|
BUCKET, original_name, image_file,
|
|
length=-1, part_size=10*1024*1024,
|
|
content_type="image/jpeg"
|
|
)
|
|
|
|
# Upload face
|
|
face_buffer = BytesIO()
|
|
face_pil.save(face_buffer, format="JPEG")
|
|
face_buffer.seek(0)
|
|
minio_client.put_object(
|
|
BUCKET, face_name, face_buffer,
|
|
length=face_buffer.getbuffer().nbytes,
|
|
content_type="image/jpeg"
|
|
)
|
|
|
|
# Upload JSON
|
|
data = {
|
|
"person_id": person_id,
|
|
"timestamp": now.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"ip": ip_address,
|
|
"confidence": confidence,
|
|
"similarity_score": round(similarity, 4),
|
|
"duration_sec": duration,
|
|
"match": match
|
|
}
|
|
json_buffer = BytesIO(json.dumps(data).encode("utf-8"))
|
|
minio_client.put_object(
|
|
BUCKET, json_name, json_buffer,
|
|
length=json_buffer.getbuffer().nbytes,
|
|
content_type="application/json"
|
|
)
|
|
|
|
return jsonify(data), 200
|
|
|
|
except Exception as e:
|
|
logger.exception("❌ Erro ao processar check-in")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
'''
|
|
Abaixo é o endpoint que precisa que seja passado duas imagens para comparação.
|
|
'''
|
|
# @face_bp.route("/verify_face_dlib", methods=["POST"])
|
|
# def verify_face_dlib():
|
|
# logger.info("🔍 Verificação facial usando deepface_dlib com imagem cadastrada")
|
|
|
|
# person_id = request.form.get("person_id")
|
|
# image_file = request.files.get("image")
|
|
|
|
# if not person_id or not image_file:
|
|
# return jsonify({"error": "Missing person_id or image"}), 400
|
|
|
|
# try:
|
|
# img_probe = np.array(Image.open(image_file).convert("RGB"))
|
|
|
|
# objects = list(minio_client.list_objects(BUCKET, prefix=f"{person_id}/", recursive=True))
|
|
# image_objects = [obj for obj in objects if obj.object_name.endswith(".jpg")]
|
|
|
|
# if not image_objects:
|
|
# return jsonify({"error": f"No registered face found for '{person_id}'"}), 404
|
|
|
|
# image_objects.sort(key=lambda x: x.object_name, reverse=True)
|
|
# target_image_obj = image_objects[0]
|
|
# logger.debug(f"🖼 Imagem cadastrada encontrada: {target_image_obj.object_name}")
|
|
|
|
# response = minio_client.get_object(BUCKET, target_image_obj.object_name)
|
|
# img_registered = np.array(Image.open(response).convert("RGB"))
|
|
|
|
# t0 = time.time()
|
|
# result = DeepFace.verify(
|
|
# img_probe,
|
|
# img_registered,
|
|
# model_name="Dlib",
|
|
# enforce_detection=False
|
|
# )
|
|
# duration = round(time.time() - t0, 4)
|
|
|
|
# distance = result["distance"]
|
|
# similarity = 1 - distance
|
|
|
|
# # Aplica o THRESHOLD sobre a similaridade
|
|
# threshold = float(os.getenv("THRESHOLD", 0.93)) # ex: 0.85 = exige mais precisão
|
|
# confidence_high = float(os.getenv("CONFIDENCE_HIGH", 0.95))
|
|
# confidence_medium = float(os.getenv("CONFIDENCE_MEDIUM", 0.85))
|
|
|
|
# match = similarity >= threshold
|
|
|
|
# if similarity >= confidence_high:
|
|
# confidence = "high"
|
|
# elif similarity >= confidence_medium:
|
|
# confidence = "medium"
|
|
# else:
|
|
# confidence = "low"
|
|
|
|
# return jsonify({
|
|
# "person_id": person_id,
|
|
# "match": match,
|
|
# "similarity_score": round(similarity, 4),
|
|
# "threshold": threshold,
|
|
# "confidence": confidence,
|
|
# "duration_sec": duration
|
|
# }), 200
|
|
|
|
# except Exception as e:
|
|
# logger.exception("Erro na verificação facial")
|
|
# return jsonify({"error": str(e)}), 500
|