diff --git a/app/api/users.py b/app/api/users.py index 043dd98..3b51e29 100644 --- a/app/api/users.py +++ b/app/api/users.py @@ -1,5 +1,13 @@ -from fastapi import APIRouter, File, Form, UploadFile +from fastapi import APIRouter, File, Form, HTTPException, UploadFile from app.models.user import UserResponse +from app.services.user_service import ( + authenticate_user, + delete_user, + get_user, + register_user, +) + + router = APIRouter() @@ -7,22 +15,45 @@ @router.post("/users/register", response_model=UserResponse) async def register(user_id: str = Form(...), image: UploadFile = File(...)): """회원 등록 엔드포인트""" - pass + image_data = await image.read() + registered_user = register_user(user_id, image_data) + + return UserResponse( + user_id=registered_user["user_id"], + registered_at=registered_user["registered_at"], + ) @router.post("/users/authenticate") async def authenticate(image: UploadFile = File(...)): """회원 인증 엔드포인트""" - pass + image_data = await image.read() + user_id = authenticate_user(image_data) + + if user_id is None: + raise HTTPException(status_code=401, detail="인증 실패: 사용자 없음") + + return {"user_id": user_id} + @router.get("/users/{user_id}") def get_user_info(user_id: str): """회원 정보 조회 엔드포인트""" - pass + user_info = get_user(user_id) + + if user_info is None: + raise HTTPException(status_code=404, detail="사용자가 존재하지 않습니다.") + + return user_info @router.delete("/users/{user_id}") def delete_user_info(user_id: str): """회원 삭제 엔드포인트""" - pass + result = delete_user(user_id) + + if not result: + raise HTTPException(status_code=404, detail="사용자가 존재하지 않습니다.") + + return {"message": "사용자가 성공적으로 삭제되었습니다."} diff --git a/app/face/face_db.py b/app/face/face_db.py index 380ae3a..f0b2699 100644 --- a/app/face/face_db.py +++ b/app/face/face_db.py @@ -1,16 +1,29 @@ +import json +import os +from datetime import datetime + DB_PATH = "face_db.json" def save_user(user_id, embedding): """사용자 등록""" - pass + db = load_db() + db[user_id] = {"embedding": embedding, "registered_at": str(datetime.now())} + with open(DB_PATH, "w") as f: + json.dump(db, f) + def load_db(): """데이터베이스 로드""" - pass + if not os.path.exists(DB_PATH): + return {} + with open(DB_PATH, "r") as f: + return json.load(f) + def save_db(db): """데이터베이스 저장""" - pass + with open(DB_PATH, "w") as f: + json.dump(db, f) diff --git a/app/face/face_embedding.py b/app/face/face_embedding.py index 3acfb0c..92f363d 100644 --- a/app/face/face_embedding.py +++ b/app/face/face_embedding.py @@ -1,8 +1,11 @@ def extract_embedding(image): """이미지에서 임베딩 추출""" - pass + embedding = DeepFace.represent(img_path)[0]["embedding"] + return embedding def verify_embedding(embedding1, embedding2): +def verify_embedding(embedding1: list, embedding2: list) -> bool: """두 임베딩이 같은 사람인지 검증""" - pass + is_same_person = DeepFace.verify(embedding1, embedding2)["verified"] + return is_same_person diff --git a/app/services/user_service.py b/app/services/user_service.py index 410b807..8444854 100644 --- a/app/services/user_service.py +++ b/app/services/user_service.py @@ -1,18 +1,54 @@ +from datetime import datetime + +import cv2 +import numpy as np + +from app.face.face_db import load_db, save_db, save_user +from app.face.face_embedding import extract_embedding, verify_embedding + def register_user(user_id: str, image_bytes: bytes) -> dict: """이미지와 ID로 회원 등록""" - pass + + image_np = np.frombuffer(image_bytes, dtype=np.uint8) + image = cv2.imdecode(image_np, cv2.IMREAD_COLOR) + embedding = extract_embedding(image) + save_user(user_id, embedding) + return {"user_id": user_id, "registered_at": str(datetime.now())} def authenticate_user(image_bytes: bytes): """이미지로 회원 인증""" - pass + image_np = np.frombuffer(image_bytes, dtype=np.uint8) + image = cv2.imdecode(image_np, cv2.IMREAD_COLOR) + db = load_db() + + embedding_to_check = extract_embedding(image) + + for user_id, data in db.items(): + if verify_embedding(data["embedding"], embedding_to_check): + return user_id + + return None # 인증 실패 + def get_user(user_id): """user_id로 회원 정보 조회""" - pass + db = load_db() + user_data = db.get(user_id) + + if user_data: + return {"user_id": user_id, "registered_at": user_data["registered_at"]} + else: + return None # 사용자 없음 + def delete_user(user_id): """user_id로 회원 삭제""" - pass + db = load_db() + if user_id in db: + del db[user_id] + save_db(db) + return True + return False