app.py

import os
import json
import base64
import secrets
import numpy as np
import cv2
from deepface import DeepFace
from pymongo import MongoClient
import gradio as gr

# ========================
# MongoDB connection
# ========================
mongo_uri = "mongodb+srv://huyh01480_db_user:zxvAwzAhr8yk3lWe@cluster0.n8pboqq.mongodb.net/?retryWrites=true&w=majority&appName=Cluster0"
client = MongoClient(mongo_uri)
db = client["userdb"]
users_col = db["users"]

# ========================
# Helper functions
# ========================
def convert_numpy(obj):
    if isinstance(obj, dict):
        return {k: convert_numpy(v) for k, v in obj.items()}
    elif isinstance(obj, list):
        return [convert_numpy(i) for i in obj]
    elif isinstance(obj, (np.float32, np.float64)):
        return float(obj)
    elif isinstance(obj, (np.int32, np.int64)):
        return int(obj)
    else:
        return obj

# ========================
# Internal API functions
# ========================
def register_internal(data):
    username = data.get('username')
    password = data.get('password')
    img_data = data.get('img')
    fullName = data.get('fullName')
    email = data.get('email')
    phone = data.get('phone')
    gender = data.get('gender')

    if not username or not password or not img_data:
        return {'error': 'username, password và ảnh là bắt buộc'}, 400

    img_str = img_data.split(",")[1] if "," in img_data else img_data
    try:
        decoded = base64.b64decode(img_str)
        nparr = np.frombuffer(decoded, np.uint8)
        img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
        if img is None:
            return {'error': 'Ảnh không thể đọc được'}, 400
    except Exception:
        return {'error': 'Ảnh không hợp lệ'}, 400

    try:
        DeepFace.extract_faces(img, enforce_detection=True)
    except Exception:
        return {'error': 'Ảnh không có khuôn mặt hợp lệ'}, 400

    if users_col.find_one({"userName": username}):
        return {'error': 'Tên người dùng đã tồn tại'}, 409

    user_info = {
        "userName": username,
        "password": password,
        "img": img_str,
        "fullName": fullName,
        "email": email,
        "phone": phone,
        "gender": gender,
    }
    users_col.insert_one(user_info)
    return {'message': 'Đăng ký thành công!', 'user': username}, 201


def login_internal(data):
    username = data.get('username')
    password = data.get('password')
    img_data = data.get('img')

    if not username or not password or not img_data:
        return {'error': 'username, password và ảnh là bắt buộc'}, 400

    img_str = img_data.split(",")[1] if "," in img_data else img_data
    user = users_col.find_one({"userName": username, "password": password})
    if not user:
        return {'error': 'Username hoặc password sai'}, 401

    try:
        nparr_input = np.frombuffer(base64.b64decode(img_str), np.uint8)
        img_input = cv2.imdecode(nparr_input, cv2.IMREAD_COLOR)
        nparr_db = np.frombuffer(base64.b64decode(user['img']), np.uint8)
        img_db = cv2.imdecode(nparr_db, cv2.IMREAD_COLOR)
        result = DeepFace.verify(img_input, img_db, enforce_detection=True)
    except Exception as e:
        return {'error': f'Lỗi khi nhận diện khuôn mặt: {e}'}, 400

    distance = result.get("distance", 1)
    similarity = max(0, (1 - distance)) * 100
    similarity = round(similarity, 2)

    if result['verified']:
        token = secrets.token_hex(16)
        user_info = {k: user[k] for k in ["userName", "fullName", "email", "phone", "gender", "img"]}
        return {'message': 'Login thành công!', 'token': token, 'similarity': similarity, 'user': user_info}, 200
    else:
        return {'error': 'Khuôn mặt không trùng khớp', 'similarity': similarity}, 401


def analyze_internal(data):
    img_data = data.get('img')
    if not img_data:
        return {'error': 'Ảnh là bắt buộc'}, 400
    img_str = img_data.split(",")[1] if "," in img_data else img_data
    try:
        nparr = np.frombuffer(base64.b64decode(img_str), np.uint8)
        img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
        result = DeepFace.analyze(img, actions=['age','gender','emotion','race'], enforce_detection=True)
        result = convert_numpy(result)
        return {'result': result}, 200
    except Exception as e:
        return {'error': f'Lỗi khi phân tích khuôn mặt: {e}'}, 400


def compare_internal(data):
    img1_data = data.get("img1")
    img2_data = data.get("img2")
    if not img1_data or not img2_data:
        return {'error': 'Cần 2 ảnh base64 để so sánh'}, 400
    try:
        nparr1 = np.frombuffer(base64.b64decode(img1_data.split(",")[1] if "," in img1_data else img1_data), np.uint8)
        img1 = cv2.imdecode(nparr1, cv2.IMREAD_COLOR)
        nparr2 = np.frombuffer(base64.b64decode(img2_data.split(",")[1] if "," in img2_data else img2_data), np.uint8)
        img2 = cv2.imdecode(nparr2, cv2.IMREAD_COLOR)
        result = DeepFace.verify(img1, img2, enforce_detection=True)
        distance = result.get("distance", 1)
        similarity = max(0, (1 - distance)) * 100
        similarity = round(similarity, 2)
        return {"verified": result.get("verified", False), "distance": distance, "similarity": similarity}, 200
    except Exception as e:
        return {'error': str(e)}, 500

# ========================
# Gradio wrapper
# ========================
def gradio_handler(json_str, action="login"):
    try:
        data = json.loads(json_str)
    except Exception:
        return json.dumps({'error': 'Invalid JSON input'})
    
    if action == "register":
        res, status = register_internal(data)
    elif action == "login":
        res, status = login_internal(data)
    elif action == "analyze":
        res, status = analyze_internal(data)
    elif action == "compare":
        res, status = compare_internal(data)
    else:
        res = {'error': 'Unknown action'}
    return json.dumps(res)


iface = gr.Interface(
    fn=gradio_handler,
    inputs=[gr.Textbox(label="JSON input"), gr.Dropdown(["register","login","analyze","compare"], label="Action")],
    outputs="textbox",
    live=False
)

iface.launch(server_name="0.0.0.0", server_port=int(os.environ.get("PORT", 7860)))