Spaces:
Sleeping
Sleeping
from datetime import datetime, timedelta | |
from typing import Optional | |
from fastapi import APIRouter, File, UploadFile, HTTPException | |
from ..utils.db import tinydb_helper, chromadb_face_helper | |
from ..utils.jwt_utils import create_access_token | |
from ..utils.ec_image_utils import get_user_cropped_image_from_photo | |
import os | |
import uuid | |
router = APIRouter() | |
L2_FACE_THRESHOLD = 0.85 # distance value closer to 0 =>best match, >1 =>poor match | |
async def verify_user_face(file_path: str) -> Optional[dict]: | |
# Assuming `get_user_cropped_image_from_photo` returns the cropped face as expected by ChromaDB | |
face_img = get_user_cropped_image_from_photo(file_path) | |
if face_img is None: | |
return None | |
# Query the user's face in ChromaDB | |
query_results = chromadb_face_helper.query_user_face(face_img) | |
if query_results and len(query_results["ids"][0]) > 0: | |
chromadb_face_helper.print_query_results(query_results) | |
# Assuming the first result is the best match | |
l2_distance = query_results["distances"][0][0] | |
if l2_distance < L2_FACE_THRESHOLD: # l2 distance threshold for top matched face | |
user_id = query_results["ids"][0][0] | |
metadata = query_results["metadatas"][0][0] | |
return {"user_id": user_id, "metadata":metadata} | |
return None | |
async def user_login(file: UploadFile = File(...)): | |
file_path = f"/tmp/{uuid.uuid4()}.jpg" # Generates a unique filename | |
with open(file_path, "wb") as buffer: | |
contents = await file.read() | |
buffer.write(contents) | |
# Perform face verification | |
verification_result = await verify_user_face(file_path) | |
if verification_result: | |
user_id = verification_result["user_id"] | |
metadata = verification_result["metadata"] | |
# Generate JWT token with user information | |
access_token = create_access_token(data={"sub": user_id, "name": metadata["name"], "role": metadata["role"]}) | |
# Calculate expiration time for the token | |
expires_at = (datetime.utcnow() + timedelta(minutes=30)).isoformat() # Example expiration time | |
# Store the token in TinyDB | |
tinydb_helper.insert_token(user_id, access_token, expires_at) | |
return {"access_token": access_token, "token_type": "bearer"} | |
else: | |
raise HTTPException(status_code=400, detail="Face not recognized") | |